依赖注入

|
from fastapi import FastAPI, Depends, Header, HTTPException
from typing import Optional
 
app = FastAPI()
 
def get_common_parameters(
    skip: int = 0,
    limit: int = 10
):
    """一个简单的依赖函数，提供通用参数"""
    return {"skip": skip, "limit": limit}
 
@app.get("/items/")
def read_items(commons: dict = Depends(get_common_parameters)):
    """使用依赖注入获取通用参数"""
    return {"items": [], **commons}

|
async def get_current_user(
    authorization: Optional[str] = Header(None)
):
    """从请求头中获取当前用户"""
    if not authorization:
        raise HTTPException(status_code=401, detail="未提供认证信息")
    
    # 这里应该验证token并查询用户
    # 为了示例，我们简化处理
    token = authorization.replace("Bearer ", "")
    if token == "valid_token":
        return {"id": 1, "username": "admin", "email": "admin@example.com"}
    else:
        raise HTTPException(status_code=401, detail="无效的认证token")
 
@app.get("/users/me")
async def get_my_info(current_user: dict = Depends(get_current_user)):
    """获取当前用户信息"""
    return current_user

|
def get_db_connection():
    """模拟数据库连接依赖"""
    return {"connection": "db_connection"}
 
async def get_user_service(db: dict = Depends(get_db_connection)):
    """用户服务依赖，依赖于数据库连接"""
    return {
        "db": db,
        "service": "user_service"
    }
 
@app.get("/users/")
async def get_users(service: dict = Depends(get_user_service)):
    """获取用户列表，使用用户服务依赖"""
    return {"users": [], "service": service}

|
from fastapi import FastAPI, Depends, Query, HTTPException
from typing import Optional
 
app = FastAPI()
 
def verify_token(token: str = Query(...)):
    """同步函数依赖：验证token"""
    if token != "secret_token":
        raise HTTPException(status_code=401, detail="无效的token")
    return {"user_id": 1, "username": "admin"}
 
@app.get("/protected/")
def protected_route(user: dict = Depends(verify_token)):
    """受保护的路由，需要token验证"""
    return {"message": f"欢迎, {user['username']}"}

|
import asyncio
 
async def get_user_from_db(user_id: int):
    """异步函数依赖：从数据库获取用户"""
    # 模拟异步数据库查询
    await asyncio.sleep(0.1)
    if user_id == 1:
        return {"id": 1, "name": "John Doe", "email": "john@example.com"}
    raise HTTPException(status_code=404, detail="用户不存在")
 
@app.get("/users/{user_id}")
async def get_user(user_data: dict = Depends(get_user_from_db)):
    """获取用户信息，使用异步依赖"""
    return user_data

|
def check_permission(required_permission: str):
    """返回一个依赖函数，检查特定权限"""
    def permission_checker(user: dict = Depends(verify_token)):
        if required_permission not in user.get("permissions", []):
            raise HTTPException(
                status_code=403,
                detail=f"需要权限: {required_permission}"
            )
        return user
    return permission_checker
 
@app.get("/admin/")
def admin_route(user: dict = Depends(check_permission("admin"))):
    """管理员路由，需要admin权限"""
    return {"message": "管理员页面", "user": user}

|
from fastapi import FastAPI, Depends
from typing import Dict
 
app = FastAPI()
 
class DatabaseConnection:
    """数据库连接类依赖"""
    def __init__(self):
        self.connection_string = "postgresql://localhost/mydb"
        self.connected = False
    
    def __call__(self):
        """当被作为依赖调用时执行"""
        if not self.connected:
            # 模拟连接数据库
            self.connected = True
            print(f"连接到数据库: {self.connection_string}")
        return self
 
# 创建单例实例
db_connection = DatabaseConnection()
 
@app.get("/data/")
def get_data(db: DatabaseConnection = Depends(db_connection)):
    """使用类依赖获取数据"""
    return {"message": "数据获取成功", "db_connected": db.connected}

|
class UserService:
    """用户服务类依赖"""
    def __init__(self, db: DatabaseConnection = Depends(db_connection)):
        self.db = db
        self.cache = {}
    
    def __call__(self, user_id: int):
        """根据用户ID获取用户"""
        if user_id in self.cache:
            return self.cache[user_id]
        
        # 模拟从数据库查询
        user = {"id": user_id, "name": f"User {user_id}"}
        self.cache[user_id] = user
        return user
 
