FastAPI 是当前 Python 生态中增长最快的 Web 框架，以高性能、自动文档和类型安全著称。本文从环境搭建开始，带你掌握路径参数、查询参数、请求体、响应模型等核心特性，最终完成一个完整的 CRUD API。
2.1 为什么选择 FastAPI
2.1.1 FastAPI 的诞生背景
在 FastAPI 出现之前，Python Web 开发主要依赖两个阵营：
- Flask：轻量灵活，但缺乏类型安全、异步支持弱、API 文档需手写
- Django / DRF：功能齐全，但笨重、学习曲线陡、异步支持姗姗来迟
FastAPI 的作者 Sebastian Ramirez 看到了一个机会：利用 Python 的类型提示（Type Hints）和现代异步特性，打造一个既快又好用的 API 框架。
2.1.2 核心优势一览
| 特性 | FastAPI | Flask | Django |
|---|---|---|---|
| 性能 | 接近 Node.js / Go | 中等 | 中等偏下 |
| 自动 API 文档 | 内置 Swagger + ReDoc | 需插件 | 需额外配置 |
| 类型安全 | 原生 Pydantic | 需插件 | 部分支持 |
| 异步支持 | 原生 async/await | 需扩展 | Django 4.1+ |
| 依赖注入 | 内置 | 无 | 无 |
| 学习曲线 | 中等 | 低 | 高 |
2.1.3 技术栈
FastAPI 不是从零造轮子，而是站在巨人肩膀上：
FastAPI
├── Starlette -- ASGI Web 框架（处理路由、请求、中间件）
├── Pydantic -- 数据验证库（处理请求体、响应模型）
├── Uvicorn -- ASGI 服务器（运行 FastAPI 应用）
└── Python 类型提示 -- 驱动自动文档和 IDE 支持
2.2 环境搭建
2.2.1 使用 uv（推荐）
uv 是 Astral 团队打造的 Python 包管理器，速度极快。
# 安装 uv（macOS / Linux）
curl -LsSf https://astral.sh/uv/install.sh | sh
# 创建项目
uv init my-api
cd my-api
# 添加依赖
uv add fastapi uvicorn[standard] pydantic[email]
# 验证安装
uv run python -c "import fastapi; print(fastapi.__version__)"
2.2.2 使用 pip + venv
# 创建虚拟环境
python -m venv .venv
source .venv/bin/activate # macOS / Linux
# .venv\Scripts\activate # Windows
# 安装依赖
pip install fastapi uvicorn[standard] pydantic[email]
2.2.3 项目初始结构
my-api/
├── main.py # 应用入口
├── pyproject.toml # 项目配置
└── .venv/ # 虚拟环境
2.3 第一个 FastAPI 应用
2.3.1 Hello World
创建 main.py：
from fastapi import FastAPI
app = FastAPI(
title="My API",
description="FastAPI 学习示例",
version="0.1.0",
)
@app.get("/")
async def root():
return {"message": "Hello, FastAPI!"}
启动服务：
uvicorn main:app --reload --host 0.0.0.0 --port 8000
| 参数 | 说明 |
|---|---|
main:app | main.py 文件中的 app 对象 |
--reload | 代码变更自动重启（开发用） |
--host 0.0.0.0 | 监听所有网络接口 |
--port 8000 | 端口号 |
访问：
- API：
http://localhost:8000/ - Swagger 文档：
http://localhost:8000/docs - ReDoc 文档：
http://localhost:8000/redoc
2.3.2 自动文档的魅力
不需要写任何额外代码，FastAPI 自动生成了：
- Swagger UI（
/docs）：交互式文档，可直接在浏览器中测试 API - ReDoc（
/redoc）：更美观的文档，适合对外展示
这是 FastAPI 相比 Flask 最大的优势之一。后续添加的每个路由、参数、模型都会自动反映在文档中。
2.4 路径参数（Path Parameters）
路径参数是 URL 中的变量部分，用于标识具体资源。
2.4.1 基础用法
@app.get("/users/{user_id}")
async def get_user(user_id: int):
return {"user_id": user_id, "type": type(user_id).__name__}
访问 GET /users/123，返回 {"user_id": 123, "type": "int"}。
FastAPI 自动完成：
- 从 URL 中提取
user_id - 根据类型提示转换为
int - 如果传入非数字（如
/users/abc），自动返回 422 错误
2.4.2 使用 Path 添加验证
from fastapi import Path
@app.get("/users/{user_id}")
async def get_user(
user_id: int = Path(
..., # ... 表示必填
title="用户ID",
description="正整数",
gt=0, # 大于 0
examples=[1, 42, 100],
),
):
return {"user_id": user_id}
Path 支持的验证参数：
| 参数 | 含义 | 示例 |
|---|---|---|
gt | 大于（greater than） | gt=0 |
ge | 大于等于（greater or equal） | ge=1 |
lt | 小于（less than） | lt=1000 |
le | 小于等于（less or equal） | le=999 |
min_length | 最小长度 | min_length=2 |
max_length | 最大长度 | max_length=50 |
regex / pattern | 正则匹配 | pattern="^[a-z]+$" |
examples | 示例值（文档用） | examples=["alice"] |
2.4.3 多个路径参数
@app.get("/users/{user_id}/posts/{post_id}")
async def get_user_post(user_id: int, post_id: int):
return {"user_id": user_id, "post_id": post_id}
2.4.4 枚举类型约束
from enum import Enum
class ModelName(str, Enum):
ALEXNET = "alexnet"
RESNET = "resnet"
LENET = "lenet"
@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
if model_name is ModelName.ALEXNET:
return {"model": model_name, "message": "Deep Learning FTW!"}
return {"model": model_name, "message": "Have some residuals"}
访问 /models/invalid 会返回 422 错误，并在文档中显示可选值。
2.5 查询参数（Query Parameters）
查询参数是 URL 中 ? 后面的键值对，用于过滤、排序、分页等。
2.5.1 基础用法
fake_items_db = [
{"item_name": "Foo"},
{"item_name": "Bar"},
{"item_name": "Baz"},
]
@app.get("/items/")
async def list_items(skip: int = 0, limit: int = 10):
return fake_items_db[skip : skip + limit]
访问 GET /items/?skip=2&limit=5。
FastAPI 根据以下规则区分路径参数和查询参数：
- 在路径中声明的 -> 路径参数
- 不在路径中声明的 -> 查询参数
- 有默认值 -> 可选查询参数
- 无默认值 -> 必填查询参数
2.5.2 必填查询参数
@app.get("/items/")
async def list_items(
keyword: str, # 没有默认值，必填
page: int = 1,
):
return {"keyword": keyword, "page": page}
访问 GET /items/ 不带 keyword 会返回 422 错误。
2.5.3 使用 Query 添加验证
from fastapi import Query
@app.get("/items/")
async def list_items(
keyword: str | None = Query(
None, # 默认值 None（可选）
title="搜索关键词",
min_length=1,
max_length=50,
),
page: int = Query(1, ge=1, description="页码"),
page_size: int = Query(20, ge=1, le=100, description="每页数量"),
):
return {
"keyword": keyword,
"page": page,
"page_size": page_size,
}
2.5.4 多值查询参数
@app.get("/items/")
async def list_items(
tags: list[str] = Query([], description="标签列表"),
):
return {"tags": tags}
访问 GET /items/?tags=python&tags=fastapi&tags=rest，返回 {"tags": ["python", "fastapi", "rest"]}。
2.6 请求体（Request Body）
请求体用于向服务器发送结构化数据，通常配合 POST、PUT、PATCH 使用。
2.6.1 使用 Pydantic 模型
from pydantic import BaseModel, Field, EmailStr
from datetime import datetime
class UserCreate(BaseModel):
"""创建用户的请求体"""
name: str = Field(
...,
min_length=2,
max_length=50,
description="用户名",
examples=["Alice"],
)
email: EmailStr = Field(
...,
description="邮箱地址",
examples=["alice@example.com"],
)
age: int | None = Field(
None,
ge=0,
le=150,
description="年龄",
examples=[25],
)
bio: str = Field(
"",
max_length=500,
description="个人简介",
)
@app.post("/users/", status_code=201)
async def create_user(user: UserCreate):
# user 对象已经通过 Pydantic 自动验证
return {
"id": 1,
**user.model_dump(),
"created_at": datetime.now().isoformat(),
}
FastAPI 自动完成：
- 读取请求体 JSON
- 验证所有字段（类型、范围、格式）
- 转换为 Python 对象
- 生成 OpenAPI schema（文档中显示字段说明）
- 验证失败时返回 422 错误和详细的错误信息
2.6.2 请求体 + 路径参数 + 查询参数
FastAPI 可以同时使用三种参数来源：
@app.put("/users/{user_id}")
async def update_user(
user_id: int, # 路径参数
notify: bool = Query(False), # 查询参数
user: UserCreate = Body(...), # 请求体
):
return {
"user_id": user_id,
"notify": notify,
"user_data": user.model_dump(),
}
FastAPI 的参数识别规则：
- 声明了
Path/ 在路径中 -> 路径参数 - 声明了
Query/ 有默认值的简单类型 -> 查询参数 - 声明了
Body/ Pydantic 模型 -> 请求体
2.6.3 嵌套模型
from pydantic import BaseModel, HttpUrl
class Address(BaseModel):
street: str
city: str
country: str = "CN"
zip_code: str | None = None
class UserCreate(BaseModel):
name: str
email: EmailStr
address: Address # 嵌套模型
tags: list[str] = [] # 列表
metadata: dict[str, str] = {} # 字典
# JSON 示例
{
"name": "Alice",
"email": "alice@example.com",
"address": {
"street": "123 Main St",
"city": "Shanghai",
"country": "CN",
"zip_code": "200000"
},
"tags": ["admin", "user"],
"metadata": {"department": "engineering"}
}
2.6.4 Field 验证速查
from pydantic import BaseModel, Field
class Example(BaseModel):
# 字符串
name: str = Field(..., min_length=1, max_length=100)
email: str = Field(..., pattern=r"^[\w.-]+@[\w.-]+\.\w+$")
# 数字
age: int = Field(..., gt=0, le=150)
score: float = Field(0.0, ge=0.0, le=100.0)
# 列表
tags: list[str] = Field([], min_length=0, max_length=10)
# 可选
bio: str | None = Field(None, max_length=500)
# 默认值
role: str = Field("user", pattern="^(user|admin|moderator)$")
2.7 响应模型（Response Model）
响应模型控制 API 返回的数据结构，可以隐藏敏感字段、过滤输出。
2.7.1 定义响应模型
from datetime import datetime
class UserResponse(BaseModel):
"""用户响应模型"""
id: int
name: str
email: str
age: int | None
bio: str
is_active: bool = True
created_at: datetime
model_config = {"from_attributes": True} # 支持 ORM 对象转换
2.7.2 使用 response_model
@app.post("/users/", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
# 返回值会被 response_model 过滤
# password 等敏感字段不会出现在响应中
return {
"id": 1,
"name": user.name,
"email": user.email,
"age": user.age,
"bio": user.bio,
"is_active": True,
"created_at": datetime.now(),
"password_hash": "xxx", # 这个字段会被 response_model 过滤掉
}
response_model 的作用：
- 过滤输出：只返回模型中定义的字段
- 序列化：自动将 Python 对象转为 JSON（包括 datetime、UUID 等）
- 文档生成：Swagger 文档中显示正确的响应结构
- 数据验证：确保返回数据符合模型定义
2.7.3 响应模型与请求模型分离
# 请求模型：包含客户端提交的字段
class UserCreate(BaseModel):
name: str
email: EmailStr
password: str = Field(..., min_length=8)
# 响应模型：包含返回给客户端的字段（无密码）
class UserResponse(BaseModel):
id: int
name: str
email: str
created_at: datetime
# 更新模型：所有字段可选
class UserUpdate(BaseModel):
name: str | None = None
email: EmailStr | None = None
@app.post("/users/", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
... # password 不会出现在响应中
@app.patch("/users/{user_id}", response_model=UserResponse)
async def update_user(user_id: int, user: UserUpdate):
... # 只更新传入的字段
2.7.4 列表响应
@app.get("/users/", response_model=list[UserResponse])
async def list_users():
return [...] # 返回用户列表
配合统一分页响应：
class PaginatedResponse(BaseModel):
items: list[UserResponse]
total: int
page: int
page_size: int
total_pages: int
@app.get("/users/", response_model=PaginatedResponse)
async def list_users(page: int = 1, page_size: int = 20):
...
2.8 状态码与错误响应
2.8.1 设置状态码
@app.post("/users/", status_code=201)
async def create_user(user: UserCreate):
...
@app.delete("/users/{user_id}", status_code=204)
async def delete_user(user_id: int):
return None # 204 No Content
常用状态码常量：
from fastapi import status
@app.post("/users/", status_code=status.HTTP_201_CREATED)
async def create_user(user: UserCreate):
...
2.8.2 HTTPException
from fastapi import HTTPException
@app.get("/users/{user_id}")
async def get_user(user_id: int):
if user_id not in fake_db:
raise HTTPException(
status_code=404,
detail="用户不存在",
headers={"X-Error-Code": "USER_NOT_FOUND"},
)
return fake_db[user_id]
2.8.3 自定义错误响应
from fastapi.responses import JSONResponse
class AppError(Exception):
def __init__(self, code: int, message: str, errors: list | None = None):
self.code = code
self.message = message
self.errors = errors or []
@app.exception_handler(AppError)
async def app_error_handler(request, exc: AppError):
return JSONResponse(
status_code=exc.code,
content={
"code": exc.code,
"message": exc.message,
"errors": exc.errors,
"data": None,
},
)
2.9 完整 CRUD 示例
下面整合以上所有知识点，构建一个完整的用户管理 API。
from fastapi import FastAPI, HTTPException, Query, Path, status
from pydantic import BaseModel, Field, EmailStr
from datetime import datetime
app = FastAPI(title="用户管理 API", version="1.0.0")
# ========== 模型定义 ==========
class UserCreate(BaseModel):
name: str = Field(..., min_length=2, max_length=50, examples=["Alice"])
email: EmailStr = Field(..., examples=["alice@example.com"])
age: int | None = Field(None, ge=0, le=150)
bio: str = Field("", max_length=500)
class UserUpdate(BaseModel):
name: str | None = Field(None, min_length=2, max_length=50)
email: EmailStr | None = None
age: int | None = Field(None, ge=0, le=150)
bio: str | None = Field(None, max_length=500)
class UserResponse(BaseModel):
id: int
name: str
email: str
age: int | None
bio: str
created_at: datetime
updated_at: datetime | None = None
# ========== 内存存储 ==========
fake_db: dict[int, dict] = {}
next_id = 1
# ========== 路由 ==========
@app.get("/api/v1/users", response_model=list[UserResponse], tags=["用户"])
async def list_users(
page: int = Query(1, ge=1),
page_size: int = Query(20, ge=1, le=100),
keyword: str | None = Query(None, max_length=50),
):
"""获取用户列表"""
users = list(fake_db.values())
if keyword:
users = [u for u in users if keyword.lower() in u["name"].lower()]
start = (page - 1) * page_size
return users[start : start + page_size]
@app.post(
"/api/v1/users",
response_model=UserResponse,
status_code=status.HTTP_201_CREATED,
tags=["用户"],
)
async def create_user(body: UserCreate):
"""创建新用户"""
global next_id
if any(u["email"] == body.email for u in fake_db.values()):
raise HTTPException(status_code=409, detail="邮箱已存在")
now = datetime.now()
user = {
"id": next_id,
**body.model_dump(),
"created_at": now,
"updated_at": None,
}
fake_db[next_id] = user
next_id += 1
return user
@app.get("/api/v1/users/{user_id}", response_model=UserResponse, tags=["用户"])
async def get_user(user_id: int = Path(..., gt=0)):
"""获取用户详情"""
if user_id not in fake_db:
raise HTTPException(status_code=404, detail="用户不存在")
return fake_db[user_id]
@app.patch("/api/v1/users/{user_id}", response_model=UserResponse, tags=["用户"])
async def update_user(user_id: int, body: UserUpdate):
"""更新用户信息"""
if user_id not in fake_db:
raise HTTPException(status_code=404, detail="用户不存在")
user = fake_db[user_id]
update_data = body.model_dump(exclude_unset=True)
if "email" in update_data:
if any(
u["email"] == update_data["email"]
for uid, u in fake_db.items()
if uid != user_id
):
raise HTTPException(status_code=409, detail="邮箱已存在")
user.update(update_data)
user["updated_at"] = datetime.now()
return user
@app.delete("/api/v1/users/{user_id}", status_code=204, tags=["用户"])
async def delete_user(user_id: int):
"""删除用户"""
if user_id not in fake_db:
raise HTTPException(status_code=404, detail="用户不存在")
del fake_db[user_id]
启动后访问 http://localhost:8000/docs，即可看到完整的交互式 API 文档，直接在浏览器中测试所有接口。
2.10 小结
| 知识点 | 关键 API |
|---|---|
| 路径参数 | @app.get("/users/{user_id}") + Path() |
| 查询参数 | @app.get("/users/?page=1") + Query() |
| 请求体 | Pydantic BaseModel |
| 响应模型 | response_model=UserResponse |
| 状态码 | status_code=201 / status.HTTP_201_CREATED |
| 错误处理 | HTTPException(status_code=404, detail="...") |
| 自动文档 | /docs（Swagger）、/redoc（ReDoc） |
