Automação Resiliente com Python: Retries, Lockfile e Idempotência Sem Drama

Automação resiliente com Python não é sobre escrever um script bonito que funciona na sua máquina às 2h da manhã. É sobre escrever um fluxo que continua honesto quando a API cai, quando o servidor reinicia no meio do processo, quando o cron roda duas vezes e quando você descobre, tarde demais, que seu robô enviou a mesma cobrança para a mesma pessoa três vezes. Delicioso. O tipo de perrengue que ensina humildade no grito.

O problema real deste artigo é simples: você tem uma automação que busca dados, transforma algo e executa uma ação com efeito colateral. Pode ser enviar e-mail, publicar conteúdo, atualizar planilha, chamar um webhook, gerar fatura, sincronizar leads ou mover arquivos. No primeiro dia ela parece mágica. No terceiro, ela falha no meio. No quarto, você adiciona retry. No quinto, o retry duplica tudo. Parabéns, agora você tem um sistema distribuído disfarçado de script.

Vamos montar uma base prática de automação resiliente com Python usando três peças que evitam a maior parte do caos: lockfile, retries com limite e idempotência. Não é arquitetura astronauta. É engenharia de sobrevivência para scripts que precisam rodar todo dia sem virar um incêndio recorrente.

1. O sintoma: o script funciona, mas não é confiável

Todo mundo começa com uma versão parecida com esta:

def main():
    dados = buscar_dados()
    resultado = processar(dados)
    enviar(resultado)

if __name__ == "__main__":
    main()

É limpo. É didático. É também inocente. Esse script não sabe se já está rodando. Não sabe se falhou depois de processar e antes de enviar. Não sabe se o destino recebeu a ação, mas a resposta HTTP morreu no caminho. Não sabe se o mesmo item apareceu duas vezes na fonte. Não sabe se o cron disparou uma segunda instância porque a primeira atrasou.

Em automação pequena, essa ignorância parece aceitável. Em automação diária, ela vira dívida técnica com juros compostos. A primeira regra é parar de tratar scripts como eventos descartáveis. Um script que roda em produção precisa lembrar do que fez, impedir concorrência acidental e ter uma política clara para falhas.

2. A tríade mínima: lock, retry e idempotência

Se eu tivesse que escolher só três proteções para uma automação Python que roda por cron, escolheria estas:

• Lockfile: impede duas execuções simultâneas do mesmo job.
• Retry com backoff: tenta de novo quando uma falha temporária acontece, mas sem martelar o serviço.
• Idempotência: garante que repetir a mesma operação não produza um segundo efeito colateral.

O lockfile protege contra concorrência local. O retry protege contra instabilidade externa. A idempotência protege contra o pior caso: quando você não sabe se a ação anterior completou ou não. Sem idempotência, retry é só duplicação com autoestima.

3. Antes de codar: defina o que é uma unidade de trabalho

Uma automação resiliente começa com uma pergunta sem glamour: qual é o menor item que pode ser processado de forma independente? Esse item é a sua unidade de trabalho. Pode ser um lead, uma linha de CSV, uma URL, um pedido, uma mensagem, um arquivo ou uma data de relatório.

Para cada unidade de trabalho, você precisa conseguir responder:

• Qual é o identificador estável desse item?
• Qual ação será executada?
• Como saber se essa ação já foi feita?
• O que pode ser tentado novamente?
• O que exige intervenção humana?

Sem identificador estável, você não tem idempotência. Tem esperança. E esperança é um péssimo banco de dados.

4. Um exemplo realista: sincronizar eventos para um webhook

Vamos imaginar uma automação que pega eventos de uma fonte e envia cada evento para um webhook. A versão ingênua faria um loop e dispararia requisições. A versão decente registra cada item antes de agir, usa um identificador determinístico e mantém estado local em SQLite.

import hashlib
import json
import sqlite3
from datetime import datetime, timezone

DB_PATH = "jobs.sqlite3"


