Estoy tratando de procesar un archivo subido por el usuario. Sin embargo, quiero que el usuario obtenga una respuesta una vez que se complete la carga y finalice la conexión pero continúe procesando el archivo. Por lo tanto, estoy usando BackgroundTasks.add_tasks y mi código se parece a esto:

 class Line(BaseModel): line: str @app.post("/foo") async def foo(line: Line): """ Processing line generate results""" ... result = ... # processing line.line print(results) return results @app.post("/upload") async def upload(background_tasks: BackgroundTasks, csv: UploadFile = File(...)): background_tasks.add_task(process, csv) return response.text("CSV has been uploaded successfully") async def process(csv): """ Processing CSV and generate data""" tasks = [foo(line) for line in csv] result = await asyncio.gather(*tasks)

Desafortunadamente, el código anterior solo se ejecuta uno por uno. Además, tengo que esperar hasta que se procesen todos los resultados y luego imprimir la declaración en foo funciona, es decir, digamos que tengo n líneas en el csv, después de que se procesen todas las n es cuando veo las declaraciones de impresión para todas. Mi programa se ejecuta en 20 trabajadores, pero mientras se ejecuta este proceso, solo utiliza alrededor del 1% de la CPU (foo no es una tarea de cálculo, es más una tarea vinculada a IO/Red). Esto me hace pensar que el proceso en segundo plano se ejecuta solo en 1 trabajador. Probé ProcessPoolExecutor de la siguiente manera:

 loop = asyncio.get_event_loop() lines = [line_0, line_1, ..., line_n] # Extracted all lines from CSV with ProcessPoolExecutor() as executor: results = [loop.run_in_executor(executor, lambda: foo(line)) for line in lines] results = loop.run_until_complete(*results)

Sin embargo, me sale el siguiente error:

processpoolexecutor no puede encurtir el objeto local

Logré superar ese error cambiando mi enfoque de:

 results = [loop.run_in_executor(executor, lambda: foo(line)) for line in lines]

a:

 results = [asyncio.ensure_future(foo(line=Line(line)) for line in lines]

Sin embargo, luego me sale este error:

Archivo "uvloop/loop.pyx", línea 2658, en uvloop.loop.Loop.run_in_executor AttributeError: el objeto 'Loop' no tiene el atributo 'submit'

Para resumir: para procesar una línea, puedo presionar el punto final "/ foo" . Ahora, quiero procesar un csv de 200 líneas. Entonces, primero acepto el archivo del usuario y devuelvo un mensaje de éxito y termino esa conexión. Luego, el csv se agrega a una tarea en segundo plano que debe asignar cada línea al punto final "/ foo" y darme los resultados para cada línea. Sin embargo, todos los enfoques que he probado hasta ahora parecen estar usando solo un hilo y están procesando cada línea una por una. Me gustaría un enfoque en el que pueda procesar varias líneas juntas, casi como si estuviera presionando el punto final "/ foo" varias veces simultáneamente, como podemos usar herramientas como Apache JMeter.