Skip to main content

​
Overview

Middleware in Framefox allows you to process HTTP requests and responses at various stages. Middleware can modify requests, add headers, handle authentication, log activity, and more.

​
Middleware System

Framefox’s middleware system provides:
  • Request/Response Processing: Intercept and modify HTTP traffic
  • Ordering Control: Execute middleware in specific sequence
  • Built-in Middleware: Pre-configured common functionality
  • Custom Middleware: Easy creation of your own middleware

​
Built-in Middleware

Framefox includes several built-in middleware components: 
from framefox.core.middleware.middleware_manager import MiddlewareManager

manager = MiddlewareManager(app)
manager.setup_middlewares()
This adds (in order):
  1. ProfilerMiddleware: Performance profiling and debugging
  2. ExceptionMiddleware: Centralized exception handling
  3. RequestMiddleware: Request context management
  4. EntityManagerMiddleware: Database session handling
  5. CsrfMiddleware: CSRF token management
  6. FirewallMiddleware: Security rules and IP filtering
  7. SessionMiddleware: Session management
  8. CustomCORSMiddleware: CORS headers

​
Middleware Execution Order

Middleware executes in a specific order: 
Request Flow:
  ↓
ProfilerMiddleware
  ↓
ExceptionMiddleware
  ↓
RequestMiddleware
  ↓
EntityManagerMiddleware
  ↓
CsrfMiddleware
  ↓
FirewallMiddleware
  ↓
SessionMiddleware
  ↓
CustomCORSMiddleware
  ↓
Your Application
  ↓
Response Flow (reverse order)

​
Creating Custom Middleware

​
Basic Middleware

Create middleware using Starlette’s BaseHTTPMiddleware: 
from starlette.middleware.base import BaseHTTPMiddleware
from fastapi import Request
import time

class TimingMiddleware(BaseHTTPMiddleware):
    """Measure request processing time"""
    
    async def dispatch(self, request: Request, call_next):
        start_time = time.time()
        
        # Process request
        response = await call_next(request)
        
        # Calculate duration
        duration = time.time() - start_time
        
        # Add header
        response.headers["X-Process-Time"] = str(duration)
        
        return response

​
Middleware with Dependencies

Access services from the container: 
from starlette.middleware.base import BaseHTTPMiddleware
from framefox.core.di.service_container import ServiceContainer
import logging

class LoggingMiddleware(BaseHTTPMiddleware):
    """Log all requests and responses"""
    
    def __init__(self, app):
        super().__init__(app)
        self.logger = logging.getLogger("REQUEST")
        self.container = ServiceContainer()
    
    async def dispatch(self, request: Request, call_next):
        # Log request
        self.logger.info(
            f"{request.method} {request.url.path} from {request.client.host}"
        )
        
        # Process request
        response = await call_next(request)
        
        # Log response
        self.logger.info(
            f"Response: {response.status_code} for {request.url.path}"
        )
        
        return response

​
Middleware with Configuration

Use settings for configuration: 
from starlette.middleware.base import BaseHTTPMiddleware
from framefox.core.config.settings import Settings
from fastapi import Request
from fastapi.responses import JSONResponse

class RateLimitMiddleware(BaseHTTPMiddleware):
    """Rate limit requests per IP address"""
    
    def __init__(self, app):
        super().__init__(app)
        self.settings = Settings()
        self.requests = {}  # IP -> request count
        self.limit = self.settings.rate_limit or 100
    
    async def dispatch(self, request: Request, call_next):
        client_ip = request.client.host
        
        # Check rate limit
        if client_ip in self.requests:
            if self.requests[client_ip] >= self.limit:
                return JSONResponse(
                    {"error": "Rate limit exceeded"},
                    status_code=429
                )
        
        # Increment counter
        self.requests[client_ip] = self.requests.get(client_ip, 0) + 1
        
        # Process request
        response = await call_next(request)
        
        return response

​
Registering Middleware

​
Application-Level

Add middleware to your FastAPI application: 
from fastapi import FastAPI
from myapp.middleware import TimingMiddleware, LoggingMiddleware

app = FastAPI()

# Add custom middleware
app.add_middleware(TimingMiddleware)
app.add_middleware(LoggingMiddleware)

​
In MiddlewareManager

Extend the MiddlewareManager for centralized configuration: 
from framefox.core.middleware.middleware_manager import MiddlewareManager
from myapp.middleware import CustomAuthMiddleware, AuditMiddleware

class MyMiddlewareManager(MiddlewareManager):
    def setup_middlewares(self):
        # Call parent to add built-in middleware
        super().setup_middlewares()
        
        # Add custom middleware
        self.app.add_middleware(CustomAuthMiddleware)
        self.app.add_middleware(AuditMiddleware)

​
Middleware Examples

​
Authentication Middleware

from starlette.middleware.base import BaseHTTPMiddleware
from fastapi import Request
from fastapi.responses import JSONResponse

