FastAPI 使用 Depends() 注入返回或产生的变量。例如,快速API/SQL https://fastapi.tiangolo.com/tutorial/sql-databases/?h=sqlalchemy#main-fastapi-app:
# 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()
这不是正确的方法。有任何想法吗?
您可以使用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
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)