Ein typsicherer Client fängt Konfigurationsfehler dort ab, wo sie günstig sind: im Speicher, bevor überhaupt eine Anfrage an die CaptchaAI-API rausgeht. Genau das leistet Pydantic v2 in einem Python-Client. Statt einen abgeschnittenen Sitekey oder eine pageurl ohne Schema erst nach dem Roundtrip als kryptischen Fehlercode wie ERROR_WRONG_CAPTCHA_ID zurückzubekommen, meldet Pydantic sofort einen lesbaren ValidationError – mit Feldname, Regel und erwartetem Wert.
Dieser Leitfaden zeigt den vollständigen Aufbau: validierte Anfrage-Modelle pro CAPTCHA-Typ, eine Client-Klasse für Übermittlung und Status-Abfrage sowie typisierte Antwort-Modelle. Ziel ist ein Client, der ungültige Eingaben ablehnt, bevor sie Zeit und Guthaben kosten – und der jedes Feld mit vollständigen Typhinweisen an Ihre IDE weiterreicht.
Was Pydantic im API-Client absichert
CAPTCHA-Solving läuft asynchron: Sie übermitteln eine Aufgabe, warten und fragen das Ergebnis ab. Jeder Fehler in den Eingabeparametern kostet deshalb einen kompletten Roundtrip, bevor er überhaupt sichtbar wird. Pydantic verschiebt diese Prüfung nach vorn – vom Netzwerk in den Konstruktor Ihrer Modelle. Der Unterschied im Alltag:
| Ohne Pydantic | Mit Pydantic |
|---|---|
| Leerer Sitekey – API-Fehler nach 5 Sekunden | ValidationError sofort |
Antwortparsing über dict["key"] → KeyError |
Typisiertes Modell mit Standardwerten und Validierung |
| Keine IDE-Autovervollständigung für Parameter | Vollständige Typhinweise zu allen Feldern |
Anfrage- und Antwort-Modelle definieren
Jeder unterstützte CAPTCHA-Typ bekommt ein eigenes Anfrage-Modell mit den Feldregeln, die die CaptchaAI-API tatsächlich erwartet: ein sitekey mit Mindestlänge, eine pageurl als echte HttpUrl, optionale Felder mit Standardwerten. Das Modul models.py deckt damit den gesamten Lebenszyklus einer Abfrage ab:
- Anfrage-Modelle wie
RecaptchaV2Request,RecaptchaV3Request,TurnstileRequestundImageRequestprüfen die Eingaben und liefern überto_params()fertige Formularparameter. - Antwort-Modelle
SubmitResponseundPollResponseparsen die Roh-Antwort in typisierte Objekte mit sprechenden Properties wiesuccess,readyundtoken– keindict["key"], das im Zweifel einenKeyErrorwirft. - Ergebnis-Modell
SolveResultbündelt Token, Task-ID und Lösungszeit an einer Stelle.
# models.py
from pydantic import BaseModel, Field, field_validator, HttpUrl
from enum import Enum
from typing import Optional
class CaptchaMethod(str, Enum):
RECAPTCHA_V2 = "userrecaptcha"
RECAPTCHA_V3 = "userrecaptcha" # Differentiated by version field
TURNSTILE = "turnstile"
HCAPTCHA = "hcaptcha"
IMAGE = "base64"
GEETEST = "geetest"
class RecaptchaV2Request(BaseModel):
"""Parameters for solving reCAPTCHA v2."""
sitekey: str = Field(min_length=20, max_length=100, description="Site's reCAPTCHA sitekey")
pageurl: HttpUrl = Field(description="URL where CAPTCHA appears")
invisible: bool = False
cookies: Optional[str] = None
@field_validator("sitekey")
@classmethod
def validate_sitekey(cls, v: str) -> str:
if v.strip() != v:
raise ValueError("Sitekey must not have leading/trailing whitespace")
return v
def to_params(self) -> dict:
params = {
"method": "userrecaptcha",
"googlekey": self.sitekey,
"pageurl": str(self.pageurl),
}
if self.invisible:
params["invisible"] = "1"
if self.cookies:
params["cookies"] = self.cookies
return params
class RecaptchaV3Request(BaseModel):
"""Parameters for solving reCAPTCHA v3."""
sitekey: str = Field(min_length=20, max_length=100)
pageurl: HttpUrl
action: str = Field(default="verify", min_length=1, max_length=100)
def to_params(self) -> dict:
return {
"method": "userrecaptcha",
"version": "v3",
"googlekey": self.sitekey,
"pageurl": str(self.pageurl),
"action": self.action,
}
class TurnstileRequest(BaseModel):
"""Parameters for solving Cloudflare Turnstile."""
sitekey: str = Field(min_length=10, max_length=100)
pageurl: HttpUrl
action: Optional[str] = None
cdata: Optional[str] = None
def to_params(self) -> dict:
params = {
"method": "turnstile",
"sitekey": self.sitekey,
"pageurl": str(self.pageurl),
}
if self.action:
params["action"] = self.action
if self.cdata:
params["data"] = self.cdata
return params
class ImageRequest(BaseModel):
"""Parameters for solving image/text CAPTCHA."""
base64_image: str = Field(min_length=100, description="Base64-encoded image")
case_sensitive: bool = False
min_length: Optional[int] = Field(default=None, ge=1, le=50)
max_length: Optional[int] = Field(default=None, ge=1, le=50)
@field_validator("base64_image")
@classmethod
def validate_base64(cls, v: str) -> str:
# Strip data URI prefix if present
if v.startswith("data:"):
parts = v.split(",", 1)
if len(parts) == 2:
return parts[1]
return v
def to_params(self) -> dict:
params = {
"method": "base64",
"body": self.base64_image,
}
if self.case_sensitive:
params["regsense"] = "1"
if self.min_length is not None:
params["min_len"] = str(self.min_length)
if self.max_length is not None:
params["max_len"] = str(self.max_length)
return params
class SubmitResponse(BaseModel):
"""Parsed API submit response."""
status: int
request: str
@property
def success(self) -> bool:
return self.status == 1
@property
def task_id(self) -> str:
if not self.success:
raise ValueError(f"No task ID — submission failed: {self.request}")
return self.request
class PollResponse(BaseModel):
"""Parsed API poll response."""
status: int
request: str
@property
def ready(self) -> bool:
return self.request != "CAPCHA_NOT_READY"
@property
def success(self) -> bool:
return self.status == 1
@property
def token(self) -> str:
if not self.success:
raise ValueError(f"No token — solve failed: {self.request}")
return self.request
class SolveResult(BaseModel):
"""Result of a successful solve."""
token: str
task_id: str
solve_time: float = Field(description="Solve time in seconds")
Die Client-Klasse: übermitteln und Status abfragen
Die Klasse CaptchaAI bündelt den Ablauf hinter den Modellen. _submit schickt die Parameter an in.php und validiert die Antwort direkt gegen SubmitResponse; _poll fragt res.php im festen Intervall ab, bis ein Token vorliegt oder das Timeout greift. Echte API-Fehlercodes wandern in eine eigene CaptchaAIError-Ausnahme und bleiben so sauber von Validierungsfehlern getrennt. Die öffentlichen Methoden solve_recaptcha_v2, solve_recaptcha_v3, solve_turnstile und solve_image nehmen einfache Argumente entgegen, bauen daraus das passende Modell und lösen die Abfrage.
# client.py
import time
import requests
from pydantic import ValidationError
from models import (
RecaptchaV2Request,
RecaptchaV3Request,
TurnstileRequest,
ImageRequest,
SubmitResponse,
PollResponse,
SolveResult,
)
SUBMIT_URL = "https://ocr.captchaai.com/in.php"
RESULT_URL = "https://ocr.captchaai.com/res.php"
class CaptchaAIError(Exception):
def __init__(self, code: str, message: str = ""):
self.code = code
super().__init__(f"{code}: {message}" if message else code)
class CaptchaAI:
def __init__(self, api_key: str, poll_interval: int = 5, timeout: int = 180):
if not api_key or len(api_key) < 10:
raise ValueError("Invalid API key")
self.api_key = api_key
self.poll_interval = poll_interval
self.timeout = timeout
def _submit(self, params: dict) -> str:
params["key"] = self.api_key
params["json"] = 1
resp = requests.post(SUBMIT_URL, data=params, timeout=30)
result = SubmitResponse.model_validate(resp.json())
if not result.success:
raise CaptchaAIError(result.request, "Submit failed")
return result.task_id
def _poll(self, task_id: str) -> str:
start = time.monotonic()
while time.monotonic() - start < self.timeout:
time.sleep(self.poll_interval)
resp = requests.get(RESULT_URL, params={
"key": self.api_key,
"action": "get",
"id": task_id,
"json": 1,
}, timeout=15)
result = PollResponse.model_validate(resp.json())
if not result.ready:
continue
if result.success:
return result.token
raise CaptchaAIError(result.request, "Solve failed")
raise CaptchaAIError("TIMEOUT", f"Task {task_id} timed out after {self.timeout}s")
def _solve(self, params: dict) -> SolveResult:
start = time.monotonic()
task_id = self._submit(params)
token = self._poll(task_id)
elapsed = time.monotonic() - start
return SolveResult(
token=token,
task_id=task_id,
solve_time=round(elapsed, 1),
)
def solve_recaptcha_v2(self, sitekey: str, pageurl: str, **kwargs) -> SolveResult:
"""Solve reCAPTCHA v2 with validated parameters."""
req = RecaptchaV2Request(sitekey=sitekey, pageurl=pageurl, **kwargs)
return self._solve(req.to_params())
def solve_recaptcha_v3(self, sitekey: str, pageurl: str, **kwargs) -> SolveResult:
"""Solve reCAPTCHA v3 with validated parameters."""
req = RecaptchaV3Request(sitekey=sitekey, pageurl=pageurl, **kwargs)
return self._solve(req.to_params())
def solve_turnstile(self, sitekey: str, pageurl: str, **kwargs) -> SolveResult:
"""Solve Cloudflare Turnstile with validated parameters."""
req = TurnstileRequest(sitekey=sitekey, pageurl=pageurl, **kwargs)
return self._solve(req.to_params())
def solve_image(self, base64_image: str, **kwargs) -> SolveResult:
"""Solve image/text CAPTCHA with validated parameters."""
req = ImageRequest(base64_image=base64_image, **kwargs)
return self._solve(req.to_params())
def get_balance(self) -> float:
"""Get current account balance."""
resp = requests.get(RESULT_URL, params={
"key": self.api_key,
"action": "getbalance",
"json": 1,
}, timeout=10)
result = SubmitResponse.model_validate(resp.json())
return float(result.request)
Der Client im Einsatz
In der Praxis reduziert sich der Aufruf auf wenige Zeilen. Eine gültige Anfrage durchläuft die Validierung und geht an die API; eine ungültige – etwa ein leerer sitekey – scheitert sofort mit ValidationError, ohne dass ein einziger HTTP-Aufruf erfolgt. So trennen Sie im Code klar zwischen „falsch konfiguriert“ (fangen Sie ValidationError) und „API meldet ein Problem“ (fangen Sie CaptchaAIError).
from pydantic import ValidationError
from client import CaptchaAI, CaptchaAIError
client = CaptchaAI("YOUR_API_KEY", timeout=120)
# Valid request — passes validation, calls API
result = client.solve_recaptcha_v2(
sitekey="6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
pageurl="https://example.com/login",
)
print(f"Token: {result.token[:40]}...")
print(f"Solved in {result.solve_time}s")
# Invalid sitekey — caught immediately, no API call
try:
client.solve_recaptcha_v2(sitekey="", pageurl="https://example.com")
except ValidationError as e:
print(e)
# sitekey: String should have at least 20 characters
# Invalid score — caught before API call
try:
client.solve_recaptcha_v3(
sitekey="6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
pageurl="https://example.com",
)
except ValidationError as e:
print(e)
# API error — caught during request
try:
result = client.solve_turnstile(
sitekey="0x4AAAAAAADnPIDROrmt1Wwj",
pageurl="https://example.com",
)
except CaptchaAIError as e:
print(f"API error: {e.code}")
Abhängigkeiten installieren:
pip install pydantic requests
Praxisbeispiel: Validierung in der GitLab-CI-Pipeline
Ein typisches DACH-Setup: Der Scraper läuft als Worker auf einem Hetzner-Server, deployt über GitLab CI. Genau dort spielt die Validierung ihren Wert aus. Ziehen Sie sitekey und pageurl aus der Umgebung oder einer Konfigurationsdatei, instanziieren Sie das passende Modell und lassen Sie Pydantic prüfen, bevor der Job überhaupt eine Solve-Anfrage stellt. Ist eine Variable in der CI-Konfiguration falsch gesetzt, bricht der Pipeline-Schritt mit einem eindeutigen ValidationError ab – nicht erst nach einem Timeout in der Nacht-Batch-Verarbeitung, wenn niemand auf die Logs schaut. Für Scraping-Workflows lohnt zusätzlich ein Blick auf die DSGVO: Prüfen Sie eigenständig, ob die abgerufenen Daten personenbezogen sind und auf welcher Rechtsgrundlage Sie sie verarbeiten.
Fehlerbehebung
Die häufigsten Stolpersteine lassen sich meist am Fehlertyp ablesen – ein ValidationError deutet auf die Eingabeparameter, ein CaptchaAIError auf die API-Antwort:
| Problem | Ursache | Lösung |
|---|---|---|
| Token wird erzeugt, aber vom Ziel abgelehnt | sitekey, pageurl oder Session-Kontext stimmen nicht |
Erfassen Sie die Parameter erneut und verwenden Sie den Token in derselben Browser- oder HTTP-Sitzung |
| Polling endet im Timeout | Intervall, Wartezeit oder Fehlerbehandlung sind zu eng gesetzt | Fragen Sie alle 5–10 Sekunden ab, trennen Sie das Timeout von echten Fehlercodes und loggen Sie die Ursache |
| Beispiel funktioniert lokal, aber nicht im Workflow | Callback oder Form-Feld fehlt in der echten Zielkette | Prüfen Sie den exakten Übergabepfad vom Solver bis zur finalen Zielanfrage und tragen Sie den Token in das richtige Formularfeld ein |
Häufige Fragen
Warum Pydantic v2 und nicht v1?
Pydantic v2 ist die aktuelle, deutlich schnellere Generation und die Grundlage für die hier gezeigten Modelle. Der Validierungskern ist in Rust umgesetzt, field_validator ersetzt das alte @validator, und model_validate löst parse_obj ab. Sehen Sie einen v1-Importfehler, installieren Sie gezielt die aktuelle Version: pip install 'pydantic>=2.0'.
Welche Fehler fängt die Validierung ab – und welche nicht?
Pydantic prüft alles, was strukturell erkennbar ist: leere oder zu kurze Sitekeys, eine pageurl ohne https://-Schema, ungültige Feldtypen oder Werte außerhalb der erlaubten Grenzen. Es prüft nicht, ob ein formal korrekter Sitekey auch der richtige für die Zielseite ist – das zeigt sich erst in der Antwort der API. Genau dafür bleibt CaptchaAIError als zweite Verteidigungslinie bestehen.
Verlangsamt die Validierung meinen Durchsatz?
Nein. Eine Validierung liegt im Mikrosekundenbereich, während ein Solve-Roundtrip Sekunden dauert. Der Aufwand fällt gegenüber der Netzwerklatenz nicht ins Gewicht – und ein früh abgefangener Fehler spart einen kompletten, verschwendeten Roundtrip. Bei hohem Durchsatz zahlt sich die Prüfung damit sogar aus.
Kann ich reCAPTCHA v2, v3, Turnstile und Bild-CAPTCHAs mit demselben Client lösen?
Ja. Jeder Typ hat sein eigenes Anfrage-Modell, aber alle laufen durch dieselbe CaptchaAI-Klasse. CaptchaAI unterstützt reCAPTCHA v2 und v3, Cloudflare Turnstile und Challenge, GeeTest v3 sowie Bild- und Raster-CAPTCHAs. Für einen weiteren Typ ergänzen Sie ein BaseModel mit to_params() und eine passende solve_-Methode.
Verwandte Leitfäden
- Asynchrones CAPTCHA-Solving in Python mit asyncio
- CaptchaAI in einen FastAPI-Microservice einbinden
- API-Antwortformate und Fehlercodes im Überblick
