Middleware & Hooks

Intercept and transform requests and responses with composable middleware and lifecycle hooks.

Overview

Middleware wraps the entire request-handling pipeline. Each middleware receives a ctx (RequestContext) and a next_handler callable, and can:

• Inspect or modify the request before the handler runs
• Short-circuit the request (return early without calling the handler)
• Modify the response after the handler runs
• Add headers, log requests, enforce security policies

from lcore import Lcore, CORSMiddleware, SecurityHeadersMiddleware

app = Lcore()

# Register middleware with app.use()
app.use(CORSMiddleware(allow_origins='*'))
app.use(SecurityHeadersMiddleware(hsts=True))

Middleware execution order is controlled by the order attribute (lowest runs first).

Built-in Middleware

RequestIDMiddleware

Adds a unique X-Request-ID header to every response and sets ctx.request_id:

from lcore import RequestIDMiddleware
app.use(RequestIDMiddleware())

RequestLoggerMiddleware

Logs every request as structured JSON with method, path, status, and duration:

from lcore import RequestLoggerMiddleware
import logging

logger = logging.getLogger('myapp')
app.use(RequestLoggerMiddleware(logger=logger))

CORSMiddleware

Full CORS support with origin validation and preflight handling:

from lcore import CORSMiddleware

# Allow all origins
app.use(CORSMiddleware(allow_origins='*'))

# Restrict to specific origins
app.use(CORSMiddleware(
    allow_origins=['https://myapp.com', 'https://admin.myapp.com'],
    allow_methods=['GET', 'POST', 'PUT', 'DELETE'],
    allow_headers=['Content-Type', 'Authorization'],
    expose_headers=['X-Request-ID'],
    allow_credentials=True,
    max_age=86400
))

SecurityHeadersMiddleware

Adds best-practice security headers to every response:

from lcore import SecurityHeadersMiddleware

app.use(SecurityHeadersMiddleware(hsts=True, hsts_max_age=31536000))

# Override specific headers
app.use(SecurityHeadersMiddleware(
    hsts=True,
    **{'Content-Security-Policy': "default-src 'self'"}
))

Default headers set: X-Content-Type-Options: nosniff, X-Frame-Options: DENY, X-XSS-Protection: 1; mode=block, Referrer-Policy: strict-origin-when-cross-origin, Content-Security-Policy: default-src 'self'.

CSRFMiddleware

Validates CSRF tokens on state-changing requests. Tokens are checked in the header first, then in form data:

from lcore import CSRFMiddleware

app.use(CSRFMiddleware(
    secret='my-csrf-secret',
    cookie_name='_csrf_token',
    header_name='X-CSRF-Token',
    form_field='_csrf_token',
    safe_methods=('GET', 'HEAD', 'OPTIONS'),
    secure=True   # Set cookie with Secure flag (HTTPS only)
))

BodyLimitMiddleware

Rejects requests with bodies exceeding the limit (returns 413):

from lcore import BodyLimitMiddleware

# Limit request bodies to 5MB
app.use(BodyLimitMiddleware(max_size=5 * 1024 * 1024))

CompressionMiddleware

Gzip-compresses responses when the client accepts it:

from lcore import CompressionMiddleware

app.use(CompressionMiddleware(
    min_size=256,       # Don't compress small responses
    level=6,            # Compression level (1-9)
    content_types=None  # Compress all content types
))

Custom Middleware

Create custom middleware by subclassing Middleware:

from lcore import Middleware
import time

class TimingMiddleware(Middleware):
    name = 'timing'
    order = 2

    def __call__(self, ctx, next_handler):
        start = time.time()
        result = next_handler(ctx)  # Continue the chain
        duration = time.time() - start
        ctx.response.set_header('X-Response-Time',
            f'{duration*1000:.2f}ms')
        return result

app.use(TimingMiddleware())

Short-Circuiting

Return a response early without calling next_handler:

from lcore import Middleware, HTTPResponse

class AuthMiddleware(Middleware):
    name = 'auth'
    order = 5

    def __call__(self, ctx, next_handler):
        token = ctx.request.get_header('Authorization')
        if not token or not self.validate(token):
            ctx.response.status = 401
            return {'error': 'Unauthorized'}
        ctx.user = self.decode(token)
        return next_handler(ctx)

    def validate(self, token):
        return token.startswith('Bearer ')

    def decode(self, token):
        return {'sub': token.split(' ')[1]}

Route-Specific Middleware

Apply middleware only to specific route patterns:

# Only applies to /api/* routes
app.use(AuthMiddleware(), routes='/api/*')

# Multiple patterns
app.use(RateLimitMiddleware(), routes=['/api/*', '/webhook/*'])

Middleware Hooks

For a cleaner separation of concerns, use MiddlewareHook with explicit pre() and post() methods:

from lcore import MiddlewareHook, HTTPResponse

class AuditHook(MiddlewareHook):
    name = 'audit'
    order = 3

    def pre(self, ctx):
        # Runs before the handler
        # Return HTTPResponse to short-circuit
        if ctx.request.path.startswith('/admin'):
            if not ctx.request.get_header('X-Admin-Key'):
                return HTTPResponse('Forbidden', status=403)

    def post(self, ctx, result):
        # Runs after the handler
        # Can modify and return the result
        log_access(ctx.request.path, ctx.response.status_code)
        return result

app.use(AuditHook())

Lifecycle Hooks

Hooks are lightweight callbacks fired at specific points during request processing:

@app.hook('before_request')
def log_request():
    print(f'{request.method} {request.path}')

@app.hook('after_request')
def add_header():
    response.set_header('X-Powered-By', 'Lcore')

All Hook Points

Programmatic Hook API

# Register
app.add_hook('on_request_start', my_callback)

# Remove
app.remove_hook('on_request_start', my_callback)

# Trigger manually
results = app.trigger_hook('on_request_start')

Module-Specific Hooks

Scope hooks to specific mounted sub-applications:

# Fire only for requests to /api/* routes
app.add_module_hook('/api/', 'before_request', check_api_key)

# Decorator form
@app.module_hook('/admin/', 'after_request')
def admin_log():
    log_admin_action(request.path)

Middleware Pipeline

The MiddlewarePipeline manages middleware execution order and caching:

• Ordering: Middleware is sorted by order attribute (lowest first)
• Caching: The global middleware chain is cached after first build and invalidated when middleware is added or removed
• Route matching: Middleware with routes= parameter is only included for matching paths

Inspect the Stack

for mw in app.show_middleware():
    print(f"[{mw['order']:3d}] {mw['name']:20s} {mw['type']}")

# [  0] body_limit           BodyLimitMiddleware
# [  1] request_id           RequestIDMiddleware
# [  3] cors                 CORSMiddleware
# [  5] security_headers     SecurityHeadersMiddleware
# [ 90] compression          CompressionMiddleware

Remove Middleware

# Remove by instance
cors = CORSMiddleware()
app.use(cors)
app.middleware.remove(cors)