FastAPI:(8)依赖项


FastAPI:(8)依赖项

概述

文章内容概括

graph TD
    A[依赖项] --> B[函数依赖]
    A --> C[类依赖]
    A --> D[子依赖]
    A --> E[全局依赖]
    A --> F[资源依赖]
    
    B -->|普通函数| G[Depends]
    C -->|封装逻辑| H[自动注入]
    D -->|嵌套调用| I[依赖链]
    E -->|应用范围| J[所有路由]
    F -->|yield语法| K[自动清理]
    
    G -->|支持| L[认证/数据库]
    H -->|用于| M[分页/校验]
    I -->|形成| N[依赖树]
    K -->|管理| O[文件/连接]

1.依赖项

可调用对象

在 FastAPI 中,可调用对象是指可以像函数一样被调用的任何对象。除了普通函数以外,类实现了 __call__ 方法的实例lambda 表达式、functools.partial 返回值等都属于可调用对象。FastAPI 在依赖注入系统和路由处理逻辑中支持将这些可调用对象作为处理函数或依赖项,确保灵活性和高可扩展性。

重要特征:

  • 函数式接口:必须实现可调用接口(例如函数或定义了 __call__ 方法的类实例)。
  • 兼容依赖注入:可被 Depends() 等机制识别并执行。
  • 可状态化:类实例可维护内部状态,实现参数化行为。
  • 可复用与组合:可调用对象可以被多次复用,并组合使用。

图像尺寸校验(正例)

可调用对象

图像处理系统中实现 __call__ 的类对象用于图像尺寸校验
现象:
某图像上传系统要求用户上传的图片必须符合特定的尺寸限制。开发者定义了一个带有 __init__ 参数的类,该类通过 __call__ 实现具体的尺寸校验逻辑。多个接口可共享这一逻辑,仅需以类实例作为依赖传入。

特征对比

  • 函数式接口:通过 __call__ 实现了函数行为
  • 可状态化:类接收 max_size 等参数,灵活性强
  • 兼容注入机制:可直接用于 Depends()
  • 高复用性:多个上传接口复用此逻辑
from fastapi import FastAPI, File, UploadFile, Depends, HTTPException

app = FastAPI()

class SizeValidator:
    def __init__(self, max_size_kb: int):
        self.max_size_kb = max_size_kb

    def __call__(self, file: UploadFile = File(...)):
        if file.spool_max_size > self.max_size_kb * 1024:
            raise HTTPException(status_code=400, detail="File too large")
        return file

@app.post("/upload/")
def upload_file(file: UploadFile = Depends(SizeValidator(max_size_kb=100))):
    return {"filename": file.filename}

functools.partial(正例)

可调用对象

财务审批系统中使用 functools.partial 构造具有固定参数的审计函数

例子描述
在企业审批系统中,多个接口需要进行审计日志记录。日志函数结构相似,但模块名不同。开发者用 functools.partial 固定部分参数(如模块名),生成多个定制化的可调用对象,用于依赖注入,实现代码简洁复用。

特征对比

  • 函数式接口:partial 对象仍可被调用
  • 兼容注入机制:FastAPI 识别并执行 partial 返回对象
  • 可复用与组合:基于相同模板创建多个不同版本函数
  • 简化调用逻辑:无需重复写多个函数
from fastapi import FastAPI, Depends
from functools import partial

app = FastAPI()

def audit_log(module: str, action: str):
    print(f"[Audit] Module: {module}, Action: {action}")
    return {"module": module, "action": action}

# 构造可调用对象
log_sales = partial(audit_log, module="sales")
log_hr = partial(audit_log, module="hr")

@app.get("/sales/")
def sales_view(log=Depends(log_sales)):
    return {"info": "Sales Access", "log": log}

@app.get("/hr/")
def hr_view(log=Depends(log_hr)):
    return {"info": "HR Access", "log": log}

依赖项

FastAPI 中的依赖项是一种声明式机制,用于将某些通用功能(如认证、数据库连接、配置共享等)提取为可复用的组件自动注入到路径操作函数、路径参数、请求体等中。依赖项的核心思想是“依赖注入”,即通过函数参数的方式自动提供某些值或执行某些逻辑,提升模块解耦性、复用性和代码清晰度。

依赖项就是一个函数,且可以使用与「路径操作函数」相同的参数,依赖项函数形式和结构和「路劲操作函数」一样,因此可以把依赖项当作没有「装饰器」(即,没有 @app.get("/some-path") )的路径操作函数。

