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

@app.middleware("http") async def middleware(request, call_next): print('中间件开始处理 -- start') response = await call_next(request) print('中间件处理完成 -- end') return response

多个中间件的执行顺序

多个中间件的执行顺序是自下而上的：

客户端 → 中间件 B → 中间件 A → 路由处理函数 → 中间件 A → 中间件 B → 客户端

二、依赖注入（Dependency Injection）

为什么需要依赖注入？

在开发过程中，我们经常会遇到多个接口需要相同参数或逻辑的情况：

@app.get('/news/news_list') async def get_news_list( skip: int = Query(0, ge=0), limit: int = Query(10, le=60) ): return {"list": "新闻列表"} @app.get("/users/user_list") async def get_user_list( skip: int = Query(0, ge=0), limit: int = Query(10, le=60) ): return {"users": "用户列表"}

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

什么是依赖注入？

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

依赖注入的优势

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

依赖注入的应用场景

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

如何使用依赖注入

步骤 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 ORM	功能最强、最灵活、企业级	各类 API、微服务、数据应用
2	Django ORM	封装好、上手快	Django 项目、管理后台
3	Tortoise ORM	全异步	异步 Web 服务、高并发 API

ORM 使用流程

安装依赖

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

建库、建表

run_sync(Base.metadata.create_all)

操作数据

查询
新增
修改
删除

实战示例

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 应用更加优雅、高效和易于维护！

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

Ne0inhk

一、中间件（Middleware）

什么是中间件？

中间件的作用

如何定义中间件