数据锲约层:请求数据的解析/校验,响应数据的序列化/过滤,以及生成OpenAPI的文档,
一 序列化
序列化的本质解决三个问题:内存对象->可传输字节(网络)、可存储字节(磁盘/缓存)、跨语言交换(系统边界),不同场景对‘体积/速度/可读性/兼容性’的权重完全不同。
对比如浏览器/移动端/第三方的数据交互选择—json标准库,
API性能敏感:无损替换,FastAPI原生支持,
redis缓存值:msgpack(省内存),orjson(可调试),
微服务RPC:高性能—pickle(仅限可信数据),
大数据管道:avroparquet—(带schema演进),
二 FastAPI场景
官方推荐使用FastAPI+pydantic的场景,Pydantic会将数据序列直接转换成JSON,Pydantic底层是由Rust写的,因此很快。
FastAPI会自动读取请求体,校验字段类型,将字段转换为Item实例,在实际的开发中使用的场景主要有:
请求体解析与校验:这是最常用的场景,用Pydantic声明请求体结构,FastAPI自动完成JSON解析、类型转换和校验,
from fastapi import FastAPI from pydantic import BaseModel, Field app = FastAPI() class Item(BaseModel): name: str = Field(..., min_length=1, max_length=100) price: float = Field(..., gt=0, description="价格必须大于0") tax: float | None = None @app.post("/items/") async def create_item(item: Item): total = item.price + (item.tax or 0) return {"name": item.name, "total": total}响应模型与数据脱敏:用response_model声明输出结构,FastAPI会校验返回值并过滤未声明的字段
from fastapi import FastAPI from pydantic import BaseModel, EmailStr app = FastAPI() class UserBase(BaseModel): username: str email: EmailStr full_name: str | None = None class UserIn(UserBase): password: str class UserOut(UserBase): pass class UserInDB(UserBase): hashed_password: str def fake_password_hasher(raw: str) -> str: return "supersecret" + raw def fake_save_user(user_in: UserIn) -> UserInDB: hashed = fake_password_hasher(user_in.password) return UserInDB(**user_in.model_dump(), hashed_password=hashed) @app.post("/user/", response_model=UserOut) async def create_user(user_in: UserIn): return fake_save_user(user_in)嵌套模型与列表:Pydantic支持任意深度的模型,FastAPI会自动校验外层或内层所有字段,在实际的开发中,很多业务场景给前端返回的数据是聚合的,前端送上来的业务数据也是复合的,此时需要一个数据model嵌套另一个数据model,这个很常用!
from fastapi import FastAPI from pydantic import BaseModel, EmailStr app = FastAPI() class Address(BaseModel): street: str city: str zip_code: str class Tag(BaseModel): name: str value: str class User(BaseModel): username: str email: EmailStr address: Address tags: list[Tag] = [] @app.post("/users/") async def create_user(user: User): return user以及一些其它常用的场景:自定义验证器、ORM模式与数据库集成、依赖注入与Pydantic模型、配置管理—PydanticSettings
三 FastAPI序列化提速
FastAPI已经内置了对orjson的支持,可以通过两种方式启动:全局启动、指定路径启动
#全局启动 from fastapi import FastAPI from fastapi.responses import ORJSONResponse app = FastAPI(default_response_class=ORJSONResponse) @app.get("/items/") async def read_items(): return [{"item_id": "Foo", "value": 42}]#在装饰器中声明指定路径下,使用orjson from fastapi import FastAPI from fastapi.responses import ORJSONResponse app = FastAPI() @app.get("/items/", response_class=ORJSONResponse) async def read_items(): return [{"item_id": "Foo", "value": 42}]四 一些优化思路
小结—按照ROI从高到低顺序优化,依次为:
字段裁剪,
消除n+1,
不要阻塞loop(提高并发能力),
orjson序列化能提高很多(上面有说FastAPI的使用),
数据传输切分后使用stream流,
gzip带宽降下来(有一点点提升,减小cpu用量)