import os
import time
import random
import requests
import logging
from typing import Optional, Dict, Any
class JuditAPIClient:
def __init__(self, api_key: str):
self.api_key = api_key
self.base_headers = {'api-key': api_key}
self.logger = logging.getLogger(__name__)
def _make_request(
self,
method: str,
url: str,
**kwargs
) -> Optional[requests.Response]:
"""
Faz requisição com tratamento de erro e retry
"""
max_retries = 3
base_delay = 1.0
for attempt in range(max_retries + 1):
try:
response = requests.request(
method=method,
url=url,
headers=self.base_headers,
timeout=30,
**kwargs
)
# Log da requisição
self._log_request(response, attempt + 1)
# Sucesso
if response.status_code < 400:
return response
# Tratamento de erros
if not self._should_retry(response, attempt, max_retries):
return response
# Aguardar antes do retry
delay = self._calculate_delay(response, attempt, base_delay)
time.sleep(delay)
except requests.exceptions.RequestException as e:
self.logger.error(f"Erro de conexão (tentativa {attempt + 1}): {e}")
if attempt < max_retries:
delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
time.sleep(delay)
else:
raise e
return None
def _should_retry(
self,
response: requests.Response,
attempt: int,
max_retries: int
) -> bool:
"""
Determina se deve fazer retry
"""
if attempt >= max_retries:
return False
# Retry para rate limit
if response.status_code == 429:
return True
# Retry para erros 5xx
if response.status_code >= 500:
return True
# Não fazer retry para outros erros
return False
def _calculate_delay(
self,
response: requests.Response,
attempt: int,
base_delay: float
) -> float:
"""
Calcula delay para retry
"""
if response.status_code == 429:
# Usar Retry-After se disponível
retry_after = response.headers.get('Retry-After')
if retry_after:
return int(retry_after)
# Backoff exponencial com jitter
return base_delay * (2 ** attempt) + random.uniform(0, 1)
def _log_request(self, response: requests.Response, attempt: int) -> None:
"""
Registra detalhes da requisição
"""
if response.status_code >= 400:
try:
error_data = response.json().get('error', {})
self.logger.error(
f"Erro API (tentativa {attempt}): "
f"Status {response.status_code}, "
f"Código: {error_data.get('code', 'UNKNOWN')}, "
f"Request ID: {error_data.get('request_id', 'N/A')}"
)
except:
self.logger.error(
f"Erro API (tentativa {attempt}): Status {response.status_code}"
)
else:
self.logger.info(f"Requisição bem-sucedida (tentativa {attempt})")
# Uso
client = JuditAPIClient(os.getenv('JUDIT_API_KEY'))
response = client._make_request('GET', 'https://requests.prod.judit.io/requests')
