Python

httpx é a biblioteca recomendada: tem manutenção ativa, timeout explícito e a mesma interface para código síncrono e assíncrono. Esta página monta um cliente pequeno sobre ela e usa esse cliente em todos os exemplos seguintes.

Configuração

pip install httpx

export LIGGA_API_KEY='ligga_live_…'
export LIGGA_BASE='https://api.ligga.app/functions/v1/api'

O módulo abaixo concentra autenticação, montagem de URL e tradução do corpo de erro em exceção.

ligga.py

import os
import uuid

import httpx

API_KEY = os.environ["LIGGA_API_KEY"]
BASE = os.environ.get("LIGGA_BASE", "https://api.ligga.app/functions/v1/api")


class LiggaError(Exception):
    def __init__(self, status: int, problem: dict | None):
        self.status = status
        self.problem = problem or {}
        super().__init__(f"HTTP {status}: {self.problem.get('title', 'erro sem corpo')}")


client = httpx.Client(
    base_url=BASE,
    headers={"Authorization": f"Bearer {API_KEY}"},
    timeout=30.0,
)


def api(
    path: str,
    method: str = "GET",
    params: dict | None = None,
    json=None,
    idempotency_key: str | None = None,
):
    headers = {}
    if idempotency_key:
        headers["Idempotency-Key"] = idempotency_key

    res = client.request(method, path, params=params, json=json, headers=headers)

    if res.status_code == 204:
        return None

    data = res.json() if res.content else None
    if res.is_error:
        raise LiggaError(res.status_code, data)
    return data


def nova_chave_idempotente() -> str:
    return str(uuid.uuid4())

Nota

api() recebe o caminho completo, com o prefixo de versão: api("/v1/me"). O caminho é sempre idêntico ao da tabela de endpoints do recurso.

Primeira requisição

GET /v1/me descreve a chave que fez a chamada e confirma que a configuração está correta.

me.py

from ligga import api

me = api("/v1/me")
print(me["data"]["team_name"], me["data"]["scopes"])

{
  "data": {
    "team_id": "bca09e2f-41ab-4417-b84b-f2ff50f991ca",
    "team_name": "Pet Shop Aurora",
    "scopes": ["*:read", "sales:write"],
    "plan_code": "pro"
  },
  "meta": { "request_id": "req_01JZXQ8V2K5M7N9P0R1S2T3U4V" }
}

Listar com paginação

Uma listagem devolve os itens em data e o estado da paginação em pagination. O cursor da próxima página está em pagination.next_cursor e vem None quando acabou.

clientes.py

from collections.abc import Iterator

from ligga import api


def listar_clientes(**filtros) -> Iterator[dict]:
    cursor = None

    while True:
        params = {"limit": 100, **filtros}
        if cursor:
            params["cursor"] = cursor

        pagina = api("/v1/customers", params=params)
        yield from pagina["data"]

        cursor = pagina["pagination"]["next_cursor"]
        if not cursor:
            break


for cliente in listar_clientes(**{"filter[is_active][eq]": "true"}):
    print(cliente["id"], cliente["full_name"])

Cuidado

O cursor está em pagination.next_cursor, não em meta. Um laço que lê meta["next_cursor"] levanta KeyError ou para na primeira página, dependendo de como você acessa o dicionário.

O limit vai de 1 a 100, com padrão 20. pagination.has_more responde a mesma pergunta que o cursor nulo, e serve quando você quer só saber se existe mais alguma coisa.

Escrita idempotente

Gere a chave antes da primeira tentativa e reaproveite o mesmo valor em todas as repetições. Assim, uma requisição perdida no caminho não vira dois registros.

criar_venda.py

import time

from ligga import LiggaError, api, nova_chave_idempotente

RETENTAVEIS = {408, 429, 500, 502, 503, 504}


def criar_venda(payload: dict, tentativas: int = 3) -> dict:
    chave = nova_chave_idempotente()

    for tentativa in range(1, tentativas + 1):
        try:
            return api("/v1/sales", method="POST", json=payload, idempotency_key=chave)
        except LiggaError as erro:
            if erro.status not in RETENTAVEIS or tentativa == tentativas:
                raise
            time.sleep(0.25 * 2 ** (tentativa - 1))


venda = criar_venda({
    "customer_id": "7c4b0a19-2f36-4d1e-9b58-1f0a6c3e5d24",
    "transaction_type": "sale",
    "sale_date": "2026-05-18",
    "items": [
        {
            "catalog_type": "product",
            "catalog_item_id": "3d61f8ba-9c04-4a77-8e12-5b7a0d9f2c68",
            "quantity": 2,
            "unit_price": 29.90,
        }
    ],
})

print(venda["data"]["id"])

Reusar a mesma chave com um corpo diferente devolve 409 idempotency_conflict: é sinal de que a chave vazou para outra operação.

Tratamento de erros

Respostas de erro vêm como application/problem+json. O cliente acima já as converte em LiggaError, com o corpo inteiro em erro.problem.

tratar_erro.py

from ligga import LiggaError, api

try:
    api("/v1/customers", method="POST", json={})
except LiggaError as erro:
    print(erro.problem["status"], erro.problem["type"])
    print(erro.problem.get("detail", erro.problem["title"]))
    print("request_id:", erro.problem["request_id"])

    for campo in erro.problem.get("errors", []):
        print(" -", campo["field"], campo["code"], campo["message"])

{
  "type": "https://api.ligga.app/errors/validation",
  "title": "Validation failed",
  "status": 422,
  "detail": "full_name is required",
  "request_id": "req_01JZXQ8V2K5M7N9P0R1S2T3U4V"
}

type, title, status e request_id estão sempre presentes; detail, instance e errors aparecem quando fazem sentido. Guarde o request_id no seu log: é o que o time da Ligga pede para investigar uma chamada.

HTTP	type	O que fazer
401	invalid_token	Confira LIGGA_API_KEY e o cabeçalho Authorization
403	insufficient_scope	A chave não carrega o escopo do recurso
403	module_not_enabled	O módulo do recurso não está no contrato
409	idempotency_conflict	A mesma Idempotency-Key foi reusada com outro corpo
422	validation	Campo obrigatório ausente ou com tipo errado
429	rate-limit-exceeded	Espere o tempo do cabeçalho Retry-After

Extras

Versão assíncrona

Para buscar vários recursos em paralelo, use httpx.AsyncClient com a mesma configuração de cabeçalhos:

ligga_async.py

import asyncio

import httpx

from ligga import API_KEY, BASE


async def buscar_em_paralelo(caminhos: list[str]) -> list[dict]:
    async with httpx.AsyncClient(
        base_url=BASE,
        headers={"Authorization": f"Bearer {API_KEY}"},
        timeout=30.0,
    ) as ac:
        respostas = await asyncio.gather(*(ac.get(caminho) for caminho in caminhos))

    for res in respostas:
        res.raise_for_status()
    return [res.json() for res in respostas]


paginas = asyncio.run(buscar_em_paralelo(["/v1/me", "/v1/customers?limit=1"]))
print(paginas[0]["data"]["team_name"])

Respeite os limites do seu plano ao paralelizar: rajadas acima do teto voltam como 429 rate-limit-exceeded.

Veja também

• Início rápido — a primeira chamada, passo a passo
• Paginação, filtros e ordenação — operadores e sort
• Idempotência — o ciclo de vida da Idempotency-Key
• Limites de requisições — tetos por plano e cabeçalhos