虽然,在路径操作函数的参数中使用 Depends 的方式与 BodyQuery 相同,但 Depends 的工作方式略有不同。这里只能传给 Depends 一个参数。且该参数必须是「可调用对象」,比如函数。该函数接收的参数和「路径操作函数」的参数一样。

接收到新的请求时,FastAPI 执行如下操作:

  • 用正确的参数调用依赖项函数(「可依赖项」)
  • 获取函数返回的结果
  • 把函数返回的结果赋值给「路径操作函数」的参数

重要特征:

  • 声明式注入:开发者只需在函数参数中声明依赖项,由 FastAPI 自动解析与注入,无需显式调用。
  • 高内聚低耦合:依赖项封装了独立逻辑,使主函数关注其核心职责,提升可维护性与复用性。
  • 支持层层依赖(嵌套):一个依赖项可以依赖其他依赖项,形成依赖树结构。
  • 作用域控制:依赖可以按请求、会话或生命周期进行作用域管理。
  • 与安全机制无缝结合:依赖项是实现认证、授权、权限控制的推荐方式。

用户认证校验函数(正例)

依赖项

现象:
一个 Web 应用的接口需要验证用户身份。开发者将“从请求头中获取并验证 Token”的逻辑封装为一个独立的函数,作为依赖项注入到需要认证的接口中。该函数独立于主业务逻辑,接口函数只需接收“当前用户”对象并执行业务处理。

特征对比

  • 声明式注入:使用 Depends(get_current_user) 方式注入,无需在主函数中显式调用验证逻辑
  • 高内聚低耦合:认证逻辑独立封装,与业务逻辑分离
  • 嵌套依赖支持:Token 验证可能依赖数据库连接或加密函数
  • 安全机制结合:是 FastAPI 安全机制的标准实现方式
from fastapi import FastAPI, Depends, HTTPException, Header

app = FastAPI()

def verify_token(token: str = Header(...)):  # 依赖项函数
    if token != "secret-token":
        raise HTTPException(status_code=403, detail="Invalid token")
    return {"user_id": 1, "role": "admin"}

@app.get("/protected")
def read_protected(current_user: dict = Depends(verify_token)): # 声明依赖项
    return {"message": "Hello, user", "user": current_user}

数据库Session(正例)

依赖项

现象:
在构建 API 时,每次请求都需要连接数据库。开发者将数据库 Session 的创建与关闭逻辑封装为依赖项,并注入到需要数据库访问的路径操作函数中。这确保了每次请求都能自动管理资源,避免连接泄露。

特征对比

  • 声明式注入:路径函数通过 db: Session = Depends(get_db) 获取数据库连接
  • 高内聚低耦合:路径函数只处理业务,不关心连接如何创建
  • 作用域控制:依赖项控制连接生命周期(按请求范围自动关闭)
  • 嵌套依赖支持:数据库连接依赖配置依赖(如 DB URI)
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

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

app = FastAPI()

def get_db(): # 依赖项函数
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/items/")
def read_items(db: Session = Depends(get_db)): # 声明依赖项
    return {"message": "Use db session here"}

手动实现校验(反例)

依赖项

现象:
某接口需要判断用户是否为管理员。开发者在接口函数内部显式写死权限逻辑,如手动提取 Header、验证 Token、查询权限。虽然功能完成,但重复出现于多个接口,难以维护。

特征对比

  • 声明式注入缺失:未使用 Depends 注入,而是手动处理请求内容
  • 高耦合:权限逻辑与业务逻辑混杂,降低代码可读性与复用性
  • 缺乏作用域控制与依赖封装:资源与流程混乱,难以集中管理
  • 无法复用:其他接口无法方便复用该权限逻辑

2.类依赖项

类依赖项

类依赖项是 FastAPI 提供的一种「依赖注入」机制,它允许开发者通过定义类的方式封装多个依赖逻辑。在「路径操作函数」中引入该类时,FastAPI 会自动调用类的 __init__ 方法并注入其参数,之后可以将类的属性用于后续的逻辑处理。它通常用于封装多个参数、共享逻辑或资源管理(如认证、分页、查询参数聚合等)。

重要特征:

  • 特征1:结构化封装
    类依赖项能够将多个逻辑参数或行为整合为一个对象,使代码更具可维护性和模块化。
  • 特征2:自动依赖注入机制
    FastAPI 能够自动解析类构造函数的参数并注入依赖,不需要显式调用。
  • 特征3:支持嵌套依赖
    类依赖项本身可以依赖其他依赖项或类,支持复杂的依赖图构建。
  • 特征4:复用性高
    适合将逻辑独立封装后跨多个路径或服务重用。

分页请求对象封装(正例)

类依赖项

