依赖注入与路由分组

写给初学者:每个概念都从”为什么需要它”开始讲,配合通俗比喻和完整可运行代码。


一、依赖注入(Dependencies)

1.1 为什么需要依赖注入?

假设你有 10 个接口都需要校验用户身份,每个接口都写一遍校验逻辑:

 
# 假设你有 10 个接口,每个都要写一遍,改起来要改 10 处——很容易漏
 
@app.get("/users")
 
async def get_users(token: str = Header(...)):
 
    # Header(...) 表示从请求头取 token 字段
 
    user = verify_token(token)   # 校验 token 的逻辑
 
    if not user:                 # token 无效
 
        raise HTTPException(status_code=401)  # 返回 401 未授权
 
    return ...                   # token 有效,继续处理
 
@app.get("/items")
 
async def get_items(token: str = Header(...)):
 
    user = verify_token(token)   # 又写一遍!和上面一模一样
 
    if not user:
 
        raise HTTPException(status_code=401)  # 又写一遍!
 
    return ...
 
# 如果有 10 个接口,就要写 10 遍——这就是"重复代码"的问题
 

依赖注入的作用:把公共逻辑抽出来,声明”我需要什么”,FastAPI 自动帮你注入。

1.2 基本用法

 
from fastapi import FastAPI, Depends, HTTPException, Header
 
app = FastAPI()
 
# 它不是接口,不能直接访问,而是被接口"依赖"调用
 
async def get_current_user(token: str = Header(...)):
 
    # Header(...) 表示从请求头里取 token 字段
 
    # ... 表示这个字段是必填的,不传就报 422 错误
 
    """校验 token,返回当前用户"""
 
    if token != "secret":
 
        # raise HTTPException 会直接中断请求,返回 401 错误给客户端
 
        # 接口函数不会执行
 
        raise HTTPException(status_code=401, detail="无效的 token")
 
    # 校验通过,返回用户信息(这个返回值会传给接口函数的参数)
 
    return {"username": "张三", "role": "admin"}
 
# get_current_user 的返回值会作为 current_user 参数传进来
 
@app.get("/users")
 
async def get_users(current_user: dict = Depends(get_current_user)):
 
    # 到这里时,current_user 一定是合法用户(不合法的已经在上面被 401 拦截了)
 
    return {"msg": f"你好, {current_user['username']}", "user": current_user}
 
@app.get("/items")
 
async def get_items(current_user: dict = Depends(get_current_user)):
 
    # 不用再写一遍校验逻辑!依赖函数已经帮你做了
 
    return {"msg": "这是你的商品列表", "owner": current_user["username"]}
 

发生了什么?

  1. 请求进来时,FastAPI 先调用 get_current_user

  2. get_current_user 返回的结果,作为 current_user 参数传入你的函数

  3. 如果 get_current_user 抛了 HTTPException,请求直接中断

通俗比喻:依赖注入就像餐厅的”食材采购”——厨师(接口函数)说”我需要鸡蛋”(Depends(get_egg)),采购员(FastAPI)自动把鸡蛋买好放桌上,厨师直接用就行。

1.3 带参数的依赖

依赖函数也可以有自己的参数:

 
# 这个依赖的作用:通用的"查不到就 404"逻辑,任何需要查商品的接口都能复用
 
async def get_item_or_404(item_id: int):
 
    """通用依赖:查不到就 404"""
 
    items = {1: "键盘", 2: "鼠标", 3: "显示器"}  # 模拟数据库
 
    if item_id not in items:
 
        # 商品不存在 → 抛 404,请求到此结束
 
        raise HTTPException(status_code=404, detail=f"商品 {item_id} 不存在")
 
    return items[item_id]  # 商品存在 → 返回商品名
 
@app.get("/items/{item_id}")
 
async def read_item(item_name: str = Depends(get_item_or_404)):
 
    return {"item": item_name}
 

1.4 类作为依赖

除了函数,类也可以当依赖:

 
# 类也可以当依赖!好处是可以把多个参数封装成一个对象
 
class Pagination:
 
    """分页参数类"""
 
    def __init__(self, skip: int = 0, limit: int = 10):
 
        self.skip = skip
 
        self.limit = limit
 
@app.get("/users")
 
