如何扩展FastAPI-boilerplate：自定义中间件与依赖注入详解

【免费下载链接】FastAPI-boilerplate An extendable async API using FastAPI, Pydantic V2, SQLAlchemy 2.0, PostgreSQL and Redis. 项目地址: https://gitcode.com/gh_mirrors/fastap/FastAPI-boilerplate

FastAPI-boilerplate是一个基于FastAPI、Pydantic V2、SQLAlchemy 2.0、PostgreSQL和Redis构建的可扩展异步API框架。本文将详细介绍如何通过自定义中间件和依赖注入来扩展该框架的功能，帮助开发者轻松实现请求处理流程的定制化和业务逻辑的复用。

一、自定义中间件：拦截请求与响应的利器

中间件是FastAPI处理请求-响应周期的重要组件，它可以在请求到达路由处理函数之前或响应返回客户端之前执行自定义逻辑。FastAPI-boilerplate已内置了日志中间件和客户端缓存中间件，我们可以以此为基础创建自己的中间件。

1.1 中间件的基本结构

所有中间件都需要继承BaseHTTPMiddleware并实现dispatch方法。以下是框架中LoggerMiddleware的核心代码：

class LoggerMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
        # 请求处理前的逻辑
        request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
        structlog.contextvars.bind_contextvars(request_id=request_id)
        
        # 调用下一个中间件或路由处理函数
        response = await call_next(request)
        
        # 响应处理逻辑
        response.headers["X-Request-ID"] = request_id
        return response

1.2 创建自定义计时中间件

假设我们需要记录每个请求的处理时间，可以创建一个TimingMiddleware：

# src/app/middleware/timing_middleware.py
import time
from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint

class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
        start_time = time.time()
        response = await call_next(request)
        process_time = time.time() - start_time
        response.headers["X-Process-Time"] = str(process_time)
        return response

1.3 注册中间件到应用

创建好中间件后，需要在应用初始化时注册：

# src/app/core/setup.py
from fastapi import FastAPI
from app.middleware.timing_middleware import TimingMiddleware

def setup_middlewares(app: FastAPI) -> None:
    app.add_middleware(TimingMiddleware)
    # 其他中间件注册...

二、依赖注入：实现业务逻辑复用

依赖注入是FastAPI的核心特性之一，它允许我们将通用逻辑抽象为可重用的依赖项，在路由函数中通过Depends进行注入。FastAPI-boilerplate在src/app/api/dependencies.py中定义了多种常用依赖。

2.1 依赖注入的基本用法

以下是框架中获取当前用户的依赖示例：

async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)], 
    db: Annotated[AsyncSession, Depends(async_get_db)]
) -> dict[str, Any]:
    token_data = await verify_token(token, TokenType.ACCESS, db)
    # 用户验证逻辑...
    return user

在路由中使用：

@router.get("/users/me")
async def read_users_me(current_user: Annotated[dict, Depends(get_current_user)]):
    return current_user

2.2 创建自定义权限依赖

假设我们需要限制某些接口只能被特定角色访问，可以创建角色验证依赖：

# src/app/api/dependencies.py
async def require_role(current_user: Annotated[dict, Depends(get_current_user)], required_role: str):
    if current_user["role"] != required_role:
        raise ForbiddenException("Insufficient permissions")
    return current_user

# 在路由中使用
@router.post("/admin/settings")
async def update_settings(
    current_user: Annotated[dict, Depends(require_role(required_role="admin"))],
    settings: SettingsUpdate
):
    # 处理逻辑...

2.3 依赖链与异步依赖

FastAPI支持依赖的嵌套和异步操作。例如框架中的速率限制依赖rate_limiter_dependency就依赖了数据库会话和用户信息：

async def rate_limiter_dependency(
    request: Request, 
    db: Annotated[AsyncSession, Depends(async_get_db)], 
    user: dict | None = Depends(get_optional_user)
) -> None:
    # 速率限制逻辑...

三、最佳实践与扩展建议

3.1 中间件开发建议

• 保持中间件职责单一，避免在一个中间件中实现过多功能
• 通过配置参数使中间件更灵活，如ClientCacheMiddleware的max_age参数
• 注意中间件的执行顺序，先注册的中间件会先执行请求处理，后执行响应处理

3.2 依赖注入最佳实践

• 将复杂依赖拆分为多个简单依赖，提高复用性
• 使用Annotated类型提示增强代码可读性
• 为依赖添加文档字符串，说明其用途和参数

3.3 项目结构与扩展位置

FastAPI-boilerplate的中间件位于src/app/middleware/目录，依赖项定义在src/app/api/dependencies.py。建议遵循此结构，将新创建的中间件和依赖放在相应位置，保持项目结构清晰。

通过自定义中间件和依赖注入，我们可以轻松扩展FastAPI-boilerplate的功能，使其更好地满足特定业务需求。无论是请求处理流程的定制，还是业务逻辑的复用，这两个特性都能极大提高开发效率和代码质量。

【免费下载链接】FastAPI-boilerplate An extendable async API using FastAPI, Pydantic V2, SQLAlchemy 2.0, PostgreSQL and Redis. 项目地址: https://gitcode.com/gh_mirrors/fastap/FastAPI-boilerplate