Fila de Tarefas Assíncronas com Python: O Controlador de Tráfego Que Sua Automação Precisa (Antes do Caos)

Às 3h47 da manhã, meu pipeline de scraping disparou. Três automações começaram ao mesmo tempo. A API do fornecedor respondeu com HTTP 429. A segunda automação, sem saber disso, tentou de novo. A terceira também. Em dois minutos, o IP da minha VPS estava banido. E eu descobri isso às 8h, quando o cliente ligou perguntando por que os dados não tinham sido entregues.

Aquele dia me ensinou uma lição que nenhum tutorial ensina: automação sem controle de concorrência é só uma forma chique de se sabotar em escala.

Se você tem mais de um script rodando — e quem não tem? — mais cedo ou mais tarde eles vão colidir. Scripts que acessam o mesmo banco. Robôs que chamam a mesma API. Cron jobs que disparam no mesmo minuto. O resultado não é bonito.

Neste artigo, eu vou te mostrar como construir uma fila de tarefas assíncronas em Python que funciona como um semáforo inteligente: decide quem passa, quando, e em que ordem. Sem depender de Redis, RabbitMQ ou qualquer infraestrutura que você não tem tempo de manter.

O stack é simples: Python, SQLite e o módulo asyncio. Tudo que você já tem instalado. E no final, você vai ter um sistema que gerencia dezenas de automações simultâneas sem que nenhuma delas pise no pé da outra.

O problema que ninguém te conta sobre automações concorrentes

Você cria um script. Funciona. Cria outro. Também funciona. Coloca os dois no cron. E de repente…

• Dois scripts gravam no mesmo arquivo de log ao mesmo tempo → log corrompido
• Dois scrapers batem na mesma API simultaneamente → rate limit ativado
• Um processo de backup roda enquanto outro tenta migrar o banco → lock de transação
• Três workers consomem 100% da CPU → servidor engasga e o SSH cai

O padrão é sempre o mesmo: funciona isolado, quebra junto.

A solução óbvia seria “não rode ao mesmo tempo”. Só que automações têm tempos diferentes. Uma API pode responder em 200ms ou 30 segundos dependendo do horário. Você não consegue prever. E escalonar tudo manualmente é insustentável.

É aqui que entra a fila de tarefas.

O que é uma fila de tarefas (e por que asyncio muda o jogo)

Uma fila de tarefas é, na essência, um sistema de despacho. Você coloca tarefas nela (enqueue), e workers pegam tarefas para executar (dequeue). A mágica está no controle:

• Ordem: primeiro a entrar, primeiro a sair (FIFO) — ou com prioridade, se você preferir
• Limite de concorrência: no máximo N tarefas rodando ao mesmo tempo
• Retry: se uma tarefa falha, volta para a fila
• Persistência: se o processo morre, a fila sobrevive

O asyncio do Python resolve isso de forma elegante porque permite centenas de tarefas “rodando” simultaneamente sem criar centenas de threads. Tudo roda no mesmo thread, e o event loop decide quando cada tarefa avança. O resultado? Consumo mínimo de memória, controle total de concorrência, e código que é mais legível que qualquer callback hell de antigamente.

Arquitetura: o mínimo que funciona

A arquitetura que eu uso em produção tem três componentes:

1. Fila persistente (SQLite): armazena as tarefas com status, prioridade e payload
2. Dispatcher (asyncio): lê a fila, respeita o limite de concorrência, dispara os workers
3. Workers (corrotinas): executam as tarefas e reportam resultado (sucesso, falha, retry)

Sem Redis. Sem Celery. Sem “precisa de um container extra”. Um arquivo Python, um banco SQLite, e pronto.

Passo 1: a tabela de tarefas no SQLite

Precisamos de uma tabela que guarde o estado de cada tarefa. Aqui está o schema que eu uso:

import sqlite3
import json
from datetime import datetime

