FastAPI 路径操作依赖项

作者likuolei

FastAPI 路径操作依赖项（Dependencies）完全指南

依赖注入（Dependency Injection） 是 FastAPI 的核心特性之一，让代码 可复用、可测试、可维护。

一、什么是依赖项？

依赖项 = 一个函数，可以被多个路由 重复使用。

def get_db():
    db = DBSession()
    try:
        yield db
    finally:
        db.close()

在路由中使用：

@app.get("/items/")
def read_items(db: Session = Depends(get_db)):
    return db.query(Item).all()

二、核心优势

优势说明
复用数据库、认证、权限等逻辑写一次用多处
测试友好轻松 mock 依赖
自动文档依赖参数也显示在 /docs
层级依赖依赖可以依赖其他依赖
缓存同一个请求中，依赖只执行一次

三、基本语法：Depends()

from fastapi import Depends

def common_parameters(q: str | None = None, skip: int = 0, limit: int = 10):
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
def read_items(commons: dict = Depends(common_parameters)):
    return commons

访问：GET /items/?q=book&skip=5&limit=10

四、类作为依赖项（推荐）

class CommonQueryParams:
    def __init__(self, q: str | None = None, skip: int = 0, limit: int = 10):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
def read_items(commons: CommonQueryParams = Depends()):
    return {"q": commons.q, "skip": commons.skip, "limit": commons.limit}

更清晰、更易测试、支持 IDE 提示

五、数据库依赖（经典示例）

1. 模拟数据库会话

from contextlib import contextmanager

fake_db = {}

@contextmanager
def get_db():
    db = fake_db
    try:
        yield db
    finally:
        pass  # 关闭连接

2. 路由中使用

@app.post("/users/")
def create_user(user: UserCreate, db: dict = Depends(get_db)):
    db[user.username] = user.dict()
    return {"message": "用户创建成功"}

六、真实数据库：SQLAlchemy + 依赖

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db() -> Session:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
# main.py
from fastapi import Depends
from sqlalchemy.orm import Session

@app.post("/users/")
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = User(**user.dict())
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

七、用户认证依赖（JWT 示例）

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def get_current_user(token: str = Depends(oauth2_scheme)):
    # 模拟验证
    if token != "secret-token":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="无效的认证凭据",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return {"username": "alice"}

@app.get("/profile/")
def read_profile(current_user: dict = Depends(get_current_user)):
    return current_user

八、权限控制依赖

from fastapi import Depends

def require_admin(user: dict = Depends(get_current_user)):
    if user.get("role") != "admin":
        raise HTTPException(status_code=403, detail="权限不足")
    return user

@app.delete("/users/{user_id}")
def delete_user(
    user_id: int,
    admin: dict = Depends(require_admin),  # 必须是管理员
    db: Session = Depends(get_db)
):
    # 删除逻辑
    return {"message": f"用户 {user_id} 已删除"}

九、依赖项层级（依赖的依赖）

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def get_current_user(token: str = Depends(oauth2_scheme)):
    # 使用 db 查询用户
    db = next(get_db())  # 也可以通过 Depends(get_db)
    user = db.query(User).filter(User.token == token).first()
    if not user:
        raise HTTPException(401)
    return user

def get_current_active_user(current_user: dict = Depends(get_current_user)):
    if not current_user.get("is_active"):
        raise HTTPException(400, "用户未激活")
    return current_user

使用：

@app.get("/me/")
def read_me(user: dict = Depends(get_current_active_user)):
    return user

十、全局依赖（所有路由都生效）

app = FastAPI(dependencies=[Depends(log_requests)])

def log_requests(request: Request):
    print(f"请求: {request.method} {request.url}")
    return None

十一、依赖项缓存（同一个请求只执行一次）

def expensive_operation():
    print("执行耗时操作...")
    return {"data": "cached"}

@app.get("/a/")
def route_a(dep = Depends(expensive_operation)):
    return dep  # 打印一次

@app.get("/b/")
def route_b(dep = Depends(expensive_operation)):
    return dep  # 同一个请求中，不会再打印

十二、子依赖（Sub-dependencies）

def verify_token(token: str = Depends(oauth2_scheme)):
    if token != "valid":
        raise HTTPException(401)
    return {"sub": "user123"}

