Signals

from flask import request_started, request_finished
from flask import request_tearing_down, got_request_exception

# Sent before request processing begins
request_started

# Sent after the response is constructed
request_finished

# Sent when the request context is torn down
request_tearing_down

# Sent when an exception occurs during request handling
got_request_exception

from flask import appcontext_pushed, appcontext_popped
from flask import appcontext_tearing_down

# Sent when an application context is pushed
appcontext_pushed

# Sent when an application context is popped
appcontext_popped

# Sent when the application context is torn down
appcontext_tearing_down

from flask import template_rendered, before_render_template

# Sent before a template is rendered
before_render_template

# Sent after a template is rendered
template_rendered

from flask import message_flashed

# Sent when a message is flashed
message_flashed

from flask import Flask, request_finished

app = Flask(__name__)

def log_request(sender, response, **extra):
    """Log each request when it completes."""
    sender.logger.info(
        f'{request.method} {request.path} - {response.status_code}'
    )

request_finished.connect(log_request, app)

from flask import request_started

@request_started.connect_via(app)
def on_request_started(sender, **extra):
    """Called before each request is processed."""
    print(f'Request started: {request.path}')

@request_started.connect_via(app)
def on_request_started(sender, **extra):
    """Sender is the app, no extra parameters."""
    pass

@request_finished.connect_via(app)
def on_request_finished(sender, response, **extra):
    """Sender is the app, response is the Response object."""
    response.headers['X-Custom-Header'] = 'value'

@request_tearing_down.connect_via(app)
def on_request_teardown(sender, exc=None, **extra):
    """Sender is the app, exc is any exception that occurred."""
    if exc is not None:
        sender.logger.error(f'Request failed: {exc}')

@got_request_exception.connect_via(app)
def on_request_exception(sender, exception, **extra):
    """Sender is the app, exception is the exception instance."""
    # Log to external service
    log_exception_to_sentry(exception)

@template_rendered.connect_via(app)
def on_template_rendered(sender, template, context, **extra):
    """Sender is the app, template is the Template object, context is the render context."""
    sender.logger.debug(f'Rendered template: {template.name}')

@message_flashed.connect_via(app)
def on_message_flashed(sender, message, category, **extra):
    """Sender is the app, message is the flashed message, category is the message category."""
    sender.logger.info(f'Flash message [{category}]: {message}')

import time
from flask import g, request_started, request_finished

@request_started.connect_via(app)
def start_timer(sender, **extra):
    """Record request start time."""
    g.start_time = time.time()

@request_finished.connect_via(app)
def log_request(sender, response, **extra):
    """Log request duration and details."""
    duration = time.time() - g.start_time
    sender.logger.info(
        f'{request.method} {request.path} '
        f'{response.status_code} {duration:.3f}s'
    )

from flask import request_finished

@request_finished.connect_via(app)
def track_analytics(sender, response, **extra):
    """Send analytics data for successful requests."""
    if 200 <= response.status_code < 300:
        analytics.track(
            user_id=getattr(g, 'user_id', None),
            event='page_view',
            properties={
                'path': request.path,
                'method': request.method,
                'status': response.status_code
            }
        )

from flask import appcontext_tearing_down

@appcontext_tearing_down.connect_via(app)
def close_db_connection(sender, exc=None, **extra):
    """Close database connection when context tears down."""
    db = g.pop('db', None)
    if db is not None:
        db.close()

from flask import got_request_exception
import traceback

@got_request_exception.connect_via(app)
def log_exception(sender, exception, **extra):
    """Report exceptions to external service."""
    error_data = {
        'exception': str(exception),
        'traceback': traceback.format_exc(),
        'url': request.url,
        'method': request.method,
        'user_agent': request.user_agent.string
    }
    
    # Send to error tracking service
    error_tracker.report(error_data)

from flask import request_finished

