5

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()不是正确的方法。有任何想法吗？

4

1 回答 1

7

您不能将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
于 2021-03-10T09:00:12.740 回答