例子描述:
在一个新闻网站的 API 中,使用类依赖项 PaginationParamspagesize 封装为类的两个字段,用于统一处理分页查询请求。

特征对比:

  • ✅ 符合结构化封装:将多个查询参数封装为分页类。
  • ✅ 使用自动依赖注入机制:类初始化由 FastAPI 自动完成。
  • ✅ 支持复用:分页类可用于多个接口,如新闻列表、评论列表等。
from fastapi import FastAPI, Depends, Query

app = FastAPI()

class PaginationParams:
    def __init__(self, page: int = Query(1, ge=1), size: int = Query(10, ge=1, le=100)):
        self.page = page
        self.size = size

@app.get("/items/")
def list_items(pagination: PaginationParams = Depends()):
    start = (pagination.page - 1) * pagination.size
    end = start + pagination.size
    return {"start": start, "end": end, "message": f"Listing items {start} to {end}"}

身份验证逻辑封装(正例)

类依赖项

例子描述:
在一个医疗系统中,将用户身份验证封装为类 AuthUser,在其构造函数中解析请求头中的 token,并解析为用户对象传入接口处理函数。

特征对比:

  • ✅ 结构清晰地封装了身份校验与用户提取逻辑。
  • ✅ 利用了 FastAPI 的自动注入机制从头信息中提取 token。
  • ✅ 高复用性,所有需要身份认证的 API 都可复用该类。
from fastapi import FastAPI, Depends, Header, HTTPException

app = FastAPI()

class AuthUser:
    def __init__(self, token: str = Header(...)):
        if token != "secrettoken123":
            raise HTTPException(status_code=401, detail="Invalid token")
        self.username = "authenticated_user"

@app.get("/secure-data/")
def get_secure_data(user: AuthUser = Depends()):
    return {"message": f"Hello {user.username}, here is your secure data."}

仅用于存储返回数据结构的类(反例)

类依赖项

例子描述:
有开发者将 Pydantic 模型 UserOut 类误用为类依赖项,希望通过它生成响应数据,直接写入依赖中,但这个类并无输入依赖,也不含业务逻辑,仅为结构定义。

特征对比:

  • ❌ 没有体现结构化封装的依赖逻辑:类不包含任何参数逻辑,仅为数据返回结构定义。
  • ❌ 无自动注入输入参数的需要。
  • ❌ 不适合作为依赖复用:它是数据模型,非行为或依赖封装体。

3.子依赖项

子依赖项

指的是在一个依赖项中,再使用其他依赖项作为其参数或组件。FastAPI 会自动解析这些嵌套依赖关系,构建出依赖图,从最底层依赖项开始依次解析。这使得开发者可以将复杂逻辑拆分为多个功能明确、可组合的模块,并形成依赖链进行自动注入。

重要特征:

  • 特征1:嵌套结构
    子依赖项体现为一个依赖项依赖另一个依赖项,支持构建复杂的依赖图。
  • 特征2:自动解析传递链
    FastAPI 会从下至上递归地解析依赖,开发者无需手动传递。
  • 特征3:模块化解耦
    子依赖允许将不同功能分离为独立函数,提高可维护性和可复用性。
  • 特征4:支持缓存与复用
    FastAPI 默认缓存同一请求中相同的依赖项实例,避免重复执行。

数据库访问权限依赖链(正例)

子依赖项

例子描述:
在企业管理系统中,API 先使用子依赖项 get_current_user 从 token 中提取用户,再由 get_db_session(user) 验证该用户是否有访问数据库权限。get_db_session 是子依赖,依赖 get_current_user

特征对比:

  • ✅ 存在明确嵌套结构:数据库访问依赖用户身份。
  • ✅ 使用 FastAPI 的自动解析链机制构建依赖图。
  • ✅ 每个函数模块化解耦,职责单一。
  • ✅ 多接口可复用同一子依赖项链
from fastapi import FastAPI, Depends, Header, HTTPException

app = FastAPI()

def get_current_user(token: str = Header(...)):
    if token != "valid-token":
        raise HTTPException(status_code=401, detail="Unauthorized")
    return {"username": "admin", "role": "manager"}

def get_db_session(user: dict = Depends(get_current_user)):
    if user["role"] != "manager":
        raise HTTPException(status_code=403, detail="Forbidden")
    return "fake_db_session"

@app.get("/data/")
def read_data(db = Depends(get_db_session)):
    return {"db_session": db, "data": "Important data"}
graph TD
    A[read_data] --> B[get_current_user]
    B --> C[get_db_session]
    C --> D[/data/]

参数验证与权限检查分离(正例)

子依赖项

