FastAPI-02.路由与数据模型

本系列:00 导读 · 01补 网络基础 · 01 心智模型 · 02 路由与数据模型(本文) · 02补 HTTP 方法对比 · 02补2 状态码共识 · 03 依赖注入与分层 · 04 中间件异常日志 · 05 异步后台与流式 · 06 鉴权与安全 · 07 测试与项目骨架 · 08 实战 HTTP↔MCP

行文:T2 模式篇 | 本篇方法:组块化 + 刻意练习 | 辅助:主动回忆、精细加工(挂 Pydantic

方法语义:见 02补状态码共识:见 02补2


组块清单

组块 触发条件 核心写法
G1 Path 资源标识在 URL 路径里 {item_id} + 类型注解
G2 Query 过滤、分页、可选参数 函数参数默认值 + 注解
G3 Body JSON 请求体 BaseModel 单对象或列表
G4 响应模型 约束输出形状、隐藏字段 response_model=
G5 状态码 创建/删除等语义 status_code= + Response
G6 APIRouter 模块拆分、统一前缀 APIRouter(prefix=...)

段末注释Pydantic 是 Python 数据验证库;FastAPI 用它解析请求并生成 OpenAPI 文档。字段校验细节见 Pydantic 笔记,本篇只讲「在路由里怎么用」。


路由Path

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 根据路径匹配对应的函数调用
# localhost:8000/ 调用根目录对应函数
@app.get("/")
async def read_item():
return {"message": "Hello World"}

# 通过路由进行调用的同时进行参数传递
# localhost:8000/items/99 其中的 99 会被解析为 item_id
@app.get("/items/{item_id}")
async def read_item(item_id: int")):
return {"item_id": item_id}

# 也可以支持通过路由的多参数传递,fastapi会自动匹配最适合的路由,并填充参数
@app.get("/users/{user_id}/items/{item_id}")
async def read_user_item(user_id: int, item_id: str):
item = {"item_id": item_id, "owner_id": user_id}
return item

可以看到通过路由地址,fastapi可以实现不同函数的调用选择,同时也可以简单的实现参数的传递。

  • 误用:路径写成 /items/{item_id}/{item_id} 却期望两个不同变量——同名占位符会冲突;用不同名字。

参数传递

我们可以直接通过路由路径进行传参,但是在参数多、结构复杂的情况下,使用体验会不友好,fastapi本身也提供多重参数传递方式:

通过查询参数

http网页如果我们关注过请求地址,我们一定看到过类似这样的地址,例如:https://cn.bing.com/search?pc=MOZI&form=MOZLBR&q=antibody(这是一个搜索浏览的搜索页面,可以看到后面 search确定搜索函数,然后通过pc、form、q传递参数),而fastapi也支持这样的查询模式。

1
2
3
4
5
# 多参数传递
@app.get("/users/{user_id}")
async def read_user_item(user_id: int, item_id: str,):
item = {"item_id": item_id, "owner_id": user_id}
return item

这时候,我们访问 http://127.0.0.1:8000/users/1/items/tem_id?item_id=13 item_id=13就会被解析并赋值给item_id传递到调用的函数中。

  • 注意: 如果参数 async def read_user_item(item_id: Annotated[list[str], Query()]): 则可以在查询中出现多次item_id 将参数构建出一个列表。

通过请求体

请求体是客户端发送给 API 的数据(也可能没有,直接通过)。响应体是 API 发送给客户端的数据。相当于使用自定义的数据结构,使用请求体和响应体都需要进行声明。
如果要使用请求体,需要先进行请求体数据类型的声明(继承 BaseModel )。

1
2
3
4
5
6
7
8
9
10
11
12
13
from fastapi import FastAPI
from pydantic import BaseModel
# 声明请求体
class Body(BaseModel):
name: str
description: str | None = None
price: float
tax: float | None = None

app = FastAPI()
@app.post("/items/")
async def create_item(item: Body): #定义输入参数是自己定义的Item
return item
  • 误用:Body 模型里塞 20 个可选字段当「万能 DTO」——应拆 Create / Update / Read 模型。

多请求体

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()

class Item(BaseModel):
name: str
description: str | None = None
price: float
tax: float | None = None

class User(BaseModel):
username: str
full_name: str | None = None

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, user: User):
results = {"item_id": item_id, "item": item, "user": user}
return results

在这种情况下,FastAPI 会注意到函数中有不止一个请求体参数(有两个参数是 Pydantic 模型)。
因此,它会将参数名作为请求体中的键(字段名),并期望请求体格式如下:

1
2
3
4
5
6
7
8
9
10
11
12
{
"item": {
"name": "Foo",
"description": "The pretender",
"price": 42.0,
"tax": 3.2
},
"user": {
"username": "dave",
"full_name": "Dave Grohl"
}
}

混合传参

之前我们已经看到可以自由地混合使用 Path、Query 和请求体参数声明,FastAPI 知道该如何处理。
但是我们之前可以看到 PATH、Query都是通过访问地址传输的,请求体是通过请求体进行传输的,有时候我们需要一起用,也是可以的

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from typing import Annotated
from fastapi import Body, FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
name: str
description: str | None = None
price: float
tax: float | None = None


class User(BaseModel):
username: str
full_name: str | None = None


@app.put("/items/{item_id}")
async def update_item(
item_id: int, item: Item, user: User, importance: Annotated[int, Body()]
):
results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
return results

通过importance注释为Body(),虽然没有声明请求体,但是他也会去请求体中获取参数,预测的请求体如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
{
"item": {
"name": "Foo",
"description": "The pretender",
"price": 42.0,
"tax": 3.2
},
"user": {
"username": "dave",
"full_name": "Dave Grohl"
},
"importance": 5
}

参数校验

在我们进行函数开发阶段,会初步定义每个参数多数据类型(Str、Int、Float、Enum,List、Dict、Tuple、Union、Optional、Any等),在请求时,会自动进行类型的校验。但是实际开发中,我们可能会有更详细的校验需求,比如国家、电话号码,邮箱地址,手机号等等。

有限枚举值限制

1
2
3
4
5
6
7
8
9
10
# 定义一个枚举类,
class ModelName(str, Enum):
alexnet = "alexnet"
resnet = "resnet"
lenet = "lenet"

# 访问/models/xxx 时,会将 xxx 赋值给变量 {model_name} 实现基于路径路由的传参。
@app.get("/models/{model_name}")
def get_model(model_name: ModelName): # 接受参数定义为一个枚举类时,传参会进行校验,如果不在枚举类中,会返回错误。
return {"model_name": model_name}

Annotated详细校验

Annotated 可以用于为参数添加元数据(Annotated[type, metadata]), 在Fastapi中,我们可以通过 Annotated 进行更精细的数据校验包括通过 Query提供的功能进行一些标准化的校验,也可以通过自定义函数进行自定义校验。

Query进行查询校验

Query除了进行参数的校验,还支持更多类型的元数据的补充(title、description、alias、deprecated、include_in_schema等等)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
async def read_items(
q: Annotated[
str | None,
Query(
title="Query string", # 标题
alias="item-query", # 别名
description="Query string for the items to search in the database that have a good match", # 描述
min_length=3, # 最小长度
max_length=50, # 最大长度
pattern="^fixedquery$", # 正则表达式
deprecated=True, # 是否弃用
include_in_schema=False, # 是否在schema中显示
gt=0, # 大于
),
] = None,
):
results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
if q:
results.update({"q": q})
return results

这样数据会校验长度,正则。

PATH进行路由参数校验

起始Query也就基本够用了,PATH和Query基本一样,只是PATH会校验路由参数。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from typing import Annotated
from fastapi import FastAPI, Path, Query
app = FastAPI()

@app.get("/items/{item_id}")
async def read_items(
*,
item_id: Annotated[int, Path(title="The ID of the item to get", ge=0, le=1000)],
size: Annotated[float, Query(gt=0, lt=10.5)],
):
results = {"item_id": item_id}
if size:
results.update({"size": size})
return results

AfterValidator 自定义校验

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
#自定义验证
from pydantic import AfterValidator

def check_valid_id(id: str):
if not id.startswith(("isbn-", "imdb-")):
raise ValueError('Invalid ID format, it must start with "isbn-" or "imdb-"')
return id

@app.get("/items/")
async def read_items(
q: Annotated[
str | None, Query(
AfterValidator(check_valid_id), # 对变量运行函数进行校验。
alias="item-query",

)
] = None,
):
results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}

