Painel de disjuntor elétrico com fiação colorida - implementação de circuit breaker em Python puro

Circuit Breaker em Python Puro: O Disjuntor Que Impede Uma API Fora do Ar de Derrubar Todo o Seu Sistema em Cascata

PorOlivetto

Você já viu um sistema inteiro cair porque uma única API externa decidiu morrer no pior momento possível? Eu já. E não foi bonito.

Era uma terça-feira, 14h47. O sistema de cobrança do meu cliente parou de responder. Não tinha fallback. Não tinha proteção. Cada retry desesperado do nosso backend só piorava a situação — como tentar reanimar um paciente dando socos no peito com mais força. Em 4 minutos, 237 threads travadas, consumo de memória em 94%, e o load balancer começando a derrubar instâncias saudáveis por timeout.

Falência em cascata. O pesadelo de todo engenheiro de sistemas distribuídos.

Hoje eu vou te mostrar como implementar o padrão que impede exatamente isso: o Circuit Breaker em Python puro. Sem bibliotecas externas. Sem dependências exóticas. Apenas código que você pode colar no seu projeto agora e dormir tranquilo hoje à noite.

O Que É Um Circuit Breaker (E Por Que Você Precisa Dele Ontem)

O nome vem da engenharia elétrica. Sabe aquele disjuntor na sua caixa de luz? Quando a corrente passa do limite, ele desliga automaticamente. Não espera queimar. Não espera pegar fogo. Ele corta. Protege o resto do circuito.

No mundo de software, o Circuit Breaker faz exatamente a mesma coisa, só que com requisições HTTP:

  • Quando um serviço externo começa a falhar, o circuit breaker detecta o padrão de erros.
  • Após um threshold configurável, ele "abre o circuito" — para de enviar requisições para o serviço problemático.
  • As requisições falham imediatamente, sem gastar timeout, sem consumir threads, sem drenar recursos.
  • Após um tempo de cooldown, ele tenta uma requisição de teste (half-open). Se funcionar, fecha o circuito. Se falhar, abre de novo.

É simples. É elegante. E salva sistemas em produção todos os dias.

Os Três Estados do Circuit Breaker

Antes de codar, vamos entender a máquina de estados. Um circuit breaker tem exatamente três estados:

1. CLOSED (Fechado) — Operação Normal

O circuito está fechado, as requisições passam normalmente. O breaker está contando as falhas silenciosamente. Enquanto a taxa de erro estiver abaixo do threshold, nada acontece. É o estado de "confio que está tudo bem, mas estou de olho".

2. OPEN (Aberto) — Proteção Ativada

O threshold de falhas foi atingido. O circuito abre. Todas as requisições subsequentes falham imediatamente — sem chamar o serviço externo, sem gastar timeout, sem bloquear threads. É o equivalente digital a puxar o plugue da tomada.

3. HALF-OPEN (Semi-aberto) — Testando as Águas

Após um período de cooldown, o breaker permite uma única requisição de teste. Se ela succeed, o circuito fecha e a operação normal retorna. Se falhar, o circuito abre novamente e o timer de cooldown recomeça.

Rede complexa de cabeamento e painéis elétricos industriais - analogia com arquitetura de microsserviços

Assim como uma rede elétrica complexa precisa de proteção contra sobrecarga, sua arquitetura de microsserviços precisa de circuit breakers.

Implementação: Circuit Breaker do Zero em Python Puro

Vamos construir um circuit breaker completo, thread-safe, com contagem de falhas deslizante, estados bem definidos e métricas de observabilidade. Tudo em Python padrão — sem pip install de nada.

Primeiro, os estados:

from enum import Enum
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Callable, Any, Optional
import time
import threading


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

Simples, explícito, sem mágica. Três estados, três valores. Se alguém olhar esse código daqui a 6 meses, vai entender em 2 segundos.

Agora, o Circuit Breaker propriamente dito:

