Este artigo aborda os conceitos essenciais da programação assíncrona em Python, incluindo coroutines, tarefas e o event loop. Para entender a programação assíncrona, é importante definir esses componentes. Em versões anteriores ao Python 3.7, o Future era um conceito relevante, mas as APIs modernas simplificaram seu uso.

Entendendo Coroutines

Uma coroutine pode ser comparada a uma pessoa que alterna entre diferentes tarefas, enquanto uma thread seria como duas pessoas trabalhando simultaneamente (ignorando o GIL por enquanto). Imagine gerenciar tarefas como enviar um email, processar um arquivo e calcular um resultado. Com uma única pessoa (coroutine), você inicia o envio do email, então enquanto espera pela resposta, passa a processar o arquivo, e assim por diante. Se alguém insistir em esperar o email ser enviado antes de fazer outra coisa, seria necessário adicionar outra pessoa (thread) para otimizar o tempo.

Operações de IO

O processo de espera no envio de email é uma operação de IO. Em Python, operações como solicitações de rede e leitura/escrita de arquivos são exemplos de IO, especialmente as requisições de rede, que consomem tempo ao esperar dados do servidor sem exigir CPU. Embora print também seja uma operação de IO, seu tempo de execução é muito curto, então não há necesidade de uma versão assíncrona como aioprint. Em contraste, tarefas como processar um grande conjunto de dados (cálculos intensivos) são do tipo CPU-bound, onde a programação assíncrona não oferece benefício; seria necessário usar múltiplas threads ou processos.

Definindo Coroutines

Uma coroutine é definida usando async def. Chamar a função assíncrona retorna um objeto coroutine sem executá-lo imediatamente.

async def minha_coroutine():
    # Implementação aqui
    pass

Tarefas

Tarefas encapsulam coroutines, facilitando o gerenciamento de cancelamento e obtenção de resultados. Elas são geralmente criadas com asyncio.create_task. Assim que uma tarefa é criada, a coroutine começa a executar. Você pode esperar pela conclusão com await ou cancelar com task.cancel().

tarefa = asyncio.create_task(minha_coroutine())

Event Loop

O event loop é o núcleo da execução assíncrona, representando uma entidade que coordena as coroutines. Cada thread pode ter um event loop, e é possível criar múltiplos loops em threads separados, especificando onde executar coroutines com asyncio.run_coroutine_threadsafe.

Utilização Básica

Iniciando o Event Loop

A maneira comum de iniciar o event loop é com asyncio.run, mas em casos específicos, pode-se usar APIs de baixo nível.

import asyncio

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(minha_coroutine_principal())

Exemplo Sequencial

O código a seguir executa coroutines de forma síncrona, onde cada uma espera a anterior terminar.

import asyncio
import time

async def executar_apos(atraso, mensagem):
    await asyncio.sleep(atraso)
    print(mensagem)

async def principal():
    print(f"Início em {time.strftime('%X')}")
    await executar_apos(1.5, 'primeiro')
    await executar_apos(2.5, 'segundo')
    print(f"Fim em {time.strftime('%X')}")

asyncio.run(principal())

Execução Concorrente

Para executar coroutines simultaneamente, use tarefas ou asyncio.gather.

import asyncio
import time

async def executar_apos(atraso, mensagem):
    await asyncio.sleep(atraso)
    print(mensagem)
    return atraso

async def principal():
    tarefa1 = asyncio.create_task(executar_apos(1, 'ola'))
    tarefa2 = asyncio.create_task(executar_apos(2, 'mundo'))
    print(f"Início em {time.strftime('%X')}")
    resultado1 = await tarefa1
    resultado2 = await tarefa2
    print(f"Resultados: {resultado1}, {resultado2}")
    print(f"Fim em {time.strftime('%X')}")

asyncio.run(principal())

Usando asyncio.gather

asyncio.gather permite executar múltiplas coroutines e coletar seus resultados na ordem de chamada.

import asyncio
import time

async def executar_apos(atraso, mensagem):
    await asyncio.sleep(atraso)
    print(mensagem)
    return atraso

async def principal():
    print(f"Início em {time.strftime('%X')}")
    resultados = await asyncio.gather(executar_apos(1, 'primeiro'), executar_apos(2, 'segundo'))
    print(f"Resultados: {resultados}")
    print(f"Fim em {time.strftime('%X')}")

asyncio.run(principal())

Timeout com asyncio.wait_for

asyncio.wait_for adiciona um tempo limite à execução de uma coroutine, lançando asyncio.TimeoutError se excedido.

import asyncio

async def tarefa_longa():
    await asyncio.sleep(100)

async def principal():
    try:
        await asyncio.wait_for(tarefa_longa(), timeout=5.0)
    except asyncio.TimeoutError:
        print("Tempo limite atingido!")

asyncio.run(principal())

Gerenciamento com asyncio.wait

asyncio.wait é usado para esperar múltiplas tarefas com opções de retorno, como quando a primeira conclui ou todas terminam.

import asyncio

async def executar_apos(atraso, mensagem):
    await asyncio.sleep(atraso)
    print(mensagem)
    return atraso

async def principal():
    tarefa1 = asyncio.create_task(executar_apos(1, 'a'))
    tarefa2 = asyncio.create_task(executar_apos(2, 'b'))
    concluidas, pendentes = await asyncio.wait([tarefa1, tarefa2], return_when=asyncio.FIRST_COMPLETED)
    for t in concluidas:
        print(f"Resultado: {t.result()}")

asyncio.run(principal())

Iteração por Conclusão com asyncio.as_completed

asyncio.as_completed fornece um iterador que yields futures na ordem em que são concluídos.

import asyncio

async def executar_apos(atraso, mensagem):
    await asyncio.sleep(atraso)
    mensagem
    return atraso

async def principal():
    tarefas = [asyncio.create_task(executar_apos(i, f"tarefa_{i}")) for i in range(3, 0, -1)]
    for coro in asyncio.as_completed(tarefas):
        resultado = await coro
        print(f"Concluído: {resultado}")

asyncio.run(principal())

Listando Tarefas Ativas

asyncio.all_tasks retorna todas as tarefas atualmente agendadas no event loop.

import asyncio

async def tarefa():
    await asyncio.sleep(1)

async def principal():
    print("Tarefas antes:", asyncio.all_tasks())
    t1 = asyncio.create_task(tarefa())
    print("Tarefas após criação:", asyncio.all_tasks())
    await t1
    print("Tarefas após conclusão:", asyncio.all_tasks())

asyncio.run(principal())

Funções Úteis do asyncio.Task

A clase asyncio.Task oferece métodos para gerenciar tarefas, como cancel(), done(), result(), e add_done_callback().

Verificação de Tipos

Funções como asyncio.iscoroutine e asyncio.iscoroutinefunction ajudam a identificar coroutines e funções assíncronas.

import asyncio

async def minha_func():
    pass

async def principal():
    print("É coroutine?", asyncio.iscoroutine(minha_func))
    obj = minha_func()
    print("É objeto coroutine?", asyncio.iscoroutine(obj))

asyncio.run(principal())