convertaja (Python)

SDK oficial da API do Convertajá para Python: comprimir e converter PDF, gerar DANFE a partir do XML da NF-e, converter etiquetas ZPL, assinatura eletrônica, Central de Expedição e consultas de CNPJ, CEP, boleto e Pix.

• Só biblioteca padrão (urllib), sem requests. Python 3.9 ou mais novo, com type hints (py.typed).
• Repete sozinho em falha de rede, 429 (respeitando Retry-After) e 5xx, sem cobrar duas vezes (Idempotency-Key automática).

Ainda não publicado no PyPI. Até a publicação, instale pelo caminho local ou pelo repositório git (veja abaixo).

Instalação

# caminho local
pip install ./caminho/para/integrations/sdk-python

# ou direto do git (quando o repositório estiver disponível)
pip install "git+https://github.com/convertaja/sdk-python.git"

# depois da publicação
pip install convertaja

Primeiros passos

Crie a chave em Minha conta → API (/conta/api). Use cj_test_… para testar (não consome créditos, até 20 chamadas por dia) e cj_live_… em produção.

import os
from convertaja import Convertaja

cj = Convertaja(api_key=os.environ["CONVERTAJA_API_KEY"])  # ou só Convertaja() lendo CONVERTAJA_API_KEY

me = cj.me()
print(me["plan"]["label"], me["credits"]["remaining"])
print("Créditos depois da última chamada:", cj.credits_remaining)