def stable_key(event: dict) -> str:
    raw = json.dumps({
        "source": event["source"],
        "external_id": event["external_id"],
        "action": "send_webhook_v1",
    }, sort_keys=True).encode("utf-8")
    return hashlib.sha256(raw).hexdigest()


def connect():
    conn = sqlite3.connect(DB_PATH)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS job_state (
            job_key TEXT PRIMARY KEY,
            status TEXT NOT NULL,
            attempts INTEGER NOT NULL DEFAULT 0,
            last_error TEXT,
            updated_at TEXT NOT NULL
        )
    """)
    return conn

O detalhe importante está no stable_key. Ele não depende da hora atual, nem da ordem em que o script rodou. Ele representa a intenção: para este evento externo, executar esta ação. Se o script encontrar o mesmo evento amanhã, a chave será igual. Isso permite bloquear duplicidade.

5. Lockfile: o cinto de segurança do cron

Cron é obediente, não é inteligente. Se você mandar rodar a cada cinco minutos, ele roda a cada cinco minutos mesmo que a execução anterior esteja presa numa API lenta desde a era do PHP 5. O lockfile evita que duas instâncias mexam no mesmo estado ao mesmo tempo.

import os
import fcntl
from contextlib import contextmanager

@contextmanager
def single_instance(lock_path="/tmp/minha-automacao.lock"):
    fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o644)
    try:
        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        yield
    except BlockingIOError:
        raise SystemExit("Outra execução já está em andamento.")
    finally:
        os.close(fd)

Esse exemplo usa fcntl, então é adequado para Linux. Em container, VPS ou servidor comum, resolve a vida. Se você roda em ambiente distribuído com múltiplas máquinas, lockfile local não basta. Aí você precisa de lock no banco, Redis, fila ou outro coordenador. Mas para muita automação pessoal e operacional, o lock local já elimina uma classe inteira de problemas bobos.

6. Retry bom não é “tenta até dar certo”

Retry sem limite é teimosia automatizada. Retry bom tem três características: limite de tentativas, espera crescente e filtro de erros. Você tenta de novo em timeout, 429, 502, 503. Você não tenta de novo quando recebeu 400 porque mandou payload inválido. A API já te respondeu que você fez besteira; insistir só transforma erro em spam.

import random
import time
import requests

RETRYABLE_STATUS = {429, 500, 502, 503, 504}


def post_with_retry(url, payload, headers=None, max_attempts=4):
    for attempt in range(1, max_attempts + 1):
        try:
            response = requests.post(url, json=payload, headers=headers, timeout=20)
            if response.status_code not in RETRYABLE_STATUS:
                response.raise_for_status()
                return response
        except requests.Timeout as exc:
            last_error = exc
        except requests.HTTPError:
            raise
        except requests.RequestException as exc:
            last_error = exc
        else:
            last_error = RuntimeError(f"HTTP {response.status_code}: {response.text[:300]}")

        if attempt == max_attempts:
            raise last_error

        sleep_for = min(60, (2 ** attempt) + random.uniform(0, 1.5))
        time.sleep(sleep_for)

O jitter no tempo de espera evita que várias execuções voltem ao mesmo tempo depois de uma queda. É aquele detalhe pequeno que parece frescura até você derrubar a API de destino junto com outros clientes igualmente entusiasmados.

7. Idempotência: o antídoto contra duplicação

Idempotência significa que executar a mesma operação uma ou várias vezes produz o mesmo resultado final. Em APIs bem desenhadas, você pode enviar um cabeçalho como Idempotency-Key. Em automações caseiras, muitas vezes você precisa construir isso na marra: registrar estado antes e depois, consultar se já fez, e nunca depender só da memória do processo.

from datetime import datetime, timezone


def now():
    return datetime.now(timezone.utc).isoformat()


def reserve_job(conn, job_key):
    row = conn.execute(
        "SELECT status FROM job_state WHERE job_key = ?",
        (job_key,)
    ).fetchone()

    if row and row[0] == "done":
        return False

    conn.execute("""
        INSERT INTO job_state (job_key, status, attempts, updated_at)
        VALUES (?, 'pending', 0, ?)
        ON CONFLICT(job_key) DO NOTHING
    """, (job_key, now()))
    conn.commit()
    return True


def mark_done(conn, job_key):
    conn.execute("""
        UPDATE job_state
        SET status = 'done', updated_at = ?, last_error = NULL
        WHERE job_key = ?
    """, (now(), job_key))
    conn.commit()


def mark_failed(conn, job_key, error):
    conn.execute("""
        UPDATE job_state
        SET status = 'failed', attempts = attempts + 1,
            last_error = ?, updated_at = ?
        WHERE job_key = ?
    """, (str(error)[:1000], now(), job_key))
    conn.commit()

Repare na ordem: antes de enviar, a automação reserva o job. Depois de enviar com sucesso, marca como concluído. Se falhar, marca como falho. Na próxima execução, ela consegue decidir se ignora, retenta ou manda para uma fila de revisão.

8. O fluxo completo em uma função principal

Agora juntamos as peças. O objetivo não é criar um framework. É deixar claro onde cada responsabilidade fica. O lock envolve a execução inteira. O banco guarda estado por item. O retry fica perto da chamada externa. A chave estável amarra tudo.

WEBHOOK_URL = "https://exemplo.com/webhook"


def process_event(conn, event):
    key = stable_key(event)

    if not reserve_job(conn, key):
        print(f"skip done: {key}")
        return

    payload = {
        "external_id": event["external_id"],
        "name": event["name"],
        "value": event["value"],
    }

    headers = {
        "Idempotency-Key": key,
        "User-Agent": "automacao-resiliente/1.0",
    }

    try:
        post_with_retry(WEBHOOK_URL, payload, headers=headers)
    except Exception as exc:
        mark_failed(conn, key, exc)
        raise
    else:
        mark_done(conn, key)


def main():
    with single_instance():
        conn = connect()
        for event in fetch_events():
            process_event(conn, event)

if __name__ == "__main__":
    main()

Esse desenho permite uma coisa essencial: parar e continuar. O processo pode morrer no evento 37 de 100. Na próxima execução, os 36 concluídos são ignorados. O 37 fica marcado conforme o ponto de falha. O restante segue. É assim que um script começa a se comportar como um trabalhador minimamente confiável, não como uma roleta com acesso à internet.

9. Como decidir o que retentar e o que travar

Nem toda falha merece retry. Algumas merecem pausa. Outras merecem alarme. Uma política simples pode separar erros em três grupos:

• Transitórios: timeout, rate limit, erro 5xx, DNS instável. Retente com backoff.
• Definitivos: payload inválido, autenticação negada, recurso inexistente. Pare e registre.
• Ambíguos: conexão caiu depois do envio, resposta truncada, destino não documenta idempotência. Use chave idempotente e verificação posterior.

O erro definitivo é onde muita automação mostra falta de maturidade. Se o token expirou, tentar cinquenta vezes não renova token por osmose. Se o payload está inválido, retry só produz logs maiores. Um bom job falha alto quando precisa falhar alto.

10. Observabilidade: log que ajuda, não log que decora

Log útil responde: qual item falhou, em qual etapa, com qual erro, depois de quantas tentativas e qual decisão foi tomada. “Erro ao processar” é quase uma ofensa. Não é log; é bilhete anônimo deixado na cena do crime.

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s job=%(job_key)s %(message)s"
)


def log_job(level, job_key, message, **extra):
    logging.log(level, message, extra={"job_key": job_key, **extra})

Se você quiser evoluir, grave também uma tabela de execuções: início, fim, quantidade de itens, concluídos, ignorados, falhos. Isso facilita responder se a automação está saudável sem abrir o terminal como quem consulta oráculo.

11. Onde guardar estado: arquivo, SQLite, Postgres ou fila?

Para automações pequenas, SQLite é uma escolha excelente. É simples, transacional, local e fácil de inspecionar. Para automações com múltiplos workers, alta concorrência ou execução distribuída, Postgres, Redis ou uma fila de verdade começam a fazer sentido.

Minha regra prática:

• Um processo em uma máquina: SQLite + lockfile resolve muita coisa.
• Vários processos na mesma máquina: SQLite ainda pode servir, mas pense bem no lock e no tempo de transação.
• Várias máquinas: use banco central, fila ou lock distribuído.
• Ação financeira ou irreversível: idempotência no destino não é opcional.

Não comece com Kubernetes para enviar três relatórios. Também não use um CSV como banco de estado para uma automação que emite cobrança. A maturidade está em calibrar o peso da solução ao tamanho do estrago possível.

12. Checklist antes de colocar no cron

Antes de confiar sua automação ao cron, passe por este checklist sem autoengano:

• Existe um lock para impedir execução simultânea?
• Cada unidade de trabalho tem uma chave estável?
• Itens concluídos são ignorados em execuções futuras?
• Retries têm limite, backoff e filtro de erro?
• Falhas definitivas param ou entram em revisão?
• Logs mostram item, etapa, tentativa e erro?
• Há um jeito simples de reprocessar apenas falhas?
• Segredos ficam fora do código?
• A automação pode ser interrompida e retomada sem duplicar ações?

Se você respondeu “não” para várias perguntas, não significa que seu script é ruim. Significa que ele ainda está na fase “funciona quando observado”. O próximo passo é torná-lo confiável quando ninguém está olhando.

13. Um comando de inspeção para recuperar falhas

Depois de armazenar estado, você ganha uma vantagem bonita: dá para consultar a bagunça. Por exemplo:

SELECT job_key, attempts, last_error, updated_at
FROM job_state
WHERE status = 'failed'
ORDER BY updated_at DESC
LIMIT 20;

Isso parece simples, mas muda a operação. Em vez de reler um log gigante, você enxerga a fila de falhas. Pode criar um modo --retry-failed, um relatório diário ou uma notificação quando o número de falhas passar de um limite. A automação deixa de ser um script jogado no servidor e vira um processo observável.

14. Conectando com outras camadas do AutoMente

Este tema conversa diretamente com a ideia de fila local que apareceu em Fila de Automação com SQLite. Também complementa a abordagem de diagnóstico em Triagem de Logs de Erro em Automações Python. E se você está blindando servidor, vale revisitar Chaves FIDO2 SSH, porque automação resiliente em servidor largado com senha fraca é só otimismo com cron.

Na categoria Lab da Garra, a régua é essa: menos teoria ornamental, mais mecanismo que sobrevive a segunda-feira. Uma automação boa não precisa ser grande. Precisa saber o que fez, o que não fez e o que não deve fazer duas vezes.

15. Fechamento: o script adulto é meio paranoico

O script adulto desconfia da rede, do relógio, do cron, do usuário, da API e principalmente de si mesmo. Ele registra estado. Ele evita rodar duplicado. Ele retenta com educação. Ele sabe quando parar. Ele permite auditoria. Ele não transforma uma falha pequena em uma duplicação enorme.

Automação resiliente com Python é menos sobre escrever código brilhante e mais sobre aceitar que o mundo externo é instável. A boa notícia é que a base não é complicada: lockfile, retry e idempotência já colocam você muitos quilômetros à frente do script ingênuo.

Agora eu quero saber: qual automação você quer ver desmontada aqui no AutoMente? Pode ser integração com planilha, bot de WhatsApp, monitor de site, triagem de e-mail, backup, publicação automática ou aquele script que você tem medo de rodar porque ele parece simples demais para ser confiável. Me diga o caso real; a graça está justamente onde dá errado.