Flask

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

import json
import logging
import os

from flask import Flask, abort, jsonify, request

import constaia
from constaia import Constaia, ConstaiaError, InsufficientCreditsError, InvalidRequestError, RateLimitError

app = Flask(__name__)
app.config["MAX_CONTENT_LENGTH"] = 21 * 1024 * 1024
app.config["CONSTAIA_WEBHOOK_SECRET"] = os.environ["CONSTAIA_WEBHOOK_SECRET"]

client = Constaia(timeout=60.0, max_retries=2)
logger = logging.getLogger(__name__)

ALLOWED_EXTENSIONS = {"jpg", "jpeg", "png", "webp", "heic", "pdf"}
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.",
}


def error(message: str, status: int):
    return jsonify({"error": {"message": message}}), status


@app.errorhandler(413)
def too_large(_):
    return error("The file is larger than 20 MB.", 413)


@app.post("/documents")
def documents():
    # Protect this route with your login (for example flask_login.login_required): every live analysis consumes credits.
    upload = request.files.get("file")
    if upload is None or not upload.filename:
        return error("No document received.", 400)
    if upload.filename.rsplit(".", 1)[-1].lower() not in ALLOWED_EXTENSIONS:
        return error(USER_MESSAGES["unsupported_file_type"], 400)

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

    try:
        analysis = client.analyze(
            upload.stream,
            filename=upload.filename,
            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)

    # Store analysis["id"] and the verdict in your database here.
    return jsonify({key: analysis.get(key) for key in ("id", "object", "status", "document", "verdict", "warnings")})


@app.post("/webhooks/constaia")
def constaia_webhook():
    try:
        event = constaia.webhooks.verify(request.get_data(), request.headers, app.config["CONSTAIA_WEBHOOK_SECRET"])
    except constaia.WebhookVerificationError:
        abort(400)

    # Deduplicate on request.headers["webhook-id"] (Redis, database) and queue heavy work.
    match event["type"]:
        case "analysis.completed" | "analysis.review_required" | "analysis.failed":
            analysis = event["data"]
            logger.info("Constaia %s: %s", analysis["id"], (analysis.get("verdict") or {}).get("status"))
        case "credits.low":
            logger.warning("Constaia credits low: %s", event["data"]["credits_available"])

    return "", 204

<script type="module" src="https://cdn.jsdelivr.net/npm/@constaia/widget@0.1"></script>
<constaia-upload endpoint="{{ url_for('documents') }}" document="es_dni" lang="en"></constaia-upload>

import io
import json
import os
from pathlib import Path

import pytest

import constaia
from app import 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():
    return app.test_client()


def post_file(http, name: str):
    data = {"file": (io.BytesIO((FIXTURES / name).read_bytes()), name)}
    return http.post("/documents", data=data, content_type="multipart/form-data")


@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 = post_file(http, name)
    assert response.status_code == 200
    assert response.get_json()["verdict"]["status"] == status


def test_webhook_signature(http):
    body = json.dumps({"type": "analysis.completed", "data": {"id": "an_test", "status": "completed"}})
    headers = constaia.webhooks.sign(body, app.config["CONSTAIA_WEBHOOK_SECRET"])

    assert http.post("/webhooks/constaia", data=body, headers=headers, content_type="application/json").status_code == 204
    headers["webhook-signature"] = "v1,bad"
    assert http.post("/webhooks/constaia", data=body, headers=headers, content_type="application/json").status_code == 400

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