async def list_users(pagination: Pagination = Depends()):
 
    # Depends() 里不写类名!FastAPI 会从参数类型注解 Pagination 自动推断
 
    return {
 
        "skip": pagination.skip,
 
        "limit": pagination.limit,
 
        "msg": f"返回 {pagination.limit} 条,跳过前 {pagination.skip} 条"
 
    }
 

1.5 多层依赖嵌套

依赖函数里还可以用其他依赖:

 
# ==================== 第一层依赖:校验 token ====================
 
async def get_current_user(token: str = Header(...)):
 
    if token != "secret":
 
        raise HTTPException(status_code=401)
 
    return {"username": "张三", "role": "admin"}
 
# ==================== 第二层依赖:检查权限 ====================
 
async def require_admin(user: dict = Depends(get_current_user)):
 
    """只有管理员才能访问"""
 
    if user["role"] != "admin":
 
        raise HTTPException(status_code=403, detail="需要管理员权限")
 
    return user
 
# ==================== 接口:依赖第二层 ====================
 
@app.get("/admin/dashboard")
 
async def admin_dashboard(admin: dict = Depends(require_admin)):
 
    return {"msg": f"欢迎进入管理后台, {admin['username']}"}
 

401 vs 403 区别:401 是”我不认识你”(未认证),403 是”我认识你,但你没权限”(未授权)。

1.6 yield 依赖 —— 带”收尾”的依赖

有时候依赖函数需要”做完之后清理一下”,比如数据库连接用完要关闭。这时候用 yield 代替 return

 
async def get_db():
 
    db = connect_to_db()   # 第一步:创建连接(请求进来时执行)
 
    yield db               # 第二步:把连接"借"出去给接口用
 
    db.close()             # 第三步:关闭连接(响应返回后执行)
 

yield 的执行时机

  1. yield 之前的代码 → 请求进来时执行(准备资源)

  2. yield 把值传出去 → 接口函数拿到这个值

  3. yield 之后的代码 → 响应返回后执行(清理资源)

正确写法:用 try...finally

 
async def get_db():
 
    db = connect_to_db()   # ① 创建连接
 
    try:
 
        yield db           # ② 把连接借出去给接口用
 
    finally:
 
        db.close()         # ③ 不管有没有异常,一定关闭连接
 

二、路由分组(APIRouter)

2.1 为什么需要路由分组?

假设你在做一个用户管理系统,有这些接口:

 
GET    /users          获取用户列表
 
GET    /users/{id}     获取单个用户
 
POST   /users          创建用户
 
PUT    /users/{id}     更新用户
 
DELETE /users/{id}     删除用户
 
GET    /items          获取商品列表
 
POST   /items          创建商品
 

如果全部写在一个 main.py 里,文件会越来越大。APIRouter 让你按功能拆分成多个文件。

2.2 基本用法

第一步:创建 router 文件

 
# routers/users.py —— 用户相关的接口都放这里
 
from fastapi import APIRouter
 
router = APIRouter()
 
@router.get("/")
 
async def list_users():
 
    """获取用户列表"""
 
    return [{"username": "张三"}, {"username": "李四"}]
 
@router.get("/{user_id}")
 
async def get_user(user_id: int):
 
    """获取单个用户"""
 
    return {"username": "张三", "id": user_id}
 
@router.post("/")
 
async def create_user(username: str):
 
    """创建用户"""
 
    return {"msg": f"用户 {username} 创建成功"}
 

第二步:在主文件里注册 router

 
# main.py —— 主文件,负责组装各个 router
 
from fastapi import FastAPI
 
from routers.users import router as users_router
 
app = FastAPI()
 
# tags=["用户"] 表示在 API 文档里归类到"用户"分组
 
app.include_router(users_router, prefix="/users", tags=["用户"])
 

访问路径对照

routers/users.py 里写的实际访问路径说明
@router.get("/")GET /usersprefix 自动拼在前面
@router.get("/{user_id}")GET /users/123路径参数也正常工作
@router.post("/")POST /usersPOST 请求也一样

2.3 完整项目结构

一个典型的 FastAPI 项目长这样:


