Flask provides a publish-subscribe event system through the Blinker library, enabling loose coupling between aplication components. This mechainsm allows handlers to respond to specific application events without direct integration between emitters and receivers.

Install the dependency:

pip install blinker

Creating Custom Signals

Implementing a custom signal requires three phases: declaration, subscription, and emission.

First, instantiate a namespace and define the signal:

from blinker import Namespace

event_space = Namespace()
user_action_triggered = event_space.signal('user-action')

Next, register a receiver function using the connect method. The receiver must accept a sender argument plus any additional keyword arguments passed during emission:

def audit_user_activity(sender, **payload):
    user_id = payload.get('user_id')
    action_type = payload.get('action')
    print(f"Audit: {user_id} performed {action_type} from {sender}")

user_action_triggered.connect(audit_user_activity)

Emit the signal using send, optionally passing contextual data:

user_action_triggered.send('payment_gateway', user_id='usr_123', action='checkout')

Complete Implementation Example

The following demonstrates an audit logging system that records request metadata:

# audit_events.py
from blinker import Namespace
from datetime import datetime
from flask import request, g

event_bus = Namespace()
access_logged = event_bus.signal('access-event')

def persist_access_record(sender):
    entry_time = datetime.utcnow().isoformat()
    client_ip = request.headers.get('X-Forwarded-For', request.remote_addr)
    user_identifier = getattr(g, 'current_user', 'anonymous')
    
    log_entry = f"[{entry_time}] {user_identifier}@{client_ip} accessed {request.path}"
    
    with open('access_audit.log', 'a') as log_file:
        log_file.write(log_entry + '\n')

access_logged.connect(persist_access_record)

# application.py
from flask import Flask, request, g
from audit_events import access_logged

app = Flask(__name__)

@app.route('/dashboard')
def dashboard():
    g.current_user = request.args.get('user', 'guest')
    access_logged.send('dashboard_module')
    return {'status': 'active', 'user': g.current_user}

Built-in Lifecycle Signals

Flask exposes predefined signals for request and application context management:

Signal	Trigger Condition
request_started	Before request processing begins
request_finished	After response generation completes
before_render_template	Prior to template rendering
template_rendered	After template rendering completes
got_request_exception	When unhandled exceptions occur
request_tearing_down	During request cleanup (success or failure)
appcontext_tearing_down	During application context teardown
appcontext_pushed	When application context is pushed to stack
appcontext_popped	When application context is removed from stack
message_flashed	Upon adding messages to the flash store

Exception Monitoring Example

Capture and log runtime errors using the got_request_exception signal:

from flask import Flask, got_request_exception
import traceback

app = Flask(__name__)

def exception_reporter(sender, exception):
    error_details = traceback.format_exception_only(type(exception), exception)
    with open('error_journal.log', 'a') as error_log:
        error_log.write(f"Critical error in {sender}: {''.join(error_details)}\n")

got_request_exception.connect(exception_reporter)

@app.route('/compute')
def compute():
    result = 10 / 0  # Intentional error
    return {'result': result}