从 FastAPI 中的后台任务获取返回状态

Question

我有一个 API，它发布创建后台作业的作业，我想在另一个 GET api 上发送作业状态。如何实现这一目标？在background_work()函数中，我将使用多处理作为内部subprocess.call()调用的目标调用。

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def background_work(data: str):
    # some computation on data and return it
    return status

@app.post("/post_job", status_code=HTTP_201_CREATED)
async def send_notification(data: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(background_work, data)
    return {"message": "Job Created, check status after some time!"}

@app.get("/get_status")
def status():
    #how to return status of job submitted to background task

Answer 1

目前，使用 FastAPI 无法实现这一点，因为后台任务只是对发送响应后要调用的可调用对象的引用，它们不存储任何类型的状态。

您将必须使用 Celery 或其他库。

Answer 2

我正在像这样使用 fastAPI，结合使用concurrent.futures.ProcessPoolExecutor()asyncio 来管理长时间运行的作业。

If you don't want to rely on other modules (celery etc), you need to manage yourself the state of your job, and store it somewhere. I store it in the DB so that pending jobs can be resumed after a restart of the server.

Note that you must NOT perform CPU intensive computations in the background_tasks of the app, because it runs in the same async event loop that serves the requests and it will stall your app. Instead submit them to a thread pool or a process pool.

Answer 3

试试这个模式：

import time
from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

class TaskState:

    def __init__(self):
        self.counter = 0

    def background_work(self):
        while True:
            self.counter += 1
            time.sleep(1)

    def get_state(self):
        return self.counter

state = TaskState()

@app.post("/post_job", status_code=HTTP_201_CREATED)
async def send_notification(background_tasks: BackgroundTasks):
    background_tasks.add_task(state.background_work)
    return {"message": "Job Created, check status after some time!"}

@app.get("/get_status")
def status():
    return state.get_state()