@request_finished.connect_via(app)
def invalidate_cache(sender, response, **extra):
    """Invalidate cache on POST/PUT/DELETE requests."""
    if request.method in ['POST', 'PUT', 'DELETE']:
        if 200 <= response.status_code < 300:
            cache_key = f'resource:{request.path}'
            cache.delete(cache_key)

import time
from flask import before_render_template, template_rendered

render_times = {}

@before_render_template.connect_via(app)
def before_render(sender, template, context, **extra):
    """Record template render start time."""
    render_times[template.name] = time.time()

@template_rendered.connect_via(app)
def after_render(sender, template, context, **extra):
    """Log slow template renders."""
    duration = time.time() - render_times.pop(template.name, time.time())
    if duration > 0.5:  # Log if render takes > 500ms
        sender.logger.warning(
            f'Slow template render: {template.name} took {duration:.3f}s'
        )

from flask.signals import Namespace

# Create a namespace for your signals
my_signals = Namespace()

# Define signals
user_registered = my_signals.signal('user-registered')
user_logged_in = my_signals.signal('user-logged-in')
order_placed = my_signals.signal('order-placed')

from flask import current_app

@app.route('/register', methods=['POST'])
def register():
    user = create_user(request.form)
    
    # Send signal
    user_registered.send(
        current_app._get_current_object(),
        user=user
    )
    
    return redirect(url_for('login'))

@user_registered.connect_via(app)
def send_welcome_email(sender, user, **extra):
    """Send welcome email when user registers."""
    send_email(
        to=user.email,
        subject='Welcome!',
        template='welcome',
        user=user
    )

@user_registered.connect_via(app)
def create_default_preferences(sender, user, **extra):
    """Create default preferences for new user."""
    Preferences.create(user_id=user.id, defaults=DEFAULT_PREFS)

# Good: Lightweight handler
@request_finished.connect_via(app)
def log_request(sender, response, **extra):
    sender.logger.info(f'{request.path} - {response.status_code}')

# Bad: Heavy processing in handler
@request_finished.connect_via(app)
def process_analytics(sender, response, **extra):
    # This blocks the response!
    heavy_analytics_processing()
    sync_to_external_service()

from flask import request_finished
from myapp.tasks import process_analytics

@request_finished.connect_via(app)
def queue_analytics(sender, response, **extra):
    """Queue analytics processing instead of doing it inline."""
    process_analytics.delay(
        path=request.path,
        status=response.status_code
    )

@request_finished.connect_via(app)
def safe_handler(sender, response, **extra):
    """Don't let signal handler errors break the app."""
    try:
        # Handler logic
        track_analytics(request, response)
    except Exception as e:
        sender.logger.error(f'Signal handler error: {e}')
        # Don't re-raise - signals should not break the app

# Disconnect a specific handler
request_finished.disconnect(log_request, app)

# Disconnect all handlers for an app
request_finished.disconnect(sender=app)

# Weak reference (default) - handler can be garbage collected
request_finished.connect(log_request, app)

# Strong reference - handler won't be garbage collected
request_finished.connect(log_request, app, weak=False)

from flask import template_rendered

def test_template_rendered_signal(app, client):
    """Test that template_rendered signal is sent."""
    templates = []
    
    def record_template(sender, template, context, **extra):
        templates.append(template.name)
    
    template_rendered.connect(record_template, app)
    
    try:
        client.get('/page')
        assert 'page.html' in templates
    finally:
        template_rendered.disconnect(record_template, app)

from contextlib import contextmanager

@contextmanager
def capture_signals(signal, app):
    """Context manager to capture signal emissions."""
    captured = []
    
    def handler(sender, **extra):
        captured.append(extra)
    
    signal.connect(handler, app)
    try:
        yield captured
    finally:
        signal.disconnect(handler, app)

def test_user_registration(app):
    with capture_signals(user_registered, app) as signals:
        register_user('[email protected]')
        assert len(signals) == 1
        assert signals[0]['user'].email == '[email protected]'