SuperDB com Python.

pip install requests pyjwt python-dotenv

# Auth: auth.superdb.com.br  |  Dados (PostgREST): api.superdb.com.br
SUPERDB_AUTH_URL=https://auth.superdb.com.br
SUPERDB_DATA_URL=https://api.superdb.com.br
SUPERDB_SERVICE_ROLE_KEY=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...SERVICE...
SUPERDB_ANON_KEY=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...ANON...

import os
import requests
from dotenv import load_dotenv

load_dotenv()

SUPERDB_URL = os.environ["SUPERDB_URL"]
# URL de dados — sempre api.superdb.com.br (sem /rest/v1/)
SUPERDB_DATA_URL = "https://api.superdb.com.br"
SERVICE_ROLE_KEY = os.environ["SUPERDB_SERVICE_ROLE_KEY"]
ANON_KEY = os.environ["SUPERDB_ANON_KEY"]


def service_headers() -> dict:
    """Headers com Service Role — bypassa RLS. Use apenas server-side."""
    return {
        "Authorization": f"Bearer {SERVICE_ROLE_KEY}",
        "apikey": SERVICE_ROLE_KEY,
        "Content-Type": "application/json",
    }


def user_headers(user_token: str) -> dict:
    """Headers com JWT do usuário — respeita RLS."""
    return {
        "Authorization": f"Bearer {user_token}",
        "apikey": ANON_KEY,
        "Content-Type": "application/json",
    }


def list_rows(table: str, token: str | None = None, query: str = "") -> list:
    """Busca linhas de uma tabela via PostgREST.

    Endpoint: api.superdb.com.br/ (sem prefixo /rest/v1/).
    """
    headers = user_headers(token) if token else service_headers()
    url = f"{SUPERDB_DATA_URL}/{table}{query}"
    resp = requests.get(url, headers=headers)
    resp.raise_for_status()
    return resp.json()


def insert_row(table: str, data: dict, token: str | None = None) -> dict:
    """Insere uma linha em uma tabela."""
    headers = user_headers(token) if token else service_headers()
    headers["Prefer"] = "return=representation"
    url = f"{SUPERDB_DATA_URL}/{table}"
    resp = requests.post(url, headers=headers, json=data)
    resp.raise_for_status()
    return resp.json()[0] if resp.json() else {}


def delete_rows(table: str, filters: str, token: str | None = None) -> None:
    """Deleta linhas com filtro PostgREST. Ex: filters='id=eq.42'"""
    headers = user_headers(token) if token else service_headers()
    url = f"{SUPERDB_DATA_URL}/{table}?{filters}"
    resp = requests.delete(url, headers=headers)
    resp.raise_for_status()

import os
import jwt
from fastapi import HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

# JWT secret do projeto — pegar em app.superdb.com.br → Settings → JWT
SUPERDB_JWT_SECRET = os.environ["SUPERDB_JWT_SECRET"]


def verify_token(
    credentials: HTTPAuthorizationCredentials = Security(security)
) -> dict:
    """
    Verifica e decodifica o JWT do SuperDB.
    Use como dependência nas rotas protegidas.
    """
    token = credentials.credentials
    try:
        payload = jwt.decode(
            token,
            SUPERDB_JWT_SECRET,
            algorithms=["HS256"],
            options={"verify_aud": False}  # SuperDB não usa audience
        )
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expirado")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Token inválido")


# Uso em rota protegida:
from fastapi import FastAPI, Depends

app = FastAPI()

@app.get("/minha-api/dados")
async def get_dados(user: dict = Depends(verify_token)):
    user_id = user.get("sub")  # UUID do usuário no SuperDB
    # Busca dados desse usuário específico
    posts = list_rows("posts", query=f"?user_id=eq.{user_id}")
    return {"user_id": user_id, "posts": posts}

import csv
import requests
from superdb_client import service_headers, SUPERDB_DATA_URL


def migrate_users_from_csv(csv_path: str) -> None:
    """Importa usuários de um CSV para a tabela profiles do SuperDB.

    Endpoint: api.superdb.com.br/profiles (sem prefixo /rest/v1/).
    """
    headers = service_headers()
    headers["Prefer"] = "return=minimal,resolution=ignore-duplicates"

    with open(csv_path, newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        rows = list(reader)

    # Inserção em batch de 100 em 100
    batch_size = 100
    for i in range(0, len(rows), batch_size):
        batch = rows[i:i + batch_size]
        resp = requests.post(
            f"{SUPERDB_DATA_URL}/profiles",
            headers=headers,
            json=batch,
        )
        resp.raise_for_status()
        print(f"Inseridos {min(i + batch_size, len(rows))}/{len(rows)}")

    print(f"Migração completa: {len(rows)} registros.")


if __name__ == "__main__":
    migrate_users_from_csv("usuarios.csv")

from fastapi import FastAPI, Request, HTTPException
import hmac, hashlib, os

app = FastAPI()
WEBHOOK_SECRET = os.environ["SUPERDB_WEBHOOK_SECRET"]


@app.post("/webhooks/superdb")
async def handle_webhook(request: Request):
    """Recebe eventos do SuperDB (ex: novo usuário criado)."""
    body = await request.body()
    signature = request.headers.get("x-superdb-signature", "")

    # Verificar assinatura HMAC
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        body,
        hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(f"sha256={expected}", signature):
        raise HTTPException(status_code=401, detail="Assinatura inválida")

    event = await request.json()
    event_type = event.get("type")

    if event_type == "INSERT" and event.get("table") == "users":
        new_user = event.get("record", {})
        print(f"Novo usuário: {new_user.get('email')}")
        # Enviar email de boas-vindas, criar recursos iniciais, etc.

    return {"ok": True}