Flask uses signals to notify subscribers when certain events occur. Signals are provided by the Blinker library.
Core Signals
All Flask signals are defined in the flask.signals module.
from flask import signals
template_rendered
template_rendered = _signals.signal("template-rendered")
Sent when a template is successfully rendered.
Sender: The application instance
Arguments:
template - The template object that was rendered
context - The context dictionary passed to the template
Example
from flask import template_rendered
def log_template_renders(sender, template, context, **extra):
print(f"Template rendered: {template.name}")
print(f"Context keys: {list(context.keys())}")
template_rendered.connect(log_template_renders, app)
before_render_template
before_render_template = _signals.signal("before-render-template")
Sent before a template is rendered.
Sender: The application instance
Arguments:
template - The template object that will be rendered
context - The context dictionary that will be passed to the template
Example
from flask import before_render_template
def inject_global_context(sender, template, context, **extra):
context['app_version'] = '1.0.0'
context['build_date'] = '2024-03-09'
before_render_template.connect(inject_global_context, app)
request_started
request_started = _signals.signal("request-started")
Sent when a request context is set up, before any request processing happens.
Sender: The application instance
Arguments: None
Example
from flask import request_started, request
def log_request(sender, **extra):
print(f"Request started: {request.method} {request.path}")
print(f"Remote address: {request.remote_addr}")
request_started.connect(log_request, app)
request_finished
request_finished = _signals.signal("request-finished")
Sent right before the response is sent to the client.
Sender: The application instance
Arguments:
response - The response object that will be sent
Example
from flask import request_finished
def log_response(sender, response, **extra):
print(f"Response status: {response.status_code}")
print(f"Response headers: {dict(response.headers)}")
request_finished.connect(log_response, app)
request_tearing_down
request_tearing_down = _signals.signal("request-tearing-down")
Sent when the request context is being torn down, even if an exception occurred.
Sender: The application instance
Arguments:
exc - The exception that caused the teardown, if any (otherwise None)
Example
from flask import request_tearing_down
def cleanup_request(sender, exc=None, **extra):
if exc is not None:
print(f"Request failed with exception: {exc}")
# Perform cleanup (close database connections, etc.)
db.session.remove()
request_tearing_down.connect(cleanup_request, app)
got_request_exception
got_request_exception = _signals.signal("got-request-exception")
Sent when an exception occurs during request processing. This is sent even if the exception is handled by an error handler.
Sender: The application instance
Arguments:
exception - The exception object
Example
from flask import got_request_exception
import logging
def log_exception(sender, exception, **extra):
logging.exception(f"Exception occurred: {exception}")
# Send to error tracking service
error_tracker.capture_exception(exception)
got_request_exception.connect(log_exception, app)
appcontext_pushed
appcontext_pushed = _signals.signal("appcontext-pushed")
Sent when an application context is pushed.
Sender: The application instance
Arguments: None
Example
from flask import appcontext_pushed
def setup_app_context(sender, **extra):
print(f"Application context pushed for {sender.name}")
# Initialize resources that need app context
appcontext_pushed.connect(setup_app_context, app)
appcontext_popped
appcontext_popped = _signals.signal("appcontext-popped")
Sent when an application context is popped.
Sender: The application instance
Arguments:
exc - The exception that caused the context to be popped, if any
Example
from flask import appcontext_popped
def teardown_app_context(sender, exc=None, **extra):
if exc is not None:
print(f"App context popped due to exception: {exc}")
# Clean up resources
appcontext_popped.connect(teardown_app_context, app)
appcontext_tearing_down
appcontext_tearing_down = _signals.signal("appcontext-tearing-down")
Sent when the application context is being torn down, even if an exception occurred.
Sender: The application instance
Arguments:
exc - The exception that caused the teardown, if any
Example
from flask import appcontext_tearing_down
def cleanup_app(sender, exc=None, **extra):
print("Cleaning up application context")
cache.clear()
db.session.remove()
appcontext_tearing_down.connect(cleanup_app, app)
message_flashed
message_flashed = _signals.signal("message-flashed")
Sent when a message is flashed using the flash() function.
Sender: The application instance
Arguments:
message - The message that was flashed
category - The category of the message
Example
from flask import message_flashed
def log_flash(sender, message, category, **extra):
print(f"Flash message [{category}]: {message}")
# Track analytics
analytics.track('message_flashed', {
'category': category,
'message_length': len(message)
})
message_flashed.connect(log_flash, app)
Using Signals
Subscribing to Signals
Connect a function to a signal to be notified when the signal is sent:
from flask import request_started
def on_request_started(sender, **extra):
print(f"Request started for app: {sender.name}")
# Subscribe to signal for specific app
request_started.connect(on_request_started, app)
# Subscribe to signal for all apps
request_started.connect(on_request_started)
Unsubscribing from Signals
from flask import request_started
def handler(sender, **extra):
pass
# Subscribe
request_started.connect(handler, app)
# Unsubscribe
request_started.disconnect(handler, app)
Creating Custom Signals
Create your own signals for custom events:
from blinker import Namespace
# Create a namespace for your signals
my_signals = Namespace()
# Create custom signals
user_logged_in = my_signals.signal('user-logged-in')
user_logged_out = my_signals.signal('user-logged-out')
order_placed = my_signals.signal('order-placed')
# Send signals
@app.route('/login', methods=['POST'])
def login():
user = authenticate(request.form['username'], request.form['password'])
if user:
login_user(user)
user_logged_in.send(current_app._get_current_object(), user=user)
return redirect(url_for('dashboard'))
return redirect(url_for('login'))
# Subscribe to custom signals
def on_user_login(sender, user, **extra):
print(f"User {user.username} logged in")
# Update last login time
user.last_login = datetime.utcnow()
db.session.commit()
user_logged_in.connect(on_user_login)
Decorator Syntax
Use the decorator syntax for cleaner subscription:
from flask import request_finished
@request_finished.connect_via(app)
def on_request_finished(sender, response, **extra):
print(f"Request finished with status {response.status_code}")
Temporary Subscriptions
Subscribe to signals only within a context:
from flask import template_rendered
def test_template():
recorded = []
def record(sender, template, context, **extra):
recorded.append(template)
# Temporarily subscribe
with template_rendered.connected_to(record, app):
render_template('test.html')
assert len(recorded) == 1
Signal Best Practices
Signal handlers should not modify the application state in ways that affect the request/response flow. Use them primarily for logging, monitoring, and side effects.
Best Practices:
- Keep handlers fast - Signal handlers run synchronously and can slow down requests
- Handle exceptions - Always wrap handler code in try/except to prevent failures
- Use weak references - Subscribers are stored weakly by default to prevent memory leaks
- Be specific - Connect to specific app instances when possible
- Document signals - Clearly document custom signals and their arguments
Example
from flask import request_finished
import logging
@request_finished.connect_via(app)
def log_request_finished(sender, response, **extra):
try:
# Fast, non-blocking operations only
logging.info(f"Request completed: {response.status_code}")
except Exception as e:
# Don't let signal handler errors affect the request
logging.exception(f"Error in signal handler: {e}")