graph TD

    root["my_project/"] --> main["main.py<br/>主文件:创建 app,注册所有 router"]

    root --> routers["routers/"]

    root --> deps["dependencies.py<br/>公共依赖函数"]

    root --> models["models.py<br/>数据模型(Pydantic)"]

    root --> db["database.py<br/>数据库连接"]

    routers --> init["__init__.py<br/>空文件,让 Python 把 routers 当作包"]

    routers --> users["users.py<br/>用户相关接口"]

    routers --> items["items.py<br/>商品相关接口"]

    routers --> admin["admin.py<br/>管理后台接口"]

2.4 APIRouter 的常用参数

 
from fastapi import APIRouter, Depends
 
router = APIRouter(
 
    prefix="/users",      # 统一前缀
 
    tags=["用户"],         # API 文档标签
 
    dependencies=[        # 公共依赖
 
        Depends(verify_token)
 
    ],
 
    responses={           # 公共响应描述(给 API 文档看的)
 
        404: {"description": "未找到"},
 
        401: {"description": "未授权"},
 
    },
 
)
 

推荐参数写在 router 文件里

  • 路由的配置跟路由定义放在一起,看 router 文件就知道完整情况

  • main.py 保持简洁,只负责组装


速查表

概念一句话解释关键代码
Depends声明依赖,FastAPI 自动注入def f(user = Depends(get_user))
依赖函数抽出来的公共逻辑async def get_user(token): ...
类依赖用类当依赖,自动解析查询参数pagination: Pagination = Depends()
多层依赖依赖里再依赖require_admin 依赖 get_current_user
yield 依赖带资源清理的依赖yield db + try...finally
APIRouter把接口按功能拆分成多个文件router = APIRouter()
include_router在主文件里注册 routerapp.include_router(router, prefix="/users")
prefix给 router 的所有接口加统一前缀prefix="/api/v1/users"
tagsAPI 文档里的分组标签tags=["用户"]

▶ 对应原理:02-依赖注入Depends原理与生命周期

速记卡(面试闪卡)

Q1:一句话讲清「依赖注入与路由分组」到底是什么?

A:依赖注入用 Depends 把鉴权/分页等公共逻辑抽出来自动注入;APIRouter 把接口按功能拆文件再组装。

Q2:为什么需要依赖注入 —— 怎么理解?

A:10 个接口都要校验身份,手写 10 遍 verify_token,改要改 10 处极易漏。DI 把公共逻辑抽成 get_current_user,路由里写 current_user: dict = Depends(get_current_user),FastAPI 请求进来先调依赖、返回值塞进参数,抛异常直接中断。就像厨师说”我需要鸡蛋”,采购员自动买好放桌上。这叫 dependency injection(依赖注入)。

Q3:yield 依赖与多层嵌套 —— 怎么理解?

A:依赖要”用完清理”就用 yield 代替 return:yield db 前备资源、后收摊,包 try…finally 保证异常也关。依赖还能嵌套——require_admin 依赖 get_current_user 搭出权限链,分别抛 401(不认识你)和 403(没权限)。这叫 yield dependency(带生命周期的依赖)。

Q4:APIRouter 路由分组 —— 怎么理解?

A:项目变大全写 main.py 会爆炸。APIRouter 让你在 routers/users.py 里 router = APIRouter() 定义接口,主文件 app.include_router(users_router, prefix="/users", tags=["用户"]) 注册并加统一前缀。还能给整组加 dependencies 公共依赖。这叫 APIRouter(路由分组)+ include_router(注册组装)。

Q5:依赖注入 vs 中间件 —— 怎么理解?

A:依赖注入管”路由专属资源”(每个接口各自要的:当前用户、分页、DB 连接),生命周期随请求进出;中间件管”全局横切”(日志、CORS、鉴权统一套在所有请求上)。职责分离:局部复用用 DI,全局一刀切用中间件。这叫 middleware(中间件)与 DI 的分工。

Q6:核心速记主线有哪些?

  • DI:Depends 抽公共逻辑自动注入,抛异常即中断

  • yield 依赖:前备料后收摊;多层嵌套搭权限链(401/403)

  • APIRouter:按功能拆文件,include_router + prefix/tags 组装

  • 分工:DI 管路由专属资源,中间件管全局横切

口诀

A:公共逻辑用注入,

Depends 声明框架顾;

yield 收尾链路固,

路由分组前缀布。

相关链接


技术学习路线图 > 工程化篇

相关链接