代码之家  ›  专栏  ›  技术社区  ›  STOPIMACODER

FastAPI:路由外的后台函数表现不同?

  •  1
  • STOPIMACODER  · 技术社区  · 2 年前

    在FastAPI中,我试图理解为什么后台实例没有在路由函数处理程序之外创建,以及它的不同行为。

    示例:

    按预期工作的标准文档示例:

    @app.get('/')
    async def index(background_tasks: BackgroundTasks):
        background_tasks.add_task(some_function_reference)
        #Executes non-blocking in the same asyncio loop without any issues
        return "Hello"
    

    在路由函数之外添加background_tasks时,其行为有所不同:

    async def some_logic(background_tasks: BackgroundTasks):
        #Throws a "required positional argument missing" error
        background_tasks.add_task(some_function_reference)
    
    
    @app.get('/')
    async def index():
        await some_logic()
        #Executes non-blocking in the same asyncio loop
        return "Hello"
    

    同时,如果我们尝试初始化 BackgroundTasks some_logic 函数,则任务不会按以下方式运行:

    async def some_logic():
        #Does not Run
        background_tasks = BackgroundTasks()
        background_tasks.add_task(some_function_reference)
    
    
    @app.get('/')
    async def index(background_tasks: BackgroundTasks):
        await some_logic()
        #Executes non-blocking in the same asyncio loop
        return "Hello"
    

    为什么这三种情况会有所不同?为什么我需要将后台任务从路由函数传递到下面的调用函数?

    1 回复  |  直到 2 年前
        1
  •  2
  •   flakes    2 年前

    这个 BackgroundTasks 实例由FastAPI框架填充并注入到调用者中。在第一个示例中,您使用的路由具有由FastAPI注入的依赖项。

    第二个示例,您正在调用 some_logic 作为正常功能。因为它不是由FastAPI直接调用的,所以没有发生依赖注入。

    在第三个示例中,您创建了一个新的后台任务实例,但这与FastAPI关心的实例不同。您在其中创建任务,但没有注册处理程序来对其执行操作。

    在这种情况下,如果目标是使后台任务更容易应用于方法,则可以添加 某些逻辑 作为路由的依赖项。FastAPI调用依赖项,因此将注入参数。

    示例:

    from fastapi import FastAPI, Depends
    from starlette.background import BackgroundTasks
    from starlette.testclient import TestClient
    
    app = FastAPI()
    
    
    def task():
        print("Hello from the Background!")
    
    
    async def some_logic(background_tasks: BackgroundTasks):
        background_tasks.add_task(task)
    
    
    @app.get('/', dependencies=[Depends(some_logic)])
    async def index():
        return "Hello"
    
    
    with TestClient(app) as client:
        client.get("/")
    
    Hello from the Background!