FastAPI 基础篇：类型注解驱动的 Python Web 开发范式

FastAPI 基础篇类型注解驱动的 Python Web 开发范式1. 引子写 API 为什么这么累先问一个问题用 Flask 或 Django 写一个带参数校验和文档的 POST 接口需要写多少代码# Flask 写一个带校验的 POST 接口fromflaskimportFlask,request,jsonifyfrommarshmallowimportSchema,fields,ValidationError appFlask(__name__)classBookSchema(Schema):titlefields.String(requiredTrue,validatevalidate.Length(min2,max50))pricefields.Float(requiredTrue,validatevalidate.Range(min0))app.route(/books,methods[POST])defcreate_book():# 手动解析 JSONdatarequest.get_json()# 手动校验schemaBookSchema()try:bookschema.load(data)exceptValidationErrorase:returnjsonify({errors:e.messages}),422# 业务逻辑returnjsonify({title:book[title],price:book[price]})# 还没写文档呢……这份代码暴露了传统框架的典型痛点手动解析请求体手动定义校验规则且校验和路由是分离的手动维护API 文档或者额外装 flasgger/Swagger手动序列化响应如果每个接口都这样写项目中 30% 的代码都在做参数搬运。更要命的是校验代码、文档注释、业务逻辑散落在三个地方修改一个字段要改三处。FastAPI 解决这个问题的思路很激进你只需要写 Python 类型注解剩下的框架替你搞定。# FastAPI 做同样的事fromfastapiimportFastAPIfrompydanticimportBaseModel,Field appFastAPI()classBookIn(BaseModel):title:strField(...,min_length2,max_length50)price:floatField(...,gt0)app.post(/books)asyncdefcreate_book(book:BookIn):return{title:book.title,price:book.price}做完了。校验、文档、序列化全部自动完成打开http://127.0.0.1:8000/docs就能看到交互式文档。2. 核心概念FastAPI 凭什么能这么干FastAPI 不是凭空冒出来的东西它的能力建立在三个技术支柱上FastAPI Starlette异步网络能力 Pydantic数据校验 类型注解驱动2.1 ASGI 与异步在 FastAPI 之前Python Web 框架基本都跑在 WSGIWeb Server Gateway Interface上这是 Python 在 2003 年提出的标准。WSGI 的工作方式是来一个请求开一个线程处理完再响应——同步阻塞。ASGIAsynchronous Server Gateway Interface是 2016 年提出的新一代标准它允许服务器在处理一个请求的同时去处理另一个请求。这对 IO 密集型场景数据库查询、外部 API 调用、文件读写提升巨大。对比WSGIFlask/Django 传统模式ASGIFastAPI/Starlette处理方式同步阻塞一个请求占一个线程异步非阻塞事件循环调度并发能力靠多线程线程切换开销大靠协程轻量级切换长连接不支持 WebSocket 原生原生支持 WebSocket/SSE性能中等高接近 Node.js/Go 的水平FastAPI 跑在 Starlette 之上Starlette 是一个轻量级的 ASGI 框架/工具包提供了路由、中间件、WebSocket 等底层能力。FastAPI 在 Starlette 之上加了一层类型注解驱动的 API 层这才是它的核心竞争力。2.2 类型注解驱动Python 的类型注解Type Hints从 Python 3.5 开始引入最初只是为了给 IDE 做静态检查。FastAPI 把这个特性玩出了新高度类型注解不再是注释而是框架行为的指令。app.get(/items/{item_id})asyncdefread_item(item_id:int,# 路径参数自动转 intq:str|NoneNone,# 查询参数可选book:BookIn,# 请求体自动解析 JSONtoken:strHeader(None),# 请求头session_id:strCookie(None),# Cookie):return{item_id:item_id,q:q}每一处注解都在告诉 FastAPI 三件事参数从哪里来路径、查询、请求体、请求头……应该是什么类型自动校验 转换是否可选有默认值 可选没有 必填2.3 Pydantic背后的守门员Pydantic 是 FastAPI 用来做数据校验的引擎。每当 FastAPI 接收到请求数据都会交给 Pydantic 模型去做校验。Pydantic 的核心机制是BaseModel——你定义一个类声明字段和类型Pydantic 自动完成校验、转换和序列化frompydanticimportBaseModel,Field,EmailStrfromdatetimeimportdatetimeclassUser(BaseModel):id:intname:strField(...,min_length2,max_length20)email:EmailStr created_at:datetime|NoneNone# 传入的 JSON 会自动校验和转换# {id: 123, name: 张三, email: testexample.com}# → id 自动从字符串 123 转为整数 123# → email 格式不对直接返回 422 错误# 输出的对象自动序列化为 JSON# → 日期时间自动转为 ISO 格式字符串# → 定义 response_model 时可以过滤敏感字段2.4 自动文档的原理FastAPI 在应用启动时遍历所有注册的路由根据你的类型注解自动生成 OpenAPI原 Swagger规范的 JSON 文件。然后基于这个 JSON 渲染出两个交互式文档Swagger UI/docs可以在这个页面直接调试接口ReDoc/redoc更清晰的可读性文档这意味着你不需要单独维护一份文档。修改了参数类型文档自动更新。代码即文档。3. 核心体系FastAPI 的请求生命周期理解 FastAPI 的最好方式是跟踪一个请求从进入到返回的全过程。下图展示了这个生命周期在这个生命周期中核心知识点分布在五个层面下面逐一拆解。3.1 路由与参数体系FastAPI 的路由定义非常直观app.get()、app.post()等装饰器绑定 URL 路径和 HTTP 方法。参数从哪里来由函数签名中的类型和默认值决定路径参数用{}包裹变量名FastAPI 自动从 URL 中提取按类型注解做转换。app.get(/users/{user_id})asyncdefget_user(user_id:int):# 访问 /users/abc 自动返回 422return{user_id:user_id}查询参数函数参数中不属于路径占位符的自动识别为查询参数。app.get(/items/)asyncdeflist_items(skip:int0,# 可选默认 0limit:int10,# 可选默认 10category:str,# 必选没有默认值):return{skip:skip,limit:limit,category:category}请求体参数Pydantic 模型类型的参数自动从 JSON 请求体中解析。app.post(/books/)asyncdefcreate_book(book:BookIn):# BookIn 继承自 BaseModelreturn{id:1,**book.model_dump()}请求头与 Cookie使用Header()和Cookie()显式声明。app.get(/secure)asyncdefsecure_endpoint(token:strHeader(...,aliasAuthorization),session_id:strCookie(None),):return{valid:True}表单与文件使用Form()和UploadFile。fromfastapiimportForm,File,UploadFileapp.post(/login)asyncdeflogin(username:strForm(...),password:strForm(...),avatar:UploadFileFile(None),):return{username:username}3.2 参数校验体系FastAPI 的参数校验分两层第一层类型注解自带的校验item_id:int# 自动校验是否为整数price:float# 自动校验是否为浮点数第二层Query()/Path()/Field()提供的增强校验fromtypingimportAnnotatedfromfastapiimportQuery,Path# Annotated 写法推荐Python 3.9app.get(/items/)asyncdefread_items(q:Annotated[str|None,Query(min_length3,max_length50,pattern^[a-zA-Z0-9]$,description搜索关键词,)]None,page:Annotated[int,Query(ge1)]1,):pass注意Annotated写法的优势是类型信息完整IDE 能正确提示。旧的q: str Query(defaultNone, min_length3)写法把默认值和校验规则混在一起IDE 可能会误以为q总是字符串。在 Pydantic 模型中Field()承担同样的职责classBook(BaseModel):title:strField(...,min_length2,max_length100,description书名)price:floatField(...,gt0,le10000,description价格)tags:list[str]Field(default[],max_length5)gt大于、ge大于等于、lt小于、le小于等于、min_length、max_length、pattern正则——这些校验参数覆盖了 90% 的日常需求。如果还不够可以用AfterValidator写自定义校验函数。3.3 响应处理FastAPI 的响应处理有三大机制响应模型response_model这是 FastAPI 的杀手锏之一。通过声明response_model框架会自动过滤掉不在模型中的字段自动做类型转换和校验自动生成文档classUserIn(BaseModel):username:strpassword:str# 输入时需要email:strclassUserOut(BaseModel):username:stremail:str# 响应时不暴露密码app.post(/users/,response_modelUserOut)asyncdefcreate_user(user:UserIn):returnuser# password 会被自动过滤掉响应状态码通过status_code指定推荐使用fastapi.status中的常量。fromfastapiimportstatusapp.post(/items/,status_codestatus.HTTP_201_CREATED)asyncdefcreate_item():return{message:created}响应类型家族FastAPI 提供了多种响应类覆盖不同场景响应类Content-Type适用场景JSONResponse默认application/json常规 JSON 数据HTMLResponsetext/html返回 HTML 页面PlainTextResponsetext/plain返回纯文本RedirectResponse3xx 状态码URL 重定向FileResponse自动识别文件下载StreamingResponse自定义流式输出、大文件下载、SSE流式响应StreamingResponse值得单独拿出来讲因为它在 LLM 应用、大文件下载场景下非常实用fromfastapi.responsesimportStreamingResponseasyncdeffile_iterator(file_path:str,chunk_size:int8192):逐块读取文件避免内存溢出withopen(file_path,rb)asf:whilechunk:f.read(chunk_size):yieldchunkapp.get(/stream/file)asyncdefstream_large_file():returnStreamingResponse(contentfile_iterator(./large_file.zip),media_typeapplication/octet-stream,)核心思想是不用一次性把整个文件读到内存而是边读边发。文件大小和内存占用无关。SSEServer-Sent Events是另一种流式响应用于服务端向客户端单向推送app.get(/stream/sse)asyncdefsse_stream():asyncdefsse_generator():foriinrange(10):yieldfdata: 第{i}条消息\n\n.encode(utf-8)returnStreamingResponse(contentsse_generator(),media_typetext/event-stream,headers{Cache-Control:no-cache,Connection:keep-alive,})3.4 异常处理FastAPI 的异常处理遵循即抛即停原则raise HTTPException后当前请求立即终止返回指定的状态码和错误信息。fromfastapiimportHTTPExceptionapp.get(/items/{item_id})asyncdefread_item(item_id:str):ifitem_idnotinitems:raiseHTTPException(status_code404,detailItem not found,headers{X-Error:not_found},)return{item:items[item_id]}自定义异常处理器可以统一处理业务异常fromfastapiimportRequestfromfastapi.responsesimportJSONResponseclassBusinessError(Exception):def__init__(self,code:int,message:str):self.codecode self.messagemessageapp.exception_handler(BusinessError)asyncdefbusiness_error_handler(request:Request,exc:BusinessError):returnJSONResponse(status_codeexc.code,content{code:exc.code,message:exc.message},)# 使用raiseBusinessError(400,库存不足)这样整个应用的错误响应格式就统一了。3.5 依赖注入FastAPI 的设计精髓依赖注入是 FastAPI 与其他框架拉开差距的关键特性。一句话解释“你的函数需要什么就声明什么FastAPI 负责帮你取来。”基础用法定义一个普通的函数作为依赖通过Depends()注入到路径操作中。fromtypingimportAnnotatedfromfastapiimportDepends# 定义依赖函数asyncdefcommon_params(skip:int0,limit:int10):return{skip:skip,limit:limit}# 注入到路径操作app.get(/items/)asyncdefread_items(params:Annotated[dict,Depends(common_params)]):returnparams依赖链依赖可以依赖其他依赖形成链式调用。defget_query(q:str|NoneNone):returnqdefget_query_or_empty(q:Annotated[str,Depends(get_query)]):returnqoremptyapp.get(/search/)asyncdefsearch(query:Annotated[str,Depends(get_query_or_empty)]):return{query:query}yield 依赖资源管理当一个依赖需要管理资源如数据库连接时用yield代替returnyield之前的代码在请求前执行之后的代码在响应后执行。asyncdefget_db():dbDatabaseSession()try:yielddb# 请求期间使用这个 dbfinally:awaitdb.close()# 请求结束后自动关闭app.get(/users/)asyncdefread_users(db:Annotated[DatabaseSession,Depends(get_db)]):returndb.query(User).all()依赖的作用范围范围写法生效范围全局app FastAPI(dependencies[Depends(auth)])所有路由模块级APIRouter(dependencies[Depends(auth)])该模块所有路由路由级router.get(/, dependencies[Depends(auth)])单个路由不注入返回值参数级func(param Depends(func))单个函数参数依赖覆盖测试用测试时可以替换依赖比如用 Mock 数据库替换真实数据库app.dependency_overrides[get_db]override_get_db# 测试期间所有用到 get_db 的地方都会使用 override_get_db4. 避坑指南 / 最佳实践4.1 推荐使用 Annotated 写法新旧两种写法对比# 旧写法默认值和校验混在一起q:strQuery(defaultNone,min_length3,max_length50)# 新写法类型、校验、默认值位置清晰q:Annotated[str|None,Query(min_length3,max_length50)]None新写法有两大好处IDE 类型提示更准确旧写法中 IDE 可能认为q总是str导致后续代码误报参数结构更清晰类型注解中的Query()只负责校验规则最后的 None才是默认值4.2 response_model 的安全问题用response_model过滤敏感字段时要注意FastAPI 是在数据返回前做过滤的如果你在路径操作函数内部就把用户密码打印到了日志里response_model可帮不了你。# 错误做法敏感字段输出了才过滤classUserOut(BaseModel):username:stremail:strapp.post(/users/,response_modelUserOut)asyncdefcreate_user(user:UserIn):print(user.password)# 密码已经在内存中了returnuser最佳实践在定义 Pydantic 模型时输入模型和输出模型分开设计输入模型包含所有字段包括敏感信息输出模型只包含需要暴露的字段。4.3 Depends 的作用范围选择遵循最小范围原则能用参数级就不用全局。全局依赖会影响到所有路由包括健康检查接口、静态文件等本不需要鉴权的路径。# 比较好的分层做法# 1. 公开路由不需要依赖app.get(/health)asyncdefhealth_check():return{status:ok}# 2. 业务路由模块统一加模块级鉴权admin_routerAPIRouter(prefix/admin,dependencies[Depends(verify_admin_token)],)4.4 用 APIRouter 组织项目不要让main.py变成一个上千行的文件。按业务模块拆分app/ ├── main.py # 入口初始化 App挂载路由 ├── api/ │ └── v1/ │ ├── api.py # 汇总层 │ └── endpoints/ │ ├── books.py # 图书模块 │ └── users.py # 用户模块 ├── schemas/ # Pydantic 模型 ├── models/ # SQLAlchemy 模型后续数据库篇会讲 └── core/ # 配置、安全等4.5 SSE vs WebSocket 选型场景选哪个原因推送通知、实时数据SSE单向足够浏览器自动重连实现简单聊天、协作编辑WebSocket双向通信低延迟LLM 流式输出SSE服务端单向推送客户端用 EventSource 接收在线游戏WebSocket需要高频双向交互4.6 官方文档FastAPI 官方文档中文Pydantic 官方文档Starlette 官方文档5. 总结下表总结了 FastAPI 基础篇的核心概念和对应的 API你要做什么用 FastAPI 的什么一句话定义路由app.get()/app.post()装饰器绑定 URL HTTP 方法路径参数{id} 类型注解自动提取 URL 变量并校验类型查询参数Query()/ 默认值自动提取?keyvalue请求体验证PydanticBaseModel自动解析 JSON 校验 文档响应过滤response_model自动剔除敏感字段流式输出StreamingResponse逐块返回不占内存异常处理HTTPException即抛即停返回标准错误依赖管理Depends()一次定义随处注入模块化路由APIRouter按业务拆分include_router聚合跨域CORSMiddlewareadd_middleware一行配置FastAPI 的核心设计哲学可以用一句话概括让类型注解替你干活。当你习惯用Annotated、BaseModel、Depends来表达意图时你会发现写 API 不再是参数搬运工而是真正在写业务逻辑。下篇预告FastAPI 进阶篇——中间件机制、安全认证OAuth2 JWT、后台任务、WebSocket 实时通信与项目工程化。这些内容会让你的 FastAPI 项目从能用变成能上线。