FastAPI 依赖注入实战:三个让我翻过车的 Depends 场景

FastAPI Depends 三个翻车场景封面图

前言

FastAPI 的依赖注入用起来很顺手——路径函数参数里加个 Depends(get_current_user),认证就自动做了。但我用了一年多,真踩过三个让我debug到凌晨的坑:Depends 的参数在路由匹配前就解析了、yield 漏写导致连接泄漏、全局变量存 Session 导致用户数据串台。这三个场景各有各的根因,修复方法也不一样,说清楚方便大家对号入座。

坑一:Depends 的参数在路由匹配前就解析了

我原来的理解是:请求进来 → 匹配路由 → 执行处理器 → 处理器里执行 Depends。但实际完全不是这样。

from fastapi import FastAPI, Depends, HTTPException

app = FastAPI()

def get_user_from_db(user_id: int):
    if user_id != 1:
        return None
    return {"id": 1, "username": "admin"}

# 错误写法:依赖函数直接声明路径参数
def get_user(user_id: int = Depends(get_user_from_db)):
    if user_id is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user_id

@app.get("/users/{user_id}")
def read_user(user = Depends(get_user)):
    return user

运行时报错:TypeError: get_user_from_db() missing 1 required positional argument: 'user_id'

根因:FastAPI 在路由匹配之前就会解析所有 Depends 的参数user_id 作为路径参数还没绑定,FastAPI 只找到函数签名里的 user_id: int,没有默认值,直接报错了。

正确写法是用 Request 对象从路径参数里手动拿:

from fastapi import FastAPI, Depends, HTTPException, Request

app = FastAPI()

def get_user_from_db(user_id: int):
    if user_id != 1:
        return None
    return {"id": 1, "username": "admin"}

async def get_current_user(request: Request):
    user_id = int(request.path_params.get("user_id"))
    user = get_user_from_db(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.get("/users/{user_id}")
def read_user(user = Depends(get_current_user)):
    return user

记住:依赖函数如果需要路径参数,只能通过 Request.path_params 拿,不能用 Depends 声明路径参数。

坑二:yield 漏写,连接泄漏到连接池爆

FastAPI 官方推荐用 yield 做资源清理:

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session

app = FastAPI()

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

@app.get("/items/{item_id}")
def get_item(item_id: int, db: Session = Depends(get_db)):
    return db.query(Item).filter(Item.id == item_id).first()

我照着这个模式写,但生产环境跑一段时间后,连接池爆了:QueuePool limit of size 10 overflow 10 reached, connection timed out

查了半天,发现是我有时候忘记写 yield

# 错误:没有 yield,连接创建后永远不会关闭
def get_db():
    db = SessionLocal()
    return db  # 习惯了 try...yield...finally,有时候漏了 yield

# 正确:必须有 yield
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

另一个高频错误是依赖注入的 session 和手动事务混在一起

from contextlib import contextmanager

@contextmanager
def get_transaction():
    db = SessionLocal()
    try:
        with db.begin():
            yield db
    except:
        db.rollback()
        raise
    finally:
        db.close()

# 这个函数没有 yield,和 Depends 混用会出问题
def get_db_with_tx():
    return get_transaction().__enter__()

当不同路由混用有无 yield 的版本时,部分连接的回收时序不对,导致连接慢慢耗尽。

教训:所有资源依赖都用 yield,不混用有无 yield 的版本。写完检查一遍每个依赖函数是否有 yield。

坑三:全局变量存 Session,用户数据串台

这是损失最严重的一个坑:用户A登录后,能看到用户B的数据。

# 全局变量——这是错的
current_user = None

def get_current_user():
    return current_user

@app.get("/me")
def get_me():
    return get_current_user()

@app.post("/login")
def login(username: str):
    global current_user
    current_user = {"username": username}
    return {"status": "ok"}

多用户并发时,User A 登录后,User B 的请求可能读到 User A 的 session。这不是 FastAPI 本身的 bug,是我自己写了错误的全局状态。

FastAPI 的依赖注入是请求级别的——每次请求都会重新调用依赖函数创建新对象。只要不用全局变量存请求状态,就不会有这个问题。

from fastapi import FastAPI, Depends, HTTPException, Header

app = FastAPI()

fake_users_db = {
    "alice-token": {"id": 1, "username": "alice"},
    "bob-token": {"id": 2, "username": "bob"},
}

async def get_current_user(authorization: str = Header(None)):
    if not authorization:
        raise HTTPException(status_code=401, detail="Not authenticated")
    
    token = authorization.replace("Bearer ", "")
    user = fake_users_db.get(token)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

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

@app.post("/login")
def login(username: str):
    token = username + "-token"
    return {"access_token": token}

记住:永远不用全局变量存请求状态。所有请求状态都通过依赖注入传递,FastAPI 保证不同请求之间完全隔离。

总结

场景 错误写法 正确写法
依赖需要路径参数 Depends(get_user(user_id)) 在依赖里用 Request.path_params
数据库连接清理 直接 return db,没有 yield yield + finally: db.close()
请求状态存储 全局变量 current_user = None 依赖注入,每次请求重建

三个坑的核心:依赖注入的参数在路由匹配前解析、资源必须用 yield 清理、请求状态永远不碰全局变量。记住这三条,能避开 90% 的 FastAPI 生产级 bug。

© 版权声明
THE END
喜欢就支持一下吧
点赞9 分享