Django

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

import os

CONSTAIA_API_KEY = os.environ["CONSTAIA_API_KEY"]
CONSTAIA_WEBHOOK_SECRET = os.environ["CONSTAIA_WEBHOOK_SECRET"]

CONSTAIA_MAX_UPLOAD_BYTES = 20 * 1024 * 1024

from django.conf import settings

from constaia import Constaia
from constaia.types import Analysis

client = Constaia(api_key=settings.CONSTAIA_API_KEY, timeout=60.0, max_retries=2)


def verify_identity(uploaded_file, user, language: str = "es") -> Analysis:
    return client.analyze(
        uploaded_file,
        filename=uploaded_file.name,
        expect=["es_dni", "es_nie", "passport"],
        checks={
            "not_expired": True,
            "min_age_years": 18,
            "holder": {"full_name": user.get_full_name()},
        },
        storage="none",
        language=language if language in ("es", "en", "pt", "fr") else "es",
        metadata={"user_id": str(user.pk)},
    )

from django.conf import settings
from django.db import models


class IdentityDocument(models.Model):
    user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
    constaia_id = models.CharField(max_length=64, unique=True)
    status = models.CharField(max_length=16)  # valid | invalid | review | queued | processing | failed
    document_number = models.CharField(max_length=32, blank=True)
    created_at = models.DateTimeField(auto_now_add=True)

from django import forms
from django.conf import settings
from django.core.validators import FileExtensionValidator


class IdentityDocumentForm(forms.Form):
    document = forms.FileField(
        label="DNI, NIE o pasaporte",
        validators=[FileExtensionValidator(["jpg", "jpeg", "png", "webp", "heic", "pdf"])],
    )

    def clean_document(self):
        document = self.cleaned_data["document"]
        if document.size > settings.CONSTAIA_MAX_UPLOAD_BYTES:
            raise forms.ValidationError("El archivo supera los 20 MB.")
        return document

import logging

from django.contrib import messages
from django.contrib.auth.decorators import login_required
from django.shortcuts import redirect, render

from constaia import ConstaiaError, InsufficientCreditsError, InvalidRequestError, RateLimitError

from .constaia_client import verify_identity
from .forms import IdentityDocumentForm
from .models import IdentityDocument

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.",
}


@login_required
def upload_identity(request):
    form = IdentityDocumentForm(request.POST or None, request.FILES or None)
    if request.method == "POST" and form.is_valid():
        try:
            analysis = verify_identity(form.cleaned_data["document"], request.user)
        except InvalidRequestError as exc:
            form.add_error("document", USER_MESSAGES.get(exc.code or "", "El documento no se ha podido procesar."))
        except (InsufficientCreditsError, RateLimitError) as exc:
            logger.error("Constaia %s %s request_id=%s", exc.status, exc.code, exc.request_id)
            form.add_error("document", "El servicio está ocupado. Inténtalo en unos minutos.")
        except ConstaiaError as exc:
            logger.exception("Constaia %s %s request_id=%s", exc.status, exc.code, exc.request_id)
            form.add_error("document", "No hemos podido comprobar el documento. Inténtalo más tarde.")
        else:
            verdict = analysis["verdict"]
            if verdict and verdict["status"] == "invalid":
                for reason in verdict["reasons"]:
                    if reason["severity"] == "error":
                        form.add_error("document", reason["message"])
            else:
                number = analysis["fields"].get("document_number") or analysis["fields"].get("nie_number") or {}
                IdentityDocument.objects.create(
                    user=request.user,
                    constaia_id=analysis["id"],
                    status=verdict["status"] if verdict else analysis["status"],
                    document_number=number.get("value") or "",
                )
                messages.success(request, "Documento recibido.")
                return redirect("identity-upload")
    return render(request, "documents/upload.html", {"form": form})

<form method="post" enctype="multipart/form-data">
  {% csrf_token %}
  {{ form.as_p }}
  <button type="submit">Comprobar</button>
</form>

import json
import logging

from django.conf import settings
from rest_framework import serializers, status
from rest_framework.parsers import MultiPartParser
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.throttling import UserRateThrottle
from rest_framework.views import APIView

from constaia import ConstaiaError, InvalidRequestError, RateLimitError

from .constaia_client import verify_identity

logger = logging.getLogger(__name__)


class DocumentUploadSerializer(serializers.Serializer):
    file = serializers.FileField()
    options = serializers.CharField(required=False, allow_blank=True)

    def validate_file(self, value):
        if value.size > settings.CONSTAIA_MAX_UPLOAD_BYTES:
            raise serializers.ValidationError("El archivo supera los 20 MB.")
        return value