例子描述:
在教育平台中,获取课程内容的接口通过 get_course_id 验证 query 参数是否合法,再通过子依赖 check_user_permission(course_id) 判断当前用户是否能访问该课程。

特征对比:

  • ✅ 明确的功能拆分:一个负责参数校验,一个负责权限验证。
  • ✅ 利用了自动传参机制:无需手动将参数从一个依赖传到另一个。
  • ✅ 各函数职责清晰,高内聚低耦合
  • ✅ 可以将 check_user_permission 复用于多个资源权限场景。
from fastapi import FastAPI, Depends, Query, HTTPException

app = FastAPI()

def get_course_id(course_id: int = Query(..., gt=0)):
    return course_id

def check_user_permission(course_id: int = Depends(get_course_id)):
    if course_id not in [101, 202]:
        raise HTTPException(status_code=403, detail="Access denied to course")
    return f"course_{course_id}"

@app.get("/course/")
def get_course(course = Depends(check_user_permission)):
    return {"message": f"Access granted to {course}"}
graph TD
    A[get_course] --> B[get_course_id]
    B --> C[check_user_permission]
    C --> D[/course/]

手动嵌套调用的依赖函数(反例)

子依赖项

例子描述:
开发者在接口函数中先手动调用 parse_token 获取用户,再显式调用 check_auth(user),而不是通过 FastAPI 的依赖系统注册嵌套。这种方式绕过了依赖图机制,失去自动注入优势。

特征对比:

  • ❌ 没有利用 FastAPI 的自动依赖解析
  • ❌ 不体现嵌套依赖结构,而是显式函数调用。
  • ❌ 模块不可自动组合,失去解耦和缓存复用能力

4.路径装饰器依赖项

路径装饰器依赖项

FastAPI 的路径装饰器(如 @app.get()@app.post() 等)允许通过 dependencies 参数声明全局性或结构性依赖项,这些依赖项不是直接传递给路径操作函数的参数,而是作为执行前的前置逻辑执行。这类依赖项常用于权限校验、日志记录、速率限制等无返回值或非业务输入输出相关的控制性逻辑。就算这些依赖项会返回值,它们的值也不会传递给「路径操作函数」,因此,可以复用在其他位置使用过的、(能返回值的)普通依赖项,即使没有使用这个值,也会执行该依赖项。

重要特征:

  • 特征1:不注入函数参数
    依赖项只在装饰器中声明,不会作为路径操作函数的参数被传入。
  • 特征2:常用于副作用逻辑
    如权限验证、资源控制、速率限制等功能,依赖的副作用(如抛异常)即可影响请求流程。
  • 特征3:适合路由级或全局控制
    用于统一为某些路由加控制逻辑,避免每个接口重复依赖声明。
  • 特征4:执行顺序明确
    依赖函数在路径函数执行前自动触发,且可以层层嵌套其他依赖。

统一身份认证检查(正例)

路径装饰器依赖项

例子描述:
在金融系统的多个 API 路由上使用 dependencies=[Depends(verify_token)] 来统一进行 token 身份验证,verify_token 不传入路径函数,仅用于提前拒绝未授权请求。

特征对比:

  • ✅ 使用 装饰器层级声明,而非函数参数。
  • verify_token 用于副作用控制(异常处理),不返回数据。
  • ✅ 符合控制性逻辑目的,在函数调用前验证合法性。
  • ✅ 提升路由模块结构性与安全一致性。
from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI()

async def verify_token(x_token: str = Header()):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid") # 路径装饰器依赖项与正常的依赖项一样,可以 `raise` 异常

async def verify_key(x_key: str = Header()):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid") # 路径装饰器依赖项与正常的依赖项一样,可以 `raise` 异常
    return x_key # 复用在其他位置使用过的、(能返回值的)普通依赖项,即使没有使用这个值,也会执行该依赖项

@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
    return [{"item": "Foo"}, {"item": "Bar"}]

请求记录审计日志(正例)

路径装饰器依赖项

例子描述:
在医疗系统的敏感数据访问路由中使用 dependencies=[Depends(log_access)] 来记录访问日志,log_access 会记录用户与时间,但不传给接口处理逻辑。

特征对比:

  • log_access 不参与路径函数输入。
  • ✅ 用于副作用性审计行为,如记录日志。
  • ✅ 在多个路径下通用复用,减少重复性代码。
  • ✅ 明确分离业务逻辑与审计逻辑。
from fastapi import FastAPI, Depends, Request

app = FastAPI()

def log_access(request: Request):
    print(f"User accessed {request.url} from {request.client.host}")

@app.get("/medical/record", dependencies=[Depends(log_access)])
def get_medical_record():
    return {"record": "patient MRI data"}

