Tutorials

Abstract CAPTCHA Solver Interface: Provider-Agnostic Design

Hardcoding CaptchaAI API calls throughout your scraping code creates tight coupling. If you swap providers, add a mock for testing, or want to route between services based on price, you rewrite everything. An abstract interface defines the contract once — implementations handle the details.

Why Abstract the Solver?

Scenario Without abstraction With abstraction
Switch providers Modify every file that calls the API Swap one implementation class
Run tests Mock HTTP calls or hit the live API Inject a mock solver
A/B test providers Duplicate all solving code Create two implementations, route via config
Add logging/metrics Modify every solve call Wrap the interface with a decorator

Python: Protocol-Based Interface

Python's Protocol defines structural typing — any class with matching methods satisfies the interface without explicit inheritance.

import requests
import time
from typing import Protocol, runtime_checkable
from dataclasses import dataclass

SUBMIT_URL = "https://ocr.captchaai.com/in.php"
RESULT_URL = "https://ocr.captchaai.com/res.php"


@dataclass
class SolveResult:
    """Standardised result from any solver."""
    token: str
    solve_time_ms: int
    provider: str
    cost_usd: float | None = None


@runtime_checkable
class CaptchaSolver(Protocol):
    """Contract that any CAPTCHA solving provider must satisfy."""

    def solve_recaptcha_v2(self, sitekey: str, pageurl: str) -> SolveResult: ...
    def solve_recaptcha_v3(self, sitekey: str, pageurl: str,
                           action: str, min_score: float) -> SolveResult: ...
    def solve_turnstile(self, sitekey: str, pageurl: str) -> SolveResult: ...
    def solve_hcaptcha(self, sitekey: str, pageurl: str) -> SolveResult: ...
    def solve_image(self, base64_image: str) -> SolveResult: ...
    def get_balance(self) -> float: ...


# --- CaptchaAI implementation ---

class CaptchaAISolver:
    """CaptchaAI implementation of the solver interface."""

    def __init__(self, api_key: str):
        self._api_key = api_key

    def _submit_and_poll(self, params: dict, timeout: int = 180) -> tuple[str, int]:
        params["key"] = self._api_key
        params["json"] = 1

        start = time.monotonic()
        resp = requests.post(SUBMIT_URL, data=params, timeout=30).json()
        if resp.get("status") != 1:
            raise RuntimeError(f"Submit failed: {resp.get('request')}")

        task_id = resp["request"]
        while time.monotonic() - start < timeout:
            time.sleep(5)
            poll = requests.get(RESULT_URL, params={
                "key": self._api_key, "action": "get",
                "id": task_id, "json": 1,
            }, timeout=15).json()

            if poll.get("request") == "CAPCHA_NOT_READY":
                continue
            if poll.get("status") == 1:
                elapsed_ms = int((time.monotonic() - start) * 1000)
                return poll["request"], elapsed_ms
            raise RuntimeError(f"Solve failed: {poll.get('request')}")

        raise RuntimeError("Timeout")

    def _result(self, token: str, solve_time_ms: int) -> SolveResult:
        return SolveResult(token=token, solve_time_ms=solve_time_ms, provider="captchaai")

    def solve_recaptcha_v2(self, sitekey: str, pageurl: str) -> SolveResult:
        token, ms = self._submit_and_poll({
            "method": "userrecaptcha", "googlekey": sitekey, "pageurl": pageurl,
        })
        return self._result(token, ms)

    def solve_recaptcha_v3(self, sitekey: str, pageurl: str,
                           action: str = "verify", min_score: float = 0.3) -> SolveResult:
        token, ms = self._submit_and_poll({
            "method": "userrecaptcha", "version": "v3",
            "googlekey": sitekey, "pageurl": pageurl,
            "action": action, "min_score": str(min_score),
        })
        return self._result(token, ms)

    def solve_turnstile(self, sitekey: str, pageurl: str) -> SolveResult:
        token, ms = self._submit_and_poll({
            "method": "turnstile", "sitekey": sitekey, "pageurl": pageurl,
        })
        return self._result(token, ms)

    def solve_hcaptcha(self, sitekey: str, pageurl: str) -> SolveResult:
        token, ms = self._submit_and_poll({
            "method": "hcaptcha", "sitekey": sitekey, "pageurl": pageurl,
        })
        return self._result(token, ms)

    def solve_image(self, base64_image: str) -> SolveResult:
        token, ms = self._submit_and_poll({"method": "base64", "body": base64_image})
        return self._result(token, ms)

    def get_balance(self) -> float:
        resp = requests.get(RESULT_URL, params={
            "key": self._api_key, "action": "getbalance", "json": 1,
        }, timeout=15).json()
        return float(resp.get("request", 0))