请求体创建阶段,制定校验

刚才的校验,更多是在调用阶段进行的,但是针对一个请求体,校验规则一般一样,而且不回随着调用环境产生差别,所以在请求体构造阶段进行校验规则的确定会更有更好的可迁移性。而pydantic也支持这样的方案。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from typing import Annotated, Literal

from fastapi import FastAPI, Query
from pydantic import BaseModel, Field

app = FastAPI()


class FilterParams(BaseModel):
model_config = {"extra": "forbid"} # 限制为制定的关键字传参。
limit: int = Field(100, gt=0, le=100)
offset: int = Field(0, ge=0)
order_by: Literal["created_at", "updated_at"] = "created_at"
tags: list[str] = []


@app.get("/items/")
async def read_items(filter_query: Annotated[FilterParams, Query()]):
return filter_query

在请求体构建阶段,就限制了每个变量的校验规则。从而避免每次调用都要重新写校验规则。

优雅递归处理-DEPEND复用参数处理

有时候有些不同的操作,需要相同的参数处理,这时DEPEND就派上了用场。不同的参数请求可以服用一个相同的参数处理逻辑。重点在于解决相同的参数处理实现可服用。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from typing import Annotated
from fastapi import Depends, FastAPI
app = FastAPI()
#一个可以服用的参数处理逻辑,接受参数,返回参数处理结果。
async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}

