Tutorials

Dependency Injection for CAPTCHA Solving Services

When CAPTCHA solving logic is hardcoded into your scraper, testing means hitting the API and spending credits. Dependency injection lets you swap the real solver for a mock in tests, swap CaptchaAI for another provider without touching business logic, and configure different solving strategies per environment.

The Problem: Tightly Coupled Solving

# Tightly coupled — can't test without API calls
class Scraper:
    def scrape(self, url):
        token = solve_recaptcha_via_captchaai(sitekey, url)  # Hardcoded
        return self.fetch_with_token(url, token)

Python: Dependency Injection with Protocols

from typing import Protocol, runtime_checkable
import requests
import time

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


# --- Interface ---

@runtime_checkable
class CaptchaSolver(Protocol):
    """Protocol defining what a CAPTCHA solver must implement."""

    def solve_recaptcha_v2(self, sitekey: str, pageurl: str) -> str: ...
    def solve_turnstile(self, sitekey: str, pageurl: str) -> str: ...
    def get_balance(self) -> float: ...


# --- Real implementation ---

class CaptchaAISolver:
    """Production solver using CaptchaAI API."""

    def __init__(self, api_key: str, timeout: int = 180):
        self.api_key = api_key
        self.timeout = timeout

    def _submit_and_poll(self, params: dict) -> str:
        params["key"] = self.api_key
        params["json"] = 1

        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"]
        start = time.monotonic()

        while time.monotonic() - start < self.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:
                return poll["request"]
            raise RuntimeError(f"Solve failed: {poll.get('request')}")

        raise RuntimeError("Timeout")

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

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

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


# --- Mock implementation for testing ---

class MockCaptchaSolver:
    """Test solver that returns predictable tokens without API calls."""

    def __init__(self):
        self.calls = []

    def solve_recaptcha_v2(self, sitekey: str, pageurl: str) -> str:
        self.calls.append(("recaptcha_v2", sitekey, pageurl))
        return "mock-recaptcha-token-" + sitekey[:10]

    def solve_turnstile(self, sitekey: str, pageurl: str) -> str:
        self.calls.append(("turnstile", sitekey, pageurl))
        return "mock-turnstile-token-" + sitekey[:10]

    def get_balance(self) -> float:
        return 99.99


# --- Business logic with injected solver ---

class FormSubmitter:
    """Submits forms behind CAPTCHAs. Solver is injected, not hardcoded."""

    def __init__(self, solver: CaptchaSolver):
        self.solver = solver

    def submit_login(self, email: str, password: str, sitekey: str, url: str) -> dict:
        token = self.solver.solve_recaptcha_v2(sitekey, url)

        # Use token to submit form
        response = requests.post(url, data={
            "email": email,
            "password": password,
            "g-recaptcha-response": token,
        })

        return {"status": response.status_code, "token_used": token[:30]}


# --- Production usage ---
solver = CaptchaAISolver(api_key="YOUR_API_KEY")
submitter = FormSubmitter(solver=solver)
result = submitter.submit_login("user@example.com", "pass", "SITEKEY", "https://example.com/login")

# --- Test usage ---
mock_solver = MockCaptchaSolver()
submitter = FormSubmitter(solver=mock_solver)
result = submitter.submit_login("test@test.com", "test", "SITEKEY", "https://example.com/login")
assert len(mock_solver.calls) == 1
assert mock_solver.calls[0][0] == "recaptcha_v2"

JavaScript: Dependency Injection with Interfaces

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

// --- Real implementation ---

class CaptchaAISolver {
  constructor(apiKey) {
    this.apiKey = apiKey;
  }

  async _submitAndPoll(params) {
    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 poll.request;
      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 getBalance() {
    const url = `${RESULT_URL}?key=${this.apiKey}&action=getbalance&json=1`;
    const resp = await (await fetch(url)).json();
    return parseFloat(resp.request);
  }
}

// --- Mock implementation ---

class MockCaptchaSolver {
  constructor() {
    this.calls = [];
  }

  async solveRecaptchaV2(sitekey, pageurl) {
    this.calls.push({ type: "recaptcha_v2", sitekey, pageurl });
    return `mock-token-${sitekey.substring(0, 10)}`;
  }

  async solveTurnstile(sitekey, pageurl) {
    this.calls.push({ type: "turnstile", sitekey, pageurl });
    return `mock-turnstile-${sitekey.substring(0, 10)}`;
  }

  async getBalance() {
    return 99.99;
  }
}

// --- Business logic with injected solver ---

class FormSubmitter {
  constructor(solver) {
    this.solver = solver;
  }

  async submitLogin(email, password, sitekey, url) {
    const token = await this.solver.solveRecaptchaV2(sitekey, url);
    // Use token...
    return { status: "ok", tokenPreview: token.substring(0, 30) };
  }
}

// Production
const prodSolver = new CaptchaAISolver("YOUR_API_KEY");
const prodSubmitter = new FormSubmitter(prodSolver);

// Testing
const mockSolver = new MockCaptchaSolver();
const testSubmitter = new FormSubmitter(mockSolver);
await testSubmitter.submitLogin("test@test.com", "pass", "SITEKEY", "https://example.com");
console.assert(mockSolver.calls.length === 1);

module.exports = { CaptchaAISolver, MockCaptchaSolver, FormSubmitter };

Configuration by Environment

import os

def create_solver() -> CaptchaSolver:
    """Factory that returns the appropriate solver for the environment."""
    env = os.getenv("APP_ENV", "production")

    if env == "test":
        return MockCaptchaSolver()
    elif env == "development":
        # Development solver with lower timeout
        return CaptchaAISolver(api_key=os.environ["CAPTCHA_API_KEY"], timeout=60)
    else:
        return CaptchaAISolver(api_key=os.environ["CAPTCHA_API_KEY"], timeout=180)


# Application setup
solver = create_solver()
app = FormSubmitter(solver=solver)

Troubleshooting

Issue Cause Fix
Mock and real solver have different methods Interface drift Use Protocol (Python) or TypeScript interface to enforce method signatures
Tests pass but production fails Mock doesn't simulate errors Add FailingMockSolver that raises errors for error-handling tests
Can't inject solver deeply nested Constructor injection doesn't reach Use a service locator or DI container for deeply nested dependencies
Runtime type check fails Missing @runtime_checkable Add @runtime_checkable decorator to the Protocol class
Mock token format causes downstream errors Real tokens are much longer Make mock tokens match expected format with realistic length

FAQ

When is dependency injection overkill for CAPTCHA solving?

For one-off scripts that run once and exit, DI adds unnecessary complexity. Use it when you have multiple environments (dev/test/prod), multiple possible providers, or need automated tests without API calls.

Can I inject different solvers for different CAPTCHA types?

Yes — inject a solver per type. For example, inject a RecaptchaSolver and a separate TurnstileSolver. Or use the factory pattern to select the right solver at runtime based on the detected CAPTCHA type.

How do I verify the mock matches the real implementation?

Use the Protocol/interface to define the contract. In Python, isinstance(mock, CaptchaSolver) returns True only if the mock implements all required methods. Write integration tests that run against the real API periodically.

Next Steps

Build testable CAPTCHA integrations — get your CaptchaAI API key and implement dependency injection.

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.