如何在 FastAPI 应用程序中发送操作进度? [英] How to send a progress of operation in a FastAPI app?

问题描述

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

I have deployed a fastapi endpoint,

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 端点请求输出，它对我来说工作得很好.

I can request output from the API endpoint and its working fine for me perfectly.

现在，对我来说最大的挑战是要知道每次迭代需要多少时间.因为在 UI 部分(那些访问我的 API 端点的人)，我想帮助他们为每个正在处理的迭代/文件显示一个进度条(TIME TAKEN).

Now, the biggest challenge for me to know how much time it is taking for each iteration. Because in the UI part (those who are accessing my API endpoint) I want to help them show a progress bar (TIME TAKEN) for each iteration/file being processed.

我有什么可能的方法来实现它吗?如果是这样，请帮助我了解如何进一步处理?

Is there any possible way for me to achieve it? If so, please help me out on how can I proceed further?

谢谢.

推荐答案

以下是使用 uniq 标识符和保存作业信息的全局可用字典的解决方案:

Below is solution which uses uniq identifiers and globally available dictionary which holds information about the jobs:

注意:在您使用动态键值(在使用示例 uuid 中)并将应用程序保持在单个进程中之前，下面的代码可以安全使用.

1. 要启动应用程序，请创建一个文件 main.py
2. 运行 uvicorn main:app --reload
3. 通过访问 http://127.0.0.1:8000/
创建作业条目
4. 重复第 3 步以创建多个作业
5. 转到 http://127.0.0.1/status 页面查看页面状态.
6. 转到 http://127.0.0.1/status/{identifier} 以查看作业 ID 的作业进度.

1. To start the app create a file main.py
2. Run uvicorn main:app --reload
3. Create job entry by accessing http://127.0.0.1:8000/
4. Repeat step 3 to create multiple jobs
5. Go to http://127.0.0.1/status page to see page statuses.
6. Go to http://127.0.0.1/status/{identifier} to see progression of the job by the job 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'),
    }

这篇关于如何在 FastAPI 应用程序中发送操作进度?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！