Circuit Breaker em Python Puro: O Disjuntor Que Impede Seu Serviço de Virar Efeito Dominó Quando a API Cai

PorOlivetto

Você já viu isso: uma API externa começa a responder com timeout. Seu serviço tenta de novo. E de novo. E de novo. Em 30 segundos, você tem 200 threads travadas esperando resposta, a memória estourou, e o que era um problema em um endpoint virou queda geral. O nome disso é cascading failure — e o Circuit Breaker é o disjuntor que desliga o circuito antes do incêndio.

Hoje vou construir um Circuit Breaker completo em Python puro. Sem tenacity, sem pybreaker, sem dependência nenhuma. Só a stdlib. No final, você terá um decorator pronto para produção que protege qualquer função contra falhas em cascata.

O Problema: Por Que Retry Simples Não Resolve

O instinto quando uma chamada falha é tentar de novo. E faz sentido — falhas transitórias existem. Mas existe uma diferença brutal entre:

  • Falha transitória: a API tropeçou, tenta de novo em 2 segundos.
  • Falha persistente: a API caiu de verdade. Tentar de novo só queima recurso seu.

Sem Circuit Breaker, seu código trata os dois casos igual. E é aí que o efeito dominó começa.

O Padrão: Três Estados, Uma Lógica

Um Circuit Breaker tem três estados:

CLOSED — tudo normal, chamadas passam. Se falhar além do limite, abre o circuito.

OPEN — circuito aberto, chamadas são rejeitadas imediatamente (fail fast). Após um timeout, tenta uma chamada de teste.

HALF_OPEN — a chamada de teste passou? Volta para CLOSED. Falhou? Volta para OPEN.

É exatamente como um disjuntor elétrico. Abre quando esquenta demais. Testa se pode fechar. Fecha se a corrente normalizou.

O Código: Circuit Breaker Completo

"""Circuit Breaker em Python puro.
Zero dependências. Pronto para produção.
"""
import time
import threading
from enum import Enum
from functools import wraps
from typing import Callable, Optional, Type, Tuple


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


class CircuitBreakerOpen(Exception):
    """Levantada quando o circuito está aberto e a chamada é rejeitada."""
    def __init__(self, remaining_seconds: float):
        self.remaining_seconds = remaining_seconds
        super().__init__(
            f"Circuito aberto. Tente novamente em {remaining_seconds:.1f}s"
        )


class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: float = 30.0,
        success_threshold: int = 2,
        excluded_exceptions: Optional[Tuple[Type[Exception], ...]] = None,
    ):
        """
        Args:
            failure_threshold: falhas consecutivas para abrir o circuito.
            recovery_timeout: segundos em OPEN antes de testar HALF_OPEN.
            success_threshold: sucessos consecutivos em HALF_OPEN para fechar.
            excluded_exceptions: exceções que NÃO contam como falha
                (ex: ValidationError -- é erro do cliente, não do serviço).
        """
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.success_threshold = success_threshold
        self.excluded_exceptions = excluded_exceptions or ()

        self._state = State.CLOSED
        self._failure_count = 0
        self._success_count = 0
        self._last_failure_time: Optional[float] = None
        self._lock = threading.Lock()

    @property
    def state(self) -> State:
        with self._lock:
            if self._state == State.OPEN:
                elapsed = time.monotonic() - self._last_failure_time
                if elapsed >= self.recovery_timeout:
                    self._state = State.HALF_OPEN
                    self._success_count = 0
            return self._state

    def call(self, func: Callable, *args, **kwargs):
        """Executa func dentro do Circuit Breaker."""
        current = self.state  # trigger transition check

        if current == State.OPEN:
            remaining = self.recovery_timeout - (
                time.monotonic() - self._last_failure_time
            )
            raise CircuitBreakerOpen(max(0, remaining))

        try:
            result = func(*args, **kwargs)
        except self.excluded_exceptions:
            raise  # não conta como falha do circuito
        except Exception as exc:
            self._on_failure()
            raise

        self._on_success()
        return result

    def _on_success(self):
        with self._lock:
            if self._state == State.HALF_OPEN:
                self._success_count += 1
                if self._success_count >= self.success_threshold:
                    self._reset()
            else:
                # CLOSED: sucesso reseta contador de falhas
                self._failure_count = 0

    def _on_failure(self):
        with self._lock:
            self._failure_count += 1
            self._last_failure_time = time.monotonic()

            if self._state == State.HALF_OPEN:
                self._state = State.OPEN
            elif self._failure_count >= self.failure_threshold:
                self._state = State.OPEN

    def _reset(self):
        self._state = State.CLOSED
        self._failure_count = 0
        self._success_count = 0

    def reset(self):
        """Reset manual (útil em testes)."""
        with self._lock:
            self._reset()


