深入掌握FastAPI与OpenAPI规范的高级适配技巧
title: 深入掌握FastAPI与OpenAPI规范的高级适配技巧


from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI(
title="电商平台API",
description="包含商品和订单管理的核心接口",
version="1.0.0",
openapi_tags=[{
"name": "商品",
"description": "商品信息管理相关接口"
}]
)
class Product(BaseModel):
id: int
name: str = Field(..., min_length=2, example="智能手机")
price: float = Field(gt=0, example=2999.99)
tags: list[str] = Field(default=[], example=["电子", "数码"])
@app.post("/products/", tags=["商品"])
async def create_product(product: Product):
return {"id": product.id}
@app.post(
"/products/",
tags=["商品"],
summary="创建新产品",
description="需要管理员权限,创建后自动生成库存记录",
response_description="返回创建成功的商品ID",
responses={
201: {
"description": "成功创建商品",
"content": {
"application/json": {
"example": {"id": 123}
}
}
},
403: {"description": "权限不足"}
},
openapi_extra={
"x-api-spec": {
"rateLimit": "1000/小时"
}
}
)
async def create_product(product: Product):
return {"id": product.id}
from fastapi.security import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(
tokenUrl="token",
scopes={
"products:write": "商品写入权限",
"products:read": "商品查询权限"
}
)
app = FastAPI(servers=[
{"url": "https://api.example.com", "description": "生产环境"},
{"url": "http://localhost:8000", "description": "开发环境"}
])
@app.get("/secure-data")
async def secure_data(
security_scopes: SecurityScopes = Depends(security)
):
return {"message": "安全数据"}
from typing import Any
from pydantic import BaseModel, create_model
def dynamic_model(fields: dict[str, Any]) -> type[BaseModel]:
return create_model(
'DynamicModel',
**{k: (v, Field(...)) for k, v in fields.items()}
)
@app.post("/dynamic-endpoint")
async def dynamic_endpoint(
data: dict[str, Any] = Body(...)
):
DynamicModel = dynamic_model(data["schema"])
# 使用动态模型进行校验
validated = DynamicModel(**data["payload"])
return validated.dict()
from pydantic import Field, validator
from datetime import datetime
class CustomDateTime(datetime):
@classmethod
def __get_validators__(cls):
yield cls.validate
@classmethod
def validate(cls, v):
if isinstance(v, str):
return datetime.fromisoformat(v)
return v
class Event(BaseModel):
timestamp: CustomDateTime = Field(
example="2023-07-20T14:30:00",
json_schema_extra={
"format": "iso8601"
}
)
@validator("timestamp")
def check_timezone(cls, v):
if v.tzinfo is None:
raise ValueError("必须包含时区信息")
return v
class Product(BaseModel):
price: float = Field(
...,
gt=0,
title="商品价格",
description="必须大于0的浮点数,单位:元",
example=99.9
)
# 手动生成最新文档
from fastapi.openapi.utils import get_openapi
def custom_openapi():
if app.openapi_schema:
return app.openapi_schema
openapi_schema = get_openapi(
title="Custom API",
version="1.0.0",
routes=app.routes,
)
app.openapi_schema = openapi_schema
return app.openapi_schema
app.openapi = custom_openapi
答案与解析 正确答案:B
app = FastAPI(default_response_headers={"X-API-Version": "1.0"})
答案与解析 正确答案:B
@app.get("/secret", include_in_schema=False)
async def secret_endpoint():
return {"message": "隐藏接口"}
评论
发表评论