Padrões Async em Python

Orientação abrangente para implementar aplicações Python assíncronas usando asyncio, padrões de programação concorrente e async/await para construir sistemas de alta performance e não bloqueantes.

Quando Usar Esta Habilidade

• Construir APIs web assíncronas (FastAPI, aiohttp, Sanic)
• Implementar operações de I/O concorrentes (banco de dados, arquivo, rede)
• Criar web scrapers com requisições concorrentes
• Desenvolver aplicações em tempo real (servidores WebSocket, sistemas de chat)
• Processar múltiplas tarefas independentes simultaneamente
• Construir microsserviços com comunicação assíncrona
• Otimizar cargas de trabalho limitadas por I/O
• Implementar tarefas e filas assíncronas em segundo plano

Guia de Decisão Sync vs Async

Antes de adotar o async, considere se é a escolha certa para o seu caso de uso.

Regra Chave: Mantenha-se totalmente sync ou totalmente async dentro de um caminho de chamada. Misturar cria bloqueios ocultos e complexidade.

Conceitos Principais

1. Event Loop

O event loop é o coração do asyncio, gerenciando e agendando tarefas assíncronas.

Características chave:

• Multitarefa cooperativa single-threaded
• Agenda corrotinas para execução
• Lida com operações de I/O sem bloquear
• Gerencia callbacks e futures

2. Corrotinas

Funções definidas com async def que podem ser pausadas e retomadas.

Sintaxe:

async def my_coroutine():
    result = await some_async_operation()
    return result

3. Tarefas (Tasks)

Corrotinas agendadas que rodam concorrentemente no event loop.

4. Futures

Objetos de baixo nível representando resultados eventuais de operações assíncronas.

5. Gerenciadores de Contexto Async

Recursos que suportam async with para limpeza adequada.

6. Iteradores Async

Objetos que suportam async for para iterar sobre fontes de dados assíncronas.

Início Rápido

import asyncio

async def main():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

# Python 3.7+
asyncio.run(main())

Padrões Fundamentais

Padrão 1: Async/Await Básico

import asyncio

async def fetch_data(url: str) -> dict:
    """Fetch data from URL asynchronously."""
    await asyncio.sleep(1)  # Simulate I/O
    return {"url": url, "data": "result"}

async def main():
    result = await fetch_data("https://api.example.com")
    print(result)

asyncio.run(main())

Padrão 2: Execução Concorrente com gather()

import asyncio
from typing import List

async def fetch_user(user_id: int) -> dict:
    """Fetch user data."""
    await asyncio.sleep(0.5)
    return {"id": user_id, "name": f"User {user_id}"}

async def fetch_all_users(user_ids: List[int]) -> List[dict]:
    """Fetch multiple users concurrently."""
    tasks = [fetch_user(uid) for uid in user_ids]
    results = await asyncio.gather(*tasks)
    return results

async def main():
    user_ids = [1, 2, 3, 4, 5]
    users = await fetch_all_users(user_ids)
    print(f"Fetched {len(users)} users")

asyncio.run(main())

Padrão 3: Criação e Gerenciamento de Tarefas

import asyncio

async def background_task(name: str, delay: int):
    """Long-running background task."""
    print(f"{name} started")
    await asyncio.sleep(delay)
    print(f"{name} completed")
    return f"Result from {name}"

async def main():
    # Create tasks
    task1 = asyncio.create_task(background_task("Task 1", 2))
    task2 = asyncio.create_task(background_task("Task 2", 1))

    # Do other work
    print("Main: doing other work")
    await asyncio.sleep(0.5)

    # Wait for tasks
    result1 = await task1
    result2 = await task2

    print(f"Results: {result1}, {result2}")

asyncio.run(main())

Padrão 4: Tratamento de Erros em Código Async

import asyncio
from typing import List, Optional

async def risky_operation(item_id: int) -> dict:
    """Operação que pode falhar."""
    await asyncio.sleep(0.1)
    if item_id % 3 == 0:
        raise ValueError(f"Item {item_id} falhou")
    return {"id": item_id, "status": "success"}

async def safe_operation(item_id: int) -> Optional[dict]:
    """Wrapper com tratamento de erros."""
    try:
        return await risky_operation(item_id)

…