def circuit_breaker(
    failure_threshold: int = 5,
    recovery_timeout: float = 30.0,
    success_threshold: int = 2,
    excluded_exceptions: Optional[Tuple[Type[Exception], ...]] = None,
    fallback: Optional[Callable] = None,
):
    """Decorator para proteger qualquer função com Circuit Breaker.

    Args:
        fallback: função alternativa chamada quando o circuito está aberto.
                  Recebe os mesmos args da função original.
    """
    breaker = CircuitBreaker(
        failure_threshold=failure_threshold,
        recovery_timeout=recovery_timeout,
        success_threshold=success_threshold,
        excluded_exceptions=excluded_exceptions,
    )

    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs):
            try:
                return breaker.call(func, *args, **kwargs)
            except CircuitBreakerOpen:
                if fallback is not None:
                    return fallback(*args, **kwargs)
                raise

        wrapper.breaker = breaker  # expõe o breaker para inspeção
        return wrapper

    return decorator

Usando na Prática

Protegendo uma chamada HTTP

import urllib.request
import json


def _fallback_busca(query):
    """Cache local quando a API está fora."""
    return {"results": [], "source": "fallback", "query": query}


@circuit_breaker(
    failure_threshold=3,
    recovery_timeout=60.0,
    success_threshold=2,
    fallback=_fallback_busca,
)
def buscar(query: str) -> dict:
    url = f"https://api.exemplo.com/busca?q={urllib.parse.quote(query)}"
    req = urllib.request.Request(url, headers={"Accept": "application/json"})
    with urllib.request.urlopen(req, timeout=5) as resp:
        return json.loads(resp.read().decode())


# Uso normal
resultado = buscar("python circuit breaker")
print(resultado)

# Inspecionar estado do breaker
print(buscar.breaker.state)  # State.CLOSED

Protegendo queries de banco

import sqlite3

# Exceções de validação não devem abrir o circuito
class QueryInvalida(Exception):
    pass


@circuit_breaker(
    failure_threshold=5,
    recovery_timeout=120.0,
    excluded_exceptions=(QueryInvalida,),
)
def executar_query(sql: str, params: tuple = ()) -> list:
    if not sql.strip().upper().startswith("SELECT"):
        raise QueryInvalida("Apenas SELECT permitido aqui")

    conn = sqlite3.connect("app.db")
    try:
        cursor = conn.execute(sql, params)
        return cursor.fetchall()
    finally:
        conn.close()


try:
    rows = executar_query("SELECT * FROM usuarios WHERE ativo = ?", (1,))
except CircuitBreakerOpen as e:
    print(f"Banco indisponível, retry em {e.remaining_seconds:.0f}s")
except QueryInvalida:
    print("Query inválida -- isso não afeta o circuito")

Monitoramento: O Circuit Breaker Precisa Falar

Um Circuit Breaker que não emite logs é um disjuntor no escuro. Você não sabe quando abriu até tudo parar. Vamos adicionar observabilidade:

import logging

logger = logging.getLogger("circuit_breaker")


class ObservableCircuitBreaker(CircuitBreaker):
    def __init__(self, name: str = "default", **kwargs):
        super().__init__(**kwargs)
        self.name = name

    def _on_failure(self):
        old_state = self._state
        super()._on_failure()
        logger.warning(
            "[CB:%s] falha #%d | %s -> %s",
            self.name,
            self._failure_count,
            old_state.value,
            self._state.value,
        )

    def _on_success(self):
        old_state = self._state
        super()._on_success()
        if old_state != State.CLOSED:
            logger.info(
                "[CB:%s] sucesso | %s -> %s",
                self.name,
                old_state.value,
                self._state.value,
            )


# Uso com nome para identificar nos logs
breaker_api = ObservableCircuitBreaker(
    name="api-pagamentos",
    failure_threshold=3,
    recovery_timeout=60.0,
)

Testando: Simulando Falhas em Cascata

"""Teste que demonstra o Circuit Breaker funcionando."""
import time

cb = CircuitBreaker(failure_threshold=3, recovery_timeout=2.0, success_threshold=1)

call_count = 0


def unstable_service():
    global call_count
    call_count += 1
    raise ConnectionError("API fora do ar")


# Fase 1: falhas acumulam até abrir
for i in range(3):
    try:
        cb.call(unstable_service)
    except ConnectionError:
        pass

