Pular para o conteúdo principal

SDK Python

🐍 Sync + async: SDK oficial com type hints completos, compatível com Django, Flask e FastAPI

O SDK Python permite integrar a Notifica em qualquer aplicação Python com suporte a sync e async. Este guia cobre:

  • Instalação e setup
  • Envio de notificações (email, SMS, WhatsApp, in-app, push)
  • Gerenciamento de subscribers, templates e workflows
  • Webhooks e analytics
  • Tratamento de erros

Instalação​

pip install notifica

Ou com poetry:

poetry add notifica

Requisitos: Python 3.8+ · Sem dependências obrigatórias (usa httpx opcionalmente para async)

Setup básico​

Síncrono​

from notifica import Notifica

client = Notifica("nk_live_sua_chave")

notification = client.notifications.send(
    channel="email",
    recipient="[email protected]",
    template_id="welcome",
    payload={"name": "Fernanda"},
)

print(f"Enviada: {notification.id} — {notification.status}")

Assíncrono (httpx)​

from notifica import AsyncNotifica

client = AsyncNotifica("nk_live_sua_chave")

notification = await client.notifications.send(
    channel="email",
    recipient="[email protected]",
    template_id="welcome",
    payload={"name": "Fernanda"},
)

Autenticação​

# Chave de produção
client = Notifica("nk_live_sua_chave")

# Chave de sandbox (desenvolvimento)
client = Notifica("nk_test_sua_chave")
dica

Use variáveis de ambiente para não commitar chaves no código:

import os
client = Notifica(os.environ["NOTIFICA_API_KEY"])

Configuração avançada​

client = Notifica(
    "nk_live_...",
    # URL customizada (self-hosted ou staging)
    base_url="https://api.staging.notifica.com.br",
    # Timeout em segundos
    timeout=60,
    # Retry com backoff exponencial
    max_retries=5,
)
OpçãoPadrãoDescrição
base_urlhttps://api.notifica.com.brURL base da API
timeout30Timeout em segundos
max_retries3Tentativas com backoff exponencial

Enviando notificações​

Todas as notificações são assíncronas e retornam status accepted (202).

Email​

n = client.notifications.send(
    channel="email",
    recipient="[email protected]",
    template_id="order-confirmation",
    payload={
        "name": "Roberto",
        "order_id": "ORD-12345",
    },
)

WhatsApp​

n = client.notifications.send(
    channel="whatsapp",
    recipient="+5511998765432",
    template_id="welcome",
    payload={
        "customer_name": "João",
        "company": "Acme",
    },
)

SMS​

n = client.notifications.send(
    channel="sms",
    recipient="+5511998765432",
    payload={"content": "Seu código de verificação é: 583921"},
)

In-App​

n = client.notifications.send(
    channel="in_app",
    recipient="subscriber-id-aqui",
    payload={
        "title": "Novo comentário",
        "body": "Maria comentou na sua tarefa #123",
    },
)

Com chave de idempotência​

Para evitar duplicatas em retries:

n = client.notifications.send(
    channel="email",
    recipient="[email protected]",
    template_id="payment-receipt",
    payload={"amount": "R$ 99,90"},
    idempotency_key="payment-12345",  # Use ID do seu sistema
)

Tracking de entrega​

Ver status de uma notificação​

n = client.notifications.get("notification-id")
print(n.status)        # "delivered"
print(n.delivered_at)  # 2024-01-15T14:30:05Z

Listar tentativas de entrega​

attempts = client.notifications.list_attempts("notification-id")
for a in attempts:
    print(f"Tentativa {a.attempt_number}: {a.status} ({a.provider})")

Listar notificações com filtros​

page = client.notifications.list(
    status="failed",
    channel="email",
    limit=50,
)

for n in page.data:
    print(n.id, n.status)

# Paginação via cursor
if page.meta.has_more:
    next_page = client.notifications.list(cursor=page.meta.cursor)

Subscribers (Assinantes)​

Criar ou atualizar (upsert)​

sub = client.subscribers.create(
    external_id="user-123",
    email="[email protected]",
    phone="+5511999999999",
    name="Maria Silva",
    custom_properties={
        "plan": "pro",
        "company": "Acme",
    },
)

Se o external_id já existir, o subscriber é atualizado automaticamente.