user_service = UserService()
 
@app.get("/users/{user_id}")
def get_user(user: dict = Depends(user_service)):
    """获取用户，使用类依赖"""
    return user

|
class ConfigManager:
    """配置管理类依赖，使用缓存"""
    def __init__(self):
        self.config = {
            "app_name": "MyApp",
            "version": "1.0.0",
            "debug": False
        }
        print("配置管理器初始化")
    
    def __call__(self):
        return self.config
 
config_manager = ConfigManager()
 
@app.get("/config/")
def get_config(config: dict = Depends(config_manager, use_cache=True)):
    """获取配置，使用缓存的类依赖"""
    return config

|
from fastapi import FastAPI, Depends
from typing import Generator
import sqlite3
 
app = FastAPI()
 
def get_db() -> Generator:
    """数据库连接依赖，使用生成器确保连接正确关闭"""
    db = sqlite3.connect("example.db")
    try:
        yield db
    finally:
        db.close()
 
@app.get("/items/")
def get_items(db: sqlite3.Connection = Depends(get_db)):
    """获取所有物品，使用数据库连接依赖"""
    cursor = db.cursor()
    cursor.execute("SELECT * FROM items")
    items = cursor.fetchall()
    return {"items": items}
 
@app.post("/items/")
def create_item(name: str, db: sqlite3.Connection = Depends(get_db)):
    """创建物品，使用数据库连接依赖"""
    cursor = db.cursor()
    cursor.execute("INSERT INTO items (name) VALUES (?)", (name,))
    db.commit()
    return {"message": "Item created", "id": cursor.lastrowid}

|
import asyncpg
from typing import AsyncGenerator
 
async def get_async_db() -> AsyncGenerator:
    """异步数据库连接依赖"""
    conn = await asyncpg.connect("postgresql://user:password@localhost/db")
    try:
        yield conn
    finally:
        await conn.close()
 
@app.get("/users/")
async def get_users(db: asyncpg.Connection = Depends(get_async_db)):
    """获取所有用户，使用异步数据库连接"""
    rows = await db.fetch("SELECT * FROM users")
    return {"users": [dict(row) for row in rows]}

|
from contextlib import asynccontextmanager
 
class DatabasePool:
    """数据库连接池类"""
    def __init__(self):
        self.pool = None
    
    async def initialize(self):
        """初始化连接池"""
        self.pool = await asyncpg.create_pool(
            "postgresql://user:password@localhost/db",
            min_size=5,
            max_size=20
        )
    
    async def get_connection(self):
        """从连接池获取连接"""
        async with self.pool.acquire() as connection:
            yield connection
 
db_pool = DatabasePool()
 
@app.on_event("startup")
async def startup():
    """应用启动时初始化连接池"""
    await db_pool.initialize()
 
@app.get("/data/")
async def get_data(conn = Depends(db_pool.get_connection)):
    """使用连接池获取数据"""
    rows = await conn.fetch("SELECT * FROM data")
    return {"data": [dict(row) for row in rows]}

|
from fastapi import FastAPI, Depends, HTTPException, Header
from typing import Optional
import jwt
 
app = FastAPI()
SECRET_KEY = "your-secret-key"
 
async def get_current_user(
    authorization: Optional[str] = Header(None)
) -> dict:
    """获取当前用户依赖"""
    if not authorization:
        raise HTTPException(
            status_code=401,
            detail="未提供认证信息",
            headers={"WWW-Authenticate": "Bearer"}
        )
    
    try:
        token = authorization.replace("Bearer ", "")
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        user_id = payload.get("sub")
        
        # 这里应该从数据库查询用户
        # 为了示例，我们简化处理
        return {
            "id": user_id,
            "username": payload.get("username"),
            "email": payload.get("email")
        }
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token已过期")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="无效的Token")
 
@app.get("/users/me")
async def get_my_info(current_user: dict = Depends(get_current_user)):
    """获取当前用户信息，需要认证"""
    return current_user