def init_db(db_path="task_queue.db"):
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS tasks (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            priority INTEGER DEFAULT 5,
            payload TEXT DEFAULT '{}',
            status TEXT DEFAULT 'pending',
            retries INTEGER DEFAULT 0,
            max_retries INTEGER DEFAULT 3,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            started_at TIMESTAMP,
            finished_at TIMESTAMP,
            error TEXT,
            result TEXT
        )
    """)
    conn.execute("CREATE INDEX IF NOT EXISTS idx_status_priority ON tasks(status, priority)")
    conn.commit()
    return conn

conn = init_db()
print("Banco criado. Hora de enfileirar.")

O campo priority permite que tarefas urgentes furam a fila (menor número = maior prioridade). O payload é um JSON flexível — cada tarefa carrega seus próprios parâmetros. E o status controla o ciclo de vida: pending → running → done | failed | retrying.

Passo 2: enqueue — colocando tarefas na fila

def enqueue(conn, name: str, payload: dict, priority: int = 5):
    """Adiciona uma tarefa na fila."""
    conn.execute(
        "INSERT INTO tasks (name, priority, payload) VALUES (?, ?, ?)",
        (name, priority, json.dumps(payload))
    )
    conn.commit()
    task_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
    return task_id

# Exemplo: enfileirando automações
enqueue(conn, "scraping_produtos", {"url": "https://loja.com/api/produtos", "timeout": 30}, priority=1)
enqueue(conn, "backup_diario", {"db": "main", "dest": "/backup/"}, priority=3)
enqueue(conn, "enviar_relatorio", {"email": "admin@exemplo.com"}, priority=5)
enqueue(conn, "sincronizar_estoque", {"sku_list": "todos"}, priority=2)

Repare que o scraping de produtos tem prioridade 1 (mais urgente). Se o sistema estiver sobrecarregado, ele roda primeiro. O envio de relatório pode esperar — prioridade 5.

Passo 3: o dispatcher assíncrono

Aqui é onde a coisa fica interessante. O dispatcher é o cérebro:

import asyncio
import httpx
import traceback

MAX_CONCURRENT = 3  # Máximo de tarefas rodando ao mesmo tempo

async def process_task(conn, task: dict):
    """Processa uma única tarefa. Retorna (success, result_or_error)."""
    task_id = task["id"]
    name = task["name"]
    payload = json.loads(task["payload"])

    # Marca como running
    conn.execute(
        "UPDATE tasks SET status='running', started_at=CURRENT_TIMESTAMP WHERE id=?",
        (task_id,)
    )
    conn.commit()

    try:
        if name == "scraping_produtos":
            result = await scraping_produtos(payload)
        elif name == "backup_diario":
            result = await backup_diario(payload)
        elif name == "enviar_relatorio":
            result = await enviar_relatorio(payload)
        elif name == "sincronizar_estoque":
            result = await sincronizar_estoque(payload)
        else:
            result = {"warning": "Tarefa '" + name + "' não registrada"}

        # Sucesso
        conn.execute(
            "UPDATE tasks SET status='done', finished_at=CURRENT_TIMESTAMP, result=? WHERE id=?",
            (json.dumps(result), task_id)
        )
        conn.commit()
        return True, result

    except Exception as e:
        error_msg = str(e)
        retries = task["retries"] + 1
        max_retries = task["max_retries"]

        if retries < max_retries:
            conn.execute(
                "UPDATE tasks SET status='pending', retries=?, error=? WHERE id=?",
                (retries, error_msg, task_id)
            )
            conn.commit()
            return False, "Retry " + str(retries) + "/" + str(max_retries) + ": " + error_msg
        else:
            conn.execute(
                "UPDATE tasks SET status='failed', finished_at=CURRENT_TIMESTAMP, error=? WHERE id=?",
                (error_msg, task_id)
            )
            conn.commit()
            return False, "Failed after " + str(max_retries) + " retries: " + error_msg


async def dispatcher(conn, semaphore: asyncio.Semaphore):
    """Loop principal: pega tarefas pendentes e despacha."""
    while True:
        task = conn.execute("""
            SELECT * FROM tasks
            WHERE status = 'pending'
            ORDER BY priority ASC, created_at ASC
            LIMIT 1
        """).fetchone()

        if task is None:
            await asyncio.sleep(2)
            continue

        task_dict = dict(zip(
            [col[0] for col in conn.execute("PRAGMA table_info(tasks)").fetchall()],
            task
        ))

        async with semaphore:
            success, msg = await process_task(conn, task_dict)
            status_icon = "✅" if success else "⚠️"
            ts = datetime.now().strftime("%H:%M:%S")
            print("[" + ts + "] " + status_icon + " " + task_dict["name"] + ": " + msg)


async def main():
    conn = init_db()
    semaphore = asyncio.Semaphore(MAX_CONCURRENT)
    print("Dispatcher iniciado. Concorrência máxima: " + str(MAX_CONCURRENT))
    await dispatcher(conn, semaphore)

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

O Semaphore(MAX_CONCURRENT) é o segredo. Ele garante que no máximo 3 tarefas rodam ao mesmo tempo. Se a fila tem 50 tarefas, 3 rodam e 47 esperam. Quando uma termina, a próxima entra automaticamente.

Isso resolve o problema do rate limit. Resolve o problema de CPU. Resolve o problema de lock de banco. Tudo com um número configurável.

Passo 4: workers reais — exemplos práticos

O dispatcher só funciona se os workers forem reais. Aqui estão exemplos que eu uso:

async def scraping_produtos(payload: dict):
    """Scrapa produtos de uma API com retry automático."""
    url = payload.get("url")
    timeout = payload.get("timeout", 30)

    async with httpx.AsyncClient(timeout=timeout) as client:
        response = await client.get(url)
        response.raise_for_status()
        data = response.json()

    count = len(data.get("products", []))
    return {"products_scraped": count, "source": url}


async def backup_diario(payload: dict):
    """Faz backup de um banco SQLite."""
    import shutil, os
    db_name = payload.get("db", "main")
    dest = payload.get("dest", "/backup/")

    source = db_name + ".db"
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    dest_file = os.path.join(dest, db_name + "_" + timestamp + ".db")

    os.makedirs(dest, exist_ok=True)
    shutil.copy2(source, dest_file)

    size_mb = os.path.getsize(dest_file) / (1024 * 1024)
    return {"backup_file": dest_file, "size_mb": round(size_mb, 2)}

Cada worker é uma corrotina async. Isso significa que quando ele faz uma chamada de rede (como o httpx), o event loop pode executar outras tarefas enquanto espera a resposta. Eficiência máxima, zero threads extras.

Monitoramento: como saber o que está acontecendo

Uma fila sem visibilidade é uma caixa preta — e caixas pretas assustam. Aqui está um painel simples que roda no terminal:

def queue_status(conn):
    """Mostra status atual da fila."""
    stats = conn.execute("""
        SELECT status, COUNT(*) as count,
               AVG(CAST(retries AS FLOAT)) as avg_retries
        FROM tasks
        GROUP BY status
    """).fetchall()

    print("\n📊 STATUS DA FILA")
    print("-" * 40)
    for status, count, avg_retries in stats:
        icon = {"pending": "⏳", "running": "🔄", "done": "✅",
                "failed": "❌", "retrying": "🔁"}.get(status, "❓")
        print("  " + icon + " " + status + ": " + str(count) + " tarefas")

    errors = conn.execute("""
        SELECT name, error, retries FROM tasks
        WHERE status = 'failed'
        ORDER BY finished_at DESC LIMIT 5
    """).fetchall()

    if errors:
        print("\n🔥 FALHAS RECENTES:")
        for name, error, retries in errors:
            print("  ❌ " + name + " (retries: " + str(retries) + "): " + error[:80])

Eu chamo isso a cada 30 segundos no dispatcher. Dá pra ver em tempo real o que está travando.

🔧 O perrengue que me ensinou isso tudo: Minha primeira versão da fila não tinha persistência. Tudo ficava em memória com asyncio.Queue. Funcionou lindo por 3 semanas — até um OOM killer do Linux derrubar o processo. Perdi 47 tarefas em execução. Lição aprendida: se não está no disco, não existe. O SQLite adicionou 2ms por operação e me deu paz de espírito. Vale cada milissegundo.

Prioridade dinâmica: quando urgência muda no meio do caminho

As vezes uma tarefa precisa ser executada agora. Um cliente ligou. Um deploy urgente. Um dado que expira em 10 minutos.

def urgent_enqueue(conn, name: str, payload: dict):
    """Coloca uma tarefa na frente da fila com prioridade máxima."""
    task_id = enqueue(conn, name, payload, priority=0)
    print("🚨 Tarefa urgente '" + name + "' enfileirada (id: " + str(task_id) + ")")
    return task_id

Prioridade 0 é mais alta que qualquer outra. O ORDER BY do dispatcher garante que tarefas com prioridade 0 são processadas primeiro, independente de quando foram adicionadas.

Quando usar (e quando NÃO usar)

Use quando:

• Você tem múltiplas automações que competem por recursos
• Precisa de controle de rate limit do seu lado
• Quer retry automático sem reescrever cada script
• Não quer depender de infraestrutura externa (Redis, RabbitMQ)
• Seu volume é de dezenas a centenas de tarefas por dia

Não use quando:

• Você precisa de processamento distribuído entre vários servidores
• O volume é de milhares de tarefas por minuto (use Celery + Redis)
• Precisa de agendamento complexo tipo cron distribuído
• Seu time já tem uma fila em produção — não reinvente a roda

Próximos passos: de onde crescer

Esse sistema é o mínimo que funciona. Daqui, você pode evoluir:

• Webhook de notificação: disparar um POST quando uma tarefa falha
• Dashboard HTTP: uma rota Flask/FastAPI que mostra a fila em tempo real
• Agendamento: tarefas recorrentes com base em intervalos (daily, hourly)
• Dead letter queue: tarefas que falharam 3x vão para uma fila separada de investigação
• Métricas: tempo médio de execução, taxa de falha por tipo de tarefa

Cada um desses é um artigo por si só. Se você quer que eu cubra algum, me diz nos comentários.

Conclusão: controle é melhor que sorte

Automação deveria te dar paz, não dor de cabeça às 3h da manhã. A diferença entre um sistema que funciona e um que parece que funciona é o controle de concorrência.

Com uma fila de tarefas assíncronas em Python, você transforma o caos de "vários scripts competindo por recursos" em um sistema ordenado, monitorável e previsível. E o melhor: sem depender de nada que não seja Python e SQLite.

Seu futuro eu das 3h da manhã vai agradecer.

Qual automação sua mais precisa de um controle de tráfego? Me conta aqui nos comentários — ou melhor ainda, me diz qual integração você quer ver automatizada no próximo artigo.

Artigos relacionados: Rate Limiter Local com Python e SQLite | Automação Resiliente com Python | Todos os posts de Produtividade Aumentada