依赖注入实践
实验目标
掌握 FastAPI 依赖注入系统,实现可复用的数据库连接、认证检查、通用参数。
实验 1:基础依赖
python
from fastapi import FastAPI, Depends
app = FastAPI()
# 简单依赖
def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
return commons
@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
return commons测试:
bash
curl "http://localhost:8000/items/?q=test&skip=5&limit=20"
# 输出: {"q":"test","skip":5,"limit":20}实验 2:数据库依赖
python
class Database:
"""示意代码,实际项目中使用 SQLAlchemy 等 ORM"""
def __init__(self):
self.connection = "mock_db_connection"
def query(self, sql: str):
return f"Executing: {sql}"
def close(self):
print("Database closed")
def get_db():
db = Database()
try:
yield db
finally:
db.close()
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: Database = Depends(get_db)):
result = db.query(f"SELECT * FROM users WHERE id={user_id}")
return {"user_id": user_id, "query": result}实验 3:认证依赖
python
from fastapi import Header, HTTPException
async def verify_token(x_token: str = Header()):
if x_token != "secret-token":
raise HTTPException(status_code=401, detail="Invalid token")
return x_token
@app.get("/protected")
async def protected_route(token: str = Depends(verify_token)):
return {"message": "Access granted", "token": token}测试:
bash
# 无 token
curl http://localhost:8000/protected
# 输出: {"detail":"Invalid token"}
# 有 token
curl -H "x-token: secret-token" http://localhost:8000/protected
# 输出: {"message":"Access granted","token":"secret-token"}实验 4:嵌套依赖
python
def get_db():
return Database()
def get_user_service(db: Database = Depends(get_db)):
return UserService(db)
@app.get("/users/{user_id}")
async def get_user(
user_id: int,
user_service: UserService = Depends(get_user_service)
):
return user_service.get_user(user_id)验收标准
✅ 依赖函数可以被多个路径复用
✅ yield 依赖在请求后正确清理资源
✅ 认证依赖阻止未授权访问
✅ 嵌套依赖正确传递
✅ 依赖可以注入到依赖中
返回第二章概览