FastAPI 依赖项（yield）：如何手动调用它们？

Posted 2023-03-08

【中文标题】FastAPI 依赖项（yield）：如何手动调用它们？

【英文标题】：FastAPI dependencies (yield): how to call them manually?

【发布时间】：2021-06-08 01:18:14

【问题描述】：

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 - 当我的方法返回时？在其他一些依赖项中，有需要执行的后生成代码；我需要再次调用next() 以确保代码执行吗？似乎next() 不是正确的方法。有什么想法吗？

【参考方案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

【讨论】：

喂，我从来没有想到过 - 很高兴我问了！谢谢