DevNull 知识库

Appearance

39｜权限控制与依赖注入 ​

认证解决"你是谁"的问题后，还需要解决"你能做什么"的问题。管理员可以删除服务器，普通用户只能查看；某些接口需要登录才能访问。FastAPI 的依赖注入系统（Dependency Injection）让权限校验、数据库会话获取、当前用户解析这些横切关注点从业务代码中抽离，统一复用。

一、依赖注入基础 ​

依赖注入的核心是：路由处理函数声明自己需要什么，FastAPI 自动提供。这个"提供"的过程封装在依赖函数中。

python
from fastapi import Depends, HTTPException

# 依赖函数：获取数据库会话
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 路由中使用依赖
@app.get("/servers")
async def list_servers(db: Session = Depends(get_db)):
    return db.execute(select(Server)).scalars().all()

Depends(get_db) 告诉 FastAPI："这个参数不是来自请求，而是调用 get_db() 获取"。每次请求都会创建新的会话，请求结束后自动关闭。

依赖的依赖 ​

依赖函数本身也可以依赖其他依赖：

python
async def get_current_user(token: str = Depends(oauth2_scheme)):
    payload = decode_token(token)
    return payload.get("sub")

async def get_current_admin(user_id: str = Depends(get_current_user)):
    user = get_user_by_id(user_id)
    if user.role != "admin":
        raise HTTPException(status_code=403, detail="无权限")
    return user

二、登录依赖 ​

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

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/login")

async def require_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = decode_token(token)
        user_id = payload.get("sub")
        if not user_id:
            raise HTTPException(status_code=401, detail="无效凭证")
        return user_id
    except ValueError:
        raise HTTPException(status_code=401, detail="凭证无效或已过期")

任何需要登录的接口都加上这个依赖：

python
@app.get("/servers")
async def list_servers(user_id: str = Depends(require_user)):
    return {"user_id": user_id, "servers": [...]}

未登录用户访问时，OAuth2PasswordBearer 自动返回 401，并在响应头中加入 WWW-Authenticate: Bearer。

三、角色权限控制 ​

基于角色的依赖 ​

python
class RoleChecker:
    def __init__(self, allowed_roles: list[str]):
        self.allowed_roles = allowed_roles

    def __call__(self, user: User = Depends(get_current_user)):
        if user.role not in self.allowed_roles:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="权限不足",
            )
        return user

require_admin = RoleChecker(["admin"])
require_editor = RoleChecker(["admin", "editor"])

使用：

python
@app.delete("/servers/{id}")
async def delete_server(
    id: int,
    user: User = Depends(require_admin),
):
    return {"deleted": id}

@app.put("/servers/{id}")
async def update_server(
    id: int,
    user: User = Depends(require_editor),
):
    return {"updated": id}

接口级别的权限 ​

更细粒度的控制可以在接口内部判断：

python
@app.post("/servers/{id}/tasks")
async def create_task(
    id: int,
    user: User = Depends(get_current_user),
    db: Session = Depends(get_db),
):
    server = db.get(Server, id)
    if not server:
        raise HTTPException(status_code=404)

    # 只能给自己的服务器创建任务
    if server.owner_id != user.id and user.role != "admin":
        raise HTTPException(status_code=403)

    # 创建任务...

四、全局依赖 ​

某些依赖应用于所有路由，如日志记录、请求计时：

python
async def log_requests(request: Request):
    start = time.time()
    yield
    duration = time.time() - start
    print(f"{request.method} {request.url.path} - {duration:.3f}s")

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

yield 之前的代码在路由执行前运行，yield 之后的代码在路由执行后运行。这在依赖注入中称为"退出栈"，适合日志、计时、事务管理。

五、依赖返回值的使用 ​

依赖的返回值可以直接在路由函数中使用：

python
@app.get("/servers")
async def list_servers(
    user: User = Depends(get_current_user),
    db: Session = Depends(get_db),
):
    # user 和 db 都已经通过依赖注入获取
    if user.role == "admin":
        return db.execute(select(Server)).scalars().all()
    else:
        return db.execute(
            select(Server).where(Server.owner_id == user.id)
        ).scalars().all()

六、常见错误 ​

依赖函数没有用 yield ​

python
# 错误：没有 yield，FastAPI 不会正确处理清理逻辑
def get_db():
    db = SessionLocal()
    return db   # 会话不会自动关闭

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

依赖中抛出的异常被吞掉 ​

python
def get_db():
    db = SessionLocal()
    try:
        yield db
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()

如果依赖中需要处理异常（如回滚事务），要在 yield 后用 try/except。但不要把所有异常都吞掉——应该 re-raise 让 FastAPI 的正常异常处理机制接管。

权限校验放在业务代码中而不是依赖中 ​

python
# 不推荐：每个路由都重复写权限判断
@app.delete("/servers/{id}")
async def delete_server(id: int, user: User = Depends(get_current_user)):
    if user.role != "admin":
        raise HTTPException(status_code=403)
    ...

# 推荐：权限判断封装成依赖，路由声明式使用
@app.delete("/servers/{id}")
async def delete_server(id: int, user: User = Depends(require_admin)):
    ...

依赖返回值类型没注解 ​

python
# 错误：没有类型注解，FastAPI 无法正确注入
def get_current_user(token: str = Depends(oauth2_scheme)):
    ...

# 正确：返回类型明确标注
def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    ...