Python

Integrate Constaia in Python with the official constaia SDK, sync and async, with retries, typed errors, verified webhooks and pytest tests.

This guide covers the Python integration with the official constaia SDK: analyse a file, a URL or base64, handle errors, control retries and concurrency, verify webhooks and test everything with a ck_test_ key. At the end you have the no-SDK alternative with httpx or requests. For frameworks, see Django, Flask, FastAPI and Celery; the full SDK reference is in Python SDK.

Requirements

Python ≥ 3.9 (some examples use match, from Python 3.10). The SDK only depends on httpx.
A test key ck_test_… from the dashboard (API keys). Test mode consumes no credits and the response depends on the file name: see Test mode.
The key lives only on your server. Never send it to a browser or a mobile app.

Installation

pip install constaia

Environment variables

.env

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

Constaia() reads CONSTAIA_API_KEY when you don't pass api_key. CONSTAIA_WEBHOOK_SECRET is the whsec_… secret returned when you create the webhook endpoint (in the dashboard or with client.webhook_endpoints.create(...)); it is shown only once.

First call

Your server decides which document you expect (expect) and which checks to apply (checks).

first_call.py

from constaia import Constaia

client = Constaia()  # reads CONSTAIA_API_KEY

analysis = client.analyze(
    "dni_valid.jpg",
    expect="es_dni",
    checks={"not_expired": True},
    storage="none",
    language="es",
)

print(analysis["verdict"]["status"])                    # valid
print(analysis["fields"]["document_number"]["value"])   # 12345678Z
for reason in analysis["verdict"]["reasons"]:
    print(reason["code"], reason["severity"], reason["message"])

With the test key and a real JPEG named dni_valid.jpg, the response (abridged) is:

{
  "id": "an_01J...",
  "object": "analysis",
  "status": "completed",
  "livemode": false,
  "document": { "type": "es_dni", "label": "DNI (España)", "confidence": 0.97, "side": "both", "country": "ESP" },
  "verdict": {
    "expected": ["es_dni"],
    "match": true,
    "status": "valid",
    "reasons": [
      { "code": "type_match", "severity": "info", "message": "El documento es DNI (España)." },
      { "code": "not_expired", "severity": "info", "message": "Vigente hasta el 12/03/2031." }
    ]
  },
  "fields": {
    "document_number": { "value": "12345678Z", "confidence": 0.99, "validated": true, "source": { "page": 1, "bbox": [0.61, 0.12, 0.83, 0.16] } }
  },
  "checks": [
    { "code": "nif_check_digit", "passed": true, "message": "La letra del documento 12345678Z es correcta." }
  ],
  "warnings": [],
  "usage": { "credits": 0, "pages": 1 }
}

Responses are plain dicts shaped exactly like the API JSON, typed with TypedDict (constaia.types.Analysis, Classification, Batch…), so mypy and pyright check field names. verdict is None if you don't pass expect. Decide on verdict["status"] (valid, invalid or review), use reasons[].code (stable) for your logic and reasons[].message (localised through language) for the user. More in Verdicts and Checks.

decide.py

from constaia.types import Analysis


def decide(analysis: Analysis) -> str:
    verdict = analysis["verdict"]
    if verdict is None:
        return "no verdict"
    if verdict["status"] == "valid":
        return "accepted"
    if verdict["status"] == "review":
        return "manual review"
    errors = [r["message"] for r in verdict["reasons"] if r["severity"] == "error"]
    return "rejected: " + "; ".join(errors)

Inputs

analyze() and classify() take exactly one of these inputs:

Input	Example
Path (`str` or `pathlib.Path`)	`client.analyze("scans/dni.jpg")`
`bytes`	`client.analyze(data, filename="dni.jpg")`
Open binary file	`client.analyze(open("dni.pdf", "rb"))`, Django `UploadedFile`, Flask `FileStorage.stream`
`(filename, bytes)` tuple	`client.analyze(("dni.jpg", data))`
Remote URL	`client.analyze(file_url="https://…/dni.jpg")`
Base64	`client.analyze(file_base64=b64, filename="dni.jpg")`

Always pass the original name with filename= when it is not implicit: in test mode it decides the response. The file is read into memory once so that retries resend the same bytes. file_url must be https, not a private IP, at most 20 MB and downloadable within 15 s.

inputs.py

from constaia import Constaia

client = Constaia()

