FastAPI

CONSTAIA_API_KEY=ck_test_...
CONSTAIA_WEBHOOK_SECRET=whsec_...

from typing import Literal, Optional

from pydantic import BaseModel


class Reason(BaseModel):
    code: str
    severity: Literal["info", "warning", "error"]
    message: str


class Verdict(BaseModel):
    expected: list[str]
    match: bool
    status: Literal["valid", "invalid", "review"]
    reasons: list[Reason]


class DocumentInfo(BaseModel):
    type: str
    label: str
    confidence: float


class AnalysisOut(BaseModel):
    id: str
    object: Literal["analysis"] = "analysis"
    status: Literal["queued", "processing", "completed", "failed"]
    document: Optional[DocumentInfo] = None
    verdict: Optional[Verdict] = None
    warnings: list[str] = []

import json
import logging
import os
from contextlib import asynccontextmanager
from typing import Annotated, Optional

from fastapi import BackgroundTasks, FastAPI, File, Form, Request, Response, UploadFile
from fastapi.responses import JSONResponse

import constaia
from constaia import AsyncConstaia, ConstaiaError, InsufficientCreditsError, InvalidRequestError, RateLimitError
from constaia.types import Analysis

from models import AnalysisOut

MAX_BYTES = 20 * 1024 * 1024
WEBHOOK_SECRET = os.environ["CONSTAIA_WEBHOOK_SECRET"]
logger = logging.getLogger(__name__)

USER_MESSAGES = {
    "file_too_large": "El archivo supera los 20 MB.",
    "unsupported_file_type": "Sube una imagen (JPG, PNG, WEBP, HEIC) o un PDF.",
    "unreadable_image": "No se puede leer la imagen. Haz otra foto con buena luz.",
    "unreadable_pdf": "No se puede leer el PDF.",
}


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.constaia = AsyncConstaia(timeout=60.0, max_retries=2, max_concurrency=8)
    yield
    await app.state.constaia.close()


app = FastAPI(lifespan=lifespan)


def error(message: str, status: int) -> JSONResponse:
    return JSONResponse({"error": {"message": message}}, status_code=status)


def save_analysis(analysis: Analysis) -> None:
    # Guarda en tu base de datos, envía emails, etc. Se ejecuta tras enviar la respuesta.
    logger.info("Constaia %s: %s", analysis["id"], (analysis["verdict"] or {}).get("status"))


@app.post("/documents", response_model=AnalysisOut)
async def documents(
    request: Request,
    background_tasks: BackgroundTasks,
    file: Annotated[UploadFile, File()],
    options: Annotated[Optional[str], Form()] = None,
):
    # Añade aquí tu dependencia de autenticación y un límite de frecuencia: cada análisis live consume créditos.
    content = await file.read()
    if not content:
        return error("No se ha recibido el documento.", 400)
    if len(content) > MAX_BYTES:
        return error(USER_MESSAGES["file_too_large"], 413)

    try:
        language = json.loads(options or "{}").get("language", "es")
    except (ValueError, AttributeError):
        language = "es"

    try:
        analysis = await request.app.state.constaia.analyze(
            content,
            filename=file.filename or "document",
            expect=["es_dni", "es_nie", "passport"],
            checks={"not_expired": True, "min_age_years": 18},
            storage="none",
            language=language if language in ("es", "en", "pt", "fr") else "es",
        )
    except InvalidRequestError as exc:
        return error(USER_MESSAGES.get(exc.code or "", "El documento no se ha podido procesar."), 422)
    except (InsufficientCreditsError, RateLimitError) as exc:
        logger.error("Constaia %s %s request_id=%s", exc.status, exc.code, exc.request_id)
        return error("El servicio está ocupado. Inténtalo en unos minutos.", 503)
    except ConstaiaError as exc:
        logger.error("Constaia %s %s request_id=%s", exc.status, exc.code, exc.request_id)
        return error("No hemos podido comprobar el documento. Inténtalo más tarde.", 502)

    background_tasks.add_task(save_analysis, analysis)
    return analysis


@app.post("/webhooks/constaia", status_code=204)
async def constaia_webhook(request: Request, background_tasks: BackgroundTasks):
    payload = await request.body()
    try:
        event = constaia.webhooks.verify(payload, request.headers, WEBHOOK_SECRET)
    except constaia.WebhookVerificationError:
        return Response(status_code=400)

    # Deduplica por request.headers["webhook-id"] (Redis, base de datos) antes de procesar.
    if event["type"] in ("analysis.completed", "analysis.review_required", "analysis.failed"):
        background_tasks.add_task(save_analysis, event["data"])
    return Response(status_code=204)

<script type="module" src="https://cdn.jsdelivr.net/npm/@constaia/widget@0.1"></script>
<constaia-upload endpoint="/documents" document="es_dni" lang="es"></constaia-upload>

import json
import os
from pathlib import Path

import pytest
from fastapi.testclient import TestClient

import constaia
from main import WEBHOOK_SECRET, app

FIXTURES = Path(__file__).parent / "fixtures"
needs_test_key = pytest.mark.skipif(
    not os.environ.get("CONSTAIA_API_KEY", "").startswith("ck_test_"), reason="needs a ck_test_ key"
)


@pytest.fixture
def http():
    with TestClient(app) as client:
        yield client


@needs_test_key
@pytest.mark.parametrize("name, status", [("dni_valid.jpg", "valid"), ("dni_expired.jpg", "invalid"), ("blurry.jpg", "review")])
def test_verdicts(http, name, status):
    response = http.post("/documents", files={"file": (name, (FIXTURES / name).read_bytes(), "image/jpeg")})
    assert response.status_code == 200
    body = response.json()
    assert body["verdict"]["status"] == status
    assert "fields" not in body


def test_webhook_signature(http):
    body = json.dumps({"type": "analysis.completed", "data": {"id": "an_test", "status": "completed", "verdict": None}})
    headers = constaia.webhooks.sign(body, WEBHOOK_SECRET)

    assert http.post("/webhooks/constaia", content=body, headers=headers).status_code == 204
    assert http.post("/webhooks/constaia", content=body, headers={**headers, "webhook-signature": "v1,bad"}).status_code == 400

CONSTAIA_API_KEY=ck_test_... CONSTAIA_WEBHOOK_SECRET=whsec_... pytest -q