Developers

Paginação e sincronização

Como percorrer listagens grandes e como buscar só o que mudou desde a última leitura.

Paginação

Toda listagem é paginada. Os parâmetros são sempre os mesmos:

Parâmetro Padrão Limite
page 1 a partir de 1
page_size 50 até 100

A resposta traz, junto com data, o bloco pagination:

"pagination": {
  "page": 1,
  "page_size": 100,
  "total_items": 1234,
  "total_pages": 13,
  "has_next": true
}

Regras:

  • page_size acima de 100 é recusado com 400 invalid_parameter, e não cortado em silêncio.
  • Pedir uma página depois da última também é 400 invalid_parameter. A primeira página de um resultado vazio é 200 com data: [].
  • A ordem é estável (cada recurso diz qual na referência). Percorrer as páginas em sequência não repete nem pula itens, desde que os dados não mudem no meio.

Laço de leitura

import os
import requests

BASE = "https://demonstracao.appbix.com.br/external-api/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['BIX_API_KEY']}"}


def read_all(path, **params):
    page = 1
    while True:
        response = requests.get(f"{BASE}{path}", headers=HEADERS, timeout=30,
                                params={**params, "page": page, "page_size": 100})
        response.raise_for_status()
        body = response.json()
        yield from body["data"]
        if not body["pagination"]["has_next"]:
            break
        page += 1


for row in read_all("/wms/stock-balances", store_code="0012"):
    print(row["product"]["internal_code"], row["quantity_available"])

Sincronização incremental com updated_since

Ler tudo a cada execução funciona, mas fica caro conforme o volume cresce. Os recursos que mudam com o tempo aceitam updated_since: só voltam os itens alterados a partir daquela data e hora.

A receita segura:

  1. Antes de começar a ler, anote a hora atual: inicio.
  2. Leia todas as páginas com updated_since=<ultimo_checkpoint>.
  3. Terminou sem erro? Grave checkpoint = inicio − 5 minutos.

A margem de 5 minutos cobre diferença de relógio e alterações gravadas durante a leitura. Ela faz alguns itens virem duas vezes, então processe de forma idempotente: o mesmo item lido de novo substitui o anterior, sem somar.

Primeira carga

Na primeira execução, leia sem updated_since (carga completa). A partir daí, use o incremental. Uma carga completa de vez em quando (por exemplo, semanal) corrige qualquer divergência acumulada.