APIRouter 实现大型项目的模块化结构。1. 架构基石:ASGI 与 Uvicorn#
FastAPI 本身并不直接处理 HTTP 连接,它是建立在 Starlette(负责 Web 路由与 ASGI 协议)和 Pydantic(负责数据验证)之上的框架。理解 FastAPI 的第一步,是理解 ASGI (Asynchronous Server Gateway Interface)。
1.1 同步与异步的界限#
传统的 Python Web 框架(如 Flask, Django)基于 WSGI 标准,本质上是同步阻塞的。当一个请求处理数据库查询时,该线程被挂起,无法处理其他请求。
FastAPI 基于 ASGI 标准,支持异步 I/O。这要求我们需要一个 ASGI 服务器来运行应用,最常用的是 Uvicorn。
- Uvicorn:负责监听 Socket 端口,解析 HTTP 协议,将原始字节转换为 ASGI 消息字典。
- FastAPI:消费这些消息,执行业务逻辑,返回响应。
1.2 并发模型:async def vs def#
这是工程实践中极易混淆且影响性能的关键点。FastAPI 对路由处理函数(Path Operation Functions)提供了两种定义方式,其运行机制截然不同。
机制解析#
async def(异步函数):- 运行环境:直接在主事件循环(Event Loop)中运行。
- 适用场景:执行支持
await的非阻塞 I/O 操作(如使用httpx,motor,tortoise-orm等异步库)。 - 警告:严禁在此类函数中执行阻塞操作(如
time.sleep(), 大量 CPU 计算,或使用同步的requests/pymysql库)。一旦阻塞,整个服务器将无法响应任何新请求。
def(普通函数):- 运行环境:FastAPI 会将其放入底层的 线程池(ThreadPool) 中执行。
- 适用场景:必须使用同步库(如
requests),或执行某些无法异步化的计算。 - 代价:线程切换存在上下文开销,并发上限受限于线程池大小。
代码对比#
import time
import asyncio
from fastapi import FastAPI
app = FastAPI()
# 场景A:正确使用 async
# 在 Event Loop 中挂起,不阻塞其他请求
@app.get("/async-sleep")
async def async_sleep():
await asyncio.sleep(1)
return {"status": "async done"}
# 场景B:正确使用 def (处理同步阻塞)
# FastAPI 自动将其丢入线程池,主 Loop 不受影响
@app.get("/sync-sleep")
def sync_sleep():
time.sleep(1)
return {"status": "sync threadpool done"}
# 场景C:【严重错误】在 async 中执行同步阻塞
# 整个服务将被卡死 1 秒,无法处理任何并发请求
@app.get("/blocking-error")
async def blocking_error():
time.sleep(1) # 禁止的操作
return {"status": "server blocked"}
2. 路由机制与参数解析#
FastAPI 利用 Python 的类型提示(Type Hints)进行路由参数的解析与注入。
2.1 实例化与上下文#
app = FastAPI(
title="Core API",
version="1.0.0",
docs_url="/api/docs" # 自定义文档路径
)
app 对象维护了路由表、中间件栈和异常处理器。在生产环境中,通常不需要直接运行 uvicorn.run(app, ...),而是通过命令行启动 worker 进程管理:
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4
2.2 路径参数 (Path Parameters)#
路径参数是 URL 的一部分。FastAPI 通过函数参数的类型注解自动进行类型转换。
@app.get("/items/{item_id}")
async def read_item(item_id: int):
# 如果访问 /items/foo,FastAPI 会自动返回 422 错误
# 因为 "foo" 无法转换为 int
return {"item_id": item_id}
2.3 查询参数 (Query Parameters)#
声明在函数签名中、但未包含在路由路径中的参数,自动被识别为查询参数(?key=value)。
@app.get("/files/{file_path:path}")
async def read_file(
file_path: str, # 路径参数
skip: int = 0, # 查询参数,默认 0
limit: int = 10 # 查询参数,默认 10
):
return {"file": file_path, "skip": skip, "limit": limit}
- 注意:
file_path:path是 Starlette 的特殊语法,允许路径参数包含/,例如/files/home/user/data.txt。
3. 工程化结构:APIRouter 模块化#
在实际工程中,将所有路由堆积在 main.py 是不可维护的。FastAPI 提供了 APIRouter 用于路由分发,类似于 Flask 的 Blueprint 或 Django 的 app urls。
3.1 目录结构建议#
project/
├── app/
│ ├── __init__.py
│ ├── main.py # 入口文件
│ └── api/ # 接口目录
│ ├── __init__.py
│ ├── users.py # 用户模块
│ └── orders.py # 订单模块
3.2 定义子路由 (app/api/users.py)#
from fastapi import APIRouter
# 实例化路由器
router = APIRouter(
prefix="/users",
tags=["Users"],
responses={404: {"description": "Not found"}},
)
@router.get("/")
async def read_users():
return [{"username": "Admin"}, {"username": "User1"}]
@router.get("/me")
async def read_user_me():
return {"username": "current_user"}
3.3 注册路由 (app/main.py)#
from fastapi import FastAPI
from app.api import users, orders
app = FastAPI()
# 将子路由挂载到主应用
app.include_router(users.router)
app.include_router(orders.router) # 假设 orders 模块已定义
通过这种方式,我们可以分离关注点,不同团队成员可以维护不同的业务模块,互不干扰。
4. 底层机制:OpenAPI 反射生成#
FastAPI 的自动文档(Swagger UI / ReDoc)并非魔术,而是基于 反射(Reflection) 机制。
- 解析签名:应用启动时,FastAPI 遍历所有注册的路由函数。
- 提取类型:分析函数的参数名、类型注解(
int,str,Pydantic Model)以及默认值。 - 生成 Schema:根据 OpenAPI 3.1.0 标准,生成对应的 JSON Schema 描述(包含输入约束、输出格式、HTTP 方法等)。
- 渲染 UI:
/docs端点读取上述生成的 JSON,渲染为交互式网页。
这也是为什么在 FastAPI 中必须严格编写类型提示的原因——它不仅仅是为了代码补全,更是 API 契约定义的核心来源。
总结#
- 并发选择:I/O 密集型任务且库支持异步时使用
async def;涉及同步阻塞调用时务必使用def。 - 类型驱动:FastAPI 严重依赖 Python Type Hints 进行数据解析和文档生成,需养成良好的类型标注习惯。
- 结构治理:项目起步阶段即应使用
APIRouter规划目录结构,避免后期重构成本。