1.FastAPI中的中介軟體
中間件是在每個請求之前和每個回應之後運行的程式碼。 FastAPI支援兩種類型的中間件：HTTP中間件（易於編寫）和ASGI中間件（效能更高）。
Request → Middleware 1 → Middleware 2 → Route Handler
│
Response ← Middleware 1 ← Middleware 2 ←──────┘
2.HTTP中間件（BaseHTTPMiddleware）
# app/middleware/logging.py
import time
import logging
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response
logger = logging.getLogger(__name__)
class LoggingMiddleware(BaseHTTPMiddleware):
"""Log mọi request/response với timing."""
async def dispatch(self, request: Request, call_next) -> Response:
# Before request
start_time = time.perf_counter()
request_id = getattr(request.state, "request_id", "unknown")
logger.info(
"Request started",
extra={
"request_id": request_id,
"method": request.method,
"path": request.url.path,
"client": request.client.host if request.client else "unknown",
},
)
# Process request
try:
response = await call_next(request)
except Exception as exc:
logger.error(f"Request failed: {exc}", exc_info=True)
raise
# After request
process_time = time.perf_counter() - start_time
response.headers["X-Process-Time"] = f"{process_time:.4f}"
logger.info(
"Request completed",
extra={
"request_id": request_id,
"status_code": response.status_code,
"process_time": f"{process_time:.4f}s",
},
)
return response
3.純ASGI中間件（高效能）
# app/middleware/timing.py
import time
from starlette.types import ASGIApp, Receive, Scope, Send
class TimingMiddleware:
"""Pure ASGI middleware - hiệu năng cao hơn BaseHTTPMiddleware."""
def __init__(self, app: ASGIApp):
self.app = app
async def __call__(self, scope: Scope, receive: Receive, send: Send):
if scope["type"] != "http":
await self.app(scope, receive, send)
return
start_time = time.perf_counter()
async def send_wrapper(message):
if message["type"] == "http.response.start":
process_time = time.perf_counter() - start_time
headers = list(message.get("headers", []))
headers.append(
(b"x-process-time", f"{process_time:.6f}".encode())
)
message["headers"] = headers
await send(message)
await self.app(scope, receive, send_wrapper)
4. 壽命事件
# app/main.py
from contextlib import asynccontextmanager
import redis.asyncio as redis
from fastapi import FastAPI
from app.config import settings
from app.core.database import engine, Base
@asynccontextmanager
async def lifespan(app: FastAPI):
"""
Application lifespan manager.
Code trước 'yield' chạy khi startup.
Code sau 'yield' chạy khi shutdown.
"""
# ===== STARTUP =====
print("🚀 Starting application...")
# Initialize database
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
print("✅ Database initialized")
# Initialize Redis
app.state.redis = redis.from_url(
settings.redis_url,
encoding="utf-8",
decode_responses=True,
)
print("✅ Redis connected")
# Initialize other services
# app.state.email_service = EmailService()
# app.state.storage = S3Storage()
print("✅ Application started successfully")
yield # ===== APPLICATION IS RUNNING =====
# ===== SHUTDOWN =====
print("🛑 Shutting down application...")
# Close Redis
await app.state.redis.close()
print("✅ Redis disconnected")
# Close database
await engine.dispose()
print("✅ Database connections closed")
print("✅ Application shut down successfully")
app = FastAPI(
title="FastAPI Tutorial",
lifespan=lifespan,
)
5. 應用程式狀態
# Truy cập app state trong routes
from fastapi import FastAPI, Request
app = FastAPI()
@app.get("/cache/{key}")
async def get_cache(key: str, request: Request):
"""Truy cập Redis thông qua app state."""
redis = request.app.state.redis
value = await redis.get(f"cache:{key}")
return {"key": key, "value": value}
@app.post("/cache/{key}")
async def set_cache(key: str, value: str, request: Request):
redis = request.app.state.redis
await redis.setex(f"cache:{key}", 3600, value)
return {"message": "Cached"}
6. 中介軟體順序
# app/main.py - Middleware được thêm theo thứ tự NGƯỢC
# Middleware cuối cùng được thêm sẽ chạy ĐẦU TIÊN
from starlette.middleware.cors import CORSMiddleware
from starlette.middleware.trustedhost import TrustedHostMiddleware
from app.middleware.logging import LoggingMiddleware
from app.middleware.security import SecurityHeadersMiddleware
from app.middleware.request_id import RequestIDMiddleware
# Thêm theo thứ tự ngược (last added = first executed)
app.add_middleware(LoggingMiddleware) # 3. Logging
app.add_middleware(SecurityHeadersMiddleware) # 2. Security headers
app.add_middleware(RequestIDMiddleware) # 1. Request ID (chạy đầu tiên)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
)
# Execution order:
# Request → RequestID → Security → Logging → Route
# Response ← RequestID ← Security ← Logging ← Route
總結
在這篇文章中我們了解到：
- HTTP中介軟體：使用BaseHTTPMiddleware輕鬆編寫
- ASGI中介軟體：高性能的生產
- 壽命：啟動/關閉事件管理資源
- 應用程式狀態：請求之間共享資源
- 中介軟體訂單：執行順序事項
下一篇文章將實現WebSockets和即時通訊。