输入参数错的装饰器(反例)

路径装饰器依赖项

例子描述:
在电商系统的商品接口中,将 get_product_id(用于校验并返回 product_id)放入装饰器依赖项中,结果路径函数内获取不到 product_id,导致业务无法继续执行。

特征对比:

  • ❌ 违反不注入参数原则:需要的值未传入函数参数中。
  • ❌ 功能不应是副作用控制而是业务数据传递。
  • ❌ 依赖应在函数参数中显式声明而非装饰器中隐藏,导致数据丢失。
  • ❌ 与“结构性逻辑分离”的设计目标相悖。

5.全局依赖项

全局依赖项

全局依赖项是 FastAPI 中通过在 FastAPI() 实例化时使用 dependencies 参数设置的依赖项。它们会自动作用于应用中的所有路径操作函数,无需在每个路由或函数中重复声明。这种机制通常用于系统范围内的前置操作,如统一身份验证、全局日志记录、IP 白名单检查等。

重要特征:

  • 特征1:全局适用性
    一次声明,应用中所有路由都将执行该依赖项。
  • 特征2:主要用于非返回值的副作用逻辑
    如权限验证、行为记录、速率限制等控制类操作。
  • 特征3:不注入到函数参数
    与装饰器依赖类似,不将值传递给路由函数,仅做控制或检查。
  • 特征4:配置入口在 FastAPI() 初始化处
    与局部路径依赖不同,配置统一且不依赖于具体模块定义。

统一访问令牌验证(正例)

全局依赖项

例子描述:
在一个 SaaS 平台中,每个请求都需验证 API token,开发者将验证逻辑作为全局依赖项添加到 FastAPI(dependencies=[Depends(check_token)]),保证所有请求均自动进行验证。

特征对比:

  • ✅ 使用 FastAPI 构造函数添加依赖项
  • 不传入函数参数,只通过副作用控制访问权限。
  • ✅ 体现全局一致性,无需逐路由声明。
  • ✅ 可跨模块复用,维护性强。
from fastapi import FastAPI, Depends, Header, HTTPException

def check_token(token: str = Header(...)):
    if token != "valid-api-token":
        raise HTTPException(status_code=401, detail="Invalid token")

app = FastAPI(dependencies=[Depends(check_token)]) # 全局注入依赖项

@app.get("/items/")
def read_items():
    return {"message": "Token verified, items listed."}

@app.post("/submit/")
def submit_data():
    return {"message": "Data submitted with valid token."}

访问日志系统监控(正例)

全局依赖项

例子描述:
一个 IoT 平台需要记录所有接口访问行为用于安全分析。开发者将 log_request_info(request) 添加为全局依赖,以在每次请求中记录 URL 和访问 IP。

特征对比:

  • ✅ 使用全局依赖配置日志逻辑,无需手动在每个接口声明。
  • ✅ 属于副作用操作,无需返回值或参数注入。
  • ✅ 与路径函数完全解耦,便于维护。
  • ✅ 执行时机统一,符合安全审计需求。
from fastapi import FastAPI, Depends, Request

def log_request_info(request: Request):
    print(f"[LOG] Accessed URL: {request.url} from {request.client.host}")

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

@app.get("/sensor/data")
def get_sensor_data():
    return {"sensor": "temperature", "value": 23.5}

@app.get("/device/status")
def get_device_status():
    return {"device": "sensor-1", "status": "online"}

6.yield依赖项

yield依赖项

FastAPI 支持使用 yield 语法定义依赖项,在请求开始时执行 yield 前的逻辑,在请求处理完后自动执行 yield 后的清理逻辑。这种模式类似 Python 上下文管理器,常用于创建并自动释放资源,如数据库连接、文件句柄、会话对象等。FastAPI 会在请求生命周期内自动管理 yield 的调用与清理。

重要特征:

  • 特征1:用于资源管理型依赖
    可在依赖项内部创建资源,在请求处理后自动释放,如数据库会话、网络连接等。
  • 特征2:具有生命周期感知能力
    yield 前的逻辑在请求开始时执行,yield 后的逻辑在请求结束时执行,支持清理操作。
  • 特征3:可返回值给路径函数使用
    yield 可以产生值供路径操作函数使用。
  • 特征4:仅支持 Depends(),不支持 dependencies=[Depends(...)]
    因为清理逻辑只有在有“返回值”的依赖中才能正确执行。

数据库会话的自动释放(正例)

yield依赖项

例子描述:
在 Web 应用中,使用 yield 创建数据库 Session,并在请求完成后自动关闭它,避免连接泄露。路径函数通过依赖获取数据库会话对象。

