Circuit Breaker Local com Python e SQLite: O Disjuntor Que Protege Suas Automações (Antes do Colapso)

Se suas automações dependem de APIs externas — e vamos ser honestos, todas dependem — você já passou por isso: o serviço remoto cai, seu script entra em loop de retry agressivo, a fila acumula, o disco enche, e você só descobre quando o cliente liga aos gritos.

O problema não é que APIs falham. O problema é que a maioria das automações não sabe parar quando algo está quebrado. Elas tentam, tentam, tentam — e transformam uma queda de 30 segundos num incidente de 3 horas.

Neste artigo, vou te mostrar como implementar um circuit breaker local com Python e SQLite que protege suas automações contra APIs instáveis, sem depender de nenhum serviço de monitoramento externo. Tudo roda na sua máquina, sem cloud, sem assinatura, sem complicação.

O que é um Circuit Breaker (e por que sua automação precisa de um)

O padrão Circuit Breaker vem do mundo da engenharia elétrica e foi popularizado no software por Martin Fowler. A ideia é simples e genial: quando um serviço remoto começa a falhar repetidamente, pare de chamá-lo por um tempo determinado. Dê ao serviço a chance de se recuperar — e proteja seu sistema do colapso.

O circuit breaker tem três estados:

• CLOSED (fechado): Operação normal. As chamadas fluem livremente.
• OPEN (aberto): O circuito disparou. Chamadas são bloqueadas imediatamente — sem tentar, sem esperar timeout, sem gastar recurso.
• HALF-OPEN (meio-aberto): Período de teste. Uma chamada de sondagem é permitida. Se funcionar, volta para CLOSED. Se falhar, reabre o circuito.

Traduzindo para o mundo real: é como aquele disjuntor da sua casa. Se a corrente passa do limite, ele desliga. Não adianta ficar ligando de novo a cada 2 segundos. Você espera esfriar, testa de novo, e vê se resolveu.

E o mais bonito? Você pode implementar isso inteiro em Python com SQLite, sem biblioteca externa, sem Redis, sem servidor de estado compartilhado. O SQLite gerencia o estado do circuito de forma persistente — se seu processo reiniciar, ele lembra que o circuito está aberto.

Arquitetura do Circuit Breaker Local

Antes de escrever código, vamos definir o que nosso circuit breaker precisa fazer:

1. Rastrear falhas por serviço: Cada API externa tem seu próprio circuito independente.
2. Contador de falhas consecutivas: Quando atinge um threshold (ex: 5 falhas), o circuito abre.
3. Timeout configurável: O circuito fica aberto por um tempo (ex: 60 segundos) antes de permitir uma nova tentativa.
4. Estado persistente: Se o processo morrer e voltar, o circuito mantém seu estado.
5. Registro de eventos: Log de todas as transições de estado para debug.

O banco SQLite terá duas tabelas: uma para o estado atual de cada circuito e outra para o histórico de transições. Simples, eficiente, e funciona até se você reiniciar o servidor.

Implementação: O Circuit Breaker em Python

Vamos construir passo a passo. Primeiro, a estrutura do banco:

import sqlite3
import time
import os
from datetime import datetime
from enum import Enum
from contextlib import contextmanager

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

DB_PATH = os.getenv("CB_DB_PATH", "/var/lib/automente/circuit_breaker.db")

def init_db(db_path=DB_PATH):
    """Inicializa o banco SQLite com as tabelas do circuit breaker."""
    os.makedirs(os.path.dirname(db_path), exist_ok=True)
    conn = sqlite3.connect(db_path)
    conn.executescript("""
        CREATE TABLE IF NOT EXISTS circuits (
            service_name TEXT PRIMARY KEY,
            state TEXT NOT NULL DEFAULT 'closed',
            failure_count INTEGER NOT NULL DEFAULT 0,
            last_failure_time REAL,
            last_success_time REAL,
            last_state_change REAL,
            opened_at REAL,
            threshold INTEGER NOT NULL DEFAULT 5,
            timeout_seconds REAL NOT NULL DEFAULT 60.0
        );

        CREATE TABLE IF NOT EXISTS circuit_log (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            service_name TEXT NOT NULL,
            from_state TEXT,
            to_state TEXT NOT NULL,
            reason TEXT,
            timestamp REAL NOT NULL
        );
    """)
    conn.commit()
    return conn

