python - 如何在 FastAPI 应用程序中发送操作进度？

Question

我已经部署了一个 fastapi 端点，

from fastapi import FastAPI, UploadFile
from typing import List

app = FastAPI()

@app.post('/work/test')
async def testing(files: List(UploadFile)):
    for i in files:
        .......
        # do a lot of operations on each file

        # after than I am just writing that processed data into mysql database
        # cur.execute(...)
        # cur.commit()
        .......
    
    # just returning "OK" to confirm data is written into mysql
    return {"response" : "OK"}

我可以从 API 端点请求输出，它对我来说工作得很好。
现在，对我来说最大的挑战是知道每次迭代需要多少时间。因为在 UI 部分(那些访问我的 API 端点的人)我想帮助他们为正在处理的每个迭代/文件显示一个进度条(TIME TAKEN)。
我有什么可能的方法来实现它吗？如果是这样，请帮助我了解如何进一步处理？
谢谢。

Answer 1

以下是使用 uniq 标识符和全局可用字典的解决方案，其中包含有关作业的信息:
注意:下面的代码可以安全使用，直到您使用动态键值(在使用的示例 uuid 中)并将应用程序保持在单个进程中。

要启动应用程序，请创建一个文件 main.py

运行 uvicorn main:app --reload

通过访问 http://127.0.0.1:8000/ 创建作业条目

重复步骤 3 以创建多个作业

转至 http://127.0.0.1/status页面查看页面状态。

转至 http://127.0.0.1/status/{identifier}通过作业 ID 查看作业的进度。

应用程序代码:

from fastapi import FastAPI, UploadFile
import uuid
from typing import List


import asyncio


context = {'jobs': {}}

app = FastAPI()



async def do_work(job_key, files=None):
    iter_over = files if files else range(100)
    for file, file_number in enumerate(iter_over):
        jobs = context['jobs']
        job_info = jobs[job_key]
        job_info['iteration'] = file_number
        job_info['status'] = 'inprogress'
        await asyncio.sleep(1)
    pending_jobs[job_key]['status'] = 'done'


@app.post('/work/test')
async def testing(files: List[UploadFile]):
    identifier = str(uuid.uuid4())
    context[jobs][identifier] = {}
    asyncio.run_coroutine_threadsafe(do_work(identifier, files), loop=asyncio.get_running_loop())

    return {"identifier": identifier}


@app.get('/')
async def get_testing():
    identifier = str(uuid.uuid4())
    context['jobs'][identifier] = {}
    asyncio.run_coroutine_threadsafe(do_work(identifier), loop=asyncio.get_running_loop())

    return {"identifier": identifier}

@app.get('/status')
def status():
    return {
        'all': list(context['jobs'].values()),
    }

@app.get('/status/{identifier}')
async def status(identifier):
    return {
        "status": context['jobs'].get(identifier, 'job with that identifier is undefined'),
    }