Skip to content

依赖注入实践 ​

实验目标 ​

掌握 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 依赖在请求后正确清理资源
✅ 认证依赖阻止未授权访问
✅ 嵌套依赖正确传递
✅ 依赖可以注入到依赖中

返回第二章概览

用心记录每一步成长