Celery

CONSTAIA_API_KEY=ck_test_...
CONSTAIA_WEBHOOK_SECRET=whsec_...
CELERY_BROKER_URL=redis://localhost:6379/0

import os
from pathlib import Path

from celery import Celery

from constaia import (
    APIConnectionError,
    APIError,
    Constaia,
    InvalidRequestError,
    RateLimitError,
)

app = Celery("documents", broker=os.environ["CELERY_BROKER_URL"])
app.conf.task_acks_late = True

client = Constaia(timeout=60.0, max_retries=1, max_concurrency=4)

IDENTITY = {
    "expect": ["es_dni", "es_nie", "passport"],
    "checks": {"not_expired": True, "min_age_years": 18},
    "storage": "none",
}


def save_result(document_id: int, analysis: dict) -> None:
    status = (analysis.get("verdict") or {}).get("status") or analysis["status"]
    print(f"document {document_id}: {analysis['id']} {status}")  # guarda en tu base de datos


def mark_rejected(document_id: int, code: str | None) -> None:
    print(f"document {document_id}: rejected ({code})")  # guarda en tu base de datos


@app.task(bind=True, rate_limit="8/s", max_retries=5, soft_time_limit=180)
def analyze_document_url(self, document_id: int, file_url: str) -> str:
    try:
        analysis = client.analyze(
            file_url=file_url,
            **IDENTITY,
            metadata={"document_id": str(document_id)},
            idempotency_key=f"document-{document_id}",
        )
    except RateLimitError as exc:
        raise self.retry(exc=exc, countdown=exc.retry_after or 5)
    except (APIError, APIConnectionError) as exc:
        raise self.retry(exc=exc, countdown=min(30 * 2**self.request.retries, 900))
    except InvalidRequestError as exc:
        mark_rejected(document_id, exc.code)
        return "rejected"

    save_result(document_id, analysis)
    return analysis["id"]


@app.task(bind=True, rate_limit="8/s", max_retries=5, soft_time_limit=180)
def analyze_document_file(self, document_id: int, path: str, original_name: str) -> str:
    try:
        analysis = client.analyze(
            Path(path).read_bytes(),
            filename=original_name,
            **IDENTITY,
            metadata={"document_id": str(document_id)},
            idempotency_key=f"document-{document_id}",
        )
    except RateLimitError as exc:
        raise self.retry(exc=exc, countdown=exc.retry_after or 5)
    except (APIError, APIConnectionError) as exc:
        raise self.retry(exc=exc, countdown=min(30 * 2**self.request.retries, 900))
    except InvalidRequestError as exc:
        mark_rejected(document_id, exc.code)
        return "rejected"

    save_result(document_id, analysis)
    Path(path).unlink(missing_ok=True)
    return analysis["id"]

from tasks import analyze_document_file, analyze_document_url

analyze_document_url.delay(42, "https://storage.example.com/uploads/dni_valid.jpg?X-Signature=...")
analyze_document_file.delay(43, "/srv/shared/uploads/7f3a.jpg", "dni_valid.jpg")

from constaia import APIConnectionError, APIError, RateLimitError

from tasks import app, client

CHUNK = 100


@app.task(bind=True, max_retries=5)
def create_batch(self, job_id: str, chunk_index: int, file_urls: list[str]) -> str:
    try:
        batch = client.batches.create(
            items=[{"file_url": url} for url in file_urls],
            options={"expect": "invoice", "export": ["xlsx"], "metadata": {"job_id": job_id}},
            idempotency_key=f"batch-{job_id}-{chunk_index}",
        )
    except RateLimitError as exc:
        raise self.retry(exc=exc, countdown=exc.retry_after or 5)
    except (APIError, APIConnectionError) as exc:
        raise self.retry(exc=exc, countdown=min(30 * 2**self.request.retries, 900))
    return batch["id"]


def submit_job(job_id: str, file_urls: list[str]) -> None:
    for index in range(0, len(file_urls), CHUNK):
        create_batch.delay(job_id, index // CHUNK, file_urls[index:index + CHUNK])


@app.task
def handle_batch_completed(batch: dict) -> None:
    print(batch["id"], batch["counts"])
    for analysis_id in batch["analyses"]:
        analysis = client.analyses.get(analysis_id)
        status = (analysis.get("verdict") or {}).get("status") or analysis["status"]
        print(analysis_id, status, analysis["fields"].get("total", {}).get("value"))

from celery import group

from tasks import analyze_document_url

group(analyze_document_url.s(doc_id, url) for doc_id, url in pending).apply_async()

import os

from flask import Flask, abort, request

import constaia
from batch_tasks import handle_batch_completed
from tasks import app as celery_app

web = Flask(__name__)


@celery_app.task
def handle_analysis_event(event: dict) -> None:
    analysis = event["data"]
    print(event["type"], analysis["id"], (analysis.get("verdict") or {}).get("status"))


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

    if event["type"] == "batch.completed":
        handle_batch_completed.delay(event["data"])
    elif event["type"] in ("analysis.completed", "analysis.review_required", "analysis.failed"):
        handle_analysis_event.delay(event)
    return "", 204

import os
import shutil
from pathlib import Path

import pytest

from tasks import analyze_document_file, app

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


@pytest.fixture(autouse=True)
def eager():
    app.conf.task_always_eager = True
    yield
    app.conf.task_always_eager = False


@pytest.mark.parametrize("name", ["dni_valid.jpg", "dni_expired.jpg"])
def test_analyze_document_file(tmp_path, name):
    stored = tmp_path / "upload.bin"
    shutil.copy(FIXTURES / name, stored)

    result = analyze_document_file.delay(1, str(stored), name).get()

    assert result.startswith("an_")
    assert not stored.exists()