receipt = client.analyze(
    file_url="https://files.example.com/receipts/payment_receipt.pdf",
    expect="payment_receipt",
    checks={
        "expected_amount": 45,
        "expected_iban": "ES7921000813610123456789",
        "expected_reference": "INSCRIPCION 123",
    },
)

with open("invoice.pdf", "rb") as fh:
    invoice = client.analyze(fh, expect="invoice", export=["xlsx"], metadata={"supplier": "acme"})
print(invoice["exports"].get("xlsx"))  # signed URL, valid for 24 h

kind = client.classify("unknown.pdf", expect=["invoice", "payment_receipt"])  # 0.2 credits
print(kind["document"], kind["candidates"])

Options use the API names (expect, extract, checks, storage, ttl_hours, keep_results, export, metadata, language). Since async is a Python keyword, that option is async_. Every option is in POST /v1/analyze.

Errors

Every exception inherits from constaia.ConstaiaError and exposes message, status, type, code, param and request_id.

Exception	HTTP	What to do
`InvalidRequestError`	400, 409, 413, 415, 422	Look at `code` (`file_too_large`, `unsupported_file_type`, `unreadable_image`, `too_many_pages`, `invalid_parameter` with `param`…)
`AuthenticationError`	401	Check `CONSTAIA_API_KEY`
`InsufficientCreditsError`	402	Top up credits in the dashboard; alert your team
`PermissionDeniedError`	403	The key can't perform that action
`NotFoundError`	404	Analysis deleted, from another account or created with `keep_results=False`
`RateLimitError`	429	Already retried by the SDK; `retry_after` says how long to wait
`APIError`	5xx	Already retried; `503` `live_mode_unavailable` means live mode is not available: use `ck_test_` meanwhile
`APIConnectionError`, `APITimeoutError`	—	Network or timeout after the retries
`WebhookVerificationError`	—	Invalid webhook signature

errors.py

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

client = Constaia()

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.",
    "too_many_pages": "The PDF has too many pages.",
}


def check_document(path: str) -> dict:
    try:
        analysis = client.analyze(path, expect="es_dni", checks={"not_expired": True}, language="en")
    except InvalidRequestError as exc:
        return {"error": USER_MESSAGES.get(exc.code or "", "The document could not be processed.")}
    except (InsufficientCreditsError, RateLimitError) as exc:
        print(f"Constaia {exc.status} {exc.code} request_id={exc.request_id}")
        return {"error": "The service is busy. Try again in a few minutes."}
    except ConstaiaError as exc:
        print(f"Constaia {exc.status} {exc.code} request_id={exc.request_id}: {exc.message}")
        return {"error": "We could not check the document. Please try again later."}
    return {"id": analysis["id"], "status": analysis["verdict"]["status"]}

Full list of codes in Errors.

Retries, idempotency and concurrency

The SDK already does what you would do by hand:

It retries 429, 408, 409 idempotency_in_progress, 5xx (except 501) and network errors up to max_retries times (2 by default), waiting for Retry-After (up to 60 s) or with exponential backoff and jitter.
It sends the same Idempotency-Key on every retry, so a retried analysis is never charged twice. Pass your own with idempotency_key= to deduplicate across processes (see Idempotency).
It keeps at most max_concurrency (4 by default) requests in flight per client, across threads or asyncio tasks; the rest wait in the client.

config.py

from constaia import Constaia

client = Constaia(
    timeout=60.0,        # seconds per attempt
    max_retries=3,
    max_concurrency=4,
)

analysis = client.analyze("dni_valid.jpg", expect="es_dni", idempotency_key="user-42-dni", timeout=90)

print(client.last_request_id)   # "req_…": include it when contacting support
print(client.last_rate_limit)   # RateLimit(limit=10, remaining=9, reset=1, policy='10;w=1', retry_after=None)

The limit is 2 requests per second per key on the free plan and 10 on paid (see Rate limits). With max_retries=0 you handle 429s yourself.

Async client

AsyncConstaia has the same methods with await. Reuse a single client and close it when you are done.

analyze_folder.py

import asyncio
from pathlib import Path

from constaia import AsyncConstaia, ConstaiaError


async def main() -> None:
    files = sorted(Path("inbox").glob("*.pdf"))
    async with AsyncConstaia(max_concurrency=4) as client:
        results = await asyncio.gather(
            *(client.analyze(path, expect="invoice") for path in files),
            return_exceptions=True,
        )
        for path, result in zip(files, results):
            if isinstance(result, ConstaiaError):
                print(path.name, "error", result.code, result.request_id)
            else:
                print(path.name, result["verdict"]["status"], result["fields"].get("total", {}).get("value"))

        async for analysis in client.analyses.list(status="completed", type="invoice"):
            print(analysis["id"])