assert cb.state == State.OPEN
assert call_count == 3
print(f"Circuito abriu após {call_count} falhas")

# Fase 2: chamadas são rejeitadas sem tocar o serviço
for i in range(10):
    try:
        cb.call(unstable_service)
    except CircuitBreakerOpen:
        pass

assert call_count == 3  # não aumentou!
print(f"10 chamadas rejeitadas. Serviço não foi tocado (calls={call_count})")

# Fase 3: após recovery_timeout, teste em HALF_OPEN
time.sleep(2.1)
assert cb.state == State.HALF_OPEN
print("Circuito em HALF_OPEN, testando...")

# Simula recuperação
def stable_service():
    return "ok"

result = cb.call(stable_service)
assert cb.state == State.CLOSED
print(f"Circuito fechou. Resultado: {result}")

Box Perrengue: O Dia que o Circuit Breaker Me Salvou de uma Conta de R$ 4.000

Eu tinha um serviço que consumia uma API de geolocalização. Cobrança por request. Numa terça-feira, a API começou a dar 503. Meu código fazia retry com backoff exponencial — 1s, 2s, 4s, 8s, 16s. Parecia esperto.

Só que eu tinha 50 workers processando uma fila de 10.000 endereços. Cada worker tentava, falhava, esperava, tentava de novo. Em 20 minutos, eu tinha disparado mais de 12.000 requests para a API caída. A boa notícia: a maioria foi rejeitada com 503, sem cobrança. A má notícia: quando a API voltou, os workers ainda tinham milhares de itens na fila e dispararam tudo de uma vez.

A conta veio salgada porque os picos de request foram taxados como “burst”. Com Circuit Breaker, o circuito teria aberto no worker 3. Quando a API voltasse, o HALF_OPEN teria testado com uma única chamada antes de liberar o fluxo. A diferença entre R$ 80 e R$ 4.000 era um disjuntor de 30 linhas.

Quando NÃO Usar Circuit Breaker

Nem tudo precisa de Circuit Breaker. Evite quando:

Chamadas locais sem I/O. Funções puras de CPU não travam em cascata. O Circuit Breaker adiciona latência de lock desnecessária.

Operações idempotentes com retry natural. Se já tem retry inteligente e a operação é barata, o Circuit Breaker pode ser overkill.

Comunicação síncrona single-thread. Sem concorrência, não há efeito dominó. Um simples try/except resolve.

Arquitetura: Circuit Breaker vs. Retry vs. Bulkhead

Esses três padrões são complementares, não concorrentes:

Retry resolve falhas transitórias. Circuit Breaker resolve falhas persistentes. Bulkhead isola recursos para que um serviço não consuma todos os threads dos outros.

A combinação ideal:

"""Retry + Circuit Breaker + Bulkhead (thread pool isolado)."""
import concurrent.futures

# Bulkhead: pool dedicado para esta API (não compartilha threads)
api_pool = concurrent.futures.ThreadPoolExecutor(
    max_workers=5, thread_name_prefix="api-geo"
)


@circuit_breaker(failure_threshold=3, recovery_timeout=30.0)
def geolocalizar(endereco: str) -> dict:
    req = urllib.request.Request(
        f"https://api.geo.com/v1/locate?q={urllib.parse.quote(endereco)}",
        headers={"Accept": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=5) as resp:
        return json.loads(resp.read().decode())


def geolocalizar_com_retry(endereco: str, max_retries: int = 2) -> dict:
    """Retry DENTRO do Circuit Breaker (não fora)."""
    last_exc = None
    for attempt in range(max_retries + 1):
        try:
            return geolocalizar(endereco)
        except (ConnectionError, TimeoutError) as e:
            last_exc = e
            if attempt < max_retries:
                time.sleep(1.0 * (attempt + 1))
    raise last_exc


# Uso via bulkhead
future = api_pool.submit(geolocalizar_com_retry, "Rua Augusta, SP")
resultado = future.result(timeout=15)

CTA: Qual Serviço Seu Está Desprotegido?

Abre o terminal agora. Conta quantas chamadas HTTP externas seu código faz sem proteção. Se a resposta for “mais de zero”, cola esse Circuit Breaker ali. São 80 linhas, zero dependências, e já salvou minha conta de R$ 4.000.

Qual foi a última vez que uma API externa derrubou seu serviço inteiro? E o que você fez pra impedir que acontecesse de novo? Conta aí nos comentários — bora trocar perrengue.

Posts Similares