O uso de programação concorrente em Python, especialmente com asyncio, é uma prática comum para otimizar operações de Entrada/Saída (I/O) ou tarefas computacionais intensivas. No entanto, ao gerenciar centenas ou milhares de tarefas em segundo plano, surge um desafio prático: a falta de visibilidade sobre o progresso da execução e a estimativa de tempo para conclusão (ETA).

O Desafio da Visibilidade em Fluxos Assíncronos

Sem uma interface visual, o desenvolvedor muitas vezes se depara com um terminal estático, sem saber se o código ainda está processanod, se houve um deadlock ou quanto tempo resta para o término. Abordagens tradicionais, como o uso de variáveis globais compartilhadas ou contadores manuais incrementados dentro das corrotinas, cosutmam introduzir acoplamento excessivo e problemas de concorrência que exigem mecanismos de bloqueio (locks), impactando a performance.

A biblioteca tqdm oferece uma solução elegante para esse problema, permitindo a integração de barras de progresso dinâmicas diretamente em loops e fluxos de trabalho assíncronos.

Uso Básico do Tqdm

Em sua forma mais simples, o tqdm envolve um objeto iterável. Veja um exemplo básico de um loop síncrono:

import time
from tqdm import tqdm

def processamento_simples():
    for _ in tqdm(range(50), desc="Processando"):
        time.sleep(0.05)

if __name__ == "__main__":
    processamento_simples()

Integração Nativa com Asyncio

Para fluxos assíncronos, a biblioteca disponibiliza o módulo tqdm.asyncio. Ele fornece a classe tqdm_asyncio, que encapsula os métodos tradicionais do asyncio para suportar o rastreamento automático de progresso.

Os dois métodos principais são:

• tqdm_asyncio.gather: Uma versão monitorada do asyncio.gather, que aguarda a conclusão de múltiplas corrotinas.
• tqdm_asyncio.as_completed: Um invólucro para asyncio.as_completed, ideal para quando os resultaods precisam ser processados à medida que ficam prontos.

Exemplo Prático: Monitorando Corrotinas

Considere o cenário abaixo, onde simulamos tarefas assíncronas com tempos de espera variados. Primeiro, definimos a lógica da tarefa:

import asyncio
import random
from tqdm.asyncio import tqdm_asyncio

class FalhaOperacao(Exception):
    pass

async def executar_tarefa(id_tarefa, disparar_erro=False):
    tempo_espera = round(random.uniform(0.5, 3.5), 2)
    
    if disparar_erro and tempo_espera > 3.0:
        raise FalhaOperacao(f"Erro na tarefa {id_tarefa}")
        
    await asyncio.sleep(tempo_espera)
    return f"Tarefa {id_tarefa} concluída em {tempo_espera}s"

Utilizando tqdm_asyncio.gather

O método gather é útil quando você deseja disparar todas as tarefas e aguardar o resultado final, visualizando o progresso total na barra.

async def rodar_com_gather():
    lista_tarefas = [executar_tarefa(i) for i in range(1000)]
    
    # Substitui o asyncio.gather tradicional
    resultados = await tqdm_asyncio.gather(*lista_tarefas, desc="Progresso Gather")
    print(f"Total de resultados: {len(resultados)}")

if __name__ == "__main__":
    asyncio.run(rodar_com_gather())

Utilizando tqdm_asyncio.as_completed

Se a lógica exige que você itere sobre as tarefas conforme elas terminam, o as_completed é a escolha adequada.

async def rodar_com_as_completed():
    jobs = [executar_tarefa(i) for i in range(1000)]
    
    count = 0
    for futura_tarefa in tqdm_asyncio.as_completed(jobs, desc="Progresso Iterativo"):
        resultado = await futura_tarefa
        count += 1
        
    print(f"Finalizado: {count} tarefas processadas.")

if __name__ == "__main__":
    asyncio.run(rodar_com_as_completed())

A implementação interna do tqdm_asyncio garante que a atualização da interface do terminal não bloqueie o loop de eventos do asyncio, mantendo a eficiência da aplicação enquanto fornece feedback visual em tempo real sobre a taxa de iterações por segundo e a estimativa de término.