asyncio.run(main())

max_concurrency already limits how many requests go out at once, even if you launch them all with gather. For more than a handful of files, a batch (client.batches.create(...), up to 100 documents) is usually simpler.

202 responses and long documents

A synchronous call waits up to 30 s. If the analysis is not finished, the API answers 202 with status queued or processing, and it always does so with async_=True (required for PDFs over 30 pages). The result arrives by webhook; if you cannot receive webhooks, poll the analysis:

poll.py

import time

from constaia import Constaia

client = Constaia()
analysis = client.analyze("contract.pdf", async_=True)
while analysis["status"] in ("queued", "processing"):
    time.sleep(3)
    analysis = client.analyses.get(analysis["id"])
print(analysis["status"], analysis.get("error"))

Verify webhooks

Constaia signs every delivery following Standard Webhooks (headers webhook-id, webhook-timestamp and webhook-signature). constaia.webhooks.verify() checks the signature and the timestamp (5-minute tolerance) and returns the event. Always pass the raw body, never re-serialised JSON.

handler.py

import os

import constaia

SECRET = os.environ["CONSTAIA_WEBHOOK_SECRET"]


def handle(raw_body: bytes, headers) -> int:
    try:
        event = constaia.webhooks.verify(raw_body, headers, SECRET)
    except constaia.WebhookVerificationError:
        return 400
    # Deduplicate on headers["webhook-id"]: it is the same across retries.
    match event["type"]:
        case "analysis.completed" | "analysis.review_required" | "analysis.failed":
            analysis = event["data"]
            print(event["type"], analysis["id"], (analysis.get("verdict") or {}).get("status"))
        case "batch.completed":
            print("batch", event["data"]["id"], event["data"]["counts"])
        case "credits.low":
            print("credits", event["data"]["credits_available"])
    return 204

Answer 2xx within 15 s and process in the background. Each framework guide shows how to get the raw body: request.body in Django, request.get_data() in Flask and await request.body() in FastAPI.

Tests with pytest

With a ck_test_ key the API is deterministic and free: the result depends on the file name, and the file must be a real JPEG, PNG, WEBP, HEIC or PDF (the type is sniffed from its content). Copy any JPEG into tests/fixtures/ as dni_valid.jpg, dni_expired.jpg and blurry.jpg. Messages are in Spanish because language defaults to es.

tests/test_constaia.py

import json
import os
from pathlib import Path

import pytest

import constaia
from constaia import Constaia

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


@pytest.fixture(scope="module")
def client():
    with Constaia() as c:
        yield c


@needs_test_key
def test_valid_dni(client):
    analysis = client.analyze(FIXTURES / "dni_valid.jpg", expect="es_dni")
    assert analysis["livemode"] is False
    assert analysis["usage"]["credits"] == 0
    assert analysis["verdict"]["status"] == "valid"
    assert analysis["fields"]["document_number"]["value"] == "12345678Z"


@needs_test_key
def test_expired_dni_is_invalid(client):
    analysis = client.analyze(FIXTURES / "dni_expired.jpg", expect="es_dni")
    assert analysis["verdict"]["status"] == "invalid"
    assert {"code": "not_expired", "severity": "error", "message": "Caducado el 15/06/2020."} in analysis["verdict"]["reasons"]


@needs_test_key
def test_blurry_needs_review(client):
    analysis = client.analyze(FIXTURES / "blurry.jpg", expect="es_dni")
    assert analysis["verdict"]["status"] == "review"
    assert "blurry" in analysis["warnings"]


def test_webhook_signature_roundtrip():
    secret = "whsec_c2VjcmV0LWRlLXBydWViYXMtMzItYnl0ZXMtbGFyZ28="
    body = json.dumps({"type": "analysis.completed", "data": {"id": "an_test"}})
    headers = constaia.webhooks.sign(body, secret)
    assert constaia.webhooks.verify(body.encode(), headers, secret)["data"]["id"] == "an_test"
    with pytest.raises(constaia.WebhookVerificationError):
        constaia.webhooks.verify(body.encode() + b" ", headers, secret)

CONSTAIA_API_KEY=ck_test_... pytest -q

All file names and their results are listed in Test mode.

Without the SDK: httpx or requests

