Manejo de Errores

{
  "error": "error_code",
  "message": "Descripción legible del error"
}

{
  "error": "unauthorized",
  "message": "API key is required. Provide it via X-API-Key header."
}

# Verificar que la key está incluida
headers = {"X-API-Key": "sk_live_tu_key_aqui"}

# Verificar formato correcto
assert api_key.startswith("sk_live_"), "Invalid key format"

{
  "error": "invalid_api_key",
  "message": "The API key provided is invalid."
}

{
  "error": "api_key_revoked",
  "message": "This API key has been revoked."
}

{
  "error": "forbidden",
  "message": "API access not available for your plan"
}

{
  "error": "company_suspended",
  "message": "Your company account has been suspended"
}

{
  "error": "validation_error",
  "message": "Invalid request data"
}

{
  "error": "validation_error",
  "message": "Validation failed",
  "details": [
    {
      "field": "leads[0].linkedin_url",
      "message": "Required field"
    }
  ]
}

def validate_lead(lead):
    errors = []
    if not lead.get('linkedin_url'):
        errors.append("linkedin_url is required")
    return errors

{
  "error": "max_leads_exceeded",
  "message": "Maximum 100 leads per request"
}

def create_jobs_batched(leads, batch_size=100):
    jobs = []
    for i in range(0, len(leads), batch_size):
        batch = leads[i:i + batch_size]
        job = api.create_job(batch)
        jobs.append(job)
    return jobs

{
  "error": "duplicate_linkedin_url",
  "message": "Duplicate linkedin_url found in request",
  "details": [
    {
      "linkedin_url": "https://www.linkedin.com/in/johndoe/",
      "positions": [2, 5]
    }
  ]
}

{
  "error": "insufficient_credits",
  "message": "Not enough credits. Available: 50, Required: 200"
}

def ensure_credits_available(leads):
    credits_needed = len(leads)
    usage = api.get_usage()

    if usage['credits_available'] < credits_needed:
        shortfall = credits_needed - usage['credits_available']
        raise InsufficientCreditsError(
            f"Need {shortfall} more credits."
        )

{
  "error": "job_not_found",
  "message": "Job not found"
}

{
  "error": "job_not_completed",
  "message": "Results are only available for completed jobs"
}

{
  "error": "api_key_not_found",
  "message": "API key not found"
}

{
  "error": "rate_limit_exceeded",
  "message": "Rate limit exceeded. Maximum 300 requests per minute.",
  "retry_after": 32
}

X-RateLimit-Limit: 300
X-RateLimit-Remaining: 0
X-RateLimit-Reset: 1704710460
Retry-After: 32

import time

def request_with_retry(request_func):
    try:
        return request_func()
    except RateLimitError as e:
        time.sleep(e.retry_after)
        return request_func()

{
  "error": "max_concurrent_jobs_exceeded",
  "message": "Maximum 10 concurrent jobs allowed. Wait for existing jobs to complete."
}

{
  "error": "internal_error",
  "message": "An internal error occurred. Please try again."
}

def request_with_backoff(request_func, max_retries=3):
    for attempt in range(max_retries):
        try:
            return request_func()
        except InternalServerError as e:
            if attempt == max_retries - 1:
                raise
            wait = 2 ** attempt  # 1, 2, 4 segundos
            time.sleep(wait)

{
  "error": "service_unavailable",
  "message": "Service temporarily unavailable. Please try again later."
}

import requests
import time
from typing import Any

class MasLeadsError(Exception):
    def __init__(self, error: str, message: str, **kwargs):
        self.error = error
        self.message = message
        self.extra = kwargs
        super().__init__(f"{error}: {message}")

class AuthenticationError(MasLeadsError): pass
class PaymentError(MasLeadsError): pass
class ValidationError(MasLeadsError): pass
class RateLimitError(MasLeadsError):
    def __init__(self, error, message, retry_after=60, **kwargs):
        super().__init__(error, message, **kwargs)
        self.retry_after = retry_after
class NotFoundError(MasLeadsError): pass
class ServerError(MasLeadsError): pass

class MasLeadsClient:
    BASE_URL = "https://api.masleads.com/api/v1"

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.session = requests.Session()
        self.session.headers.update({
            "X-API-Key": api_key,
            "Content-Type": "application/json"
        })

    def _handle_error(self, response: requests.Response):
        """Convierte respuestas de error en excepciones."""
        try:
            data = response.json()
            error = data.get("error", "unknown")
            message = data.get("message", "Unknown error")
        except:
            error = "unknown"
            message = response.text

        if response.status_code == 401:
            raise AuthenticationError(error, message)
        elif response.status_code == 402:
            raise PaymentError(error, message)
        elif response.status_code == 403:
            raise AuthenticationError(error, message)
        elif response.status_code == 400:
            raise ValidationError(error, message)
        elif response.status_code == 404:
            raise NotFoundError(error, message)
        elif response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", 60))
            raise RateLimitError(error, message, retry_after)
        elif response.status_code >= 500:
            raise ServerError(error, message)

    def _request(self, method: str, endpoint: str, **kwargs) -> Any:
        """Hace request con manejo de errores."""
        url = f"{self.BASE_URL}{endpoint}"
        response = self.session.request(method, url, **kwargs)

        if not response.ok:
            self._handle_error(response)

        return response.json()

    def _request_with_retry(self, method: str, endpoint: str,
                           max_retries: int = 3, **kwargs) -> Any:
        """Request con retry automático."""
        for attempt in range(max_retries):
            try:
                return self._request(method, endpoint, **kwargs)
            except RateLimitError as e:
                if attempt == max_retries - 1:
                    raise
                time.sleep(e.retry_after)
            except ServerError:
                if attempt == max_retries - 1:
                    raise
                time.sleep(2 ** attempt)

    def create_job(self, leads: list, field: str):
        return self._request_with_retry(
            "POST", "/enrich/leads",
            json={"leads": leads, "field": field}
        )

    def get_job_status(self, job_id: str):
        return self._request_with_retry("GET", f"/jobs/{job_id}")

    def get_job_results(self, job_id: str):
        return self._request_with_retry("GET", f"/jobs/{job_id}/results")

    def get_usage(self):
        return self._request_with_retry("GET", "/usage")


# Uso
client = MasLeadsClient("sk_live_tu_key")

try:
    job = client.create_job(
        leads=[{"linkedin_url": "https://linkedin.com/in/johndoe"}],
        field="email"
    )
except AuthenticationError:
    print("API Key inválida - verificar configuración")
except PaymentError as e:
    print(f"Sin créditos: {e.message}")
except ValidationError as e:
    print(f"Datos inválidos: {e.message}")
except RateLimitError as e:
    print(f"Rate limit - esperar {e.retry_after}s")

# ❌ Malo
response = api.create_job(leads)

# ✅ Bueno
try:
    response = api.create_job(leads)
except MasLeadsError as e:
    log_error(e)
    handle_gracefully(e)

import logging

logger = logging.getLogger(__name__)

try:
    job = api.create_job(leads)
except MasLeadsError as e:
    logger.error(
        "API error",
        extra={
            "error": e.error,
            "message": e.message,
            "leads_count": len(leads)
        }
    )

def handle_error(error):
    if isinstance(error, PaymentError):
        alert_admin("Credits depleted - action required")
    elif isinstance(error, AuthenticationError) and error.error == "api_key_revoked":
        alert_admin("API Key revoked - check settings")