Django + Django REST Framework

Este ejemplo crea un endpoint POST /api/assistant/ask/ en tu backend de Django. El navegador solo habla con él; Django llama a Datafala y ejecuta el SQL en un alias de base de datos de solo lectura.

Reutiliza el cliente de la página Python en assistant/datafala.py, leyendo DATAFALA_API_KEY y DATAFALA_END_USER_SECRET de las variables de entorno.

Base de datos de solo lectura

settings.py

DATABASES["readonly"] = {
    "ENGINE": "django.db.backends.postgresql",
    "NAME": env("READONLY_DB_NAME"),
    "USER": env("READONLY_DB_USER"),  # el rol datafala_ro
    "PASSWORD": env("READONLY_DB_PASSWORD"),
    "HOST": env("READONLY_DB_HOST"),
    "PORT": "5432",
    "OPTIONS": {"options": "-c statement_timeout=5000 -c default_transaction_read_only=on"},
    "TEST": {"MIRROR": "default"},
}

El alias solo se usa con un cursor manual; migrate y los models siguen usando default.

Ejecución

assistant/runner.py

from django.db import connections


def run_readonly(sql: str, params: dict, max_rows: int = 1000):
    # params SIEMPRE, incluso vacío: sin él el driver no trata el %% de los literales.
    with connections["readonly"].cursor() as cursor:
        cursor.execute(sql, params)
        columns = [column[0] for column in cursor.description]
        rows = cursor.fetchmany(max_rows)
    return columns, rows

Vista

assistant/views.py

from django.db import DatabaseError
from rest_framework import serializers, status
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.views import APIView

from . import datafala
from .runner import run_readonly

# Códigos de error cuyo mensaje (ya traducido) puede mostrarse al usuario.
USER_FACING = {"no_answer", "sql_rejected", "question_too_long", "rate_limited"}


class AskSerializer(serializers.Serializer):
    question = serializers.CharField(max_length=2000)


class AskView(APIView):
    permission_classes = [IsAuthenticated]

    def post(self, request):
        body = AskSerializer(data=request.data)
        body.is_valid(raise_exception=True)
        try:
            answer = datafala.ask(body.validated_data["question"], user_id=request.user.pk)
        except datafala.DatafalaError as error:
            if error.code in USER_FACING:
                code = status.HTTP_429_TOO_MANY_REQUESTS if error.code == "rate_limited" else 422
                return Response({"code": error.code, "message": error.message}, status=code)
            return Response(
                {"code": "assistant_unavailable"}, status=status.HTTP_503_SERVICE_UNAVAILABLE
            )

        try:
            columns, rows = run_readonly(answer["sql"], answer["params"])
        except DatabaseError:
            return Response(
                {"code": "query_failed", "request_id": answer["request_id"]}, status=422
            )

        return Response(
            {
                "request_id": answer["request_id"],
                "explanation": answer["explanation"],
                "sql": answer["sql"],
                "columns": columns,
                "rows": rows,
            }
        )


class FeedbackView(APIView):
    permission_classes = [IsAuthenticated]

    def post(self, request):
        datafala.feedback(
            request.data["request_id"], request.data["rating"], request.data.get("comment")
        )
        return Response(status=status.HTTP_204_NO_CONTENT)

Puntos de atención

• Timeout del servidor web: la llamada al LLM puede tardar decenas de segundos. Ajusta el timeout de Gunicorn (--timeout 90) y del proxy inverso para esta ruta.
• Permiso: habilita la ruta solo a quienes pueden ver los datos de las tablas registradas. En sistemas multiempresa, lee Alcance de datos.
• Auditoría: guarda en tu base de datos quién preguntó, la pregunta y el request_id. El historial de Datafala es de todo el tenant, no de cada usuario.