If you can't add dependencies, the API is JSON over HTTPS. The file goes in the file field and the options in the options field as a JSON string. Without the SDK you add the Idempotency-Key, the Retry-After retries and the error parsing (error.code, error.request_id) yourself.

raw_httpx.py

import json
import os
import time
import uuid

import httpx

options = {"expect": "es_dni", "checks": {"not_expired": True}, "storage": "none"}
headers = {
    "Authorization": f"Bearer {os.environ['CONSTAIA_API_KEY']}",
    "Idempotency-Key": str(uuid.uuid4()),
}
content = open("dni_valid.jpg", "rb").read()

for attempt in range(3):
    response = httpx.post(
        "https://api.constaia.com/v1/analyze",
        headers=headers,
        files={"file": ("dni_valid.jpg", content, "image/jpeg")},
        data={"options": json.dumps(options)},
        timeout=60,
    )
    if response.status_code in (429, 500, 502, 503, 504) and attempt < 2:
        time.sleep(min(float(response.headers.get("Retry-After", 2**attempt)), 60))
        continue
    break

body = response.json()
if response.is_error:
    error = body["error"]
    raise RuntimeError(f"{response.status_code} {error['code']} {error['message']} ({error['request_id']})")
print(body["verdict"]["status"])

For URLs or base64 send JSON: {"file_url": "https://…", "options": {…}} or {"file_base64": "…", "filename": "dni.jpg", "options": {…}}. Webhook verification without the SDK: HMAC-SHA256 over <webhook-id>.<webhook-timestamp>.<raw body> keyed with the base64-decoded secret (without the whsec_ prefix), compared in constant time:

verify_webhook.py

import base64
import hashlib
import hmac
import json
import time


def verify_webhook(payload: bytes, headers, secret: str, tolerance: int = 300) -> dict:
    msg_id = headers.get("webhook-id", "")
    timestamp = headers.get("webhook-timestamp", "")
    signatures = headers.get("webhook-signature", "")
    if not (msg_id and timestamp.isdigit() and signatures) or abs(time.time() - int(timestamp)) > tolerance:
        raise ValueError("invalid webhook headers")
    key = base64.b64decode(secret.removeprefix("whsec_"))
    expected = "v1," + base64.b64encode(
        hmac.new(key, f"{msg_id}.{timestamp}.".encode() + payload, hashlib.sha256).digest()
    ).decode()
    if not any(hmac.compare_digest(candidate, expected) for candidate in signatures.split()):
        raise ValueError("invalid webhook signature")
    return json.loads(payload)

headers must support case-insensitive lookup (as Django, Flask and Starlette headers do).

Production checklist

Switch to a ck_live_ key (requires a verified email) only in the production environment.
Server upload limit ≥ 20 MB (for example client_max_body_size 21M; in nginx) and reject larger files early in your own code.
Timeouts ≥ 60 s along the whole path: SDK (60 s per attempt), worker (gunicorn --timeout 90), proxy and load balancer. Or use async_=True and webhooks.
The route that receives user files requires authentication and has its own rate limiting: every live analysis consumes credits.
Your server sets expect and checks; don't accept arbitrary options from the client.
One client per process (connection reuse) and max_concurrency × processes close to your requests-per-second limit.
Webhook with signature verification, fast 2xx, deduplication on webhook-id and queued processing.
Log the request_id of every error; read Storage and privacy to choose storage.

Python

Requirements

Installation

Environment variables

First call

Inputs

Errors

Retries, idempotency and concurrency

Async client

202 responses and long documents

Verify webhooks

Tests with pytest

Without the SDK: httpx or requests

Production checklist

Next steps

Python SDK

Django

FastAPI

Celery

On this page

Python

RequirementsCopy Anchor Link

InstallationCopy Anchor Link

Environment variablesCopy Anchor Link

First callCopy Anchor Link

InputsCopy Anchor Link

ErrorsCopy Anchor Link

Retries, idempotency and concurrencyCopy Anchor Link

Async clientCopy Anchor Link

202 responses and long documentsCopy Anchor Link

Verify webhooksCopy Anchor Link

Tests with pytestCopy Anchor Link

Without the SDK: httpx or requestsCopy Anchor Link

Production checklistCopy Anchor Link

Next stepsCopy Anchor Link

Python SDK

Django

FastAPI

Celery

On this page

Requirements

Installation

Environment variables

First call

Inputs

Errors

Retries, idempotency and concurrency

Async client

202 responses and long documents

Verify webhooks

Tests with pytest

Without the SDK: httpx or requests

Production checklist

Next steps