Circuit Breaker em Python Puro: O Mecanismo Que Detecta Falhas em Cascata e Desliga o Serviço Problemático Antes Que Ele Derrube Todo o Sistema (Sem Istio, Sem Hystrix)
Você já teve aquela sensação de que seu sistema inteiro vai cair porque um único microsserviço resolveu tirar uma soneca? Eu já. Várias vezes. E posso te dizer: não existe nada mais humilhante do que ver sua aplicação inteira travando em cascata enquanto você assiste impotente aos logs vomitando ConnectionTimeout a cada 30 milissegundos.
Foi exatamente isso que aconteceu comigo numa sexta-feira à noite, 47 minutos antes de um deploy crítico. O serviço de pagamentos estava lento — não morto, só lento. Mas o suficiente para fazer meu serviço de pedidos esperar, esperar, esperar… até esgotar o pool de threads e derrubar tudo junto. Um dominó perfeito. Quatrocentos usuários receberam “500 Internal Server Error” enquanto eu tentava entender por que diabos um requests.get() com timeout de 5 segundos estava destruindo minha arquitetura inteira.
A solução? Um padrão que eu conhecia de nome, mas nunca tinha implementado na unha: Circuit Breaker. E não, eu não queria instalar Istio, nem Hystrix, nem nenhuma daquelas bibliotecas que adicionam 47 dependências transitivas ao meu requirements.txt. Eu queria entender o mecanismo. Querendo implementar em Python puro, sem mágica, sem framework — só a lógica nua e crua de quando desligar a tomada antes que o incêndio se espalhe.
Se você já passou por isso — ou quer estar preparado para quando passar — esse artigo é o guia completo que eu gostaria de ter tido naquela sexta-feira.
O Que É Circuit Breaker e Por Que Seu Sistema Precisa Dele
Imagine o disjuntor da sua casa. Quando algo dá errado na fiação — um curto-circuito, uma sobrecarga — o disjuntor “abre” e corta a energia antes que os fios derretam e sua casa pegue fogo. Ele não tenta consertar o problema. Ele apenas interrompe o fluxo para proteger o resto do sistema.
O Circuit Breaker Pattern faz exatamente isso para chamadas entre serviços. Ele monitora as requisições e, quando detecta que o serviço remoto está falhando consistentemente, ele “abre o circuito” e para de tentar. Em vez de esperar 30 segundos por um timeout que nunca vem, a chamada falha imediatamente — e seu sistema pode responder com um fallback, uma mensagem de erro amigável, ou simplesmente seguir com a vida.
Os três estados de um Circuit Breaker são:
- Closed (Fechado): Tudo funcionando normal. Requisições passam. O contador de falhas está zerado ou baixo.
- Open (Aberto): O serviço remoto está falhando demais. Requisições são bloqueadas imediatamente e retornam erro sem nem tentar conectar.
- Half-Open (Meio-Aberto): Depois de um tempo no estado Open, o Circuit Breaker permite uma requisição de teste passar. Se funcionar, volta para Closed. Se falhar, volta para Open.
Parece simples, né? E é. Mas a beleza está nos detalhes: quantas falhas até abrir? Quanto tempo esperar antes de tentar de novo? Como evitar race conditions quando 200 threads tentam fazer a transição de estado ao mesmo tempo?
A Arquitetura de Um Circuit Breaker em Python Puro
Antes de escrever uma linha de código, vamos desenhar a estrutura. Um Circuit Breaker precisa de:
- Estado atual: CLOSED, OPEN ou HALF_OPEN
- Contador de falhas: Quantas requisições consecutivas falharam
- Threshold de falhas: Quantas falhas até abrir o circuito (ex: 5)
- Timeout de recuperação: Quanto tempo esperar antes de tentar de novo (ex: 30 segundos)
- Timestamp da última falha: Quando foi a última vez que algo deu errado
- Função protegida: O que estamos tentando chamar (a API remota, o banco de dados, etc.)
- Fallback: O que fazer quando o circuito está aberto (opcional, mas recomendado)
Aqui está a primeira versão — simples, funcional, e sem nenhuma dependência externa:
import time
from enum import Enum
from typing import Callable, Any, Optional
class CircuitState(Enum):
CLOSED = "closed"
OPEN = "open"
HALF_OPEN = "half_open"
class CircuitBreaker:
def __init__(
self,
func: Callable,
failure_threshold: int = 5,
recovery_timeout: int = 30,
fallback: Optional[Callable] = None
):
self.func = func
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.fallback = fallback
self.state = CircuitState.CLOSED
self.failure_count = 0
self.last_failure_time = 0
def call(self, *args, **kwargs) -> Any:
if self.state == CircuitState.OPEN:
if time.time() - self.last_failure_time >= self.recovery_timeout:
self.state = CircuitState.HALF_OPEN
else:
if self.fallback:
return self.fallback(*args, **kwargs)
raise CircuitBreakerOpenError("Circuito aberto - serviço indisponível")
try:
result = self.func(*args, **kwargs)
self._on_success()
return result
except Exception as e:
self._on_failure()
if self.state == CircuitState.OPEN and self.fallback:
return self.fallback(*args, **kwargs)
raise
def _on_success(self):
self.failure_count = 0
self.state = CircuitState.CLOSED
def _on_failure(self):
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
class CircuitBreakerOpenError(Exception):
pass
Parece bom, né? Mas tem um problema que eu só descobri na prática: essa implementação não é thread-safe. Se você tiver 50 threads chamando call() ao mesmo tempo, elas podem ler self.state como HALF_OPEN, todas tentarem a requisição, e todas falharem — derrotando o propósito do Circuit Breaker.
Tornando Thread-Safe (Sem Usar Locks Desnecessários)
A solução ingênua é jogar um threading.Lock() em tudo. Funciona, mas introduz contenção desnecessária. Uma abordagem mais elegante usa atomic operations via threading.Lock() apenas nas transições de estado:
import threading
class ThreadSafeCircuitBreaker(CircuitBreaker):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._lock = threading.Lock()
def call(self, *args, **kwargs) -> Any:
# Check state with lock only on transitions
with self._lock:
if self.state == CircuitState.OPEN:
if time.time() - self.last_failure_time >= self.recovery_timeout:
self.state = CircuitState.HALF_OPEN
else:
if self.fallback:
return self.fallback(*args, **kwargs)
raise CircuitBreakerOpenError()
# Execute without lock (allow concurrent calls)
try:
result = self.func(*args, **kwargs)
self._on_success_safe()
return result
except Exception as e:
self._on_failure_safe()
if self.state == CircuitState.OPEN and self.fallback:
return self.fallback(*args, **kwargs)
raise
def _on_success_safe(self):
with self._lock:
self.failure_count = 0
self.state = CircuitState.CLOSED
def _on_failure_safe(self):
with self._lock:
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
Agora sim. O lock protege apenas as leituras e escritas do estado compartilhado, mas a execução da função em si (que é a parte lenta) acontece fora do lock. Isso permite que múltiplas threads executem a função protegida concorrentemente, mas garante que as transições de estado sejam atômicas.
Testando na Prática: O Cenário Que Me Salvou Naquela Sexta-Feira
Vamos criar um cenário realista: uma função que simula um serviço remoto instável, com 70% de chance de falhar nas primeiras 10 chamadas, e depois se estabiliza.
import random
class UnstableService:
def __init__(self):
self.call_count = 0
def __call__(self, user_id: int) -> dict:
self.call_count += 1
# Simula instabilidade nas primeiras 10 chamadas
if self.call_count <= 10:
if random.random() < 0.7: # 70% de falha
raise ConnectionError("Serviço remoto indisponível")
# Depois estabiliza
return {"user_id": user_id, "status": "success"}
def fallback_response(user_id: int) -> dict:
return {
"user_id": user_id,
"status": "fallback",
"message": "Serviço temporariamente indisponível. Tente novamente em alguns segundos."
}
# Setup
service = UnstableService()
breaker = ThreadSafeCircuitBreaker(
func=service,
failure_threshold=5,
recovery_timeout=10,
fallback=fallback_response
)
# Teste
results = []
for i in range(20):
try:
result = breaker.call(user_id=i)
results.append(result)
print(f"Call {i+1}: {result['status']}")
except Exception as e:
results.append({"error": str(e)})
print(f"Call {i+1}: ERROR - {e}")
print(f"\nTotal calls to actual service: {service.call_count}")
print(f"Fallback responses: {sum(1 for r in results if r.get('status') == 'fallback')}")
O que você vai ver nas primeiras 5-7 chamadas é uma mistura de successes e fallbacks. Depois que o Circuit Breaker abre, todas as chamadas subsequentes retornam fallback imediatamente — sem esperar timeout, sem consumir recursos, sem sobrecarregar o serviço que já está sofrendo.
Quando os 10 segundos de recovery passam, o Circuit Breaker entra em HALF_OPEN e permite uma tentativa. Se funcionar, volta para CLOSED e o fluxo normal é restaurado.
Adicionando Métricas: Porque Se Você Não Mede, Você Não Sabe Se Está Funcionando
Um Circuit Breaker sem métricas é como um disjuntor sem label — você só descobre que ele existe quando tudo para de funcionar. Vamos adicionar contadores para:
- Total de chamadas
- Chamadas bem-sucedidas
- Chamadas que falharam
- Chamadas que retornaram fallback
- Transições de estado
from dataclasses import dataclass, field
from collections import defaultdict
@dataclass
class CircuitBreakerMetrics:
total_calls: int = 0
successful_calls: int = 0
failed_calls: int = 0
fallback_calls: int = 0
state_transitions: list = field(default_factory=list)
def record_success(self):
self.total_calls += 1
self.successful_calls += 1
def record_failure(self):
self.total_calls += 1
self.failed_calls += 1
def record_fallback(self):
self.total_calls += 1
self.fallback_calls += 1
def record_transition(self, from_state: CircuitState, to_state: CircuitState):
self.state_transitions.append({
"from": from_state.value,
"to": to_state.value,
"timestamp": time.time()
})
def summary(self) -> dict:
return {
"total": self.total_calls,
"success_rate": self.successful_calls / max(1, self.total_calls),
"failure_rate": self.failed_calls / max(1, self.total_calls),
"fallback_rate": self.fallback_calls / max(1, self.total_calls),
"transitions": len(self.state_transitions)
}
class MonitoredCircuitBreaker(ThreadSafeCircuitBreaker):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.metrics = CircuitBreakerMetrics()
def call(self, *args, **kwargs) -> Any:
old_state = self.state
try:
result = super().call(*args, **kwargs)
if self.state != old_state:
self.metrics.record_transition(old_state, self.state)
self.metrics.record_success()
return result
except CircuitBreakerOpenError:
self.metrics.record_fallback()
if self.state != old_state:
self.metrics.record_transition(old_state, self.state)
raise
except Exception as e:
self.metrics.record_failure()
if self.state != old_state:
self.metrics.record_transition(old_state, self.state)
raise
# Uso
breaker = MonitoredCircuitBreaker(
func=UnstableService(),
failure_threshold=5,
recovery_timeout=10,
fallback=fallback_response
)
# ... depois de várias chamadas ...
print(breaker.metrics.summary())
# {'total': 20, 'success_rate': 0.35, 'failure_rate': 0.35, 'fallback_rate': 0.30, 'transitions': 3}
Circuit Breaker vs. Retry: Quando Usar Cada Um (E Quando Usar Ambos)
Aqui vai uma confissão: eu costumava confundir Circuit Breaker com Retry. “Ah, se a chamada falhar, é só tentar de novo, né?” Não. São ferramentas diferentes para problemas diferentes.
Retry é para falhas transitórias — timeouts, erros 503, “tente novamente em 1 segundo”. Você assume que o serviço vai se recuperar rápido e vale a pena insistir.
Circuit Breaker é para falhas persistentes — o serviço caiu, o banco está sobrecarregado, a rede está particionada. Você assume que insistir só vai piorar as coisas (para você E para o serviço que já está sofrendo).
A combinação poderosa é: Circuit Breaker + Retry com backoff exponencial. O Retry tenta algumas vezes com intervalos crescentes. Se mesmo assim não funcionar, o Circuit Breaker abre e para de tentar por um tempo.
class RetryWithCircuitBreaker:
def __init__(self, func, max_retries=3, base_delay=1, **cb_kwargs):
self.func = func
self.max_retries = max_retries
self.base_delay = base_delay
self.breaker = MonitoredCircuitBreaker(func=self._retry_wrapper, **cb_kwargs)
def _retry_wrapper(self, *args, **kwargs):
for attempt in range(self.max_retries):
try:
return self.func(*args, **kwargs)
except Exception as e:
if attempt == self.max_retries - 1:
raise
delay = self.base_delay * (2 ** attempt) # 1s, 2s, 4s
time.sleep(delay)
def call(self, *args, **kwargs):
return self.breaker.call(*args, **kwargs)
Para um guia completo sobre Retry com backoff, veja o artigo Retry com Backoff Exponencial e Jitter em Python Puro que publiquei semana passada.
Padrões Avançados: Sliding Window e Health Checks Proativos
A implementação básica conta falhas consecutivas. Mas e se você tiver 1000 requisições por segundo, e 4 falharem, depois 500 successes, e mais 1 falhar? O contador consecutivo não captura bem a “saúde” do serviço.
Uma abordagem mais sofisticada usa sliding window — conta falhas nos últimos N segundos ou nas últimas N requisições:
from collections import deque
class SlidingWindowCircuitBreaker(ThreadSafeCircuitBreaker):
def __init__(self, *args, window_size: int = 100, **kwargs):
super().__init__(*args, **kwargs)
self.window_size = window_size
self.results_window = deque(maxlen=window_size)
def _on_success(self):
self.results_window.append(True)
self._check_health()
def _on_failure(self):
self.results_window.append(False)
self._check_health()
def _check_health(self):
if len(self.results_window) < self.window_size:
return
failure_rate = sum(1 for r in self.results_window if not r) / len(self.results_window)
if failure_rate > 0.5: # Mais de 50% de falhas na janela
self.state = CircuitState.OPEN
self.last_failure_time = time.time()
elif self.state == CircuitState.HALF_OPEN and failure_rate < 0.1:
self.state = CircuitState.CLOSED
Outra melhoria: health checks proativos. Em vez de esperar uma requisição real para testar o serviço (no estado HALF_OPEN), você pode ter uma thread de background que periodicamente faz um ping leve (um GET /health ou SELECT 1) para verificar se o serviço voltou:
import threading
class ProactiveCircuitBreaker(SlidingWindowCircuitBreaker):
def __init__(self, *args, health_check: Callable, check_interval: int = 10, **kwargs):
super().__init__(*args, **kwargs)
self.health_check = health_check
self.check_interval = check_interval
self._start_health_monitor()
def _start_health_monitor(self):
def monitor():
while True:
time.sleep(self.check_interval)
if self.state == CircuitState.OPEN:
try:
self.health_check()
self.state = CircuitState.HALF_OPEN
except:
pass # Ainda não voltou
thread = threading.Thread(target=monitor, daemon=True)
thread.start()
Quando NÃO Usar Circuit Breaker (Sim, Existem Casos)
Nem tudo precisa de Circuit Breaker. Aqui vão alguns casos onde ele pode ser overkill ou até prejudicial:
- Chamadas síncronas críticas: Se a operação precisa acontecer (um pagamento, uma transação financeira), Circuit Breaker pode falhar rápido demais. Nesse caso, use Retry agressivo + fila de reprocessamento.
- Sistemas monolíticos: Se tudo está no mesmo processo, Circuit Breaker entre funções internas raramente faz sentido. Use para chamadas externas.
- Operações idempotentes com fila: Se você já tem uma fila (Celery, RabbitMQ, SQS) que reprocessa automaticamente, o Circuit Breaker pode ser redundante.
- Latência ultrabaixa: O overhead do Circuit Breaker (mesmo que mínimo, ~1μs) pode ser inaceitável em sistemas de trading ou real-time.
Deploy em Produção: O Checklist Que Eu Uso
Antes de colocar qualquer Circuit Breaker em produção, eu passo por este checklist:
- Threshold calibrado: 5 falhas é pouco? 20 é muito? Depende do seu tráfego. Eu começo com 5 e ajusto baseado nas métricas.
- Recovery timeout realista: 30 segundos é bom para APIs HTTP. Para bancos de dados, talvez 60-120 segundos.
- Fallback testado: O fallback realmente funciona? Ele não depende do mesmo serviço que está falhando?
- Métricas expostas: Você consegue ver no Grafana/Datadog quantas vezes o circuito abriu esta semana?
- Alertas configurados: Se o circuito abrir 10 vezes em 1 hora, alguém é notificado?
- Logs detalhados: Cada transição de estado é logada com timestamp, motivo, e contexto?
- Testes de caos: Você já simulou o serviço remoto caindo em staging para ver como o Circuit Breaker se comporta?
Para uma abordagem mais completa sobre observabilidade de logs, veja o artigo Log Anomaly Detector em Python Puro que mostra como detectar picos anômalos automaticamente.
Conclusão: O Padrão Que Todo Backend Deveria Conhecer
Circuit Breaker não é bala de prata. Mas é uma das ferramentas mais subutilizadas no arsenal de quem trabalha com sistemas distribuídos. A diferença entre “meu sistema caiu porque um serviço lento derrubou tudo” e “meu sistema degrada gracefully e se recupera sozinho” é, muitas vezes, apenas 50 linhas de Python.
A implementação que mostrei aqui é simples, thread-safe, e não depende de nenhuma biblioteca externa. Você pode copiá-la, adaptá-la, e colocá-la em produção hoje. Mas o mais importante não é o código — é o mindset: pare de tratar chamadas remotas como se fossem chamadas locais. Elas falham. Elas demoram. Elas desaparecem sem aviso. E seu sistema precisa estar preparado para isso.
Qual foi a última vez que um serviço remoto derrubou sua aplicação inteira? Que automação de resiliência você gostaria de ver implementada aqui? Me conta nos comentários — ou no Log de Erros, onde a gente disseca esses padrões na prática.
Se você curtiu esse artigo, compartilhe com aquele colega que ainda usa try/except pelado e reza para o serviço remoto não cair. Ele vai te agradecer depois.