特征对比:

  • ✅ 明确体现资源创建(会话)和释放(close)
  • ✅ 返回资源供业务使用(路径函数获取 session)
  • ✅ 生命周期由 FastAPI 管理,自动清理
  • ✅ 避免重复代码和资源泄漏
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

app = FastAPI()

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

@app.get("/users/")
def read_users(db: Session = Depends(get_db)):
    return db.execute("SELECT 'user1' AS name").fetchall()

文件资源读取与自动关闭(正例)

yield依赖项

例子描述:
在大数据接口中,用 yield 打开一个只读文件作为数据源,在返回数据后自动关闭文件句柄,避免文件占用。

特征对比:

  • ✅ 创建文件对象供业务使用(yield file
  • ✅ 在请求处理完成后关闭资源
  • ✅ 支持大数据文件按需打开与关闭
  • ✅ 保持系统资源可控和安全
from fastapi import FastAPI, Depends

app = FastAPI()

def get_file():
    f = open("data.txt", "r")
    try:
        yield f
    finally:
        f.close()

@app.get("/data")
def read_data(file = Depends(get_file)):
    return {"lines": file.readlines()}

副作用控制误用yield(反例)

yield依赖项

例子描述:
在权限验证中使用 yield 返回 True,而不是直接抛异常或执行副作用逻辑。这样使得本应是检查行为变成了资源依赖,导致逻辑不清晰,也无任何清理行为。

特征对比:

  • ❌ 无资源需要释放,不应使用 yield
  • ❌ 不具备生命周期管理的必要性
  • yield 后代码形同多余,违背设计目的
  • ❌ 模糊副作用逻辑与资源依赖逻辑的边界
from fastapi import FastAPI, Depends, HTTPException

app = FastAPI()

# 错误示范:使用 yield 进行权限验证
def check_admin():
    # 模拟权限检查逻辑
    user_is_admin = False  # 假设当前用户不是管理员
    if not user_is_admin:
        raise HTTPException(status_code=403, detail="Not authorized")
    
    # 没有真正资源返回,但仍使用 yield —— 这是误用
    yield True

    # 这段清理逻辑实际上无任何意义
    print("This cleanup is meaningless")

@app.get("/admin")
def get_admin_data(allowed: bool = Depends(check_admin)):
    return {"data": "secret admin info"}

yield子依赖项

yield 子依赖项指的是一个依赖项使用 yield 语法定义资源清理逻辑,并且被其他依赖项间接调用,即作为“子依赖项”存在。FastAPI 会递归解析依赖树,确保在整个请求生命周期中,即使 yield 发生在子依赖中,其清理逻辑仍然会被执行。这使得在复杂资源链或中间逻辑中能进行自动清理。

重要特征:

  • 特征1:清理逻辑位于子层但自动执行
    即使 yield 定义在子依赖中,FastAPI 会在请求完成后自动调用其 finally 部分。
  • 特征2:支持依赖链中多级嵌套调用
    yield 可以在依赖链中任何一层安全使用,不必直接绑定在路径函数上。
  • 特征3:通常用于资源初始化/封装结构
    比如数据库连接 → service 封装 → handler,yield 只需在最底层。
  • 特征4:不应将纯逻辑验证误用为 yield 子依赖
    因为没有资源需要清理,会增加无意义的生命周期负担。

三个yield子依赖(正例)

声明任意数量和层级的树状依赖,而且它们中的任何一个或所有的都可以使用 yieldFastAPI 会确保每个带有 yield 的依赖中的"退出代码"按正确顺序运行。例如,dependency_c 可以依赖于 dependency_b,而 dependency_b 则依赖于 dependency_a

在这种情况下,dependency_c 在执行其退出代码时需要 dependency_b(此处称为 dep_b)的值仍然可用。而 dependency_b 反过来则需要 dependency_a(此处称为 dep_a )的值在其退出代码中可用。

from typing import Annotated

from fastapi import Depends


async def dependency_a():
    dep_a = generate_dep_a()
    try:
        yield dep_a
    finally:
        dep_a.close()


async def dependency_b(dep_a: Annotated[DepA, Depends(dependency_a)]):
    dep_b = generate_dep_b()
    try:
        yield dep_b
    finally:
        dep_b.close(dep_a)


async def dependency_c(dep_b: Annotated[DepB, Depends(dependency_b)]):
    dep_c = generate_dep_c()
    try:
        yield dep_c
    finally:
        dep_c.close(dep_b)

数据库会话在子依赖中清理(正例)

例子描述:
在服务层中封装业务逻辑,这一服务依赖数据库会话。数据库连接通过 yield 创建并在子依赖中释放,路径函数只接收 service 实例,间接完成数据库的生命周期管理。

特征对比:

  • yield 在子层完成资源创建与清理
  • ✅ 清理逻辑被 FastAPI 自动追踪执行
  • ✅ 路径函数无感知底层依赖细节
  • ✅ 结构清晰,符合资源生命周期原则
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine)

