Python

pip install httpx

import json
import os
import time
import uuid
from pathlib import Path

import httpx

API_URL = "https://api.constaia.com/v1"


class ConstaiaError(Exception):
    def __init__(self, status: int, body: dict):
        err = body.get("error", {}) if isinstance(body, dict) else {}
        super().__init__(err.get("message") or f"HTTP {status}")
        self.status = status
        self.type = err.get("type")
        self.code = err.get("code")
        self.param = err.get("param")
        self.request_id = err.get("request_id")


class Constaia:
    def __init__(self, api_key: str | None = None, timeout: float = 60.0):
        key = api_key or os.environ["CONSTAIA_API_KEY"]
        self._http = httpx.Client(
            base_url=API_URL,
            headers={"Authorization": f"Bearer {key}"},
            timeout=timeout,
        )

    def _request(self, method: str, path: str, *, retries: int = 2, **kwargs) -> dict:
        # Misma Idempotency-Key en todos los reintentos: nunca se cobra dos veces.
        if method == "POST":
            kwargs.setdefault("headers", {})["Idempotency-Key"] = str(uuid.uuid4())
        for attempt in range(retries + 1):
            try:
                resp = self._http.request(method, path, **kwargs)
            except httpx.TransportError:
                if attempt == retries:
                    raise
                time.sleep(0.5 * 2**attempt)
                continue
            if resp.status_code in (429, 500, 502, 503, 504) and attempt < retries:
                retry_after = resp.headers.get("retry-after")
                time.sleep(min(float(retry_after), 60) if retry_after else 0.5 * 2**attempt)
                continue
            if resp.is_error:
                raise ConstaiaError(resp.status_code, resp.json() if resp.content else {})
            return resp.json()
        raise RuntimeError("unreachable")

    def analyze(self, path: str | Path, **options) -> dict:
        path = Path(path)
        return self._request(
            "POST",
            "/analyze",
            # bytes en memoria: se pueden reenviar tal cual si hay reintento
            files={"file": (path.name, path.read_bytes())},
            data={"options": json.dumps(options)},
        )

    def analyze_url(self, file_url: str, **options) -> dict:
        return self._request("POST", "/analyze", json={"file_url": file_url, **options})

    def get_analysis(self, analysis_id: str) -> dict:
        return self._request("GET", f"/analyses/{analysis_id}")

    def wait(self, analysis: dict, max_wait: float = 120) -> dict:
        """Si la API respondió 202 (queued/processing), consulta hasta que termine."""
        delay, waited = 1.0, 0.0
        while analysis["status"] in ("queued", "processing"):
            if waited > max_wait:
                raise TimeoutError(f"{analysis['id']} sigue en curso")
            time.sleep(delay)
            waited += delay
            delay = min(delay * 2, 15)
            analysis = self.get_analysis(analysis["id"])
        return analysis

from constaia_client import Constaia, ConstaiaError

constaia = Constaia()  # lee CONSTAIA_API_KEY

try:
    analysis = constaia.analyze(
        "dni_valid.jpg",
        expect="es_dni",
        checks={"not_expired": True, "holder": {"full_name": "María García López"}},
        storage="none",
        metadata={"registration_id": "123"},
    )
    analysis = constaia.wait(analysis)  # por si tardó más de 30 s
except ConstaiaError as e:
    if e.type == "insufficient_credits":
        ...  # avisa al administrador
    raise

verdict = analysis["verdict"]
if verdict["status"] == "valid":
    print("OK:", analysis["fields"]["document_number"]["value"])
elif verdict["status"] == "invalid":
    for reason in verdict["reasons"]:
        print(reason["code"], "—", reason["message"])
else:  # "review"
    print("Revisión manual. Avisos:", analysis["warnings"])

import base64
import hashlib
import hmac
import json
import time

TOLERANCE_SECONDS = 5 * 60


def verify_constaia_webhook(raw_body: bytes, headers, secret: str) -> dict:
    msg_id = headers.get("webhook-id")
    timestamp = headers.get("webhook-timestamp")
    signatures = headers.get("webhook-signature")
    if not msg_id or not timestamp or not signatures:
        raise ValueError("Faltan cabeceras de webhook")

    if abs(time.time() - int(timestamp)) > TOLERANCE_SECONDS:
        raise ValueError("Timestamp fuera de tolerancia")

    key = base64.b64decode(secret.removeprefix("whsec_"))
    signed = f"{msg_id}.{timestamp}.".encode() + raw_body
    expected = base64.b64encode(hmac.new(key, signed, hashlib.sha256).digest()).decode()

    for entry in signatures.split(" "):
        version, _, signature = entry.partition(",")
        if version == "v1" and hmac.compare_digest(signature, expected):
            return json.loads(raw_body)

    raise ValueError("Firma no válida")

import os

from fastapi import FastAPI, HTTPException, Request, Response

from constaia_webhooks import verify_constaia_webhook

app = FastAPI()
SECRET = os.environ["CONSTAIA_WEBHOOK_SECRET"]


@app.post("/webhooks/constaia")
async def constaia_webhook(request: Request):
    raw_body = await request.body()  # bytes crudos, sin parsear
    try:
        event = verify_constaia_webhook(raw_body, request.headers, SECRET)
    except ValueError:
        raise HTTPException(status_code=400, detail="Invalid signature")

    message_id = request.headers["webhook-id"]
    if already_processed(message_id):  # tu tabla o tu Redis
        return Response(status_code=204)

    if event["type"] in ("analysis.completed", "analysis.review_required"):
        enqueue_analysis(event["data"])  # tu cola: responde rápido, procesa después

    mark_processed(message_id)
    return Response(status_code=204)