CommonsDep = Annotated[dict, Depends(common_parameters)] # 提高可读性的指向封装

@app.get("/items/")
async def read_items(commons: CommonsDep): # read_items接受参数后,先传递给CommonsDep中执行的依赖common_parameters,然后经过common_parameters处理后返回的参数作为 read_items的实际输入。
return commons

@app.get("/users/")
async def read_users(commons: CommonsDep):
return commons

结果返回

返回响应模型块

和请求体一样,最好也进行返回体的定义,进行返回数据的筛选(防止内部参数暴露给外部),FastAPI 会看到返回类型,并确保你返回的内容 仅 包含类型中声明的字段。

1
2
3
4
5
6
7
8
9
class ItemRead(BaseModel):
id: int
name: str
price: float
# 不含内部字段 cost、deleted_at

@app.get("/items/{item_id}", response_model=ItemRead) # 制定返回结果是自定义类型。
async def get_item(item_id: int)->ItemRead : # 像普通函数一样定义输出对象
return {"id": item_id, "name": "tea", "price": 9.9, "cost": 3.0} #会忽略掉自定义类型中没定义的键值。
  • 效果cost 不会出现在 JSON;OpenAPI 也只展示 ItemRead 字段。
  • 误用response_modelreturn 类型不一致却不测——用 TestClient 断言响应体字段。

状态码块

作为结果返回的一部分。

1
2
3
4
5
6
7
8
9
10

from fastapi import Response, status
@app.post("/items", status_code=status.HTTP_201_CREATED)
async def create(item: ItemCreate):
return {"id": 42, **item.model_dump()}


@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int):
return Response(status_code=status.HTTP_204_NO_CONTENT)
  • 误用:204 仍 return {"ok": true}——客户端可能收到非空 body,违背语义。

APIRouter 块

1
2
3
4
5
6
7
8
from fastapi import APIRouter
router = APIRouter(prefix="/items", tags=["items"])

@router.get("/")
async def list_items():
return []

app.include_router(router)
  • 误用prefix 与路由内路径都带 /items,拼成 /items/items——约定 router 管前缀,子路由写 //{id}

刻意练习:只改一处

在空项目 main.py 上改,用 uvicorn main:app --reload + /docs 验证。

# 题面 只改什么 验收
D1 GET /users/{user_id}user_id 必须 ≥ 1 Path(ge=1) 传 0 得 422
D2 GET /items?q=...&limit=...limit 默认 10、最大 50 Query 注解 超 50 得 422
D3 POST /items 接收 name+price 新建 ItemCreate body 缺字段得 422
D4 响应里隐藏 password_hash response_model=UserPublic 响应无 hash 字段
D5 创建资源返回 201 status_code=201 状态码为 201
D6 把 items 路由拆到 routers/items.py APIRouter + include_router /items 仍可访问
D7 同一函数既要 Path item_id 又要 Query detail 两种参数同函数 OpenAPI 显示两类参数
D8 PUT /items/{id} body 字段均可选 ItemUpdateOptional 只传 price 也能过

-------------本文结束感谢您的阅读-------------