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调用 - 当我的方法返回时？在其他一些依赖项中，有需要执行的生成后代码；我需要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