class IdentityDocumentAPIView(APIView):
    parser_classes = [MultiPartParser]
    permission_classes = [IsAuthenticated]
    throttle_classes = [UserRateThrottle]

    def post(self, request):
        serializer = DocumentUploadSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        try:
            language = json.loads(serializer.validated_data.get("options") or "{}").get("language", "es")
        except (ValueError, AttributeError):
            language = "es"

        try:
            analysis = verify_identity(serializer.validated_data["file"], request.user, language)
        except InvalidRequestError as exc:
            return Response({"error": {"code": exc.code, "message": exc.message}}, status=422)
        except RateLimitError:
            return Response({"error": {"message": "Inténtalo en unos minutos."}}, status=503)
        except ConstaiaError as exc:
            logger.error("Constaia %s %s request_id=%s", exc.status, exc.code, exc.request_id)
            return Response({"error": {"message": "No hemos podido comprobar el documento."}}, status=502)

        return Response(
            {key: analysis.get(key) for key in ("id", "object", "status", "document", "verdict", "warnings")},
            status=status.HTTP_200_OK,
        )

REST_FRAMEWORK = {
    "DEFAULT_THROTTLE_RATES": {"user": "20/hour"},
}

from django.conf import settings
from django.core.cache import cache
from django.http import HttpResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST

import constaia

from .models import IdentityDocument


@csrf_exempt
@require_POST
def constaia_webhook(request):
    try:
        event = constaia.webhooks.verify(request.body, request.headers, settings.CONSTAIA_WEBHOOK_SECRET)
    except constaia.WebhookVerificationError:
        return HttpResponse(status=400)

    if not cache.add(f"constaia:webhook:{request.headers['webhook-id']}", True, timeout=4 * 86400):
        return HttpResponse(status=204)

    match event["type"]:
        case "analysis.completed" | "analysis.review_required" | "analysis.failed":
            analysis = event["data"]
            status = (analysis.get("verdict") or {}).get("status") or analysis["status"]
            IdentityDocument.objects.filter(constaia_id=analysis["id"]).update(status=status)
        case "credits.low":
            pass  # avisa a tu equipo: event["data"]["credits_available"]

    return HttpResponse(status=204)

from django.urls import path

from . import api, views, webhooks

urlpatterns = [
    path("identity/", views.upload_identity, name="identity-upload"),
    path("api/identity/", api.IdentityDocumentAPIView.as_view(), name="identity-api"),
    path("webhooks/constaia/", webhooks.constaia_webhook, name="constaia-webhook"),
]

import json
import os
from pathlib import Path
from unittest import skipUnless

from django.conf import settings
from django.contrib.auth import get_user_model
from django.core.files.uploadedfile import SimpleUploadedFile
from django.test import TestCase
from django.urls import reverse

import constaia

from documents.models import IdentityDocument

FIXTURES = Path(__file__).parent / "fixtures"
TEST_KEY = os.environ.get("CONSTAIA_API_KEY", "").startswith("ck_test_")


def upload(name: str) -> SimpleUploadedFile:
    return SimpleUploadedFile(name, (FIXTURES / name).read_bytes(), content_type="image/jpeg")


class IdentityUploadTests(TestCase):
    def setUp(self):
        self.user = get_user_model().objects.create_user(
            "maria", password="test-password", first_name="María", last_name="García López"
        )
        self.client.force_login(self.user)

    @skipUnless(TEST_KEY, "needs a ck_test_ key")
    def test_valid_dni_is_saved(self):
        self.client.post(reverse("identity-upload"), {"document": upload("dni_valid.jpg")})
        doc = IdentityDocument.objects.get(user=self.user)
        self.assertEqual(doc.status, "valid")
        self.assertEqual(doc.document_number, "12345678Z")

    @skipUnless(TEST_KEY, "needs a ck_test_ key")
    def test_expired_dni_shows_reason(self):
        response = self.client.post(reverse("identity-upload"), {"document": upload("dni_expired.jpg")})
        self.assertContains(response, "Caducado el 15/06/2020.")

    def test_webhook_signature(self):
        body = json.dumps({"type": "analysis.completed", "data": {"id": "an_test", "status": "completed"}})
        headers = constaia.webhooks.sign(body, settings.CONSTAIA_WEBHOOK_SECRET)
        url = reverse("constaia-webhook")

        ok = self.client.post(url, data=body, content_type="application/json", headers=headers)
        bad = self.client.post(url, data=body, content_type="application/json",
                               headers={**headers, "webhook-signature": "v1,bad"})
        self.assertEqual(ok.status_code, 204)
        self.assertEqual(bad.status_code, 400)