Python

Um cliente pequeno com httpx, tratamento de erros pelo code e execução no banco com usuário somente leitura.

pip install httpx "psycopg[binary]"   # ou: pip install httpx pymysql
# Firebird 2.5+: pip install fdb; Firebird 3+: pip install firebird-driver

Cliente

datafala_client.py

import hashlib
import hmac
import os
from dataclasses import dataclass, field

import httpx

DATAFALA_URL = os.environ.get("DATAFALA_URL", "https://app.datafala.com.br")
API_KEY = os.environ["DATAFALA_API_KEY"]
END_USER_SECRET = os.environ["DATAFALA_END_USER_SECRET"].encode()

# A LLM pode levar vários segundos; 60 s de leitura cobre a correção automática.
_client = httpx.Client(
    base_url=DATAFALA_URL,
    headers={"Authorization": f"Bearer {API_KEY}"},
    timeout=httpx.Timeout(60.0, connect=5.0),
)


@dataclass
class DatafalaError(Exception):
    status: int
    code: str
    message: str = ""
    details: dict = field(default_factory=dict)
    retry_after: int | None = None
    request_id: str | None = None


def end_user_id(user_id: int | str) -> str:
    return hmac.new(END_USER_SECRET, str(user_id).encode(), hashlib.sha256).hexdigest()


def _raise_for_error(response: httpx.Response) -> None:
    try:
        body = response.json()
    except ValueError:
        body = {}
    retry_after = response.headers.get("Retry-After")
    raise DatafalaError(
        status=response.status_code,
        code=body.get("code", "unknown"),
        message=body.get("message", ""),
        details=body.get("details") or {},
        retry_after=int(retry_after) if retry_after and retry_after.isdigit() else None,
        request_id=response.headers.get("X-Request-ID"),
    )


def ask(question: str, user_id: int | str, *, max_rows: int | None = None) -> dict:
    payload: dict = {"question": question, "end_user": end_user_id(user_id)}
    if max_rows is not None:
        payload["max_rows"] = max_rows
    response = _client.post("/v1/ask", json=payload)
    if response.status_code != 200:
        _raise_for_error(response)
    return response.json()  # sql, params, dialect, tables, explanation, locale, request_id


def feedback(request_id: str, rating: str, comment: str | None = None) -> None:
    payload = {"request_id": request_id, "rating": rating}
    if comment:
        payload["comment"] = comment[:500]
    response = _client.post("/v1/feedback", json=payload)
    if response.status_code != 204:
        _raise_for_error(response)


def find_in_history(text: str) -> dict | None:
    response = _client.get("/v1/history", params={"q": text[:200], "limit": 1})
    if response.status_code != 200:
        _raise_for_error(response)
    items = response.json()["items"]
    return items[0] if items else None

Executar no PostgreSQL (psycopg 3)

import os

import psycopg

from datafala_client import DatafalaError, ask

readonly = psycopg.connect(
    os.environ["READONLY_DATABASE_URL"],
    options="-c statement_timeout=5000 -c default_transaction_read_only=on",
)

try:
    answer = ask("Top 5 clientes por faturamento em 2026", user_id=42, max_rows=5)
except DatafalaError as error:
    if error.code in {"no_answer", "sql_rejected"}:
        print(error.message)  # já traduzida; peça para reformular
    else:
        raise
else:
    with readonly.cursor() as cursor:
        cursor.execute(answer["sql"], answer["params"])  # params sempre, mesmo {}
        columns = [column.name for column in cursor.description]
        rows = cursor.fetchmany(1000)
    print(answer["explanation"], columns, rows)

Executar no MySQL (PyMySQL)

import os

import pymysql

readonly = pymysql.connect(
    host=os.environ["MYSQL_HOST"],
    user="datafala_ro",
    password=os.environ["MYSQL_PASSWORD"],
    database=os.environ["MYSQL_DATABASE"],
    init_command="SET SESSION max_execution_time = 5000",
)

with readonly.cursor() as cursor:
    cursor.execute(answer["sql"], answer["params"])  # PyMySQL aceita %(nome)s
    rows = cursor.fetchmany(1000)

Executar no Firebird

Para Firebird 2.5 use FDB; firebird-driver requer Firebird 3 ou mais novo. Os dois drivers usam ?, então converta os placeholders %(nome)s e %% antes da execução e configure a transação como READ ONLY. Veja Executar o SQL para um exemplo FDB e os limites de timeout por versão.