FastAPI路由与请求处理进阶指南：解锁企业级API开发黑科技 🔥
title: FastAPI路由与请求处理进阶指南：解锁企业级API开发黑科技 🔥
from fastapi import Path
@app.get("/users/{user_id:int}")
async def get_user(
user_id: int = Path(..., regex="^[0-9]{8}$", example=10000001)
):
# 匹配8位数字ID
return db.query(User).filter(User.id == user_id).first()
# 高优先级路由
@app.get("/users/me", priority=100)
async def get_current_user():
...
# 低优先级通用路由
@app.get("/users/{user_id}", priority=10)
async def get_user(user_id: int):
...
class Address(BaseModel):
street: str
city: str
class UserProfile(BaseModel):
name: str
addresses: list[Address]
@app.post("/users")
async def create_user(profile: UserProfile):
# 自动解析嵌套结构
db.save(profile.dict())
from fastapi import UploadFile, File
@app.post("/upload")
async def upload_large_file(
chunk: UploadFile = File(...),
chunk_number: int = Form(...)
):
with open(f"temp_{chunk_number}", "wb") as buffer:
content = await chunk.read()
buffer.write(content)
return {"received_chunks": chunk_number + 1}
from fastapi import Depends
def get_db():
# 数据库连接池
return DatabasePool()
@app.get("/products")
async def list_products(
page: int = 1,
db: Database = Depends(get_db)
):
return db.query(Product).paginate(page)
# 按需加载路由模块
from fastapi import APIRouter
order_router = APIRouter()
@order_router.get("/")
async def list_orders():
...
app.include_router(order_router, prefix="/orders")
from fastapi_limiter import Limiter
limiter = Limiter(key_func=get_remote_address)
@app.get("/api/data", dependencies=[Depends(limiter.limit("100/min"))])
async def sensitive_data():
return generate_report()
# 危险：直接拼接
@app.get("/users")
async def unsafe_query(name: str):
query = f"SELECT * FROM users WHERE name = '{name}'"
# 安全：参数化查询
@app.get("/users")
async def safe_query(name: str = Query(...)):
query = "SELECT * FROM users WHERE name = :name"
params = {"name": name}
return db.execute(query, params)
# 要求：
# 1. 支持SKU编码校验（格式：ABC-12345）
# 2. 实现库存实时扣减
# 3. 处理高并发冲突
@app.put("/skus/{sku_code}")
async def update_sku(sku_code: str, stock: int):
# 你的代码
# 原代码
@app.get("/orders")
async def list_orders():
return db.query(Order).all()
# 优化目标：响应时间 <100ms（提示：添加缓存索引）
| 错误现象 | 原因 | 解决方案 |
|---|---|---|
422 Unprocessable Entity | 嵌套模型校验失败 | 使用try-except包裹模型解析 |
404 Not Found | 路由优先级冲突 | 调整priority参数或路由顺序 |
500 Internal Error | 异步未await | 检查所有IO操作是否使用async/await |
评论
发表评论