FastAPI实战：异步Web开发、依赖注入与自动文档

聊到 fastapi很多人第一反应是“又一个 Python Web 框架”。说实话我以前也这么想毕竟 Flask 在中小型项目里已经够用Django 在重型项目里也站稳了脚跟。真正让我改变看法的是一次内部工具重构那套系统要定时抓取外部数据、推送 Webhook、给前端提供实时查询接口Flask 写起来本身不难但同步阻塞的 IO、手写的请求体校验、不统一的序列化代码把维护成本一点点堆高了。后来我用 fastapi 做了一个两周的技术验证发现它能同时解决校验、序列化、异步和接口文档四件事于是新项目直接切了过去。这篇内容不是 fastapi 文档的复述而是一个在 Flask 和 Tornado 之间来回切换过的开发者在真正把 fastapi 用进生产项目之后整理出的使用心得。内容包括 FastAPI 的设计逻辑、一个带鉴权的图书管理接口完整实战、依赖注入和异步化在真实场景中的用法以及上线前后绕不开的测试部署和“坑”。如果你正在评估 fastapi 适不适合自己手里那个项目或者已经动手但遇到了一些奇怪问题这篇应该能帮上忙。1. 从Flask切到FastAPI一次被异步和序列化逼出来的迁移1.1 老项目最痛的三个点先说那个让我决定迁移的项目。它本质上是一个数据汇聚服务上游有几个数据源需要定时拉取清洗后写入本地存储同时对外提供一组查询和推送接口。最开始我用 Flask 写业务路由用 requests 做上游调用用flask-restful组织资源类。前两个月没什么问题但需求越叠越多之后麻烦开始集中爆发。第一个痛点是同步阻塞。Flask 默认的 WSGI 模型是每个请求占用一个线程遇到调用上游接口、读数据库这种 IO 操作时线程就挂在那里等。流量一旦上来线程池被占满新请求只能排队。我试着用concurrent.futures做异步任务但业务代码里到处是time.sleep和同步 requests改造起来像在拆炸弹。第二个痛点是请求体验证的重复劳动。每个 POST 接口都要写一遍“字段是否存在、类型对不对、范围合理不合理”的逻辑一个接口一套写法十个接口十种风格。第三个痛点是文档。前端同学每次联调都追着问参数格式我有时改了个字段名忘了同步到文档线上对不上排查半天才发现是文档过期。这几个问题单独看都不致命凑在一起就变得非常消耗精力。后来我在一次技术分享里看到有人演示 fastapi 写接口参数校验、自动文档、异步支持几乎是“白送”的当场就坐不住了。回到工位后我搭了一个最小 Demo用fastapi和uvicorn跑起来发现代码量和维护成本确实比预期低很多于是决定用小范围试点来验证。1.2 FastAPI带着类型提示回来了FastAPI 给我的第一观感是“这玩意儿对类型提示的依赖程度非常高”。写 Flask 路由时函数参数只是一种约定你爱怎么写就怎么写框架不会强制你标注类型。但 fastapi 的设计思路反过来你声明了参数类型框架就从类型里反推出校验规则、序列化规则和 OpenAPI 文档。这个思路让 Python Web 开发的体验往前迈了一大步。做个简单的对比。同样的用户查询接口Flask 版本通常要手动从request.args取值然后自己转类型、判空、拼错误信息FastAPI 版本直接写成函数参数类型错误自动返回 422参数缺失自动有提示文档页面自动出现这个接口的完整说明。这种“类型即契约”的体验让代码的可读性和可维护性同时提升。因为所有数据结构都通过BaseModel定义接口之间共享模型前端拿到 doc 页面就能知道字段含义后端也不用反复写文档。性能方面fastapi 底层的 Starlette 是异步框架支持 ASGI在长连接和 IO 密集型场景下明显比传统 WSGI 稳。我在技术验证时测过一个小接口单机压到每秒一千多请求P95 延迟依然可控对于内部工具来说已经完全够用。2. 类型提示怎么变成了一整套Web基础设施2.1 路径参数、查询参数和请求体的统一解析用 FastAPI 写接口最直观的变化是参数解析不再靠猜。路径参数直接在路由字符串里声明查询参数用默认值表示可选请求体用BaseModel类型声明结构。框架根据函数签名和类型注解自动完成所有解析工作。from fastapi import FastAPI, Path, Query app FastAPI() app.get(/books/{book_id}) def get_book( book_id: int Path(..., title图书ID, ge1), detail: bool Query(False, description是否返回详情), ): return {book_id: book_id, detail: detail}这里面有几个值得注意的细节。book_id声明为int如果请求传入的是字符串abcFastAPI 会直接返回 422 校验错误而不是等到业务代码里才报错。ge1是数值范围约束传0或负数也会被拦截。detail是查询参数有默认值False所以这个参数可有可无。这种声明方式最大的好处是接口的“边界”在函数签名处就画清楚了。后面维护代码的人不需要翻文档也不需要读完整函数体只看参数定义就知道这个接口接受什么、拒绝什么。我后来接手过一个旧 Flask 项目里面大部分接口都靠request.args.get手动处理参数类型混乱有的地方返回1有的地方返回1前端一直被这种不一致坑。用 fastapi 之后这类问题在入口处就被类型系统挡住了。请求体的处理也类似。定义一个 Pydantic 模型路由函数声明这个模型为参数FastAPI 会自动解析 JSON、做校验、把数据转成模型实例。写业务代码时拿到的就是一个已经验证过的对象不需要再关心“上游传少了字段怎么办”这类问题。2.2 Pydantic模型数据校验、序列化和业务结构Pydantic 是 FastAPI 的校验核心。它看起来像是一个 dataclass 增强版但能力比 dataclass 强很多支持类型转换、嵌套结构、字段约束、自定义验证器还能从模型实例直接.json()序列化。from pydantic import BaseModel, Field, validator class Author(BaseModel): name: str Field(..., min_length1, max_length50) email: str Field() class Book(BaseModel): title: str author: Author price: float Field(..., gt0, le9999) validator(title) def title_not_blank(cls, v): if not v.strip(): raise ValueError(title cannot be blank) return v.strip()在这个例子里Book嵌套了一个Author模型FastAPI 收到请求体时会自动完成嵌套校验。price必须大于 0 且不超过 9999title不能是空白字符串。这些规则全部由类型和校验器声明不需要写一长串if判断。我用 Pydantic 之后还发现一个额外好处它可以作为业务层的“结构定义”使用。数据库查询结果转成模型外部调用方传出的数据再转成模型视图层拿到的是稳定结构。就算后端数据库字段改了只要模型对外不变接口契约就不变。这比散落各处的字典和列表要可靠得多。不过刚上手时也容易踩坑。Pydantic 的默认行为是“能转就转”比如price: float收到字符串19.9会转成19.9而不会报错。有些场景希望严格类型需要额外配置。我在内部规范里会明确要求金额、ID、状态码这类字段能严格就严格避免脏数据进入业务层。按我实测的经验2.x 版本和 1.x 版本在字段定义、验证器写法上有些差异。如果是新项目直接用 v2 写法和功能如果是老项目迁移先跑一遍测试把所有validator换成新的field_validator免得升级后校验逻辑失效。2.3 自动文档背后其实是OpenAPIFastAPI 最让团队省心的一点是接口文档完全零成本生成。启动服务后访问/docs就能看到 Swagger UI访问/redoc是另一种风格的红oc文档。这两个页面背后其实是同一个 OpenAPI JSON 文件路径是/openapi.json。这里的机制并不神秘FastAPI 从路由装饰器和函数签名里收集路径、请求方法、参数类型、模型定义、描述信息最终聚合出标准的 OpenAPI 文档。你给函数写的 docstring 会成为接口描述Query、Path里的description会成为字段说明BaseModel的注释会成为 schema 描述。这些原本要单独维护的内容现在全部内嵌在代码里。所以写 FastAPI 接口时不要吝啬函数注释和字段说明。你以为只是给自己看的 docstring其实会直接出现在文档页面上前端和测试同事都能看到。我在项目里会要求每个接口至少写清楚业务用途、参数含义、返回结构这样就没有人再问“这个字段是什么意思”了。如果某些接口不想对外暴露在文档里可以在依赖注入或者路由的时候把include_in_schemaFalse传进去。比如内部健康检查接口、调试接口就可以从文档中隐藏。这点在给外部合作方开放部分接口时特别有用。3. 图书管理接口实战从零搭一套带鉴权的CRUD3.1 工程结构和数据模型理论说了不少下面用一个具体的图书管理接口来演示如何落地。这个项目我通常用来做团队新人培训因为它覆盖了 FastAPI 最基础的几个环节路由、数据模型、数据库、鉴权。工程结构不复杂一个目录放配置一个目录放数据库模型一个目录放路由主文件负责创建应用。当你项目从单个文件膨胀到十几个路由之后建议尽早拆分否则依赖关系会越来越乱。# app/models.py from sqlalchemy import Column, Integer, String, Float from sqlalchemy.ext.declarative import declarative_base Base declarative_base() class Book(Base): __tablename__ books id Column(Integer, primary_keyTrue, indexTrue) title Column(String(128), nullableFalse) author Column(String(64), nullableFalse) price Column(Float, default0)数据库层我用 SQLAlchemy 2.x 来做 ORM。之所以选它而不是直接用 SQL 拼字符串是因为模型定义可以和 Pydantic 模型互相映射写起来更顺手。要注意的是ORM 模型和 API 入参模型最好分开定义。ORM 模型里的id、数据库约束字段属于内部实现入参模型只关心用户能传什么。混在一起容易出现“用户传了 id 把主键覆盖了”之类的低级错误。3.2 路由与数据库会话FastAPI 里一个常见做法是使用Depends来提供数据库会话。每次请求进来时创建独立的 Session请求结束自动关闭。这样既保证了事务隔离也不会因为忘记关闭连接导致连接池泄漏。from fastapi import Depends, FastAPI, HTTPException from sqlalchemy.orm import Session from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker DATABASE_URL sqlite:///./books.db engine create_engine(DATABASE_URL, connect_args{check_same_thread: False}) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) def get_db(): db SessionLocal() try: yield db finally: db.close() app FastAPI() app.get(/books) def list_books(db: Session Depends(get_db)): return db.query(Book).all()路由函数里声明db: Session Depends(get_db)FastAPI 就会在请求进入时调用get_db拿到数据库会话然后把它注入到函数参数中。请求结束后执行finally关闭会话。这个模式是 FastAPI 项目最标准的数据库连接方式只要理解了Depends后面写用户鉴权、权限校验都是同一套思路。继续补全创建和查询逻辑from pydantic import BaseModel class BookCreate(BaseModel): title: str author: str price: float app.post(/books, status_code201) def create_book(payload: BookCreate, db: Session Depends(get_db)): book Book(**payload.model_dump()) db.add(book) db.commit() db.refresh(book) return book注意这里我用了model_dump()而不是dict()这是 Pydantic v2 的写法。新版里dict()还能用但model_dump()是官方推荐的 API语义更清晰。如果接手的代码是 v1 写的迁移时这一行几乎是必改项。3.3 JWT登录与用户鉴权只有一个 CRUD 接口的 API 看不出威力加上鉴权之后依赖注入的价值就体现出来了。我用 JWT 做一个简单的用户登录流程。用户登录成功后签发一个加密 token。后续请求在 Header 里带上Authorization: Bearer tokenFastAPI 通过依赖函数解析 token得到当前用户再把用户对象注入到路由函数。from datetime import datetime, timedelta, timezone import jwt from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials SECRET_KEY replace-me-in-production ALGORITHM HS256 security HTTPBearer() def create_token(user_id: int) - str: expire datetime.now(timezone.utc) timedelta(hours12) payload {sub: str(user_id), exp: expire} return jwt.encode(payload, SECRET_KEY, algorithmALGORITHM) def get_current_user( credentials: HTTPAuthorizationCredentials Depends(security), db: Session Depends(get_db), ): token credentials.credentials try: payload jwt.decode(token, SECRET_KEY, algorithms[ALGORITHM]) user_id int(payload[sub]) except jwt.PyJWTError: raise HTTPException(status_code401, detailinvalid token) user db.query(User).filter(User.id user_id).first() if user is None: raise HTTPException(status_code401, detailuser not found) return user路由里只要加上user: User Depends(get_current_user)整个接口就自动进入登录态。没有 token、token 过期或者用户被删除都会统一返回 401。你不需要在业务代码里到处写 if 判断。这个模式的扩展性很强。比如有的接口需要管理员权限就再写一个依赖函数require_admin内部先调用get_current_user再检查用户的角色字段。职责划分一清二楚测试也方便替换依赖即可。4. 依赖注入与异步化FastAPI真正值钱的设计4.1 Depends是一种可组合的管道依赖注入不是 FastAPI 发明的但它在 FastAPI 里的地位极其重要。你可以把Depends看作一个“可以组合的处理管道”一个依赖可以调用另一个依赖层层嵌套最终拼出一条完整的请求处理链路。最常见的例子是分页参数。如果每个列表接口都自己定义一个page、page_size参数代码会非常冗长。定义一个公共依赖之后所有接口都能复用from typing import Optional def pagination( page: int 1, page_size: int Query(20, ge1, le100) ): return {page: page, page_size: page_size} app.get(/books) def list_books( page: dict Depends(pagination), db: Session Depends(get_db), ): offset (page[page] - 1) * page[page_size] items db.query(Book).offset(offset).limit(page[page_size]).all() return {items: items, total: len(items), page: page}这样的好处是接口之间的行为一致。之后想改默认分页大小或者加一个最大限制只需要动一个地方。这种“组合”能力是 FastAPI 区别于 Flask 的一个重要原因。Flask 也可以用装饰器做类似的事但类型注解和 IDE 提示的体验差很多。依赖还有一个容易被忽视的用途资源清理。如果用yield写依赖函数yield之前的部分是初始化yield之后的部分是收尾。数据库连接、文件句柄、外部客户端连接都能交给依赖函数管理请求结束自动释放比在业务代码里手动 try-finally 干净得多。4.2 异步请求、async def和线程池的取舍FastAPI 同时支持def和async def路由。很多刚上手的人不清楚这两者的区别我给你一个判断标准如果路由里使用的是同步 IO 库比如 SQLAlchemy 的传统 Session、requests就老老实实写def如果使用异步库比如httpx.AsyncClient、异步数据库驱动就写async def这样才能发挥事件循环的威力。为什么这么说因为 FastAPI 的底层机制是这样设计的同步路由会被丢进线程池运行不阻塞事件循环async def路由则直接运行在事件循环上。如果你用async def路由去调用同步阻塞库整个事件循环会被卡住服务端所有请求都会排队性能反而更差。import httpx from fastapi import FastAPI app FastAPI() app.get(/proxy) async def proxy(): async with httpx.AsyncClient() as client: resp await client.get(https://example.com/data) return {status: resp.status_code}这段代码是典型的异步场景对外发起 HTTP 请求等待响应的过程不占用事件循环。同样的请求用同步写法也跑得通但在高并发时会成为瓶颈。我们项目里的经验是数据库操作如果走同步 ORM用def路由纯计算任务也用def路由让框架自动丢线程池只有网络 IO 密集型操作调用第三方 API、长连接推送才用async def。不要为了“看起来异步”而强行异步关键看底层库是否支持非阻塞。4.3 BackgroundTasks做轻量异步任务有些跟接口调用没有直接关系的动作比如发送通知邮件、写日志、更新统计缓存不适合在请求里同步执行否则接口响应时间会被拉长。FastAPI 内建的BackgroundTasks是一个轻量解决方案不需要引入 Celery 就能处理简单场景。from fastapi import BackgroundTasks def send_notification(email: str, message: str): # 这里执行真正的邮件发送逻辑 pass app.post(/books) def create_book_with_notify( payload: BookCreate, background: BackgroundTasks, db: Session Depends(get_db), ): book Book(**payload.model_dump()) db.add(book) db.commit() db.refresh(book) background.add_task(send_notification, adminexample.com, fnew book: {book.title}) return bookBackgroundTasks会在响应发送之后执行任务客户端不用等待任务完成。这个特性非常适合内部工具里的“顺手做点什么”场景。如果以后任务变复杂比如需要重试、任务编排、定时调度再迁到 Celery 也不迟。在项目早期就引入 Celery 往往属于过度设计。5. 上线前必须处理的测试、部署与“坑”5.1 用TestClient把接口行为固定下来FastAPI 的测试体验很好因为它自带一个基于httpx的TestClient可以像真实请求一样调用路由不需要额外起服务。在 CI 里跑测试非常快适合把接口行为固定下来。from fastapi.testclient import TestClient client TestClient(app) def test_create_book(): resp client.post(/books, json{ title: FastAPI in Action, author: Someone, price: 49.9, }) assert resp.status_code 201 data resp.json() assert data[title] FastAPI in Action assert id in data测试代码的写法和 requests 库几乎一样团队新人上手成本极低。我习惯在每个路由文件旁放一个test_xxx.py覆盖正常路径、参数错误、缺少鉴权三类场景。这样做的好处是以后改模型字段、加校验规则时跑一遍测试就能发现破坏性改动。测试时要注意把数据库隔离。测试用例里创建临时 SQLite 内存库或者用独立的测试库别碰开发数据。几个测试文件同时跑的时候数据库表因为共享导致互相影响是新手最常见的坑。5.2 部署时需要注意的多个进程问题FastAPI 项目部署通常用 Uvicorn 或 Gunicorn 作为进程管理器。如果是在容器里跑单进程直接uvicorn main:app就够如果要跑多 worker我建议用 Gunicorn 配合 Uvicorn worker这样能利用多核 CPU也能处理优雅重启和 worker 崩溃。gunicorn main:app -k uvicorn.workers.UvicornWorker -w 4 -b 0.0.0.0:8000 --timeout 120这里的-w 4表示启动 4 个 worker 进程通常按 CPU 核数的 1 到 2 倍设置。部署时一定要记得关闭--reload这个参数只适合开发环境生产环境开着会频繁重启还会带来代码热加载的安全隐患。多进程部署下有个很容易忽略的问题内存中的状态不共享。如果你用BackgroundTasks或全局变量存数据多个 worker 之间互相不感知。需要共享状态时上 Redis 或数据库别指望进程内字典能跨 worker。我开始做内部工具时没注意这个写了个统计在线人数的功能测试环境一个进程没事生产多个进程后发现数字来回跳排查半天才意识到是全局变量在捣乱。反向代理层我习惯用 Nginx 做 TLS 卸载和静态文件处理应用层只关注 API。需要注意 Nginx 到 Uvicorn 的超时设置长接口或文件上传接口如果超过了默认超时会被代理提前断开。调参时要前后端一起看别只调一边。5.3 我踩过的三个典型坑第一个坑是 CORS 预检请求。前端在浏览器里跨域调用接口时会先发一个OPTIONS请求。FastAPI 默认不处理 CORS需要在应用里配置CORSMiddleware。我第一版没配前端同事联调时发现所有带自定义 Header 的请求都失败日志里又看不到错误折腾了一下午。配好之后还要注意allow_headers要把Authorization加进去否则 JWT 照样会被浏览器拦截。第二个坑是 Pydantic v1 到 v2 的兼容差异。旧项目从 v1 升级时很多校验器写法会直接崩。比如原来的validator在新版本里变成了field_validator原来的dict()变成了model_dump()。这些是 API 层面的变化不是细节问题。升级前先看看官方迁移指南把测试用例跑一遍。第三个坑是async def路由里用同步 requests。我之前有个第三方 API 回调接口在测试环境表现正常上线后并发一高就出现接口超时。后来抓包发现是因为路由里用同步 requests 请求外部服务由于事件循环被阻塞整个服务的吞吐量骤降。改成httpx.AsyncClient之后问题立刻消失。还有一个容易被忽略的问题数据库连接池的配置。SQLAlchemy 默认连接池参数在小流量下没问题但长时间运行后可能出现连接失效。我遇到过“服务刚启动正常跑几天后部分请求报数据库连接断开”的情况后来加了连接回收和重连逻辑才解决。这类问题的特点是间歇性出现排查时优先看数据库连接相关的日志。