def verify_scope(claims: dict = Depends(verify_token)):
    if "admin" not in claims.get("scopes", []):
        raise HTTPException(403)
    return claims

@app.get("/admin/", dependencies=[Depends(verify_scope)])
def admin_panel():
    return {"message": "管理员面板"}

十三、依赖项返回多个值

from typing import Tuple

def get_pagination(skip: int = 0, limit: int = 10) -> Tuple[int, int]:
    return skip, limit

@app.get("/items/")
def read_items(pagination: Tuple[int, int] = Depends(get_pagination)):
    skip, limit = pagination
    return {"skip": skip, "limit": limit}

十四、异步依赖

import asyncio

async def async_dependency():
    await asyncio.sleep(1)
    return {"message": "异步依赖"}

@app.get("/async/")
async def async_route(dep: dict = Depends(async_dependency)):
    return dep

十五、依赖项在 /docs 中显示

def get_query_params(q: str | None = None):
    return {"q": q}

@app.get("/search/", dependencies=[Depends(get_query_params)])
def search():
    return {"message": "搜索"}

在 Swagger UI 中会显示 q 参数！

十六、完整实战示例

from fastapi import FastAPI, Depends, HTTPException, status, Header
from pydantic import BaseModel
from typing import Optional

app = FastAPI()

# 1. 通用查询参数
class CommonQuery:
    def __init__(self, skip: int = 0, limit: int = 10, q: Optional[str] = None):
        self.skip = skip
        self.limit = limit
        self.q = q

# 2. 模拟数据库
fake_users_db = [
    {"id": 1, "username": "alice", "role": "admin"},
    {"id": 2, "username": "bob", "role": "user"},
]

# 3. 依赖项
def get_query_params(q: Optional[str] = None, skip: int = 0, limit: int = 100):
    return CommonQuery(skip, limit, q)

def get_current_user(authorization: str = Header(...)):
    if authorization != "Bearer secret":
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "无效 token")
    return {"username": "alice", "role": "admin"}

def get_admin_user(user: dict = Depends(get_current_user)):
    if user["role"] != "admin":
        raise HTTPException(403, "需要管理员权限")
    return user

# 4. 路由
@app.get("/users/")
def read_users(
    commons: CommonQuery = Depends(get_query_params),
    user: dict = Depends(get_current_user)
):
    start = commons.skip
    end = start + commons.limit
    users = fake_users_db[start:end]
    if commons.q:
        users = [u for u in users if commons.q in u["username"]]
    return {"users": users, "current_user": user["username"]}

@app.delete("/users/{user_id}")
def delete_user(
    user_id: int,
    admin: dict = Depends(get_admin_user),
    db: list = Depends(lambda: fake_users_db)
):
    for i, user in enumerate(db):
        if user["id"] == user_id:
            db.pop(i)
            return {"message": f"用户 {user_id} 已删除"}
    raise HTTPException(404, "用户不存在")

十七、测试依赖项（Pytest）

from fastapi.testclient import TestClient

client = TestClient(app)

def test_read_users():
    response = client.get("/users/", headers={"Authorization": "Bearer secret"})
    assert response.status_code == 200
    assert "alice" in response.json()["current_user"]

覆盖依赖：

def override_get_current_user():
    return {"username": "testuser", "role": "admin"}

app.dependency_overrides[get_current_user] = override_get_current_user

# 测试完成后清除
app.dependency_overrides.clear()

总结：依赖项最佳实践

实践建议
复用逻辑数据库、认证、日志
类形式CommonQueryParams() 更清晰
返回 yield自动管理资源关闭
权限分层get_user → get_active_user → get_admin
全局依赖日志、CORS、监控
测试覆盖dependency_overrides

下一步学习

主题关键词
数据库依赖（SQLAlchemy）sqlalchemy
JWT 完整认证jwt
速率限制依赖ratelimit
缓存依赖（Redis）redis
依赖项文档docs

现在就用依赖项重构你的项目，让代码更优雅！

需要我生成一个 完整依赖项目模板（包含数据库、认证、分页、权限）吗？
回复 依赖模板 立刻获取！

类似文章

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注