|
def require_permission(required_permission: str):
    """权限检查依赖工厂"""
    async def permission_checker(
        current_user: dict = Depends(get_current_user)
    ):
        user_permissions = current_user.get("permissions", [])
        if required_permission not in user_permissions:
            raise HTTPException(
                status_code=403,
                detail=f"需要权限: {required_permission}"
            )
        return current_user
    return permission_checker
 
@app.get("/admin/")
async def admin_route(
    user: dict = Depends(require_permission("admin"))
):
    """管理员路由，需要admin权限"""
    return {"message": "管理员页面", "user": user}
 
@app.delete("/users/{user_id}")
async def delete_user(
    user_id: int,
    current_user: dict = Depends(get_current_user),
    admin_user: dict = Depends(require_permission("admin"))
):
    """删除用户，需要管理员权限"""
    # 只有管理员可以删除用户
    return {"message": f"用户 {user_id} 已删除"}

|
async def get_current_user_optional(
    authorization: Optional[str] = Header(None)
) -> Optional[dict]:
    """可选的用户认证依赖"""
    if not authorization:
        return None
    
    try:
        token = authorization.replace("Bearer ", "")
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return {
            "id": payload.get("sub"),
            "username": payload.get("username")
        }
    except:
        return None
 
@app.get("/posts/")
async def get_posts(
    current_user: Optional[dict] = Depends(get_current_user_optional)
):
    """获取帖子列表，支持匿名访问"""
    if current_user:
        return {"posts": [], "user": current_user}
    else:
        return {"posts": [], "user": None}

|
# main.py
from fastapi import FastAPI, Depends, HTTPException
 
app = FastAPI()
 
def get_db():
    """真实的数据库依赖"""
    # 实际应用中这里会连接真实数据库
    return {"connection": "real_db"}
 
def get_current_user(token: str):
    """真实的用户认证依赖"""
    if token != "valid_token":
        raise HTTPException(status_code=401, detail="Invalid token")
    return {"id": 1, "username": "admin"}
 
@app.get("/items/")
def get_items(db: dict = Depends(get_db)):
    """获取物品列表"""
    return {"items": ["item1", "item2"], "db": db}
 
@app.get("/users/me")
def get_my_info(user: dict = Depends(get_current_user)):
    """获取当前用户信息"""
    return user

|
# test_main.py
from fastapi.testclient import TestClient
from main import app
 
def override_get_db():
    """模拟数据库依赖"""
    return {"connection": "mock_db"}
 
def override_get_current_user():
    """模拟用户认证依赖"""
    return {"id": 999, "username": "test_user"}
 
# 覆盖依赖
app.dependency_overrides[app.get_db] = override_get_db
app.dependency_overrides[app.get_current_user] = override_get_current_user
 
client = TestClient(app)
 
def test_get_items():
    """测试获取物品，使用模拟数据库"""
    response = client.get("/items/")
    assert response.status_code == 200
    assert response.json()["db"]["connection"] == "mock_db"
 
def test_get_my_info():
    """测试获取用户信息，使用模拟认证"""
    response = client.get("/users/me", headers={"Authorization": "Bearer any_token"})
    assert response.status_code == 200
    assert response.json()["username"] == "test_user"
 
# 清理覆盖
def teardown():
    app.dependency_overrides.clear()

|
def override_get_current_user_error():
    """模拟认证失败的依赖"""
    from fastapi import HTTPException
    raise HTTPException(status_code=401, detail="Authentication failed")
 
def test_authentication_failure():
    """测试认证失败的情况"""
    app.dependency_overrides[app.get_current_user] = override_get_current_user_error
    response = client.get("/users/me")
    assert response.status_code == 401
    assert "Authentication failed" in response.json()["detail"]

|
def override_get_user_by_id(user_id: int):
    """模拟根据ID获取用户的依赖"""
    if user_id == 999:
        raise HTTPException(status_code=404, detail="User not found")
    return {"id": user_id, "name": f"User {user_id}"}
 
def test_get_user_success():
    """测试成功获取用户"""
    app.dependency_overrides[app.get_user_by_id] = override_get_user_by_id
    response = client.get("/users/123")
    assert response.status_code == 200
    assert response.json()["id"] == 123
 
def test_get_user_not_found():
    """测试用户不存在的情况"""
    app.dependency_overrides[app.get_user_by_id] = override_get_user_by_id
    response = client.get("/users/999")
    assert response.status_code == 404