name: api-route-design
description: |
用于在FastAPI或Python项目中设计RESTful API端点时使用。
触发场景:创建GET/POST/PUT/DELETE端点、使用Pydantic进行请求验证、
使用JSON模式格式化响应、选择状态码、分页、筛选或排序参数。
不适用于:GraphQL API、WebSocket处理程序或非RESTful端点。
API路由设计技能
具备专业设计和实现RESTful API的能力,包括适当的验证、响应格式化和HTTP语义。
快速参考
| 模式 |
示例 |
用途 |
| 列表资源 |
@router.get("/fees/", response_model=List[FeeOut]) |
检索集合 |
| 按ID获取 |
@router.get("/fees/{fee_id}") |
检索单个资源 |
| 创建 |
@router.post("/fees/", response_model=FeeOut, status_code=201) |
创建新资源 |
| 更新 |
@router.put("/fees/{fee_id}") |
完整资源更新 |
| 部分更新 |
@router.patch("/fees/{fee_id}") |
部分资源更新 |
| 删除 |
@router.delete("/fees/{fee_id}", status_code=204) |
移除资源 |
URL命名规范
/v1/{resource} # 集合端点
/v1/{resource}/{id} # 单个资源端点
/v1/{resource}/{id}/sub # 嵌套资源端点
规则:
- 使用小写,多单词用连字符:
/student-fees 而非 /studentFees
- 集合使用复数名词:
/users 而非 /user
- 语义化使用HTTP方法:GET(读取)、POST(创建)、PUT/PATCH(更新)、DELETE(删除)
HTTP状态码
| 代码 |
用途 |
示例 |
| 200 |
成功 |
成功的GET、PUT、PATCH |
| 201 |
已创建 |
成功的POST(资源已创建) |
| 202 |
已接受 |
异步操作已启动 |
| 204 |
无内容 |
成功的DELETE |
| 400 |
错误请求 |
无效输入,验证失败 |
| 401 |
未授权 |
缺少或无效的身份验证 |
| 403 |
禁止访问 |
已认证但未授权 |
| 404 |
未找到 |
资源不存在 |
| 422 |
无法处理的实体 |
验证错误(Pydantic) |
| 500 |
内部服务器错误 |
意外的服务器错误 |
请求验证模式
路径参数
from fastapi import APIRouter, HTTPException
from typing import Annotated
router = APIRouter()
@router.get("/fees/{fee_id}")
async def get_fee(fee_id: int):
fee = await get_fee_by_id(fee_id)
if not fee:
raise HTTPException(status_code=404, detail="Fee not found")
return fee
查询参数(分页、筛选、排序)
@router.get("/fees/", response_model=List[FeeOut])
async def list_fees(
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=1000),
status: str | None = Query(None, pattern="^(pending|paid|overdue)$"),
sort_by: str = Query("created_at", enum=["created_at", "amount", "due_date"]),
sort_order: str = Query("desc", enum=["asc", "desc"]),
):
return await paginate_fees(
skip=skip,
limit=limit,
status=status,
sort_by=sort_by,
sort_order=sort_order,
)
请求体(Pydantic模型)
from pydantic import BaseModel
from datetime import datetime
class FeeCreate(BaseModel):
student_id: int
amount: float = Field(..., gt=0)
due_date: datetime
description: str | None = None
class FeeUpdate(BaseModel):
amount: float | None = Field(None, gt=0)
status: str | None = Field(None, pattern="^(pending|paid|overdue)$")
due_date: datetime | None = None
@router.post("/fees/", response_model=FeeOut, status_code=201)
async def create_fee(fee_in: FeeCreate):
return await create_fee_db(fee_in)
@router.patch("/fees/{fee_id}", response_model=FeeOut)
async def update_fee(fee_id: int, fee_in: FeeUpdate):
return await update_fee_db(fee_id, fee_in)
响应模型
标准响应封装
class FeeOut(BaseModel):
id: int
student_id: int
amount: float
status: str
created_at: datetime
due_date: datetime
class PaginatedResponse(BaseModel):
data: List[FeeOut]
total: int
skip: int
limit: int
has_more: bool
错误响应
class ErrorResponse(BaseModel):
error: str
detail: str | None = None
code: str | None = None
完整端点示例
from fastapi import APIRouter, Depends, HTTPException, Query, status
from typing import List, Annotated
router = APIRouter(prefix="/v1/fees", tags=["fees"])
@router.get(
"/",
response_model=PaginatedResponse[FeeOut],
summary="列出费用",
description="检索带有可选筛选的分页费用列表。",
)
async def list_fees(
skip: Annotated[int, Query(0, ge=0)] = 0,
limit: Annotated[int, Query(100, ge=1, le=1000)] = 100,
status: Annotated[str | None, Query(pattern="^(pending|paid|overdue)$")] = None,
_current_user: User = Depends(get_current_user),
) -> PaginatedResponse[FeeOut]:
fees, total = await get_fees(
skip=skip, limit=limit, status=status, user=_current_user
)
return PaginatedResponse(
data=fees,
total=total,
skip=skip,
limit=limit,
has_more=(skip + limit) < total,
)
@router.get(
"/{fee_id}",
response_model=FeeOut,
responses={404: {"model": ErrorResponse}},
)
async def get_fee(
fee_id: int,
_current_user: User = Depends(get_current_user),
) -> FeeOut:
fee = await get_fee_by_id(fee_id)
if not fee:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Fee not found",
)
return fee
@router.post(
"/",
response_model=FeeOut,
status_code=status.HTTP_201_CREATED,
responses={400: {"model": ErrorResponse}},
)
async def create_fee(
fee_in: FeeCreate,
_current_user: User = Depends(get_current_user),
) -> FeeOut:
return await create_fee_db(fee_in, created_by=_current_user.id)
与其他技能的集成
| 技能 |
集成点 |
@fastapi-app |
在 main.py 中注册路由器 |
@sqlmodel-crud |
端点中的数据库操作 |
@jwt-auth |
受保护路由的 Depends(get_current_user) |
@api-client |
此API设计的消费者 |
质量检查清单
- [ ] 分页标准:使用带有
has_more 指示器的 skip/limit
- [ ] 筛选:常见筛选字段的查询参数
- [ ] 排序:
sort_by 和 sort_order 参数
- [ ] 状态码:POST用201,DELETE用204,未找到用404
- [ ] 响应模型:所有端点都使用
response_model
- [ ] 文档:OpenAPI的
summary 和 description
- [ ] 错误处理:一致的错误响应格式
分页标准
@router.get("/items/", response_model=PaginatedResponse[ItemOut])
async def list_items(
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=1000),
) -> PaginatedResponse[ItemOut]:
items, total = await get_items(skip=skip, limit=limit)
return PaginatedResponse(
data=items,
total=total,
skip=skip,
limit=limit,
has_more=(skip + limit) < total,
)
筛选和排序标准
@router.get("/items/")
async def list_items(
# 筛选
category: str | None = None,
status: str | None = Query(None, pattern="^(active|inactive)$"),
min_amount: float | None = Query(None, ge=0),
# 排序
sort_by: str = Query("created_at", enum=["created_at", "amount", "name"]),
sort_order: str = Query("desc", enum=["asc", "desc"]),
):
return await get_items(
filters={"category": category, "status": status, "min_amount": min_amount},
order_by=f"{sort_order} {sort_by.lstrip('-')}",
)