Fading Coder

One Final Commit for the Last Sprint

Home > Tech > Content

Advanced Flask: Dynamic Dispatching, Flash Messaging, and Middleware

Tech May 9 3

Dynamic Notification Dispatching

To build a system that supports multiple notification channels (like SMS, Email, and Push Notifications) and allows quick switching via configuration, we can use dynamic module importing combined with an abstract base class.

Create an abstract base class in services/notifier/core.py to enforce implementation of the dispatch method:

class AbstractNotifier:
    def dispatch(self, content):
        raise NotImplementedError("Subclasses must implement the dispatch method")

Implement specific channels. For email in services/notifier/email_channel.py:

from .core import AbstractNotifier

class EmailNotifier(AbstractNotifier):
    def __init__(self):
        self.credentials = 'email_config'

    def dispatch(self, content):
        print(f"Dispatching via Email: {content}")

For SMS in services/notifier/sms_channel.py:

from .core import AbstractNotifier

class SMSNotifier(AbstractNotifier):
    def __init__(self):
        self.credentials = 'sms_config'

    def dispatch(self, content):
        print(f"Dispatching via SMS: {content}")

For Push in services/notifier/push_channel.py:

from .core import AbstractNotifier

class PushNotifier(AbstractNotifier):
    def dispatch(self, content):
        print(f"Dispatching via Push: {content}")

In services/notifier/__init__.py, dynamically load the classes specified in the application configuration and trigger their dispatch methods:

import importlib
from config import NOTIFIER_CLASSES

def trigger_alerts(message):
    for class_path in NOTIFIER_CLASSES:
        module_path, class_name = class_path.rsplit('.', 1)
        module_obj = importlib.import_module(module_path)
        notifier_instance = getattr(module_obj, class_name)()
        notifier_instance.dispatch(message)

Define the channel paths in config.py:

NOTIFIER_CLASSES = [
    "services.notifier.email_channel.EmailNotifier",
    "services.notifier.sms_channel.SMSNotifier",
    "services.notifier.push_channel.PushNotifier",
]

Integrate the dispatch system into the Flask application in main.py:

from flask import Flask
from services.notifier import trigger_alerts

app = Flask(__name__)

@app.route('/')
def home():
    trigger_alerts('System alert triggered')
    return 'Alerts Sent'

if __name__ == '__main__':
    app.run()

Flash Messaging

Flash messaging provides temporary server-side data storage, built on top of sessions. Data is removed once it is retrieved.

from flask import Flask, flash, get_flashed_messages

app = Flask(__name__)
app.secret_key = 'secure_key_here'

@app.route('/retrieve')
def retrieve():
    messages = get_flashed_messages()
    print(messages)
    return 'Messages Retrieved'

@app.route('/store')
def store():
    flash('Temporary data stored')
    return 'Data Flashed'

if __name__ == '__main__':
    app.run()

Flash messages can be categorized for targeted retrieval:

from flask import Flask, flash, get_flashed_messages, request, redirect

app = Flask(__name__)
app.secret_key = 'secure_key_here'

@app.route('/dashboard')
def dashboard():
    auth_param = request.args.get('auth')
    if auth_param == 'valid_token':
        return 'Access Granted'
    flash('Session expired', category="security")
    return redirect('/error_page')

@app.route('/error_page')
def error_page():
    alerts = get_flashed_messages(category_filter=['security'])
    error_msg = alerts[0] if alerts else "Default error"
    return f"Error Display: {error_msg}"

if __name__ == '__main__':
    app.run()

Request Lifecycle Hooks

Flask provides decorators to execute logic before or after requests. before_request runs prior to view execution, while after_request runs post-execution and must return a response object. Multiple hooks execute in declaration order for before_request and reverse declaration order for after_request. If a before_request intercepts the request, all after_request hooks still execute.

from flask import Flask

app = Flask(__name__)
app.debug = True

@app.before_request
def pre_process_1():
    print('Pre-process hook 1 executing')

@app.before_request
def pre_process_2():
    print('Pre-process hook 2 executing')

@app.after_request
def post_process_1(response):
    print('Post-process hook 1 executing')
    return response

@app.after_request
def post_process_2(response):
    print('Post-process hook 2 executing')
    return response

@app.route('/home')
def home():
    print('View function executing')
    return 'Home Page'

if __name__ == '__main__':
    app.run()

A common use case for before_request is enforcing authentication:

from flask import Flask, request, redirect, session

app = Flask(__name__)
app.secret_key = 'secret_value'

@app.before_request
def verify_login():
    if request.path == '/auth':
        return None
    if session.get('user_id'): 
        return None
    return redirect('/auth')

@app.route('/profile')
def profile():
    return 'User Profile'

@app.route('/auth', methods=['GET', 'POST'])
def auth():
    session['user_id'] = 1001
    return 'Login Successful'

if __name__ == '__main__':
    app.run()

WSGI Middleware

Flask invokes the wsgi_app method inside __call__ for every request. Wrapping wsgi_app allows the creation of custom middleware.

from flask import Flask

app = Flask(__name__)

@app.route('/')
def index():
    return 'Hello Middleware!'

class RequestLogger:
    def __init__(self, original_wsgi_app):
        self.original_wsgi_app = original_wsgi_app

    def __call__(self, environ, start_response):
        print('Incoming request logged')
        response = self.original_wsgi_app(environ, start_response)
        print('Outgoing response logged')
        return response

if __name__ == '__main__':
    app.wsgi_app = RequestLogger(app.wsgi_app)
    app.run()

Custom Error Pages

Flask allows registering handlers for specific HTTP error codes:

@app.errorhandler(404)
def handle_not_found(error):
    return "Custom 404: Resource Not Found", 404

Template Global Functions and Filters

Custom logic can be injected into Jinja2 templates directly from the Flask application configuration.

from flask import Flask, render_template

app = Flask(__name__)

@app.template_global()
def add_values(num1, num2):
    return num1 + num2

@app.template_filter()
def compute_sum(num1, num2, num3):
    return num1 + num2 + num3

@app.route('/view')
def view():
    return render_template('display.html')

In templates/display.html:

<html>
<body>
    <h1>{{ 5 | compute_sum(10, 15) }}</h1>
    <h1>{{ add_values(20, 30) }}</h1>
</body>
</html>

Related Articles

Understanding Strong and Weak References in Java

Strong References Strong reference are the most prevalent type of object referencing in Java. When an object has a strong reference pointing to it, the garbage collector will not reclaim its memory. F...

Comprehensive Guide to SSTI Explained with Payload Bypass Techniques

Introduction Server-Side Template Injection (SSTI) is a vulnerability in web applications where user input is improper handled within the template engine and executed on the server. This exploit can r...

Implement Image Upload Functionality for Django Integrated TinyMCE Editor

Django’s Admin panel is highly user-friendly, and pairing it with TinyMCE, an effective rich text editor, simplifies content management significantly. Combining the two is particular useful for bloggi...

Leave a Comment

Anonymous

◎Feel free to join the discussion and share your thoughts.