app = FastAPI()

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

# 子依赖:业务服务层
class UserService:
    def __init__(self, db: Session):
        self.db = db

    def get_users(self):
        return self.db.execute("SELECT 'user1' AS name").fetchall()

def get_user_service(db: Session = Depends(get_db)):
    return UserService(db)

# 路径函数:只依赖 service,db 在子层清理
@app.get("/users")
def read_users(service: UserService = Depends(get_user_service)):
    return service.get_users()
graph TD
    A[read_users 路径函数] --> B[get_user_service]
    B --> C[get_db]
    C --> D[(yield 创建 DB 会话)]
    C --> E[(finally 关闭 DB 会话)]

文件句柄子依赖于数据处理器(正例)

例子描述:
一个文件打开器通过 yield 提供文件对象,中间逻辑处理数据并构建 handler,路径函数只依赖 handler。文件资源作为子依赖项在请求结束时关闭。

特征对比:

  • ✅ 文件资源在子层通过 yield 创建
  • ✅ 清理在路径函数外完成,无需显式控制
  • ✅ 依赖链嵌套但依然保持清理语义明确
from fastapi import FastAPI, Depends

app = FastAPI()

# yield 子依赖:打开文件
def open_file():
    f = open("data.txt", "r")
    try:
        yield f
    finally:
        f.close()

# 子依赖:数据处理逻辑
class FileProcessor:
    def __init__(self, file):
        self.lines = file.readlines()

    def get_summary(self):
        return {"line_count": len(self.lines)}

def get_processor(file=Depends(open_file)):
    return FileProcessor(file)

@app.get("/file-summary")
def read_file_summary(processor: FileProcessor = Depends(get_processor)):
    return processor.get_summary()
graph TD
    A[read_file_summary 路径函数] --> B[get_processor]
    B --> C[open_file]
    C --> D[(yield 打开文件)]
    C --> E[(finally 关闭文件)]

错误的yield

例子描述:
权限判断函数通过 yield 返回布尔值并作为子依赖,实际无资源需要清理,只用于抛出权限异常。由于无清理行为,使用 yield 仅是误用,造成依赖生命周期语义混乱。

特征对比:

  • ❌ 无资源分配,不符合 yield 场景
  • ❌ 清理逻辑无意义
  • ❌ 增加不必要的依赖复杂性
  • ❌ 模糊了依赖职责边界
from fastapi import FastAPI, Depends, HTTPException

app = FastAPI()

# 错误的子依赖:用 yield 模拟权限逻辑(无清理需要)
def check_permission():
    authorized = False
    if not authorized:
        raise HTTPException(status_code=403, detail="Not authorized")
    yield True  # 无资源,仍用了 yield
    print("no-op cleanup")  # 实际无用

# 错误:将上面逻辑当作子依赖
class ProtectedHandler:
    def __init__(self):
        self.status = "ok"

def get_handler(allowed: bool = Depends(check_permission)):
    return ProtectedHandler()

@app.get("/secure-data")
def secure_data(handler: ProtectedHandler = Depends(get_handler)):
    return {"status": handler.status}

包含ield与HTTPException依赖项

可以在 yield 之后的退出代码中抛出一个 HTTPException 或类似的异常。

from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()


data = {
    "plumbus": {"description": "Freshly pickled plumbus", "owner": "Morty"},
    "portal-gun": {"description": "Gun to create portals", "owner": "Rick"},
}


class OwnerError(Exception):
    pass


def get_username(): # 使用HTTPException抛出异常
    try:
        yield "Rick"
    except OwnerError as e:
        raise HTTPException(status_code=400, detail=f"Owner error: {e}")


@app.get("/items/{item_id}")
def get_item(item_id: str, username: Annotated[str, Depends(get_username)]):
    if item_id not in data:
        raise HTTPException(status_code=404, detail="Item not found")
    item = data[item_id]
    if item["owner"] != username:
        raise OwnerError(username)
    return item

包含yield和except的依赖项

如果在包含 yield 的依赖项中使用 except 捕获了一个异常,然后你没有重新抛出该异常(或抛出一个新异常),与在普通的Python代码中相同,FastAPI不会注意到发生了异常。
所以, 在包含 yield 和 except 的依赖项中一定要 raise,不然客户端将会收到 HTTP 500 Internal Server Error 的响应,因为我们没有抛出 HTTPException 或者类似的异常,并且服务器也 不会有任何日志 或者其他提示来告诉我们错误是什么。