@contextmanager
def get_db(db_path=DB_PATH):
    """Context manager para conexões seguras com o SQLite."""
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    try:
        yield conn
    finally:
        conn.close()

Perceba o detalhe importante: cada serviço tem seu próprio threshold e timeout_seconds. Isso significa que você pode configurar thresholds diferentes para APIs críticas vs. APIs secundárias. Uma API de pagamento pode ter threshold 3 e timeout 30s. Uma API de notificação pode ter threshold 10 e timeout 120s.

A Classe CircuitBreaker

Agora a classe principal que encapsula toda a lógica:

class CircuitBreaker:
    def __init__(self, service_name, db_path=DB_PATH):
        self.service_name = service_name
        self.db_path = db_path
        self._ensure_circuit()

    def _ensure_circuit(self):
        """Garante que existe registro para este serviço."""
        with get_db(self.db_path) as conn:
            conn.execute(
                "INSERT OR IGNORE INTO circuits (service_name) VALUES (?)",
                (self.service_name,)
            )
            conn.commit()

    def _get_state(self):
        with get_db(self.db_path) as conn:
            row = conn.execute(
                "SELECT * FROM circuits WHERE service_name = ?",
                (self.service_name,)
            ).fetchone()
            return dict(row) if row else None

    def _log_transition(self, from_state, to_state, reason=""):
        with get_db(self.db_path) as conn:
            conn.execute(
                "INSERT INTO circuit_log (service_name, from_state, to_state, reason, timestamp) VALUES (?, ?, ?, ?, ?)",
                (self.service_name, from_state, to_state, reason, time.time())
            )
            conn.commit()

    def _transition(self, new_state, reason=""):
        current = self._get_state()
        old_state = current["state"]
        now = time.time()

        with get_db(self.db_path) as conn:
            if new_state == CircuitState.OPEN.value:
                conn.execute("""
                    UPDATE circuits SET state=?, failure_count=failure_count+1,
                    last_failure_time=?, last_state_change=?, opened_at=?
                    WHERE service_name=?
                """, (new_state, now, now, now, self.service_name))
            elif new_state == CircuitState.CLOSED.value:
                conn.execute("""
                    UPDATE circuits SET state=?, failure_count=0,
                    last_success_time=?, last_state_change=?
                    WHERE service_name=?
                """, (new_state, now, now, self.service_name))
            elif new_state == CircuitState.HALF_OPEN.value:
                conn.execute("""
                    UPDATE circuits SET state=?, last_state_change=?
                    WHERE service_name=?
                """, (new_state, now, self.service_name))

            conn.commit()

        self._log_transition(old_state, new_state, reason)
        print(f"[CIRCUIT] {self.service_name}: {old_state} → {new_state} ({reason})")

    def can_execute(self):
        """Verifica se uma chamada pode ser executada."""
        state = self._get_state()

        if state["state"] == CircuitState.CLOSED.value:
            return True

        if state["state"] == CircuitState.OPEN.value:
            elapsed = time.time() - (state["opened_at"] or 0)
            if elapsed >= state["timeout_seconds"]:
                # Timeout expirou: transição para half-open
                self._transition(CircuitState.HALF_OPEN.value, "timeout expired")
                return True
            return False

        # HALF_OPEN: permite uma tentativa
        return True

    def record_success(self):
        """Registra sucesso e possivelmente fecha o circuito."""
        state = self._get_state()

        if state["state"] == CircuitState.HALF_OPEN.value:
            # Sondagem funcionou: fecha o circuito
            self._transition(CircuitState.CLOSED.value, "probe succeeded")
        elif state["state"] == CircuitState.CLOSED.value:
            # Apenas registra o sucesso
            with get_db(self.db_path) as conn:
                conn.execute(
                    "UPDATE circuits SET last_success_time=?, failure_count=0 WHERE service_name=?",
                    (time.time(), self.service_name)
                )
                conn.commit()

    def record_failure(self):
        """Registra falha e possivelmente abre o circuito."""
        state = self._get_state()

        if state["state"] == CircuitState.HALF_OPEN.value:
            # Sondagem falhou: reabre o circuito
            self._transition(CircuitState.OPEN.value, "probe failed")
            return

        if state["state"] == CircuitState.CLOSED.value:
            new_count = state["failure_count"] + 1
            if new_count >= state["threshold"]:
                self._transition(CircuitState.OPEN.value, f"threshold reached ({new_count})")
            else:
                # Só incrementa, não loga transição
                with get_db(self.db_path) as conn:
                    conn.execute(
                        "UPDATE circuits SET failure_count=?, last_failure_time=? WHERE service_name=?",
                        (new_count, time.time(), self.service_name)
                    )
                    conn.commit()