Listar com busca​

page = client.subscribers.list(search="maria", limit=20)
print(f"Encontrados: {page.total} subscribers")

Atualizar​

sub = client.subscribers.update("subscriber-id", name="Maria Santos")

Deletar (LGPD)​

Soft delete com nullificação de PII:

client.subscribers.delete("subscriber-id")

Preferências de notificação​

# Consultar
prefs = client.subscribers.get_preferences("subscriber-id")

# Atualizar
prefs = client.subscribers.set_preferences("subscriber-id", preferences=[
    {"category": "marketing", "channel": "email", "enabled": False},
    {"category": "transacional", "channel": "email", "enabled": True},
])

Importação em massa​

result = client.subscribers.bulk_import(subscribers=[
    {"external_id": "user-1", "email": "[email protected]", "name": "Ana"},
    {"external_id": "user-2", "email": "[email protected]", "name": "Bruno"},
])
print(f"Importados: {result.imported} subscribers")

Templates​

Criar template​

tmpl = client.templates.create(
    channel="email",
    slug="welcome",
    name="Email de Boas-vindas",
    content="Olá {{name}}, bem-vindo ao {{company}}!",
    status="published",
)

Listar por canal​

templates = client.templates.list(channel="email", status="published")

Preview com dados de exemplo​

result = client.templates.preview("template-id", variables={
    "name": "Maria",
    "company": "Acme",
})
print(result.output)  # "Olá Maria, bem-vindo ao Acme!"

Validar template​

result = client.templates.validate("template-id")
if not result.valid:
    print("Erros:", result.errors)

Workflows​

Criar workflow​

wf = client.workflows.create(
    slug="onboarding",
    name="Onboarding do Usuário",
    steps=[
        {"type": "send", "channel": "email", "template": "welcome"},
        {"type": "delay", "duration": "24h"},
        {"type": "send", "channel": "email", "template": "getting-started"},
        {"type": "delay", "duration": "72h"},
        {"type": "send", "channel": "whatsapp", "template": "check-in"},
    ],
)

Disparar workflow​

run = client.workflows.trigger("onboarding",
    recipient="user-123",
    data={"name": "João", "plan": "pro"},
)
print(f"Workflow run: {run.id} — {run.status}")

Cancelar workflow em execução​

run = client.workflows.cancel_run("run-id")

Webhooks​

Criar webhook​

O signing secret só é retornado na criação — guarde-o!

wh = client.webhooks.create(
    url="https://api.empresa.com/webhooks/notifica",
    events=["notification.sent", "notification.delivered", "notification.failed"],
)
print(f"Secret (guarde!): {wh.secret}")

Listar deliveries​

page = client.webhooks.list_deliveries("webhook-id", limit=50)
for d in page.data:
    print(f"Evento: {d.event_type}, Status HTTP: {d.response_status}")

Domínios de envio​

# Registrar
domain = client.domains.create(domain="mail.empresa.com.br")

# Verificar DNS
domain = client.domains.verify(domain.id)
if domain.status == "verified":
    print("Domínio verificado!")

# Health check
health = client.domains.check_health(domain.id)
print(f"SPF: {health.spf.status}, DKIM: {health.dkim.status}")

Analytics​

# Resumo dos últimos 7 dias
overview = client.analytics.overview("7d")
print(f"Enviados: {overview.sent}, Entregues: {overview.delivered}")

# Por canal
stats = client.analytics.by_channel("30d")
for s in stats:
    print(f"{s.channel}: {s.sent} enviados")

# Timeseries
points = client.analytics.timeseries(period="7d", granularity="day")

# Top templates
top = client.analytics.top_templates(period="30d", limit=5)

Notificações In-App​

# Listar não-lidas
notifications = client.in_app.list("subscriber-id", unread_only=True, limit=20)

# Contagem
count = client.in_app.get_unread_count("subscriber-id")
print(f"{count} notificações não lidas")

# Marcar como lida
client.in_app.mark_read("subscriber-id", "notification-id")

# Marcar todas como lidas
client.in_app.mark_all_read("subscriber-id")

Audit Logs (Admin Only)​

Autenticação Admin Necessária