class AuthMiddleware(BaseHTTPMiddleware):
    """Verify JWT tokens for protected routes"""
    
    def __init__(self, app):
        super().__init__(app)
        self.public_paths = ["/login", "/register", "/health"]
    
    async def dispatch(self, request: Request, call_next):
        # Skip authentication for public paths
        if request.url.path in self.public_paths:
            return await call_next(request)
        
        # Check for authorization header
        auth_header = request.headers.get("Authorization")
        
        if not auth_header or not auth_header.startswith("Bearer "):
            return JSONResponse(
                {"error": "Missing authentication token"},
                status_code=401
            )
        
        # Verify token
        token = auth_header.split(" ")[1]
        user = self.verify_token(token)
        
        if not user:
            return JSONResponse(
                {"error": "Invalid authentication token"},
                status_code=401
            )
        
        # Add user to request state
        request.state.user = user
        
        return await call_next(request)
    
    def verify_token(self, token: str):
        # Implement token verification
        # Return user object or None
        pass

​
Request ID Middleware

import uuid
from starlette.middleware.base import BaseHTTPMiddleware

class RequestIDMiddleware(BaseHTTPMiddleware):
    """Add unique request ID to all requests"""
    
    async def dispatch(self, request: Request, call_next):
        # Generate request ID
        request_id = str(uuid.uuid4())
        
        # Add to request state
        request.state.request_id = request_id
        
        # Process request
        response = await call_next(request)
        
        # Add to response headers
        response.headers["X-Request-ID"] = request_id
        
        return response

​
Caching Middleware

from starlette.middleware.base import BaseHTTPMiddleware
from fastapi.responses import Response
import hashlib

class CacheMiddleware(BaseHTTPMiddleware):
    """Cache GET requests"""
    
    def __init__(self, app):
        super().__init__(app)
        self.cache = {}  # Simple in-memory cache
        self.ttl = 300  # 5 minutes
    
    async def dispatch(self, request: Request, call_next):
        # Only cache GET requests
        if request.method != "GET":
            return await call_next(request)
        
        # Generate cache key
        cache_key = self.get_cache_key(request)
        
        # Check cache
        if cache_key in self.cache:
            cached_response = self.cache[cache_key]
            return Response(
                content=cached_response["content"],
                status_code=cached_response["status_code"],
                headers=cached_response["headers"]
            )
        
        # Process request
        response = await call_next(request)
        
        # Cache successful responses
        if response.status_code == 200:
            self.cache[cache_key] = {
                "content": response.body,
                "status_code": response.status_code,
                "headers": dict(response.headers)
            }
        
        return response
    
    def get_cache_key(self, request: Request) -> str:
        # Create cache key from URL and query params
        key_string = f"{request.url.path}?{request.url.query}"
        return hashlib.md5(key_string.encode()).hexdigest()

​
Session Middleware

Framefox’s built-in SessionMiddleware manages user sessions: 
from framefox.core.middleware.middlewares.session_middleware import SessionMiddleware

# Access session in your routes
from fastapi import Request

@app.get("/profile")
async def profile(request: Request):
    # Get session data
    user_id = request.state.session_data.get("user_id")
    
    if not user_id:
        return {"error": "Not logged in"}
    
    # Set session data
    request.state.session_data["last_visit"] = datetime.now().isoformat()
    
    return {"user_id": user_id}
Session features:
  • Automatic Management: Session created and updated automatically
  • Signed Cookies: Sessions signed for security
  • Expiration: Configurable timeout with cookie_max_age setting
  • Cleanup: Expired sessions automatically removed

​
CSRF Middleware

The CsrfMiddleware protects against CSRF attacks: 
from framefox.core.middleware.middlewares.csrf_middleware import CsrfMiddleware

# CSRF tokens automatically added to cookies for GET requests
# Validate in your forms:

@app.post("/submit-form")
async def submit_form(request: Request, csrf_token: str):
    # Get token from session
    session_token = request.state.session_data.get("csrf_token")
    
    if csrf_token != session_token:
        return {"error": "Invalid CSRF token"}, 403
    
    # Process form...
    return {"success": True}

​
Best Practices

  1. Order Matters: Register middleware in the correct order
  2. Performance: Keep middleware fast; avoid heavy operations
  3. Error Handling: Handle exceptions gracefully
  4. State Management: Use request.state for request-scoped data
  5. Testing: Test middleware independently
  6. Documentation: Document middleware behavior and configuration

​
Middleware Ordering Tips

# Correct order:
app.add_middleware(ExceptionMiddleware)  # First - catch all errors
app.add_middleware(LoggingMiddleware)    # Log all requests
app.add_middleware(AuthMiddleware)       # Authenticate users
app.add_middleware(SessionMiddleware)    # Manage sessions
app.add_middleware(CsrfMiddleware)       # CSRF protection
app.add_middleware(CORSMiddleware)       # Last - add CORS headers

​
Testing Middleware

import pytest
from fastapi.testclient import TestClient
from fastapi import FastAPI
from myapp.middleware import TimingMiddleware

def test_timing_middleware():
    app = FastAPI()
    app.add_middleware(TimingMiddleware)
    
    @app.get("/test")
    async def test_route():
        return {"message": "success"}
    
    client = TestClient(app)
    response = client.get("/test")
    
    # Verify middleware added header
    assert "X-Process-Time" in response.headers
    assert response.status_code == 200

Build docs developers (and LLMs) love

Get started for freeTalk to us