由网友(浪)分享简介:FastAPI使用Depends()注入返回或产生的变量。例如,FastAPI/SQL:# Dependencydef get_db():db = SessionLocal()try:yield dbfinally:db.close()...def create_user(db: Session = Depen...
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()
不是正确的方式。有什么想法吗?
推荐答案
您不能将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
相关推荐
最新文章