Como passar parâmetros para várias tarefas assíncronas em Python

Question

No momento, tenho um código parecido com este:

    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 você pode ver acima, tenho 3 funções assíncronas que fazem coisas separadas simultaneamente. Eu queria saber se existe alguma maneira de criar facilmente muitas funções com base na entrada do usuário? Essencialmente, o que quero que seja capaz de fazer é o seguinte:

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

E, depois de obter os dados, ele seria executado 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())

Assim, ele faria funções baseadas em certos verificáveis ​​(como userinput1) e então faria isso quantas vezes especificado (amount_of_needed_functions) e então executaria todos estes simultaneamente. Desculpe, esta é uma pergunta um pouco confusa, mas estou perdido sobre por onde começar a pesquisar isso. Obrigado!

Answer 1

Passe a entrada do usuário como um argumento para cada tarefa:

Função única para várias tarefas

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últiplas funções para múltiplas tarefas

Use um dicionário para invocar o método desejado 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

Você pode tentar algo assim se todas as funções fizerem a mesma coisa

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]
    )