Opções: base_url (padrão https://convertaja.com.br), max_retries (padrão 2), timeout em segundos (padrão 120), retry_delay (padrão 0,5 s).

Comprimir um PDF

# wait=True segura a resposta até 60 s; se ainda não terminou, use jobs.wait
job = cj.tools.run(
    "comprimir-pdf",
    files=["contrato.pdf"],          # caminho, bytes, arquivo aberto, {"data", "name"}, {"url"} ou {"base64", "name"}
    options={"level": "high"},       # low | medium | high
    wait=True,
)
if job["status"] not in ("done", "error"):
    job = cj.jobs.wait(job["id"])
if job["status"] == "error":
    raise RuntimeError(job["error"]["message"])

arquivo = cj.jobs.download(job["id"])  # um arquivo, ou ZIP se houver vários
arquivo.save(arquivo.filename or "comprimido.pdf")

Consultar CNPJ e CEP (grátis)

empresa = cj.lookups.cnpj("11.222.333/0001-81")  # aceita o CNPJ alfanumérico
print(empresa["razaoSocial"], empresa["situacao"])

endereco = cj.lookups.cep("01310-100")
print(endereco["logradouro"], endereco["cidade"], endereco["uf"])

DANFE a partir do XML da NF-e

job = cj.tools.run(
    "xml-nfe-para-pdf",
    files=["nota.xml"],              # vários XMLs ou um ZIP
    options={"layout": "a4"},        # a4 | simplificado (10×15, etiqueta)
    wait=True,
)
cj.jobs.download(job["id"]).save("danfe.pdf")

ZPL para PDF

# código ZPL direto nas opções (sem arquivo)...
job = cj.tools.run(
    "zpl-para-pdf",
    options={"zpl": "^XA^FO50,50^A0N,40,40^FDOlá^FS^XZ", "width_mm": 100, "height_mm": 150, "dpmm": "8"},
    wait=True,
)
# ...ou arquivos .zpl/.txt
job = cj.tools.run("zpl-para-pdf", files=["etiquetas.zpl"], wait=True)
cj.jobs.download(job["id"]).save("etiquetas.pdf")

Liste as ferramentas, extensões aceitas e opções (JSON Schema) com cj.tools.list() ou cj.tools.get("comprimir-pdf").

Assinatura eletrônica (envelope)

envelope = cj.esign.envelopes.create(
    "contrato.pdf",                  # ou {"url": "https://…/contrato.pdf"}
    signers=[
        {"name": "Ana Souza", "email": "ana@cliente.com.br", "role": "sign"},
        {"name": "Bruno Lima", "email": "bruno@empresa.com.br", "role": "witness"},
    ],
    fields=[
        {"signer": 0, "type": "signature", "page": 0, "x": 0.1, "y": 0.75, "w": 0.3, "h": 0.07},
        {"signer": 1, "type": "signature", "page": 0, "x": 0.55, "y": 0.75, "w": 0.3, "h": 0.07},
    ],
    name="Contrato de prestação de serviços",
    message="Olá! Segue o contrato para assinatura.",
    webhook_url="https://seusite.com.br/webhooks/convertaja",
)
print(envelope["id"], envelope["status"])  # K7QD-9WZ2-M4HT-R8PA pending

# depois que todos assinarem:
cj.esign.envelopes.download_final(envelope["id"]).save("contrato-assinado.pdf")

Também há list, list_all, get(id, events=True, links=True), send (rascunhos criados com send=False), remind, cancel, delete e download_audit.

Outros recursos

cj.validate("12.ABC.345/01DE-35", type="cnpj")
cj.validate(values=["123.456.789-09", "11.222.333/0001-81"])  # lote: 1 crédito a cada 50
cj.boleto.parse("23793.38128 86000.782713 36950.000632 1 10000000012345")
cj.pix.brcode("contato@empresa.com.br", "Empresa", "SAO PAULO", amount=10.5, format="png")
cj.qrcode("https://convertaja.com.br", format="svg")
cj.barcode("ean13", "789123456789", format="png")

# Tarefas
page = cj.jobs.list(status="done", limit=20)  # page["next_cursor"] → starting_after
for job in cj.jobs.list_all(tool="comprimir-pdf"):
    print(job["id"])
cj.jobs.download_file(job_id, 0).save("saida.pdf")
cj.jobs.delete(job_id)  # apaga os arquivos antes de 1 hora (LGPD)

# Central de Expedição: etiquetas + XMLs das notas
lote = cj.expedicao.lotes.create(["etiquetas.pdf"], ["notas.zip"], intercalar=True, wait=True)
lote = cj.expedicao.lotes.wait(lote["id"])
cj.expedicao.lotes.download(lote["id"]).save("expedicao.zip")

# Webhook de teste (envia um evento assinado para a sua URL)
cj.webhooks.test("https://seusite.com.br/webhooks/convertaja")

Webhooks: verificação da assinatura

Cada aviso chega com X-Convertaja-Signature: t=<unix>,v1=<hex>, onde v1 = HMAC-SHA256(segredo, "<t>.<corpo bruto>"). O segredo (whsec_…) fica em Minha conta → API. Verifique sempre com o corpo bruto e recuse assinaturas com mais de 300 s.

import os
from flask import Flask, request, abort
from convertaja import webhooks, ConvertajaSignatureError

app = Flask(__name__)

@app.post("/webhooks/convertaja")
def convertaja_webhook():
    try:
        event = webhooks.construct_event(
            request.get_data(),  # bytes exatos recebidos
            request.headers.get("X-Convertaja-Signature"),
            os.environ["CONVERTAJA_WEBHOOK_SECRET"],
        )
    except ConvertajaSignatureError:
        abort(400)
    if event["type"] == "job.completed":
        tarefa = event["data"]  # id, outputs, download_url…
    elif event["type"] == "esign.envelope.completed":
        envelope = event["data"]
    return "", 200  # responda 2xx rápido; o Convertajá tenta até 3 vezes

webhooks.verify(raw_body, header, secret, tolerance_sec=300) devolve True/False se você preferir tratar sozinho. Use o cabeçalho X-Convertaja-Delivery para descartar entregas repetidas.

Erros

Toda falha vira ConvertajaError:

from convertaja import ConvertajaError

try:
    cj.tools.run("comprimir-pdf", files=["grande.pdf"])
except ConvertajaError as e:
    print(e.status)      # 402
    print(e.code)        # "insufficient_credits" (estável, para tratar no código)
    print(e.message)     # mensagem em português
    print(e.request_id)  # informe ao suporte
    print(e.details)     # corpo de erro completo (ex.: credits_required)

Códigos do SDK, além dos da API: network_error, timeout, wait_timeout (em jobs.wait), missing_api_key, invalid_signature. Lista completa em /api/docs.

Novas tentativas e idempotência

• Repete até max_retries vezes (padrão 2) em falha de rede, tempo esgotado, 408, 429 (exceto test_limit e reminder_too_soon), 5xx e 409 idempotency_in_progress.
• Em 429 espera o Retry-After da API (até 60 s); nos outros casos usa recuo exponencial (0,5 s, 1 s, 2 s… até 8 s).
• Todo POST leva uma Idempotency-Key (UUID) gerada uma vez por chamada e repetida em todas as tentativas: se a primeira chegou a ser processada, a repetição devolve a mesma resposta, sem novo débito.
• Para repetir com segurança entre processos, passe a sua chave: cj.tools.run("comprimir-pdf", files=[...], idempotency_key=f"pedido-{pedido.id}"). A mesma chave com outro conteúdo em 24 h dá 409 idempotency_key_reused.
• Depois de cada chamada: cj.credits_remaining, cj.last_rate_limit, cj.last_request_id e cj.last_response (status, attempts, idempotency_key, replayed).
• timeout vale por operação de rede (conexão e cada leitura), como no urllib.

Testes

python3 -I -m unittest discover -s tests -v   # servidor HTTP local numa thread; nenhuma chamada sai da máquina

Licença

MIT