# --- Mock implementation for testing ---

class MockSolver:
    """Test double that returns predictable tokens."""

    def __init__(self, token: str = "MOCK_TOKEN_123"):
        self._token = token
        self.calls: list[dict] = []

    def _record(self, method: str, **kwargs) -> SolveResult:
        self.calls.append({"method": method, **kwargs})
        return SolveResult(token=self._token, solve_time_ms=0, provider="mock")

    def solve_recaptcha_v2(self, sitekey: str, pageurl: str) -> SolveResult:
        return self._record("recaptcha_v2", sitekey=sitekey, pageurl=pageurl)

    def solve_recaptcha_v3(self, sitekey: str, pageurl: str,
                           action: str = "verify", min_score: float = 0.3) -> SolveResult:
        return self._record("recaptcha_v3", sitekey=sitekey, pageurl=pageurl)

    def solve_turnstile(self, sitekey: str, pageurl: str) -> SolveResult:
        return self._record("turnstile", sitekey=sitekey, pageurl=pageurl)

    def solve_hcaptcha(self, sitekey: str, pageurl: str) -> SolveResult:
        return self._record("hcaptcha", sitekey=sitekey, pageurl=pageurl)

    def solve_image(self, base64_image: str) -> SolveResult:
        return self._record("image")

    def get_balance(self) -> float:
        return 999.99


# --- Consumer code stays clean ---

def scrape_page(solver: CaptchaSolver, url: str, sitekey: str) -> str:
    """Works with any solver implementation."""
    assert isinstance(solver, CaptchaSolver)  # runtime check

    result = solver.solve_recaptcha_v2(sitekey=sitekey, pageurl=url)
    print(f"Solved by {result.provider} in {result.solve_time_ms}ms")
    return result.token


# Production
solver = CaptchaAISolver("YOUR_API_KEY")
token = scrape_page(solver, "https://example.com", "SITEKEY")

# Testing
mock = MockSolver()
token = scrape_page(mock, "https://example.com", "SITEKEY")
assert mock.calls[0]["method"] == "recaptcha_v2"

JavaScript: Class-Based Interface

const SUBMIT_URL = "https://ocr.captchaai.com/in.php";
const RESULT_URL = "https://ocr.captchaai.com/res.php";

// Abstract interface (throws if not implemented)
class CaptchaSolver {
  async solveRecaptchaV2(sitekey, pageurl) { throw new Error("Not implemented"); }
  async solveRecaptchaV3(sitekey, pageurl, action, minScore) { throw new Error("Not implemented"); }
  async solveTurnstile(sitekey, pageurl) { throw new Error("Not implemented"); }
  async solveHCaptcha(sitekey, pageurl) { throw new Error("Not implemented"); }
  async solveImage(base64Image) { throw new Error("Not implemented"); }
  async getBalance() { throw new Error("Not implemented"); }
}

class CaptchaAISolver extends CaptchaSolver {
  #apiKey;

  constructor(apiKey) {
    super();
    this.#apiKey = apiKey;
  }

  async #submitAndPoll(params) {
    const start = Date.now();
    const body = new URLSearchParams({ key: this.#apiKey, json: "1", ...params });
    const resp = await (await fetch(SUBMIT_URL, { method: "POST", body })).json();
    if (resp.status !== 1) throw new Error(`Submit: ${resp.request}`);

    const taskId = resp.request;
    for (let i = 0; i < 60; i++) {
      await new Promise((r) => setTimeout(r, 5000));
      const url = `${RESULT_URL}?key=${this.#apiKey}&action=get&id=${taskId}&json=1`;
      const poll = await (await fetch(url)).json();
      if (poll.request === "CAPCHA_NOT_READY") continue;
      if (poll.status === 1) {
        return { token: poll.request, solveTimeMs: Date.now() - start, provider: "captchaai" };
      }
      throw new Error(`Solve: ${poll.request}`);
    }
    throw new Error("Timeout");
  }

  async solveRecaptchaV2(sitekey, pageurl) {
    return this.#submitAndPoll({ method: "userrecaptcha", googlekey: sitekey, pageurl });
  }

  async solveTurnstile(sitekey, pageurl) {
    return this.#submitAndPoll({ method: "turnstile", sitekey, pageurl });
  }

  async solveHCaptcha(sitekey, pageurl) {
    return this.#submitAndPoll({ method: "hcaptcha", sitekey, pageurl });
  }

  async solveImage(base64Image) {
    return this.#submitAndPoll({ method: "base64", body: base64Image });
  }

  async getBalance() {
    const url = `${RESULT_URL}?key=${this.#apiKey}&action=getbalance&json=1`;
    const resp = await (await fetch(url)).json();
    return parseFloat(resp.request);
  }
}

class MockSolver extends CaptchaSolver {
  constructor(token = "MOCK_TOKEN") {
    super();
    this.token = token;
    this.calls = [];
  }