💡 O perrengue que gerou esse código: Na primeira versão, eu usei time.sleep() dentro do circuit breaker para implementar o cooldown. Resultado? Thread travada, filas acumulando, caos. A versão correta é stateless por design: o can_execute() verifica o timestamp e decide. Quem chama é que decide se espera ou não. Nunca bloqueie dentro do breaker.

Como Usar: Wrapper para Chamadas HTTP

Agora vem a parte que realmente importa: como integrar isso com suas chamadas reais. O padrão mais limpo é um decorador (decorator) que envolve qualquer função que faça chamadas externas:

import requests
from functools import wraps

def protected_call(func):
    """Decorador que protege chamadas a APIs com circuit breaker."""
    @wraps(func)
    def wrapper(*args, **kwargs):
        service = kwargs.pop("cb_service", func.__name__)
        cb = CircuitBreaker(service)

        if not cb.can_execute():
            raise CircuitBreakerOpenError(
                f"Circuito aberto para '{service}'. "
                f"Serviço em cooldown, tentando novamente mais tarde."
            )

        try:
            result = func(*args, **kwargs)
            cb.record_success()
            return result
        except (requests.exceptions.RequestException, TimeoutError) as e:
            cb.record_failure()
            raise CircuitBreakerError(f"Falha em '{service}': {e}") from e

    return wrapper

class CircuitBreakerOpenError(Exception):
    """Erro lançado quando o circuito está aberto."""
    pass

class CircuitBreakerError(Exception):
    """Erro genérico do circuit breaker."""
    pass

# Exemplo de uso:
@protected_call
def consultar_api_pagamento(dados):
    return requests.post(
        "https://api.pagamento.com/v1/charge",
        json=dados,
        timeout=10
    )

# Chamada normal — o decorator cuida do circuit breaker:
try:
    resposta = consultar_api_pagamento(
        {"valor": 99.90, "cb_service": "api-pagamento"}
    )
    print(f"Pagamento processado: {resposta.status_code}")
except CircuitBreakerOpenError:
    print("⚡ API de pagamento em cooldown — registrando para retry posterior")
except CircuitBreakerError as e:
    print(f"❌ Falha na API: {e}")

Notou o detalhe? O decorador aceita um parâmetro cb_service que identifica qual circuito usar. Isso permite que uma mesma função tenha circuitos diferentes dependendo do contexto — útil quando você integra com múltiplos endpoints.

Dashboard CLI: Monitorando os Circuitos

De nada adianta ter um circuit breaker se você não sabe o estado dele. Aqui está um pequeno script CLI que mostra todos os circuitos ativos direto do terminal:

def status_dashboard(db_path=DB_PATH):
    """Exibe o estado atual de todos os circuitos no terminal."""
    with get_db(db_path) as conn:
        circuits = conn.execute(
            "SELECT * FROM circuits ORDER BY service_name"
        ).fetchall()

        if not circuits:
            print("Nenhum circuito registrado.")
            return

        print(f"\n{'═' * 70}")
        print(f"{'SERVIÇO':<25} {'ESTADO':<12} {'FALHAS':<8} {'ÚLTIMA FALHA':<15}")
        print(f"{'─' * 70}")

        for c in circuits:
            last_fail = "— "
            if c["last_failure_time"]:
                lt = datetime.fromtimestamp(c["last_failure_time"])
                last_fail = lt.strftime("%H:%M:%S")

            state_icon = {
                "closed": "✅",
                "open": "🔴",
                "half_open": "🟡"
            }.get(c["state"], "❓")

            print(f"{state_icon} {c['service_name']:<23} {c['state'].upper():<12} {c['failure_count']:<8} {last_fail}")

        print(f"{'═' * 70}\n")

def reset_circuit(service_name, db_path=DB_PATH):
    """Reseta manualmente um circuito para CLOSED."""
    with get_db(db_path) as conn:
        conn.execute("""
            UPDATE circuits SET state='closed', failure_count=0,
            last_state_change=?, opened_at=NULL
            WHERE service_name=?
        """, (time.time(), service_name))
        conn.commit()
        print(f"Circuito '{service_name}' resetado para CLOSED.")

Rode isso no cron a cada 5 minutos e você terá visibilidade completa do estado dos seus circuitos. Melhor que muitos dashboards SaaS que cobram por isso.

Cenários Reais de Aplicação

1. Proteção contra API de terceiros instável

Você tem uma automação que consulta a API de um serviço de clima a cada 10 minutos. O serviço tem 99% de uptime — mas aquele 1% acontece sempre na hora errada. Sem circuit breaker, seus logs ficam cheios de timeout. Com circuit breaker, depois de 5 falhas consecutivas, o circuito abre e suas outras automações continuam rodando normalmente.

2. Filas de processamento assíncrono

Seu worker consome mensagens de uma fila e chama uma API externa para cada mensagem. Se a API cair, sem circuit breaker o worker fica retrying e a fila cresce exponencialmente. Com circuit breaker, quando o circuito abre, o worker pode redirecionar as mensagens para uma dead letter queue local (outra tabela SQLite) e processar depois.

3. Webhooks e callbacks

Seu sistema envia webhooks para URLs de clientes. Alguns clientes ficam fora do ar por horas. Sem circuit breaker, você gasta recursos tentando entregar para endpoints mortos. Com circuit breaker por URL destino, você para de tentar automaticamente e notifica o cliente quando o circuito abre.

Configurações Recomendadas por Tipo de Serviço

Nem todo serviço merece o mesmo tratamento. Aqui vão minhas recomendações práticas depois de rodar isso em produção por meses:

Tipo de Serviço	Threshold	Timeout (s)	Motivo
API de Pagamento	3	30	Crítico, recupera rápido
API de Notificação	10	120	Menos crítica, tolera delay
API de Clima/Dados	5	60	Rate limiting comum
Webhooks de Cliente	8	300	Clientes demoram a voltar
API Interna (microserviço)	3	15	Deveria estar sempre no ar

Ajuste conforme sua realidade. O ponto é: threshold baixo para serviços críticos, timeout proporcional à criticidade.

Testando na Prática

Vamos simular um cenário real para ver o circuit breaker em ação:

def test_circuit_breaker():
    """Simula falhas consecutivas e verifica transições de estado."""
    cb = CircuitBreaker("api-teste")

    print("=== Simulando 6 falhas consecutivas ===")
    for i in range(6):
        cb.record_failure()
        state = cb._get_state()
        print(f"  Falha #{i+1}: estado={state['state']}, falhas={state['failure_count']}")

    print("\n=== Circuito deve estar ABERTO ===")
    print(f"  Pode executar? {cb.can_execute()}")  # False

    print("\n=== Aguardando timeout (simulado: resetando opened_at) ===")
    # Na prática, espere timeout_seconds. Aqui vamos forçar para teste:
    with get_db() as conn:
        conn.execute(
            "UPDATE circuits SET opened_at=? WHERE service_name=?",
            (time.time() - 65, "api-teste")  # 65s atrás
        )
        conn.commit()

    print(f"  Pode executar? {cb.can_execute()}")  # True → HALF_OPEN
    state = cb._get_state()
    print(f"  Estado após can_execute: {state['state']}")

    print("\n=== Sondagem funciona ===")
    cb.record_success()
    state = cb._get_state()
    print(f"  Estado final: {state['state']}")  # CLOSED

if __name__ == "__main__":
    init_db()
    test_circuit_breaker()

A saída esperada mostra claramente as transições: CLOSED → OPEN → HALF_OPEN → CLOSED. Cada transição fica registrada no circuit_log, então você pode auditar exatamente quando e por que cada circuito mudou de estado.

Erros Comuns (e como evitar)

Implementei circuit breaker três vezes antes de acertar. Aqui estão os erros que cometi para você não repetir:

1. Usar threading.Lock ao invés de SQLite para estado compartilhado. Se você tem múltiplos workers (gunicorn, celery), o lock de thread não compartilha estado entre processos. Use SQLite ou Redis. SQLite é mais simples e funciona perfeitamente para circuit breaker local.

2. Esquecer de resetar o failure_count no sucesso. Se o circuito fecha depois de 5 falhas mas o contador continua em 5, na próxima falha ele abre de novo imediatamente. Sempre zere o contador no record_success().

3. Timeout muito curto. Se sua API costuma ficar indisponível por 2 minutos e o timeout do circuit breaker é 30 segundos, ele vai abrir e fechar como pisca-pisca de árvore de Natal. Monitore o tempo real de indisponibilidade antes de configurar.

4. Não distinguir tipos de erro. Um 503 (Service Unavailable) é diferente de um 400 (Bad Request). Só conte como falha do circuito erros que indicam problema no servidor remoto (5xx, timeout, connection refused). Erros de validação (4xx) são problema seu, não do serviço.

Integração com Systemd para Resiliência Total

Para completar, rode seu serviço de automação como um unit do systemd com Restart=on-failure. Se o processador de filas morrer por qualquer motivo, o systemd reinicia. E quando reiniciar, o circuit breaker já está lá — estado preservado no SQLite. Zero configuração extra.

[Unit]
Description=Automação com Circuit Breaker
After=network.target

[Service]
Type=simple
User=automente
WorkingDirectory=/opt/automente
ExecStart=/opt/automente/venv/bin/python worker.py
Restart=on-failure
RestartSec=10

[Install]
WantedBy=multi-user.target

Conclusão: Sua Automação Merece um Disjuntor

Circuit breaker não é luxo — é higiene básica de automação. Toda vez que seu código depende de algo fora do seu controle (API externa, rede, DNS, outro serviço), você precisa de um plano para quando isso falhar.

A implementação que mostrei aqui é deliberadamente simples: Python puro, SQLite, zero dependências externas. Não precisa de Redis, não precisa de sidecar, não precisa de serviço de monitoramento. Funciona em qualquer servidor Linux, inclusive aquele VPS de 5 dólares que você usa para rodar suas automações.

O investimento é de algumas horas. O retorno é dormir tranquilo sabendo que, quando a API do parceiro cair às 3 da manhã, sua automação vai silenciosamente abrir o circuito, registrar no log, e continuar operando — em vez de gerar 47.000 linhas de erro e consumir toda a CPU em retry infinito.

E se você quer automatizar alguma coisa específica — monitoramento, deploy, integração com API, whatever — me conta aqui nos comentários. Qual automação você quer ver sendo construída do zero no próximo post?