FastAPI 依赖项产量：如何手动调用它们？

问题描述

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

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

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

with get_db() as db:
   pass

由于原始 get_db() 未包装为 @contextmanager 而失败。 （注意，我不想装饰它 - 我以 get_db 为例，我需要处理更复杂的依赖项）。最后，我尝试了 db = next(get_db()) - 它有效，但我认为这不是正确的解决方案。何时/如何调用 finally - 当我的方法返回时？在其他一些依赖项中，有需要执行的 post-yield 代码；我是否需要再次调用 next() 以确保代码执行？似乎 next() 不是正确的方法。有什么想法吗？

解决方法

您不能将 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