Cómo pasar parámetros a múltiples tareas asíncronas en Python

Question

Ahora mismo tengo un código que se parece a esto:

    userinput1 = abc.....
    userinput2 = abc.....
    userinput3 = abc.....
    
    async def task1():
        do something with userinput1...
        do another thing...
    
    async def task2():
        do something with userinput2...
        do another thing...
    
    async def task3():
        do something with userinput3...
        do another thing...
    
    async def main():
        await asyncio.wait([task1() , task2(), task3()])
    
    if __name__ == '__main__':
        asyncio.get_event_loop().run_until_complete(main())

Como puede ver arriba, tengo 3 funciones asíncronas que hacen cosas separadas simultáneamente. Me preguntaba si hay alguna forma de crear fácilmente muchas funciones basadas en la entrada del usuario. Básicamente, lo que quiero que pueda hacer es esto:

    userinput1 = abc.....
    userinput2 = abc.....
    userinput3 = abc.....
    userinput4 = abc.....
    amount_of_needed_functions = 4

Y luego, una vez que hubiera obtenido esos datos, se ejecutaría como este script:

    async def task1():
            do something with userinput1...
            do another thing...
        
    async def task2():
            do something with userinput2...
            do another thing...
        
    async def task3():
            do something with userinput3...
            do another thing...
    
    async def task4():
            do something with userinput4...
            do another thing...
        
    async def main():
            await asyncio.wait([task1() , task2(), task3(), task4()])
        
    if __name__ == '__main__':
            asyncio.get_event_loop().run_until_complete(main())

Así que prácticamente haría funciones basadas en ciertos verificables (como userinput1) y luego haría esto tantas veces como se especifique (cantidad_de_funciones_necesarias) y luego ejecutará todas estas simultáneamente. Lo siento, esta es una pregunta un poco confusa, pero estoy bastante perdido en cuanto a dónde comenzar a investigar esto. ¡Gracias!

Answer 1

Pase la entrada del usuario como argumento para cada tarea:

Función única para múltiples tareas

import asyncio

async def function(user_input, input_index):
    print(f'In {input_index} function: {user_input}')


async def main():
    tasks = []
    for input_index in range(1, 4):
        user_input = input(f"Enter input #{input_index}\n")
        tasks.append(asyncio.create_task(function(user_input, input_index)))
    await asyncio.gather(*tasks)


if __name__ == '__main__':
    asyncio.run(main())

Múltiples funciones para múltiples tareas

Utilice un diccionario para invocar su método deseado para cada entrada.

import asyncio

async def function1(user_input, input_index):
    print(f'In {input_index} function1: {user_input}')

async def function2(user_input, input_index):
    print(f'In {input_index} function2: {user_input}')

async def function3(user_input, input_index):
    print(f'In {input_index} function3: {user_input}')


FUNCTION_DICTIONARY = { 1 : function1, 2 : function2, 3 : function3 }

async def main():
    tasks = []
    for input_index in range(1, 4):
        user_input = input(f"Enter input #{input_index}\n")
        tasks.append(asyncio.create_task(FUNCTION_DICTIONARY[input_index](user_input, input_index)))
    await asyncio.gather(*tasks)


if __name__ == '__main__':
    asyncio.run(main())

Answer 2

Puedes probar algo como esto si todas las funciones hacen lo mismo

inputs = ['a', 'b', 'c']

async def task(input: str):
    # Do stuff / await stuff
    return input
   
async def main()
    await asyncio.wait(
        [task(arg) for arg in inputs]
    )