一、引言FastAPI 作为当前 Python 领域最炙手可热的异步 Web 框架凭借其高性能、自动生成 API 文档以及对异步的原生支持赢得了大量开发者的青睐。然而仅仅掌握路由和请求响应的基础用法远不足以构建一个健壮的企业级应用。中间件Middleware、依赖注入Dependency Injection和ORM对象关系映射正是进阶路上必须跨越的三座大山。本文将基于 FastAPI 官方最佳实践结合 SQLAlchemy 2.0 异步特性从零搭建一个完整的图书管理 API涵盖统一日志处理、用户认证依赖、以及完整的数据库增删改查、分页、聚合查询等核心功能。所有代码均可在实际项目中直接复用。二、中间件Middleware—— 为所有请求添加“守门员”2.1 什么是中间件中间件是一个在每次请求进入 FastAPI 应用时都会被执行的函数。它在请求到达路径操作函数之前运行并且在响应返回给客户端之前再执行一次。你可以把它想象成一个“守门员”对所有进出请求进行统一处理。2.2 中间件的典型应用场景记录每个请求的耗时和日志统一身份认证虽然更推荐用依赖注入但中间件也可做全局鉴权跨域资源共享CORS配置统一添加或修改响应头如X-Request-ID性能监控和指标收集2.3 如何定义一个中间件使用app.middleware(http)装饰器函数签名为async def middleware(request: Request, call_next)。内部通过await call_next(request)调用下一个中间件或最终的路由处理函数并获取响应对象。from fastapi import FastAPI, Request import time app FastAPI() app.middleware(http) async def log_requests(request: Request, call_next): # 请求前处理 start_time time.time() print(f收到请求: {request.method} {request.url.path}) # 执行后续逻辑路由处理函数或下一个中间件 response await call_next(request) # 响应后处理 process_time time.time() - start_time response.headers[X-Process-Time] str(process_time) print(f请求处理完成耗时: {process_time:.4f}秒) return response2.4 多个中间件的执行顺序顺序是“自下而上”—— 即先注册的后执行类似洋葱模型。例如app.middleware(http) async def middleware_a(request, call_next): print(A 开始) response await call_next(request) print(A 结束) return response app.middleware(http) async def middleware_b(request, call_next): print(B 开始) response await call_next(request) print(B 结束) return response2.4.1核心结论FastAPI 中后注册的中间件会先执行前置逻辑收尾逻辑则逆序执行。 你代码里先定义middleware_a、再定义middleware_b注册顺序A→B执行打印顺序B 开始 A 开始 # 路由接口业务逻辑执行 A 结束 B 结束2.4.2 原理说明FastAPI 会把多个 HTTP 中间件层层嵌套包装B(middleware_a(路由函数))调用call_next(request)就是进入内层中间件 / 路由先走外层 B 前置B 开始进入内层 A 前置A 开始走到真正接口视图函数接口返回后先退出 A 后置A 结束最后退出 B 后置B 结束2.4.3 直观流程图请求进来 ↓ middleware_b 【前置】打印 B 开始 ↓ call_next() middleware_a 【前置】打印 A 开始 ↓ call_next() 路由处理函数执行业务 ↑ 响应返回 middleware_a 【后置】打印 A 结束 ↑ 回到上一层 middleware_b 【后置】打印 B 结束 ↓ 返回响应给客户端三、依赖注入Dependency Injection—— 告别重复代码3.1 依赖注入系统的作用依赖注入是一种设计模式FastAPI 内置了极其优雅的实现。它的核心思想是将可复用的逻辑如数据库会话、用户身份校验、配置对象抽取为依赖项由框架自动调用并注入到路径操作函数中。主要优势代码复用一次编写多处使用解耦业务逻辑与基础设施数据库、缓存等分离易于测试可以轻松用模拟对象替换真实依赖进行单元测试3.2 使用步骤创建一个可调用对象函数或类作为依赖项在路径操作函数参数中使用Depends(依赖项)声明FastAPI 自动解析并注入依赖项的结果3.3 实战用户认证依赖项假设我们有多个接口需要验证 JWT Token并返回当前用户信息。from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer from jose import JWTError, jwt oauth2_scheme OAuth2PasswordBearer(tokenUrltoken) # 模拟用户数据 fake_users_db { admin: {username: admin, role: superadmin} } async def get_current_user(token: str Depends(oauth2_scheme)): credentials_exception HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detail无效的认证凭据, headers{WWW-Authenticate: Bearer}, ) try: payload jwt.decode(token, SECRET_KEY, algorithms[HS256]) username: str payload.get(sub) if username is None: raise credentials_exception except JWTError: raise credentials_exception user fake_users_db.get(username) if user is None: raise credentials_exception return user # 在路由中使用 app.get(/users/me) async def read_users_me(current_user: dict Depends(get_current_user)): return current_user这样一来所有需要认证的接口只需声明current_user Depends(get_current_user)即可完全避免了在每个函数里重复编写解析 Token 的代码。四、ORM对象关系映射—— 用面向对象的方式操作数据库4.1 ORM 简介ORM 将数据库表映射为 Python 类将表中的行映射为类的实例对象。开发者通过操作对象和方法即可完成增删改查无需手写 SQL。在 FastAPI 异步生态中SQLAlchemy 2.0是绝对的首选它原生支持async/await性能强大企业应用广泛。ORM 带来的好处减少重复的 SQL 拼接代码代码更简洁、可读性更强自动管理数据库连接和事务有效防止 SQL 注入攻击4.2 环境准备与引擎创建首先安装依赖pip install fastapi uvicorn sqlalchemy[asyncio] aiomysql创建异步数据库引擎以 MySQL 为例from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker, declarative_base from sqlalchemy import Column, Integer, String, DateTime, func from datetime import datetime ASYNC_DATABASE_URL mysqlaiomysql://root:123456localhost:3306/fastapi_test?charsetutf8 async_engine create_async_engine( ASYNC_DATABASE_URL, echoTrue, # 打印 SQL 日志便于调试 pool_size10, # 连接池保持的持久连接数 max_overflow20 # 允许额外创建的连接数 ) AsyncSessionLocal sessionmaker( async_engine, class_AsyncSession, expire_on_commitFalse )4.3 定义模型基类和模型类我们定义一个基类包含所有表共有的create_time和update_time字段然后为图书表创建模型。from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column from sqlalchemy import String, DateTime, func class Base(DeclarativeBase): create_time: Mapped[datetime] mapped_column( DateTime, insert_defaultfunc.now(), defaultdatetime.now, comment创建时间 ) update_time: Mapped[datetime] mapped_column( DateTime, insert_defaultfunc.now(), onupdatefunc.now(), defaultdatetime.now, comment修改时间 ) class Book(Base): __tablename__ book id: Mapped[int] mapped_column(primary_keyTrue, autoincrementTrue) bookname: Mapped[str] mapped_column(String(255), nullableFalse) author: Mapped[str] mapped_column(String(255), nullableFalse) price: Mapped[float] mapped_column(nullableTrue) # 价格字段为后续查询演示扩展4.4 创建数据库表应用启动时利用 FastAPI 的startup事件在应用启动时自动建表。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.5 获取数据库会话的依赖项每次请求需要独立的事务会话我们将其封装为依赖项并自动管理会话的关闭。4.5.1异步会话工厂 AsyncSessionLocalAsyncSessionLocal async_sessionmaker( bindasync_engine, # 绑定全局异步数据库引擎提前初始化的async数据库连接引擎 class_AsyncSession, # 指定使用SQLAlchemy异步会话类AsyncSession expire_on_commitFalse # commit之后模型对象不会过期不用重复查库方便查询后继续使用ORM对象 )async_sessionmakerSQLAlchemy 异步版会话工厂用来批量创建数据库会话expire_on_commitFalse常用场景查询拿到数据 commit 后依然能正常读取模型属性不会报lazy loading过期报错4.5.2依赖函数 get_database最关键这是 FastAPI 依赖项每来一次 HTTP 请求就执行一次生成全新独立 sessionasync def get_database(): async with AsyncSessionLocal() as session: try: yield session # 把会话给到接口函数使用yield实现依赖的上下文暂停 await session.commit() # 接口无报错自动提交事务 except Exception: await session.rollback() # 接口抛异常事务回滚保证数据一致性 raise # 重新抛出异常让FastAPI正常返回错误响应 finally: await session.close() # 无论成功失败最终强制关闭会话释放连接4.5.3路由层Depends 注入使用app.get(/book/books) async def get_book_list( db: AsyncSession Depends(get_database) ): # 使用注入进来的db会话执行查询 result await db.execute(select(Book)) book_list result.scalars().all() return book_list五、路由中使用 ORM 与数据库操作5.1 依赖注入数据库会话核心创建依赖项get_database返回异步数据库会话AsyncSession并通过Depends注入到路由处理函数。from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker # 创建异步会话工厂 AsyncSessionLocal async_sessionmaker(async_engine, expire_on_commitFalse) async def get_database() - AsyncSession: async with AsyncSessionLocal() as session: yield session在路由中from fastapi import Depends, FastAPI from sqlalchemy.ext.asyncio import AsyncSession app FastAPI() app.get(/books) async def get_books(db: AsyncSession Depends(get_database)): # 使用 db 执行查询 pass5.2 查询操作5.2.1 查询所有数据与单条数据查询所有select(模型类)→db.execute()→scalars().all()from sqlalchemy import select app.get(/book/get_books) async def get_book_list(db: AsyncSession Depends(get_database)): result await db.execute(select(Book)) books result.scalars().all() return books查询单条使用scalars().first()或db.get(模型类, 主键值)app.get(/book/get_book) async def get_book(db: AsyncSession Depends(get_database)): # 方式一提取第一条 # result await db.execute(select(Book)) # book result.scalars().first() # 方式二通过主键获取 book await db.get(Book, 1) return book5.2.2 查询条件在select().where()中添加条件支持多种比较和逻辑运算。比较判断,,,,app.get(/book/{book_id}) async def get_book_by_id(book_id: int, db: AsyncSession Depends(get_database)): result await db.execute(select(Book).where(Book.id book_id)) book result.scalar_one_or_none() return book模糊查询like()%匹配任意字符_匹配单个字符app.get(/book/get_books_by_author) async def get_books_by_author(db: AsyncSession Depends(get_database)): result await db.execute(select(Book).where(Book.author.like(曹%))) books result.scalars().all() return books与非查询使用与、|或、~非app.get(/book/get_books_filter) async def get_books_filter(db: AsyncSession Depends(get_database)): result await db.execute( select(Book).where((Book.author 曹雪芹) (Book.price 200)) ) books result.scalars().all() return books包含查询in_()app.get(/book/get_books_by_ids) async def get_books_by_ids(db: AsyncSession Depends(get_database)): id_list [1, 2, 3, 4, 5, 6] result await db.execute(select(Book).where(Book.id.in_(id_list))) books result.scalars().all() return books5.2.3 聚合查询使用func.count(),func.avg(),func.max(),func.min(),func.sum()进行聚合。from sqlalchemy import func app.get(/book/count) async def get_count(db: AsyncSession Depends(get_database)): # result await db.execute(select(func.count(Book.id))) # result await db.execute(select(func.max(Book.price))) # result await db.execute(select(func.sum(Book.price))) result await db.execute(select(func.avg(Book.price))) count result.scalar() return count5.2.4 分页查询使用.offset().limit()实现分页。offset为跳过的记录数limit为每页返回的记录数。计算公式offset (当前页码 - 1) * 每页数量当前页码每页数量(limit)跳过数量(offset)1100210103102041030app.get(/book/get_books) async def get_book_list( page: int 1, page_size: int 3, db: AsyncSession Depends(get_database) ): skip (page - 1) * page_size stmt select(Book).offset(skip).limit(page_size) result await db.execute(stmt) books result.scalars().all() return {books: books}5.2.5 查询总结核心思路select()→db.execute()→ 从 ORM 对象获取数据 → 响应结果获取数据方式scalars().all()获取所有数据scalars().first()获取第一条数据scalar_one_or_none()获取一个或 None若多于一条则报错scalar()获取标量值聚合查询常用5.3 新增操作步骤创建 ORM 对象 →db.add()→await db.commit()from pydantic import BaseModel class BookBase(BaseModel): bookname: str author: str price: float app.post(/book/add_book) async def add_book(book: BookBase, db: AsyncSession Depends(get_database)): book_obj Book(**book.__dict__) # 或 Book(booknamebook.bookname, ...) db.add(book_obj) await db.commit() return book_obj5.4 更新操作步骤查询获取对象 → 修改属性 →commitclass BookUpdate(BaseModel): bookname: str author: str price: float app.put(/book/update_book/{book_id}) async def update_book( book_id: int, data: BookUpdate, db: AsyncSession Depends(get_database) ): book await db.get(Book, book_id) if book is None: raise HTTPException(status_code404, detailBook not found) book.bookname data.bookname book.author data.author book.price data.price await db.commit() return book5.5 删除操作步骤查询获取对象 →db.delete()→commitapp.delete(/book/delete_book/{book_id}) async def delete_book(book_id: int, db: AsyncSession Depends(get_database)): db_book await db.get(Book, book_id) if db_book is None: raise HTTPException(status_code404, detailBook not found) await db.delete(db_book) await db.commit() return {message: Book deleted}六、总结本文围绕 FastAPI 进阶的三大核心内容展开中间件为每个请求/响应添加统一处理逻辑定义简单执行顺序自下而上。依赖注入抽取可复用组件如数据库会话、认证逻辑实现解耦与易测性。ORMSQLAlchemy通过对象方式操作数据库避免手写 SQL提升开发效率与安全性。ORM 使用全流程速览安装依赖pip install sqlalchemy[asyncio] aiomysql创建异步引擎create_async_engine定义基类DeclarativeBase和模型类启动时建表run_sync(Base.metadata.create_all)路由中注入会话依赖Depends(get_database)执行 CRUDselect()、add()、属性赋值、delete()最后commit()掌握这些技能即可构建健壮、可维护的 FastAPI 企业级应用。