  async solveRecaptchaV2(sitekey, pageurl) {
    this.calls.push({ method: "recaptcha_v2", sitekey, pageurl });
    return { token: this.token, solveTimeMs: 0, provider: "mock" };
  }

  async solveTurnstile(sitekey, pageurl) {
    this.calls.push({ method: "turnstile", sitekey, pageurl });
    return { token: this.token, solveTimeMs: 0, provider: "mock" };
  }

  async solveHCaptcha(sitekey, pageurl) {
    this.calls.push({ method: "hcaptcha", sitekey, pageurl });
    return { token: this.token, solveTimeMs: 0, provider: "mock" };
  }

  async solveImage(base64Image) {
    this.calls.push({ method: "image" });
    return { token: this.token, solveTimeMs: 0, provider: "mock" };
  }

  async getBalance() { return 999.99; }
}

// Consumer code — provider-agnostic
async function scrapePage(solver, url, sitekey) {
  const result = await solver.solveRecaptchaV2(sitekey, url);
  console.log(`Solved by ${result.provider} in ${result.solveTimeMs}ms`);
  return result.token;
}

Decorator Pattern: Adding Cross-Cutting Concerns

Wrap the abstract interface to add behaviour without modifying implementations:

class LoggingSolver:
    """Decorator that logs every solve call."""

    def __init__(self, inner: CaptchaSolver):
        self._inner = inner

    def solve_recaptcha_v2(self, sitekey: str, pageurl: str) -> SolveResult:
        print(f"[LOG] solve_recaptcha_v2 sitekey={sitekey[:10]}...")
        result = self._inner.solve_recaptcha_v2(sitekey, pageurl)
        print(f"[LOG] Solved in {result.solve_time_ms}ms")
        return result

    # Delegate remaining methods via __getattr__
    def __getattr__(self, name):
        return getattr(self._inner, name)


# Stack decorators
solver = LoggingSolver(CaptchaAISolver("YOUR_API_KEY"))

Troubleshooting

Issue Cause Fix
isinstance check fails with Protocol Missing @runtime_checkable Add @runtime_checkable decorator to Protocol class
Mock doesn't satisfy interface Missing method Implement all Protocol methods in the mock
__getattr__ delegates wrong method Decorator missing explicit method Implement all critical methods explicitly
Type checker warns about missing methods Partial implementation Add all methods defined in the Protocol/base class
SolveResult fields inconsistent Providers return different data Normalize in each implementation's methods

FAQ

Should I use ABC or Protocol in Python?

Use Protocol for structural typing — classes satisfy the interface if they have matching methods, without inheriting from anything. Use ABC when you want to enforce inheritance and shared base functionality. Protocol is more flexible for third-party code you don't control.

How do I handle provider-specific features?

Keep the abstract interface generic. Expose provider-specific options via keyword arguments or a params dict. The consumer code stays clean — only provider-aware code passes the extra params.

Can I combine this with the factory pattern?

Yes. The factory selects which implementation to create based on configuration or detection. The abstract interface ensures all implementations share the same contract. The factory returns CaptchaSolver, and consumer code never knows the concrete type.

Next Steps

Build provider-agnostic CAPTCHA solving — get your CaptchaAI API key and implement the abstract interface.

Related guides:

Full Working Code

Complete runnable examples for this article in Python, Node.js, PHP, Go, Java, C#, Ruby, Rust, Kotlin & Bash.

View on GitHub →

Discussions (0)

No comments yet.