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": "The file is larger than 20 MB.",
    "unsupported_file_type": "Upload an image (JPG, PNG, WEBP, HEIC) or a PDF.",
    "unreadable_image": "We can't read the image. Take another photo in good light.",
    "unreadable_pdf": "We can't read the 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:
    # Save to your database, send emails, etc. Runs after the response is sent.
    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,
):
    # Add your authentication dependency and a rate limit here: every live analysis consumes credits.
    content = await file.read()
    if not content:
        return error("No document received.", 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 "", "The document could not be processed."), 422)
    except (InsufficientCreditsError, RateLimitError) as exc:
        logger.error("Constaia %s %s request_id=%s", exc.status, exc.code, exc.request_id)
        return error("The service is busy. Try again in a few minutes.", 503)
    except ConstaiaError as exc:
        logger.error("Constaia %s %s request_id=%s", exc.status, exc.code, exc.request_id)
        return error("We could not check the document. Please try again later.", 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)

    # Deduplicate on request.headers["webhook-id"] (Redis, database) before processing.
    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="en"></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