CAPTCHA API Error Handling: Production Best Practices

import httpx
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def create_task(client, params):
    try:
        response = await client.post("/createTask", json=params)
        response.raise_for_status()
        return response.json()
    except httpx.TimeoutException:
        raise  # Let tenacity retry
    except httpx.HTTPStatusError as e:
        if e.response.status_code >= 500:
            raise  # Retry server errors
        raise e  # Don't retry client errors

class CaptchaAPIError(Exception):
    def __init__(self, error_id: int, description: str):
        self.error_id = error_id
        self.description = description
        super().__init__(f"Error {error_id}: {description}")

ERROR_CODES = {
    1: "Invalid API key",
    2: "Insufficient balance",
    3: "Task not found",
    10: "Site key invalid",
    12: "Task timeout",
}

def handle_api_response(data: dict):
    if error_id := data.get("errorId"):
        desc = ERROR_CODES.get(error_id, data.get("errorDescription", "Unknown"))
        raise CaptchaAPIError(error_id, desc)

import asyncio

async def solve_with_timeout(solver, params, timeout=120):
    try:
        return await asyncio.wait_for(
            solver.solve(**params),
            timeout=timeout
        )
    except asyncio.TimeoutError:
        # Log for metrics
        logger.warning(f"CAPTCHA solve timed out after {timeout}s")
        raise

# With fallback
async def solve_with_fallback(solvers, params):
    for solver in solvers:
        try:
            return await solve_with_timeout(solver, params)
        except Exception as e:
            logger.warning(f"Solver {solver.name} failed: {e}")
            continue
    raise Exception("All solvers failed")

from datetime import datetime, timedelta
from enum import Enum

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

class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: int = 60,
        half_open_max: int = 3
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.half_open_max = half_open_max

        self.state = CircuitState.CLOSED
        self.failures = 0
        self.last_failure_time = None
        self.half_open_calls = 0

    async def call(self, func, *args, **kwargs):
        if self.state == CircuitState.OPEN:
            if self._should_try_reset():
                self.state = CircuitState.HALF_OPEN
                self.half_open_calls = 0
            else:
                raise Exception("Circuit breaker is open")

        try:
            result = await func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise

    def _should_try_reset(self):
        return (
            self.last_failure_time and
            datetime.now() - self.last_failure_time >
            timedelta(seconds=self.recovery_timeout)
        )

    def _on_success(self):
        if self.state == CircuitState.HALF_OPEN:
            self.half_open_calls += 1
            if self.half_open_calls >= self.half_open_max:
                self.state = CircuitState.CLOSED
                self.failures = 0

    def _on_failure(self):
        self.failures += 1
        self.last_failure_time = datetime.now()

        if self.failures >= self.failure_threshold:
            self.state = CircuitState.OPEN

class CaptchaSolverWithFallback:
    def __init__(self, primary_solver, fallback_strategy):
        self.primary = primary_solver
        self.fallback = fallback_strategy
        self.circuit = CircuitBreaker()

    async def solve(self, **params):
        try:
            return await self.circuit.call(
                self.primary.solve,
                **params
            )
        except Exception as e:
            logger.warning(f"Primary solver failed: {e}")
            return await self.fallback(**params)

# Fallback strategies
async def queue_for_manual_review(**params):
    """Queue the task for manual processing"""
    await task_queue.push({
        "type": "captcha_manual",
        "params": params,
        "created_at": datetime.now()
    })
    return None

async def skip_captcha_action(**params):
    """Skip the action requiring CAPTCHA"""
    logger.info(f"Skipping CAPTCHA-protected action: {params}")
    return None

from dataclasses import dataclass
from collections import deque

@dataclass
class SolverMetrics:
    total_attempts: int = 0
    successes: int = 0
    failures: int = 0
    timeouts: int = 0
    avg_solve_time: float = 0.0

    recent_errors: deque = None

    def __post_init__(self):
        self.recent_errors = deque(maxlen=100)

    @property
    def success_rate(self):
        if self.total_attempts == 0:
            return 0.0
        return self.successes / self.total_attempts * 100

    def should_alert(self):
        # Alert if success rate drops below 80%
        return self.success_rate < 80 and self.total_attempts > 10

# Usage
metrics = SolverMetrics()

async def solve_with_metrics(solver, params):
    start = time.time()
    metrics.total_attempts += 1

    try:
        result = await solver.solve(**params)
        metrics.successes += 1
        metrics.avg_solve_time = (
            metrics.avg_solve_time * 0.9 +
            (time.time() - start) * 0.1
        )
        return result
    except asyncio.TimeoutError:
        metrics.timeouts += 1
        raise
    except Exception as e:
        metrics.failures += 1
        metrics.recent_errors.append(str(e))
        raise