Un client CaptchaAI qui contrôle ses paramètres avec Pydantic transforme chaque erreur de configuration en message clair, avant le moindre appel réseau. Au lieu de déboguer un ERROR_WRONG_CAPTCHA_ID renvoyé après plusieurs secondes d'attente, vous récupérez une ValidationError explicite au moment où vous construisez la requête.
L'enjeu est concret pour une équipe Python : les modèles typés documentent l'API, activent l'autocomplétion et déplacent la détection des fautes du réseau vers votre code. Ce guide structure ce client autour de Pydantic v2.
Pourquoi valider les paramètres avec Pydantic
Sur un pipeline en continu, chaque requête mal formée coûte un aller-retour HTTP et occupe un thread pendant l'attente. Le forfait BASIC ($15/mois, 5 threads) en compte cinq en parallèle : une sitekey vide qui consomme un thread pour finir en erreur, c'est une place perdue dans la file. Valider en amont libère ces threads pour de vraies résolutions.
Le tableau ci-dessous résume ce que Pydantic change côté client. Si ce pipeline collecte des données, gardez aussi le réflexe RGPD de minimiser les données personnelles qui passent par vos requêtes et vos logs.
| Sans Pydantic | Avec Pydantic |
|---|---|
| Clé de site vide – Erreur API après 5 s | ValidationError immédiatement |
Analyse des réponses via dict["key"] – KeyError |
Modèle typé avec valeurs par défaut et validation |
| Pas de saisie semi-automatique IDE pour les paramètres | Conseils de type complets sur tous les champs |
Définir les modèles Pydantic
Chaque type devient une sous-classe BaseModel : reCAPTCHA v2, v3, Turnstile, image/OCR. Les contraintes (min_length, HttpUrl, validateurs de champ) rejettent les valeurs impossibles dès l'instanciation, et to_params() produit le dictionnaire attendu par l'API. Les modèles de réponse parsent le JSON en objets typés, sans accès par clé fragile.
# 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")
Le validateur sur sitekey refuse les espaces parasites et HttpUrl impose un schéma https://. Ces règles couvrent les types pris en charge en production : reCAPTCHA v2 et v3, Cloudflare Turnstile, image/OCR.
Assembler le client CaptchaAI
Le client encapsule l'envoi (_submit), l'interrogation du résultat (_poll) et l'orchestration (_solve). Les méthodes publiques instancient le modèle correspondant, donc la validation se joue avant l'appel HTTP. En cas de refus côté API, une exception CaptchaAIError transporte le code d'erreur brut.
# 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)
_poll interroge le résultat toutes les poll_interval secondes jusqu'au timeout, en s'appuyant sur PollResponse.ready pour écarter les CAPCHA_NOT_READY. Le token final passe une dernière validation dans SolveResult.
Utiliser le client en pratique
L'exemple ci-dessous montre les deux chemins : la requête valide qui part vers l'API, et la requête fautive interceptée par une ValidationError, sans consommer ni thread ni solde.
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}")
Installer les dépendances :
pip install pydantic requests
Séparez toujours les deux familles d'erreurs : ValidationError signale une faute côté client, CaptchaAIError remonte un problème côté API (solde, quota, échec) à gérer avec un retry ou une alerte.
Dépannage
| Problème | Cause | Correctif |
|---|---|---|
ValidationError sur une clé de site d'apparence valide |
Clé de site trop courte (< 20 caractères) | Vérifiez la longueur de la clé du site ; ajustez min_length si votre cible utilise des clés plus courtes |
ValidationError sur l'URL de la page |
Schéma d'URL manquant | Incluez le préfixe https:// |
| La validation de l'image Base64 échoue | Chaîne trop courte ou incluant le préfixe data: |
Le validateur supprime automatiquement le préfixe data: ; assurez-vous que le contenu réel en base64 dépasse 100 caractères |
CaptchaAIError: ERROR_ZERO_BALANCE |
Fonds insuffisants | Rechargez votre solde sur le tableau de bord CaptchaAI |
| Erreurs d'import Pydantic v1 | Mauvaise version de Pydantic | Utilisez Pydantic v2 : pip install 'pydantic>=2.0' |
FAQ
Pydantic v2 est-il indispensable pour ce client ?
Oui. Le code s'appuie sur l'API v2 (field_validator, model_validate), qui diffère de Pydantic v1. Installez la bonne version avec pip install 'pydantic>=2.0' ; une v1 dans l'environnement provoquera des erreurs d'import.
Peut-on réutiliser ces modèles avec un client asynchrone (httpx) ?
Oui. Remplacez requests par httpx.AsyncClient et passez _submit, _poll et les méthodes de résolution en async. Les modèles Pydantic ne changent pas : ils valident de façon synchrone, juste avant l'appel HTTP asynchrone.
Ces modèles sont-ils réutilisables dans une API FastAPI ?
Oui, et c'est un atout. FastAPI repose déjà sur Pydantic : réutilisez ces classes comme corps d'endpoint, et une valeur validée à la frontière de votre service l'est aussi avant l'appel à CaptchaAI.
Comment gérer proprement une CaptchaAIError renvoyée par l'API ?
Interceptez-la séparément de ValidationError et inspectez son attribut code : un ERROR_ZERO_BALANCE déclenche une alerte de rechargement, un TIMEOUT justifie un retry avec backoff exponentiel.
Articles connexes
- automatiser un navigateur avec CaptchaAI
- construire des pipelines de résolution CAPTCHA côté client
- valider les callbacks webhook CaptchaAI
Prochaines étapes
Assemblez un client CaptchaAI validé : récupérez votre clé API et branchez vos modèles Pydantic.
Guides associés :
