Sufast/Documentation
Middleware
Middleware System
Process requests and responses with Sufast's powerful middleware system.
Middleware in Sufast runs in the order it's added to the application. Each middleware can modify the request before it reaches your route handler and the response before it's sent to the client.
Built-in Middleware
Ready-to-use middleware for common functionality
CORS Middleware
CORS Configurationpython
from sufast import App
from sufast.middleware import CORSMiddleware
app = App()
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["https://example.com", "https://app.example.com"],
allow_credentials=True,
allow_methods=["GET", "POST", "PUT", "DELETE"],
allow_headers=["*"],
)
# For development, allow all origins
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=False,
allow_methods=["*"],
allow_headers=["*"],
)Logging Middleware
Logging Setuppython
from sufast import App
from sufast.middleware import LoggingMiddleware
app = App()
# Add request/response logging
app.add_middleware(
LoggingMiddleware,
format="%(asctime)s - %(method)s %(path)s - %(status_code)s - %(duration)sms",
level="INFO"
)
# Custom log format
app.add_middleware(
LoggingMiddleware,
format="[%(client_ip)s] %(method)s %(path)s %(status_code)s in %(duration)sms",
include_headers=True,
include_body=False # Don't log request/response bodies
)Rate Limiting Middleware
Rate Limitingpython
from sufast import App
from sufast.middleware import RateLimitMiddleware
app = App()
# Basic rate limiting
app.add_middleware(
RateLimitMiddleware,
calls=100, # 100 requests
period=60, # per 60 seconds
per_ip=True # per IP address
)
# Advanced rate limiting with different limits
app.add_middleware(
RateLimitMiddleware,
rules=[
{"path": "/api/auth/*", "calls": 5, "period": 60}, # Auth endpoints: 5/min
{"path": "/api/upload/*", "calls": 10, "period": 300}, # Upload: 10/5min
{"path": "/api/*", "calls": 1000, "period": 3600}, # General API: 1000/hour
]
)Custom Middleware
Create your own middleware for specific needs
Basic Custom Middleware
Custom Middlewarepython
from sufast import App, Request, Response
import time
app = App()
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
start_time = time.time()
# Process the request
response = await call_next(request)
# Add processing time to response headers
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
return response
@app.middleware("http")
async def add_custom_headers(request: Request, call_next):
response = await call_next(request)
# Add security headers
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["X-Frame-Options"] = "DENY"
response.headers["X-XSS-Protection"] = "1; mode=block"
return responseAuthentication Middleware
Authentication Middlewarepython
from sufast import App, Request, Response
import jwt
app = App()
@app.middleware("http")
async def auth_middleware(request: Request, call_next):
# Skip auth for public endpoints
public_paths = ["/", "/health", "/docs", "/auth/login"]
if request.url.path in 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 Response(
{"error": "Missing or invalid authorization header"},
status_code=401
)
# Verify JWT token
token = auth_header.split(" ")[1]
try:
payload = jwt.decode(token, "your-secret-key", algorithms=["HS256"])
request.state.user = payload # Add user info to request
except jwt.InvalidTokenError:
return Response(
{"error": "Invalid token"},
status_code=401
)
return await call_next(request)
@app.get("/protected")
def protected_endpoint(request: Request):
user = request.state.user
return {"message": f"Hello, {user['username']}!"}Request Validation Middleware
Validation Middlewarepython
from sufast import App, Request, Response
import json
app = App()
@app.middleware("http")
async def validate_json_middleware(request: Request, call_next):
# Only validate POST/PUT requests with JSON content
if request.method in ["POST", "PUT", "PATCH"]:
content_type = request.headers.get("content-type", "")
if "application/json" in content_type:
try:
# Try to parse JSON body
body = await request.body()
if body:
json.loads(body)
except json.JSONDecodeError:
return Response(
{"error": "Invalid JSON in request body"},
status_code=400
)
return await call_next(request)
@app.middleware("http")
async def request_size_limit(request: Request, call_next):
# Limit request body size to 10MB
max_size = 10 * 1024 * 1024 # 10MB
content_length = request.headers.get("content-length")
if content_length and int(content_length) > max_size:
return Response(
{"error": "Request body too large"},
status_code=413
)
return await call_next(request)Middleware Execution Order
Understanding how middleware is executed
Execution Flow
Middleware Orderpython
from sufast import App, Request
app = App()
@app.middleware("http")
async def first_middleware(request: Request, call_next):
print("First middleware - before request")
response = await call_next(request)
print("First middleware - after request")
return response
@app.middleware("http")
async def second_middleware(request: Request, call_next):
print("Second middleware - before request")
response = await call_next(request)
print("Second middleware - after request")
return response
@app.get("/test")
def test_endpoint():
print("Route handler executed")
return {"message": "test"}
# Execution order for GET /test:
# 1. First middleware - before request
# 2. Second middleware - before request
# 3. Route handler executed
# 4. Second middleware - after request
# 5. First middleware - after requestImportant: Middleware is executed in the order it's added to the application. The first middleware added will be the outermost layer, and the last middleware added will be closest to your route handlers.