FastAPI依赖关系(yield):如何手动调用它们?

Question

FastAPI使用Depends()注入返回或产生的变量。例如,FastAPI/SQL:

# Dependency
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
...
def create_user(db: Session = Depends(get_db)):
...

如果我想在其他地方使用get_db()(在FastAPI路由之外)，我该怎么做?我知道这是Python的核心知识，但我似乎搞不懂。我最初的想法是db = yield from get_db()，但我不能在异步函数中调用yield from(并且不知道它是否会工作)。然后我试了:

with get_db() as db:
pass

失败，因为原来的get_db()没有包装为@contextmanager。(注意，我不想装饰这个-我使用get_db作为一个例子，我需要与更复杂的依赖关系工作)。最后，我尝试了db = next(get_db())-这是有效的，但我不认为这是正确的解决方案。何时/如何将finally被调用-当我的方法返回?在其他依赖项中，需要执行yield后代码;我需要再次调用next()来确保代码执行吗?似乎next()不是正确的方式。什么好主意吗?

Answer 1

您可以使用contextmanager不作为装饰器，而是作为返回上下文管理器的函数:

from contextlib import contextmanager
# Dependency
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()

# synchronously
with contextmanager(get_db)() as session:  # execute until yield. Session is yielded value
pass
# execute finally on exit from with

但是请记住，代码将同步执行。如果你想在线程中执行它，那么你可以使用FastAPI工具:

import asyncio
from contextlib import contextmanager
from fastapi.concurrency import contextmanager_in_threadpool

async def some_coro():
async with contextmanager_in_threadpool(contextmanager(get_db)()) as session:
pass

Answer 2

如果尝试在此上下文中使用next(get_db())，可能会遇到以下问题:

没有数据库会话:get_db()函数可能不会返回一个有效的会话，因为它依赖于FastAPI的依赖注入系统。

Session Leaks:如果忘记正确关闭Session，在请求上下文之外手动管理数据库Session的生命周期会导致资源泄漏。

线程安全:根据FastAPI应用的部署方式(例如，使用多个工作进程或线程)，手动管理会话可能不是线程安全的。

如果你需要在FastAPI路由处理程序或中间件之外执行数据库操作(例如，在脚本或后台任务中)，最好显式地创建和管理数据库会话，而不依赖于FastAPI的依赖机制。您可以使用SQLAlchemy中的sessionmaker来创建会话并根据需要处理它们的生命周期:

from sqlalchemy.orm import sessionmaker
SessionLocal =    sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Usage example
def some_function():
# Create a session
session = SessionLocal()
try:
# Perform database operations     using the session
user = session.query(User).filter_by(username="example_user").first()
# Other database operations here

# Commit the transaction (if applicable)
session.commit()
except Exception as e:
# Handle exceptions, rollback on error, and/or log
session.rollback()
finally:
# Close the session
session.close()

调用函数执行数据库操作some_function ()