Async e await são palavras-chave para programação assíncrona em Python.
Elas foram introduzidas no Python 3.5 para facilitar a concorrência.
Primeiramente, async def define uma função assíncrona (corrotina).
Dentro dela, await pausa a execução sem bloquear a thread.
Por exemplo, await asyncio.sleep(1) espera 1 segundo cooperativamente.
Além disso, você só pode usar await dentro de funções async.
Assim, o código se torna mais claro e previsível.
Consequentemente, a legibilidade melhora drasticamente.
Quando utilizar async/await? Em operações de I/O com alta concorrência.
Por exemplo, requisições HTTP, acesso a banco de dados ou WebSockets.
Por outro lado, async/await não é para tarefas com CPU intensiva.
Portanto, identifique primeiro o tipo do seu gargalo.
Vamos explorar sintaxe, padrões e boas práticas.
Três subtítulos guiarão você pelo coração da programação assíncrona.
Finalmente, você dominará async e await em Python.
Sintaxe básica: async def, await e asyncio.run
Uma corrotina é definida com async def nome().
Para chamar uma corrotina, você precisa usar await.
O ponto de entrada principal usa asyncio.run(corrotina()).
Então, isso cria um event loop e executa a corrotina até completar.
Quando usar funções async? Sempre que houver espera por I/O.
Por exemplo, leitura de arquivos ou chamadas de rede.
Exemplo básico mostrando a sintaxe fundamental:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 |
import asyncio import time # Definindo uma corrotina (função assíncrona) async def saudacao_assincrona(nome, segundos): """Função que espera e retorna uma saudação.""" print(f" Iniciando saudação para {nome}") await asyncio.sleep(segundos) # Pausa cooperativa print(f" Saudação para {nome} concluída após {segundos}s") return f"Olá, {nome}!" async def tarefa_rapida(): """Corrotina que não espera muito.""" print(" Tarefa rápida: começando") await asyncio.sleep(0.5) print(" Tarefa rápida: terminou") return "Rápido" async def tarefa_lenta(): """Corrotina que demora mais.""" print(" Tarefa lenta: começando") await asyncio.sleep(2) print(" Tarefa lenta: terminou") return "Lento" async def main(): print("=== Execução sequencial (uma await após outra) ===") inicio = time.time() resultado1 = await saudacao_assincrona("Ana", 1) resultado2 = await saudacao_assincrona("Bob", 1) print(f"Resultados sequenciais: {resultado1}, {resultado2}") print(f"Tempo sequencial: {time.time() - inicio:.1f}s") print("\n=== Execução concorrente (múltiplos awaits em tasks) ===") inicio = time.time() # Criando tasks para executar concorrentemente task1 = asyncio.create_task(saudacao_assincrona("Carlos", 1)) task2 = asyncio.create_task(saudacao_assincrona("Diana", 1)) # Aguardando ambas resultado3 = await task1 resultado4 = await task2 print(f"Resultados concorrentes: {resultado3}, {resultado4}") print(f"Tempo concorrente: {time.time() - inicio:.1f}s") print("\n=== gather: execute múltiplas corrotinas em paralelo ===") inicio = time.time() resultados = await asyncio.gather( tarefa_rapida(), tarefa_lenta(), saudacao_assincrona("Eva", 1) ) print(f"Resultados do gather: {resultados}") print(f"Tempo do gather: {time.time() - inicio:.1f}s") if __name__ == "__main__": asyncio.run(main()) |
Use create_task() para agendar corrotinas concorrentes.
Use gather() para esperar várias corrotinas simultaneamente.
Nunca esqueça de await em chamadas de corrotinas.
Assim, você evita erros comuns de iniciantes.
Padrões avançados: timeout, wait e as_completed
Async/await oferece ferramentas para controlar a execução.
asyncio.wait_for() adiciona timeout a uma corrotina.
asyncio.wait() espera múltiplas tarefas com condições.
asyncio.as_completed() processa resultados conforme terminam.
Quando usar esses padrões? Em sistemas que exigem limites de tempo.
Também em pipelines onde você quer respostas parciais.
Além disso, timeouts previnem travamentos eternos.
Exemplo prático com timeout e processamento parcial:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 |
import asyncio import random async def tarefa_instavel(nome, tempo_espera, chance_erro=0): """Tarefa que pode demorar ou falhar.""" print(f" {nome}: iniciando (vai demorar {tempo_espera}s)") await asyncio.sleep(tempo_espera) if chance_erro > 0 and random.random() < chance_erro: raise Exception(f"{nome}: Falhou após {tempo_espera}s") print(f" {nome}: concluída") return f"Resultado de {nome}" async def main(): print("=== Exemplo com timeout ===") try: # Timeout de 2 segundos para uma tarefa que demora 3 resultado = await asyncio.wait_for( tarefa_instavel("Tarefa-Lenta", 3), timeout=2 ) print(f"Resultado: {resultado}") except asyncio.TimeoutError: print("Timeout! Tarefa levou mais que 2 segundos") print("\n=== Exemplo com wait (FIRST_COMPLETED) ===") tarefas = [ tarefa_instavel("A", 3), tarefa_instavel("B", 1), tarefa_instavel("C", 2), ] # Aguarda a primeira tarefa completar concluidas, pendentes = await asyncio.wait( tarefas, return_when=asyncio.FIRST_COMPLETED ) for tarefa in concluidas: resultado = tarefa.result() print(f"Primeira tarefa concluída: {resultado}") # Cancela as tarefas pendentes for tarefa in pendentes: tarefa.cancel() print("Tarefas pendentes foram canceladas") print("\n=== Exemplo com as_completed (processamento parcial) ===") tarefas = [ tarefa_instavel("X", 2), tarefa_instavel("Y", 1), tarefa_instavel("Z", 3), ] # Processa resultados conforme terminam inicio = time.time() for corrotina in asyncio.as_completed(tarefas): try: resultado = await corrotina print(f"Resultado recebido em {time.time() - inicio:.1f}s: {resultado}") except Exception as e: print(f"Erro: {e}") print("\n=== Exemplo com wait_for e shields (proteção) ===") # asyncio.shield protege uma tarefa de ser cancelada tarefa_protegida = asyncio.create_task(tarefa_instavel("Protegida", 2)) try: resultado = await asyncio.wait_for(asyncio.shield(tarefa_protegida), timeout=1) except asyncio.TimeoutError: print("Timeout atingido, mas a tarefa protegida continua") # A tarefa ainda está rodando em background resultado_final = await tarefa_protegida print(f"Resultado final da tarefa protegida: {resultado_final}") if __name__ == "__main__": import time asyncio.run(main()) |
Timeouts previnem travamentos em operações lentas.
as_completed é ótimo para mostrar progresso gradual.
shield protege tarefas importantes de cancelamento.
Portanto, use essas ferramentas para sistemas robustos.
Async/await vs. threads vs. callbacks tradicionais
Async/await é muito mais legível que callbacks aninhados. Ele também usa menos memória que threads para muitas conexões. No entanto, threads são melhores para código síncrono legado. Quando escolher async/await? Em projetos novos com muito I/O. Também quando você precisa de milhares de conexões simultâneas. Além disso, async/await facilita o tratamento de erros. A fórmula de legibilidade mostra vantagem clara: \(L = \frac{N_{\text{linhas}}}{N_{\text{aninhamento}}}\) Async/await reduz aninhamento comparado a callbacks. Exemplo comparando callbacks vs. async/await:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 |
import asyncio import time # Versão com callbacks (estilo antigo, difícil) def baixar_com_callback(url, callback): """Simula download com callback.""" def executar(): time.sleep(1) # Simula I/O bloqueante (ruim) callback(f"Dados de {url}") import threading threading.Thread(target=executar).start() def callback_antigo(resultado): print(f"Callback recebeu: {resultado}") # Callback hell começaria aqui se tivesse mais chamadas # Versão com async/await (moderno e legível) async def baixar_assincrono(url): """Simula download com async/await.""" await asyncio.sleep(1) # I/O não bloqueante return f"Dados de {url}" async def processar_em_paralelo(urls): """Processa múltiplos downloads de forma legível.""" tarefas = [baixar_assincrono(url) for url in urls] resultados = await asyncio.gather(*tarefas) for resultado in resultados: print(f"Processando: {resultado}") # Fácil adicionar transformações dados_upper = [r.upper() for r in resultados] return dados_upper async def main(): print("=== Callbacks (difícil de ler e manter) ===") baixar_com_callback("site1.com", callback_antigo) baixar_com_callback("site2.com", callback_antigo) time.sleep(2) # Espera feia print("\n=== Async/Await (claro e legível) ===") urls = ["site1.com", "site2.com", "site3.com"] resultados = await processar_em_paralelo(urls) print(f"Resultados finais: {resultados}") print("\n=== Async/Await com tratamento de erros elegante ===") async def tarefa_segura(): try: await asyncio.sleep(0.5) return "Sucesso" except Exception as e: return f"Erro: {e}" resultados = await asyncio.gather( tarefa_segura(), tarefa_segura(), return_exceptions=True # Não levanta exceção imediatamente ) print(f"Resultados com exceptions capturadas: {resultados}") if __name__ == "__main__": asyncio.run(main()) |
Async/await torna o código assíncrono tão legível quanto síncrono.
Use return_exceptions=True no gather para lidar com falhas.
A regra de ouro: toda operação de I/O deve ser async.
Comece com async/await em novas funções de rede.
Gradualmente, converta código síncrono legado.
Você nunca mais vai querer voltar para callbacks.
Portanto, abrace async/await em seus próximos projetos.