FastAPI 进阶：中间件、依赖注入与 ORM

这种重复代码不仅冗余，也难以维护。依赖注入系统就是用来解决这个问题的。

什么是依赖注入？

• 依赖项：可重用的组件（函数/类），负责提供某种功能或数据
• 注入：FastAPI 自动帮你调用依赖项，并将结果"注入"到路径操作函数中

依赖注入的优势

1. 代码复用：一次编写，多处使用
2. 解耦：业务逻辑与基础设施代码分离
3. 易于测试：轻松地用模拟依赖替换真实依赖进行测试

依赖注入的应用场景

• 从请求中提取和验证参数
• 管理数据库会话的创建、使用、关闭
• 抽取封装多个路由公用的逻辑代码
• 验证用户身份、检查权限和角色要求等

如何使用依赖注入

步骤 1：创建依赖项

async def common_parameters(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, le=60)
):
    return {
        "skip": skip,
        "limit": limit
    }

步骤 2：导入并使用 Depends

from fastapi import Depends

@app.get("/news/news_list")
async def get_news_list(
    commons = Depends(common_parameters)
):
    return commons

中间件 vs 依赖注入

三、ORM（对象关系映射）

什么是 ORM？

ORM（Object-Relational Mapping，对象关系映射）是一种编程技术，用于在面向对象编程语言和关系型数据库之间建立映射。它允许开发者通过操作对象的方式与数据库进行交互，而无需直接编写复杂的 SQL 语句。

ORM 的优势

• ✅ 减少重复的 SQL 代码
• ✅ 代码更简洁易读
• ✅ 自动处理数据库连接和事务
• ✅ 自动防止 SQL 注入攻击

常见 ORM 工具

ORM 使用流程

1. 安装依赖

• sqlalchemy[asyncio]
• aiomysql（异步数据库驱动）

2. 建库、建表

• run_sync(Base.metadata.create_all)

3. 操作数据

• 查询
• 新增
• 修改
• 删除

实战示例

1. 创建数据库引擎

from sqlalchemy.ext.asyncio import create_async_engine

ASYNC_DATABASE_URL = "mysql+aiomysql://root:123456@localhost:3306/fastapi_test?charset=utf8"

async_engine = create_async_engine(
    ASYNC_DATABASE_URL,
    echo=True,
    pool_size=10,
    max_overflow=20
)

2. 定义模型类

from sqlalchemy.ext.declarative import DeclarativeBase
from sqlalchemy.orm import mapped_column, Mapped
from sqlalchemy import DateTime, String, func
from datetime import datetime

class Base(DeclarativeBase):
    create_time: Mapped[datetime] = mapped_column(
        DateTime, insert_default=func.now(), default=datetime.now, comment="创建时间"
    )
    update_time: Mapped[datetime] = mapped_column(
        DateTime, insert_default=func.now(), onupdate=func.now(), default=datetime.now, comment="修改时间"
    )

class User(Base):
    __tablename__ = "user"
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(255))
    password: Mapped[str] = mapped_column(String(255))

3. 创建数据库表

async def create_tables():
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

@app.on_event("startup")
async def startup_event():
    await create_tables()

4. 在路由中使用 ORM

创建异步会话工厂和依赖项：

from sqlalchemy.ext.asyncio import async_sessionmaker, AsyncSession

AsyncSessionLocal = async_sessionmaker(
    bind=async_engine,
    class_=AsyncSession,
    expire_on_commit=False
)

async def get_database():
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

查询数据：

from sqlalchemy import select

@app.get("/users")
async def get_user_list(
    db: AsyncSession = Depends(get_database)
):
    result = await db.execute(select(User))
    users = result.scalars().all()
    return users

新增数据：

@app.post("/users")
async def create_user(
    user: UserCreate,
    db: AsyncSession = Depends(get_database)
):
    db_user = User(username=user.username, password=user.password)
    db.add(db_user)
    await db.flush()
    return db_user

更新数据：

@app.put("/users/{user_id}")
async def update_user(
    user_id: int,
    user_update: UserUpdate,
    db: AsyncSession = Depends(get_database)
):
    user = await db.get(User, user_id)
    if user:
        user.username = user_update.username
        user.password = user_update.password
    return user

删除数据：

@app.delete("/users/{user_id}")
async def delete_user(
    user_id: int,
    db: AsyncSession = Depends(get_database)
):
    user = await db.get(User, user_id)
    if user:
        await db.delete(user)
    return {"message": "删除成功"}

总结

FastAPI 的这三个进阶特性各有所长：

• 中间件：适合处理全局统一的请求/响应逻辑
• 依赖注入：适合在部分路由间共享逻辑，灵活可控
• ORM：让数据库操作变得面向对象，简洁安全

合理组合使用这三个特性，可以让你的 FastAPI 应用更加优雅、高效和易于维护！