Audit logs requerem autenticação admin (Bearer token do backoffice). Não estão disponíveis com API keys regulares (nk_live_... ou nk_test_...).

Rastreie ações sensíveis na sua organização:

  • API Keys: created, rotated, revoked
  • Team Members: invited, removed, role_changed
  • Subscriptions: created, plan_changed, cancelled, payment_failed
  • Domains: added, verified, removed
  • Webhooks: created, updated, deleted
# Listar logs recentes
result = client.audit.list({"limit": 50})
for log in result["data"]:
    print(f"{log['action']} por {log['actor']['name']} em {log['created_at']}")

# Filtrar por tipo de ação
api_key_logs = client.audit.list({
    "action": "api_key.created",
    "limit": 20,
})

# Filtrar por tipo de recurso
webhook_logs = client.audit.list({
    "resource_type": "webhook",
})

# Filtrar por período
logs = client.audit.list({
    "start_date": "2024-01-01T00:00:00Z",
    "end_date": "2024-01-31T23:59:59Z",
})

# Obter um audit log específico
log = client.audit.get("audit_abc123")

# Auto-paginar por todos os logs
for log in client.audit.list_auto({"resource_type": "api_key"}):
    print(log["action"], log["resource_id"])

Filtros disponíveis​

ParâmetroTipoDescrição
actionstringFiltrar por ação (ex: api_key.created)
resource_typestringFiltrar por tipo de recurso (api_key, webhook, etc.)
resource_idstringFiltrar por ID do recurso
actor_typestringFiltrar por tipo de ator (user, api_key, system)
actor_idstringFiltrar por ID do ator
start_datestringData inicial (ISO 8601)
end_datestringData final (ISO 8601)

Tratamento de erros​

from notifica import Notifica, NotificaError, ValidationError, RateLimitError

try:
    client.notifications.send(...)
except ValidationError as e:
    # 422: Dados inválidos
    print(f"Campos inválidos: {e.details}")
except RateLimitError as e:
    # 429: Limite excedido
    print("Rate limited, tente novamente")
except NotificaError as e:
    # Qualquer erro da API
    print(f"HTTP {e.status_code} | {e.code}: {e.message}")
informação

O SDK faz retry automático com backoff exponencial para erros 429 e 5xx. Configure max_retries=0 para desabilitar.

Integração com frameworks​

FastAPI​

from fastapi import FastAPI
from notifica import AsyncNotifica

app = FastAPI()
client = AsyncNotifica(os.environ["NOTIFICA_API_KEY"])

@app.post("/send-welcome")
async def send_welcome(email: str, name: str):
    n = await client.notifications.send(
        channel="email",
        recipient=email,
        template_id="welcome",
        payload={"name": name},
    )
    return {"notification_id": n.id, "status": n.status}

Django​

# settings.py
NOTIFICA_API_KEY = os.environ["NOTIFICA_API_KEY"]

# views.py
from notifica import Notifica
from django.conf import settings

client = Notifica(settings.NOTIFICA_API_KEY)

def send_welcome(request):
    n = client.notifications.send(
        channel="email",
        recipient=request.POST["email"],
        template_id="welcome",
        payload={"name": request.POST["name"]},
    )
    return JsonResponse({"notification_id": n.id})

Flask​

from flask import Flask, request, jsonify
from notifica import Notifica

app = Flask(__name__)
client = Notifica(os.environ["NOTIFICA_API_KEY"])

@app.post("/send-welcome")
def send_welcome():
    data = request.json
    n = client.notifications.send(
        channel="email",
        recipient=data["email"],
        template_id="welcome",
        payload={"name": data["name"]},
    )
    return jsonify(notification_id=n.id, status=n.status)

FAQ​

Q: Preciso do httpx para o modo síncrono? A: Não. O modo síncrono usa urllib3 ou requests (se disponível). O httpx é necessário apenas para AsyncNotifica.

Q: Como funciona o retry automático? A: O SDK retenta automaticamente em erros 429 (rate limit) e 5xx (servidor) com backoff exponencial: 1s, 2s, 4s...

Q: Os type hints são completos? A: Sim. O SDK é totalmente tipado e compatível com mypy em modo strict.

Q: Qual a versão mínima do Python? A: Python 3.8+. Recomendamos 3.10+ para melhor suporte a type hints.