@dataclass
class CircuitBreaker:
    name: str
    failure_threshold: int = 5
    recovery_timeout: int = 30  # segundos
    success_threshold: int = 2  # sucessos consecutivos para fechar

    _failure_count: int = field(default=0, init=False)
    _success_count: int = field(default=0, init=False)
    _state: CircuitState = field(default=CircuitState.CLOSED, init=False)
    _last_failure_time: Optional[datetime] = field(default=None, init=False)
    _last_state_change: datetime = field(default_factory=datetime.now, init=False)
    _lock: threading.Lock = field(default_factory=threading.Lock, init=False)

    _total_calls: int = field(default=0, init=False)
    _total_failures: int = field(default=0, init=False)
    _total_successes: int = field(default=0, init=False)
    _total_rejected: int = field(default=0, init=False)
    _total_fallbacks: int = field(default=0, init=False)

    @property
    def state(self) -> CircuitState:
        with self._lock:
            if self._state == CircuitState.OPEN:
                if self._should_attempt_reset():
                    self._transition_to(CircuitState.HALF_OPEN)
                else:
                    return CircuitState.OPEN
            return self._state

    def _should_attempt_reset(self) -> bool:
        if self._last_failure_time is None:
            return False
        elapsed = datetime.now() - self._last_failure_time
        return elapsed.total_seconds() >= self.recovery_timeout

    def _transition_to(self, new_state: CircuitState) -> None:
        self._state = new_state
        self._last_state_change = datetime.now()
        if new_state == CircuitState.CLOSED:
            self._failure_count = 0
            self._success_count = 0
        elif new_state == CircuitState.HALF_OPEN:
            self._success_count = 0
        print(f"[{self.name}] Transição para {new_state.value}")

Repara em alguns detalhes importantes: thread-safe, auto-transição OPEN→HALF_OPEN, métricas embutidas, e success threshold evitando falsos positivos.

O método principal — call():

    def call(
        self,
        func: Callable,
        fallback: Optional[Callable] = None,
        *args,
        **kwargs
    ) -> Any:
        with self._lock:
            current_state = self.state
            self._total_calls += 1

            if current_state == CircuitState.OPEN:
                self._total_rejected += 1
                if fallback:
                    self._total_fallbacks += 1
                    return fallback(*args, **kwargs)
                raise CircuitOpenError(
                    f"Circuit '{self.name}' is OPEN. "
                    f"Rejeitando requisição para proteger o sistema."
                )

        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as exc:
            self._on_failure()
            if fallback:
                self._total_fallbacks += 1
                return fallback(*args, **kwargs)
            raise

    def _on_success(self) -> None:
        with self._lock:
            self._total_successes += 1
            if self._state == CircuitState.HALF_OPEN:
                self._success_count += 1
                if self._success_count >= self.success_threshold:
                    self._transition_to(CircuitState.CLOSED)
            elif self._state == CircuitState.CLOSED:
                self._failure_count = 0

    def _on_failure(self) -> None:
        with self._lock:
            self._total_failures += 1
            self._last_failure_time = datetime.now()

            if self._state == CircuitState.HALF_OPEN:
                self._transition_to(CircuitState.OPEN)
            elif self._state == CircuitState.CLOSED:
                self._failure_count += 1
                if self._failure_count >= self.failure_threshold:
                    self._transition_to(CircuitState.OPEN)


class CircuitOpenError(Exception):
    pass

O lock protege apenas a verificação de estado e atualização de contadores. A execução real acontece fora do lock. Isso é crucial — segurar um lock durante uma chamada HTTP é um anti-pattern que transforma proteção em gargalo.

Exemplo Prático: Circuit Breaker em um Serviço de Pagamentos

Imagine que você tem um serviço de pagamentos externo que decide ficar instável:

import requests
from dataclasses import dataclass


@dataclass
class PaymentResult:
    success: bool
    transaction_id: str
    message: str


payment_breaker = CircuitBreaker(
    name="pagamento-externo",
    failure_threshold=3,
    recovery_timeout=60,
    success_threshold=2
)


def cobrar_cliente(valor: float, cliente_id: str) -> PaymentResult:
    response = requests.post(
        "https://api.pagamento-externo.com/v1/cobrar",
        json={"valor": valor, "cliente_id": cliente_id},
        timeout=5
    )
    response.raise_for_status()
    data = response.json()
    return PaymentResult(
        success=data["status"] == "ok",
        transaction_id=data["tx_id"],
        message=data.get("message", "")
    )


def fallback_cobranca(valor: float, cliente_id: str) -> PaymentResult:
    print(f"⚠️ Fallback: cobrança de R${valor:.2f} do cliente {cliente_id} registrada para retry assíncrono")
    return PaymentResult(
        success=False,
        transaction_id="pending-retry",
        message="Cobrança enfileirada para processamento assíncrono"
    )


try:
    resultado = payment_breaker.call(
        cobrar_cliente,
        fallback=fallback_cobranca,
        valor=99.90,
        cliente_id="cli_12345"
    )
    print(f"Resultado: {resultado.message}")
except CircuitOpenError as e:
    print(f"⚡ Circuit breaker ativado: {e}")
except requests.RequestException as e:
    print(f"❌ Erro na cobrança: {e}")

