Como Evitar Quedas de Serviço ao Usar uma API Gratuita de CPF

import time
import hashlib
from collections import OrderedDict
import requests
import os

class CPFCache:
    def __init__(self, max_entries: int = 10000, ttl_seconds: int = 86400):
    self.cache = OrderedDict()
    self.max_entries = max_entries
    self.ttl = ttl_seconds
    self.hits = 0
    self.misses = 0

    def get(self, cpf: str):
    chave = self._hash(cpf)
    if chave in self.cache:
    entrada = self.cache[chave]
    if time.time() - entrada["timestamp"] < self.ttl:
    self.hits += 1
    self.cache.move_to_end(chave)
    return entrada["dados"]
    else:
    del self.cache[chave]
    self.misses += 1
    return None

    def set(self, cpf: str, dados: dict):
    chave = self._hash(cpf)
    if len(self.cache) >= self.max_entries:
    self.cache.popitem(last=False)
    self.cache[chave] = {
    "dados": dados,
    "timestamp": time.time()
    }

    def _hash(self, cpf: str) -> str:
    return hashlib.sha256(cpf.encode()).hexdigest()

    @property
    def taxa_hit(self) -> float:
    total = self.hits + self.misses
    return (self.hits / total * 100) if total > 0 else 0

class CPFServiceResilient:
    def __init__(self, api_key: str):
    self.api_key = api_key
    self.base_url = "https://api.cpfhub.io/cpf"
    self.cache = CPFCache(max_entries=10000, ttl_seconds=86400)
    self.session = requests.Session()
    self.session.headers["x-api-key"] = api_key

    def consultar(self, cpf: str) -> dict:
    cpf_limpo = "".join(c for c in cpf if c.isdigit())

    # Camada 1: Cache
    cached = self.cache.get(cpf_limpo)
    if cached:
    return {**cached, "origem": "cache"}

    # Camada 2: API com timeout
    try:
    response = self.session.get(
    f"{self.base_url}/{cpf_limpo}",
    timeout=10
    )

    if response.status_code == 200:
    dados = response.json()
    if dados.get("success"):
    self.cache.set(cpf_limpo, dados["data"])
    return {**dados["data"], "origem": "api"}

    return {"valido": False, "erro": "CPF não encontrado", "origem": "api"}

    except Exception:
    # Camada 3: Fallback local
    return self._validacao_local(cpf_limpo)

    def _validacao_local(self, cpf: str) -> dict:
    digits = [int(d) for d in cpf]
    if len(digits) != 11 or len(set(digits)) == 1:
    return {"valido": False, "erro": "Formato inválido", "origem": "local"}

    soma1 = sum(digits[i] * (10 - i) for i in range(9))
    check1 = 0 if soma1 % 11 < 2 else 11 - (soma1 % 11)
    soma2 = sum(digits[i] * (11 - i) for i in range(10))
    check2 = 0 if soma2 % 11 < 2 else 11 - (soma2 % 11)

    if digits[9] == check1 and digits[10] == check2:
    return {
    "valido": None,
    "nota": "Formato válido, dados pendentes de confirmação",
    "origem": "local"
    }
    return {"valido": False, "erro": "CPF inválido", "origem": "local"}

import time
from enum import Enum

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

class CircuitBreaker:
    def __init__(
    self,
    falhas_para_abrir: int = 5,
    tempo_recuperacao: int = 60,
    sucesso_para_fechar: int = 3
    ):
    self.estado = CircuitState.CLOSED
    self.falhas_consecutivas = 0
    self.sucessos_half_open = 0
    self.falhas_para_abrir = falhas_para_abrir
    self.tempo_recuperacao = tempo_recuperacao
    self.sucesso_para_fechar = sucesso_para_fechar
    self.ultimo_falha = 0

    def pode_executar(self) -> bool:
    if self.estado == CircuitState.CLOSED:
    return True
    if self.estado == CircuitState.OPEN:
    if time.time() - self.ultimo_falha > self.tempo_recuperacao:
    self.estado = CircuitState.HALF_OPEN
    self.sucessos_half_open = 0
    return True
    return False
    return True # HALF_OPEN permite tentativa

    def registrar_sucesso(self):
    if self.estado == CircuitState.HALF_OPEN:
    self.sucessos_half_open += 1
    if self.sucessos_half_open >= self.sucesso_para_fechar:
    self.estado = CircuitState.CLOSED
    self.falhas_consecutivas = 0
    else:
    self.falhas_consecutivas = 0

    def registrar_falha(self):
    self.falhas_consecutivas += 1
    self.ultimo_falha = time.time()
    if self.falhas_consecutivas >= self.falhas_para_abrir:
    self.estado = CircuitState.OPEN

# Integração com o serviço
class CPFServiceComCircuitBreaker:
    def __init__(self, api_key: str):
    self.service = CPFServiceResilient(api_key)
    self.breaker = CircuitBreaker(falhas_para_abrir=5, tempo_recuperacao=60)

    def consultar(self, cpf: str) -> dict:
    if not self.breaker.pode_executar():
    return self.service._validacao_local(
    "".join(c for c in cpf if c.isdigit())
    )

    resultado = self.service.consultar(cpf)

    if resultado.get("origem") == "api":
    self.breaker.registrar_sucesso()
    elif resultado.get("origem") == "local":
    self.breaker.registrar_falha()

    return resultado

class MonitorSaude:
    def __init__(self, service: CPFServiceComCircuitBreaker):
    self.service = service
    self.metricas = {
    "total": 0,
    "cache_hits": 0,
    "api_calls": 0,
    "fallbacks": 0,
    "erros": 0
    }

    def registrar(self, resultado: dict):
    self.metricas["total"] += 1
    origem = resultado.get("origem", "desconhecido")

    if origem == "cache":
    self.metricas["cache_hits"] += 1
    elif origem == "api":
    self.metricas["api_calls"] += 1
    elif origem == "local":
    self.metricas["fallbacks"] += 1

    if resultado.get("valido") is False:
    self.metricas["erros"] += 1

    def relatorio(self) -> dict:
    total = max(self.metricas["total"], 1)
    return {
    "total_consultas": self.metricas["total"],
    "taxa_cache": f"{self.metricas['cache_hits']/total*100:.1f}%",
    "taxa_api": f"{self.metricas['api_calls']/total*100:.1f}%",
    "taxa_fallback": f"{self.metricas['fallbacks']/total*100:.1f}%",
    "circuit_breaker": self.service.breaker.estado.value,
    "cache_hit_rate": f"{self.service.service.cache.taxa_hit:.1f}%"
    }