FastAPI依赖注入作用域与生命周期控制

鼠扑 · 2025-4-8 00:22:29

title: FastAPI依赖注入作用域与生命周期控制
date: 2025/04/08 00:02:10
updated: 2025/04/08 00:02:10
author: cmdragon
excerpt:
FastAPI框架中，依赖项的作用域决定了它们的创建和销毁时机，重要分为应用级和请求级两种。应用级依赖在整个应用生命周期内只初始化一次，得当长期保持的昂贵资源；请求级依赖在每个HTTP请求时创建新实例，得当必要频仍初始化的资源。通过yield语法可以实现请求级依赖的生命周期控制，确保资源在使用后准确开释。合理划分依赖项作用域和准确控制生命周期，能明显提升应用性能和资源利用率。
categories:

后端开辟
FastAPI

tags:

FastAPI
依赖注入
作用域
生命周期控制
应用级作用域
请求级作用域
资源管理

扫描二维码关注或者微信搜一搜：编程智域前端至全栈交换与成长
探索数千个预构建的 AI 应用，开启你的下一个巨大创意
FastAPI依赖注入作用域与生命周期控制详解

1. 依赖项作用域底子概念

在FastAPI框架中，依赖项的作用域决定了它们的创建和销毁时机。就像图书馆里的公共办法（应用级）与个人借阅的书籍（请求级）的区别，差别作用域的依赖项适用于差别的使用场景。
作用域重要分为两种类型：

应用级作用域（Singleton）：整个应用生命周期内只初始化一次
请求级作用域（Request）：每个HTTP请求都会创建新的实例

from fastapi import Depends, FastAPI
app = FastAPI()
# 应用级依赖示例
class DatabasePool:
def __init__(self):
print("创建数据库连接池")
self.pool = "模拟连接池"
db_pool = DatabasePool()
@app.get("/data")
async def get_data(pool: str = Depends(lambda: db_pool.pool)):
return {"pool": pool}

复制代码

2. 作用域划分实践技巧

2.1 应用级作用域配置

得当必要长期保持的昂贵资源，推荐在应用启动事故中初始化：

from contextlib import asynccontextmanager
from fastapi import FastAPI
@asynccontextmanager
async def lifespan(app: FastAPI):
# 应用启动时初始化
app.state.db_pool = await create_db_pool()
yield
# 应用关闭时清理
await app.state.db_pool.close()
app = FastAPI(lifespan=lifespan)
@app.get("/items")
async def read_items(pool=Depends(lambda: app.state.db_pool)):
return {"pool": pool.status}

复制代码

2.2 请求级作用域实现

使用yield语法实现请求级依赖的生命周期控制：

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
async def get_db():
# 每个请求创建新会话
db_session = AsyncSession(bind=engine)
try:
yield db_session
finally:
# 请求结束后关闭会话
await db_session.close()
@app.post("/users")
async def create_user(
user: UserSchema,
db: AsyncSession = Depends(get_db)
):
db.add(User(**user.dict()))
await db.commit()
return {"status": "created"}

复制代码

3. 生命周期控制模式

3.1 初始化-使用-销毁流程

from typing import Generator
from fastapi import Depends
class FileProcessor:
def __init__(self, filename):
self.file = open(filename, "r")
print(f"打开文件 {filename}")
def process(self):
return self.file.read()
def close(self):
self.file.close()
print("文件已关闭")
def get_processor() -> Generator[FileProcessor, None, None]:
processor = FileProcessor("data.txt")
try:
yield processor
finally:
processor.close()
@app.get("/process")
async def process_file(
processor: FileProcessor = Depends(get_processor)
):
content = processor.process()
return {"content": content[:100]}

复制代码

4. 混合作用域实战案例

组合差别作用域的依赖项实现高效资源管理：

from fastapi import Depends, BackgroundTasks
# 应用级缓存
cache = {}
# 请求级数据库连接
async def get_db():
...
# 缓存依赖（应用级）
def get_cache():
return cache
@app.post("/cached-data")
async def get_data(
db: AsyncSession = Depends(get_db),
cache: dict = Depends(get_cache),
bg: BackgroundTasks = Depends()
):
if "data" not in cache:
result = await db.execute("SELECT ...")
cache["data"] = result
bg.add_task(lambda: cache.pop("data", None), delay=3600)
return cache["data"]

复制代码

5. 课后Quiz

问题1：请求级依赖的yield语句必须放在try/finally块中吗？

答案：不是必须，但推荐使用。finally块确保无论是否发生非常都会执行清算操作，避免资源走漏
问题2：应用级依赖能否访问请求上下文？

答案：不能。应用级依赖在请求上下文创建之前就已经初始化，无法访问请求相干信息
6. 常见报错办理方案

错误1：RuntimeError: Dependency is not yield

缘故起因：在异步依赖项中忘记使用yield语法

# 错误示例
async def get_db():
return Session()
# 正确写法
async def get_db():
db = Session()
try:
yield db
finally:
db.close()

复制代码

错误2：DependencyOveruseWarning

现象：频仍创建昂贵资源导致性能问题
办理：检查依赖项作用域是否合理，将数据库连接池等昂贵资源改为应用级作用域
错误3：ContextVariableNotFound

场景：在应用级依赖中尝试访问请求信息
处理：将必要请求信息的依赖改为请求级作用域，或通过参数转达所需数据
7. 环境配置与运行

安装依赖：

pip install fastapi uvicorn sqlalchemy python-dotenv

复制代码

启动服务：

uvicorn main:app --reload --port 8000

复制代码

测试端点：

curl http://localhost:8000/items
curl -X POST http://localhost:8000/users -H "Content-Type: application/json" -d '{"name":"John"}'

复制代码

		自动登录	找回密码
密码			立即注册

FastAPI依赖注入作用域与生命周期控制

本帖子中包含更多资源

0 个回复

快速回复

楼主热帖

标签云

浏览过的版块