Percebe o que acontece?

  1. As primeiras 2 falhas passam — ainda é "normal".
  2. Na 3ª falha consecutiva, o circuito abre.
  3. Todas as requisições seguintes vão direto pro fallback.
  4. Após 60 segundos, uma requisição de teste é enviada.
  5. Se voltou, 2 sucessos consecutivos fecham o circuito.
  6. Se ainda falha, circuito reabre e espera mais 60 segundos.

Painel de controle elétrico com circuitos e conexões - padrão circuit breaker em sistemas distribuídos

Cada conexão num painel de controle representa uma dependência entre serviços. Uma falha mal gerenciada aqui = colapso em cascata.

Configurações que Importam (E Os Valores Que Eu Uso em Produção)

Parâmetro Valor Padrão Por quê?
failure_threshold 3-5 Pouco demais = falso positivo. Muito demais = dano antes de proteger.
recovery_timeout 30-60s Tempo razoável para um serviço se recuperar de pico de carga.
success_threshold 2 Um sucesso pode ser sorte. Dois já é tendência.

Quando Ajustar Para Baixo

Se o serviço externo é crítico e raramente falha, use failure_threshold=3.

Quando Ajustar Para Cima

Se o serviço é naturalmente instável (ex: API de terceiros com SLA de 99%), use failure_threshold=5 ou mais.

🔥 O Perrengue Real

Implementei meu primeiro circuit breaker num sistema de notificações push. Coloquei failure_threshold=2 achando que era "conservador e seguro". Resultado: o serviço de push tinha um pico de latência de 200ms a cada 10 minutos. Meu circuit breaker abria a cada 10 minutos. Notificações iam todas pro fallback (email). Os usuários recebiam emails 8 minutos depois do push original. O CTO me chamou pra conversar. Desde então, meu threshold mínimo é 3 e eu monitore a taxa de falhas antes de decidir o número.

Circuit Breaker vs Retry: Quando Usar Cada Um

  • Retry é para falhas transitórias — um timeout isolado, um 503 momentâneo.
  • Circuit Breaker é para falhas persistentes — o serviço caiu, o banco travou, a rede caiu.

Eles são complementares. O padrão mais robusto é retry DENTRO do circuit breaker.

Observabilidade: Métricas Que Salvam Vidas

Um circuit breaker sem métricas é como um disjuntor sem etiqueta na caixa de luz.

    def metrics(self) -> dict:
        with self._lock:
            return {
                "name": self.name,
                "state": self._state.value,
                "failure_count": self._failure_count,
                "total_calls": self._total_calls,
                "total_failures": self._total_failures,
                "total_successes": self._total_successes,
                "total_rejected": self._total_rejected,
                "total_fallbacks": self._total_fallbacks,
                "failure_rate": (
                    self._total_failures / self._total_calls
                    if self._total_calls > 0 else 0
                ),
                "last_state_change": self._last_state_change.isoformat(),
            }

Configure alertas para:

  • Circuit breaker aberto por mais de 5 minutos → alerta de alta prioridade
  • Taxa de rejeição > 50% → serviço dependente está em colapso
  • Fallback sendo acionado > 100x/hora → investigate a causa raiz

Quando NÃO Usar Circuit Breaker

  • Operações síncronas críticas onde fallback não é aceitável.
  • Serviços com baixa latência e alta confiabilidade (ex: localhost → banco local). O overhead pode não compensar.
  • Quando o "serviço externo" é na verdade seu próprio código. Se sua função está falhando tanto, o problema está nela.

Resumo: O Checklist do Circuit Breaker

  1. ✅ Threshold de falhas baseado em dados reais
  2. ✅ Fallback implementado e testado
  3. ✅ Métricas expostas e monitoradas
  4. ✅ Alertas configurados para estado OPEN prolongado
  5. ✅ Recovery timeout ajustado ao SLA do serviço dependente
  6. ✅ Thread-safe
  7. ✅ Logs de transição de estado
  8. ✅ Testes unitários cobrindo as 3 transições

Se você implementou até aqui, parabéns. Você agora tem uma arma a mais contra falhas em cascata.

Qual automação ou padrão de engenharia você quer ver implementado do zero em Python puro? Me conta aqui embaixo. Os mais pedidos viram post.

📂 Veja mais artigos da categoria Lab da Garra →

Posts relacionados:
Dead Letter Queue em Python Puro — A quarentena que salva mensagens perdidas
Graceful Shutdown em Python Puro — O protocolo que faz seu processo morrer sem perder dados
Rate Limiter em Python Puro — O guardião que impede sua automação de virar um ataque DDoS

Posts Similares