FastAPI依赖注入:从基础概念到应用
title: FastAPI依赖注入:从基础概念到应用


from fastapi import Depends, FastAPI
app = FastAPI()
# 基础依赖函数示例
def query_extractor(q: str | None = None):
return q
# 类形式依赖项
class Pagination:
def __init__(self, page: int = 1, size: int = 10):
self.page = page
self.size = size
# 路由中使用依赖
@app.get("/items/")
async def read_items(
q: str = Depends(query_extractor),
pagination: Pagination = Depends()
):
return {
"q": q,
"page": pagination.page,
"size": pagination.size
}
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
def get_current_user(db: Session = Depends(get_db)):
user = db.query(User).first()
if not user:
raise HTTPException(status_code=404)
return user
@app.get("/profile")
def user_profile(user: User = Depends(get_current_user)):
return {"username": user.name}
user_profile
└── get_current_user
└── get_db
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
class User(BaseModel):
username: str
is_admin: bool = False
def validate_token(token: str = Depends(oauth2_scheme)) -> str:
if token != "secret_token":
raise HTTPException(status_code=403)
return token
def get_user(token: str = Depends(validate_token)) -> User:
return User(username="admin", is_admin=True)
@app.get("/admin")
def admin_dashboard(
user: User = Depends(get_user),
db: Session = Depends(get_db)
):
if not user.is_admin:
raise HTTPException(status_code=403)
return {"message": "Admin console"}
{
"detail": [
{
"loc": ["query", "page"],
"msg": "value is not a valid integer",
"type": "type_error.integer"
}
]
}
from pydantic import BaseModel
class PaginationParams(BaseModel):
page: int = 1
size: int = 10
@validator("page")
def validate_page(cls, v):
if v < 1:
raise ValueError("Page must be ≥1")
return v
# 错误示例
def dep_a(b: str = Depends(dep_b)): ...
def dep_b(a: int = Depends(dep_a)): ...
def dep_a(b: str = Depends(lambda: dep_b)): ...
# 方法1:通过Request对象
def get_ua(request: Request):
return request.headers.get("user-agent")
# 方法2:使用Header参数
def get_ua(user_agent: str | None = Header(None)):
return user_agent
pip install fastapi uvicorn sqlalchemy python-multipart
uvicorn main:app --reload
curl -X GET "http://localhost:8000/items/?q=test&page=2&size=20"
评论
发表评论