FastAPI依赖注入:链式调用与多级参数传递
title: FastAPI依赖注入:链式调用与多级参数传递


from fastapi import Depends, FastAPI
app = FastAPI()
# 第一级依赖
def get_query():
return "search_query"
# 第二级依赖(依赖第一级)
def get_filter(q: str = Depends(get_query)):
return f"filter:{q}"
@app.get("/search/")
async def search(filter_str: str = Depends(get_filter)):
return {"result": filter_str}
from fastapi import Depends, HTTPException
from pydantic import BaseModel
class User(BaseModel):
id: int
username: str
is_vip: bool = False
class Item(BaseModel):
item_id: int
stock: int
price: float
# 模拟数据库
fake_db = {
"users": {
1: User(id=1, username="vip_user", is_vip=True),
2: User(id=2, username="normal_user")
},
"items": {
101: Item(item_id=101, stock=10, price=99.9),
102: Item(item_id=102, stock=0, price=199.9)
}
}
# 第一级:用户认证
async def get_current_user():
user = fake_db["users"].get(1) # 模拟登录用户
if not user:
raise HTTPException(status_code=401)
return user
# 第二级:VIP校验
async def check_vip_status(
user: User = Depends(get_current_user)
):
if not user.is_vip:
raise HTTPException(
status_code=403,
detail="VIP会员专属功能"
)
return {"user": user, "discount": 0.8}
# 第三级:库存检查
async def check_inventory(
item_id: int,
vip_info: dict = Depends(check_vip_status)
):
item = fake_db["items"].get(item_id)
if not item or item.stock <= 0:
raise HTTPException(
status_code=400,
detail="商品库存不足"
)
return {
**vip_info,
"item": item,
"final_price": item.price * vip_info["discount"]
}
@app.post("/orders/{item_id}")
async def create_order(
order_data: dict,
inventory: dict = Depends(check_inventory)
):
"""最终订单创建接口"""
return {
"user": inventory["user"].username,
"item": inventory["item"].item_id,
"price": inventory["final_price"],
"order_data": order_data
}
def dep1(): return "data1"
def dep2(d1: str = Depends(dep1)):
return d1 + "_data2"
def dep3(d2: str = Depends(dep2)):
return d2.upper()
def config1(): return {"setting1": True}
def config2(): return {"setting2": 100}
@app.get("/settings")
def get_settings(
c1: dict = Depends(config1),
c2: dict = Depends(config2)
):
return {**c1, **c2}
def pagination_params(
page: int = 1,
size: int = 10
):
return {"offset": (page - 1) * size, "limit": size}
@app.get("/products")
def get_products(
pagination: dict = Depends(pagination_params)
):
return f"Showing {pagination['limit']} items"
common_deps = Depends(dep1) & Depends(dep2)
@app.get("/route1", dependencies=[common_deps])
{
"detail": [
{
"loc": [
"query",
"q"
],
"msg": "field required",
"type": "value_error.missing"
}
]
}
def dep_a(d=Depends(dep_b)): ...
def dep_b(d=Depends(dep_a)): ...
from fastapi import Depends
from sqlalchemy.orm import Session
# 使用lru_cache缓存数据库会话
def get_db():
return SessionLocal()
@app.get("/items")
def read_items(db: Session = Depends(get_db)):
...
security = Depends(authenticate) & Depends(authorize)
async def async_dep():
await some_io_operation()
pip install fastapi uvicorn pydantic python-multipart
uvicorn main:app --reload
curl -X POST "http://localhost:8000/orders/101" \
-H "Content-Type: application/json" \
-d '{"remark":"urgent"}'
评论
发表评论