Django + Django REST Framework

Este exemplo cria um endpoint POST /api/assistant/ask/ no seu backend Django. O navegador fala só com ele; o Django chama o Datafala e executa o SQL num alias de banco somente leitura.

Reaproveite o cliente da página Python em assistant/datafala.py, lendo DATAFALA_API_KEY e DATAFALA_END_USER_SECRET das variáveis de ambiente.

Banco somente leitura

settings.py

DATABASES["readonly"] = {
    "ENGINE": "django.db.backends.postgresql",
    "NAME": env("READONLY_DB_NAME"),
    "USER": env("READONLY_DB_USER"),  # o papel 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"},
}

O alias só é usado com cursor manual; migrate e os models continuam no default.

Execução

assistant/runner.py

from django.db import connections


def run_readonly(sql: str, params: dict, max_rows: int = 1000):
    # params SEMPRE, mesmo vazio: sem ele o driver não trata o %% dos literais.
    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

View

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

# Erros cuja mensagem (já traduzida) pode ser mostrada ao usuário.
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)

Pontos de atenção

• Timeout do servidor web: a chamada à LLM pode levar até dezenas de segundos. Ajuste o timeout do Gunicorn (--timeout 90) e do proxy reverso para essa rota.
• Permissão: libere a rota só para quem pode ver os dados das tabelas cadastradas. Em sistemas multiempresa, leia Escopo de dados.
• Auditoria: guarde no seu banco quem perguntou, a pergunta e o request_id. O histórico do Datafala é do tenant inteiro, não por usuário.