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


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}
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}
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"}
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]}
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"]
# 错误示例
async def get_db():
return Session()
# 正确写法
async def get_db():
db = Session()
try:
yield db
finally:
db.close()
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"}'
评论
发表评论