from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()


class InternalError(Exception):
    pass


def get_username():
    try:
        yield "Rick"
    except InternalError: # 自定义报错
        print("We don't swallow the internal error here, we raise again 😎")
        raise # 使用raise来发出错误


@app.get("/items/{item_id}")
def get_item(item_id: str, username: Annotated[str, Depends(get_username)]):
    if item_id == "portal-gun":
        raise InternalError(
            f"The portal gun is too dangerous to be owned by {username}"
        )
    if item_id != "plumbus":
        raise HTTPException(
            status_code=404, detail="Item not found, there's only a plumbus here"
        )
    return item_id

上下文管理器

「上下文管理器」是一种用于资源管理的编程结构,通常结合 Python 的 with 语句或 @contextmanager 装饰器(或 asynccontextmanager),确保在资源使用完成后,自动执行清理操作。在依赖项(Dependency)中使用上下文管理器,FastAPI 会自动识别并正确执行资源初始化和释放,适用于数据库连接、文件操作、网络连接等场景。

重要特征

  • 特征1:结构上使用 withyield + finally,确保进入/退出有界定
  • 特征2:FastAPI 可以与 @contextmanager@asynccontextmanager 一起使用依赖项,实现资源自动管理
  • 特征3:适用于需要显式释放资源的任务,例如关闭连接、清除缓存、释放文件锁
  • 特征4:保持依赖项结构清晰,职责分离(资源获取 vs 使用)
  • 特征5:FastAPI 会在请求生命周期结束后自动处理上下文退出逻辑

数据库连接(正例)

上下文管理器

例子描述
在一个金融服务应用中,用户请求必须获取数据库会话才能执行账户余额查询。开发者使用 @contextmanager 创建了一个 get_db() 函数,封装数据库连接和关闭逻辑,并将其作为依赖项注入给业务层查询逻辑。

特征对比

  • ✅ 使用了 @contextmanager 创建明确的资源生命周期
  • ✅ FastAPI 自动处理请求后关闭连接
  • ✅ 业务逻辑与资源管理解耦
  • ✅ 满足资源安全释放的关键需
from fastapi import FastAPI, Depends
from contextlib import contextmanager

app = FastAPI()

@contextmanager
def get_db():
    db = {"conn": "数据库连接"}
    try:
        yield db
    finally:
        print("关闭数据库连接")

@app.get("/users/")
def read_users(db=Depends(get_db)):
    return {"msg": "读取成功", "db": db}

临时配置文件(正例)

上下文管理器

例子描述
在一款图像处理服务中,每次上传图像时,系统会临时生成一个配置文件用于处理参数。该文件在处理完成后应立即删除。开发者使用 @contextmanager 实现 temp_config(),用于生成和清理配置文件。

特征对比

  • ✅ 控制文件生命周期,资源释放及时
  • ✅ 使用 with 结构封装文件创建和删除
  • ✅ 应用内只需关注业务处理,清理自动完成
from fastapi import FastAPI, Depends
from contextlib import contextmanager
import tempfile
import os

app = FastAPI()

@contextmanager
def temp_config():
    path = tempfile.NamedTemporaryFile(delete=False).name
    try:
        with open(path, 'w') as f:
            f.write("临时配置参数")
        yield path
    finally:
        os.remove(path)

@app.get("/process/")
def process_image(config_path=Depends(temp_config)):
    return {"config_path": config_path}

未封装资源释放的临时文件(反例)

上下文管理器

例子描述
在一个 NLP 训练接口中,上传的语料数据保存为临时文件,供模型读取。然而,开发者直接在路径函数中打开文件,并未封装为上下文管理器,也未在结束时删除或关闭资源,导致多个请求后服务器空间被占满。

特征对比

  • ❌ 没有定义上下文管理结构,没有资源清理保证
  • ❌ 文件打开后未关闭,资源泄漏
  • ❌ 不具备自动退出机制
  • ❌ 依赖和处理逻辑耦合,代码可维护性差
from fastapi import FastAPI
import tempfile

app = FastAPI()

@app.get("/upload/")
def upload_file():
    path = tempfile.NamedTemporaryFile(delete=False).name
    with open(path, 'w') as f:
        f.write("用户数据")
    # ❌ 文件未删除,也未用上下文管理
    return {"temp_file": path}

文章作者: Hkini
版权声明: 本博客所有文章除特別声明外,均采用 CC BY 4.0 许可协议。转载请注明来源 Hkini !
评论
  目录