如何在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屋！