API Conta Azul muda em 10/11: ajuste sua carga no Power BI

1 Quebre o período em janelas de 1 ano

A regra prática é a do próprio exemplo da Conta Azul: uma chamada por ano-calendário, de 01/01 a 31/12. Para buscar de 2024 a 2026:

?data_vencimento_de=2024-01-01&data_vencimento_ate=2024-12-31
?data_vencimento_de=2025-01-01&data_vencimento_ate=2025-12-31
?data_vencimento_de=2026-01-01&data_vencimento_ate=2026-12-31

Por que ano-calendário e não "hoje menos 365 dias"? Porque as janelas ficam fixas, fáceis de conferir e sem sobreposição. E 2024, que é bissexto, está no exemplo oficial.

2 Use um tamanho de página válido e percorra as páginas

tamanho_pagina passa a aceitar só 10, 20, 50, 100, 200, 500 ou 1000. Se você usa outro valor (300, 5000...), troque pelo permitido mais próximo, até 1000. Para carga de BI, use 1000: menos chamadas por janela.

A resposta traz itens_totais, itens e totais. Com itens_totais você sabe quantas páginas existem e para de chamar na hora certa:

páginas = arredonda_para_cima(itens_totais / tamanho_pagina)

3 No Power BI: consulta M com janelas e paginação

Abra o Power Query (Página Inicial > Transformar dados), crie uma Consulta Nula (Nova Fonte > Consulta Nula), abra o Editor Avançado e cole:

let
    // Ajuste o período e o token (veja o artigo de token da Conta Azul)
    Token = "SEU_TOKEN",
    Inicio = #date(2024, 1, 1),
    Fim = #date(Date.Year(DateTime.Date(DateTime.LocalNow())) + 1, 12, 31),
    TamanhoPagina = 1000,

    // 1. Uma janela por ano-calendário (máximo de 1 ano por busca)
    Janelas = List.Transform(
        {Date.Year(Inicio) .. Date.Year(Fim)},
        each [
            de  = Date.ToText(List.Max({#date(_, 1, 1), Inicio}), "yyyy-MM-dd"),
            ate = Date.ToText(List.Min({#date(_, 12, 31), Fim}), "yyyy-MM-dd")
        ]
    ),

    // 2. Busca uma página de uma janela
    BuscarPagina = (janela as record, pagina as number) as record =>
        Json.Document(
            Web.Contents(
                "https://api-v2.contaazul.com",
                [
                    RelativePath = "v1/financeiro/eventos-financeiros/contas-a-receber/buscar",
                    Query = [
                        pagina = Text.From(pagina),
                        tamanho_pagina = Text.From(TamanhoPagina),
                        data_vencimento_de = janela[de],
                        data_vencimento_ate = janela[ate]
                    ],
                    Headers = [Authorization = "Bearer " & Token]
                ]
            )
        ),

    // 3. Todas as páginas de uma janela, usando itens_totais
    BuscarJanela = (janela as record) as list =>
        let
            Primeira = BuscarPagina(janela, 1),
            TotalPaginas = Number.RoundUp(Primeira[itens_totais] / TamanhoPagina),
            Demais = List.Transform({2 .. TotalPaginas}, each BuscarPagina(janela, _)[itens])
        in
            List.Combine({Primeira[itens]} & Demais),

    // 4. Junta todas as janelas numa tabela
    Itens = List.Combine(List.Transform(Janelas, BuscarJanela)),
    Tabela = Table.FromRecords(Itens, null, MissingField.UseNull)
in
    Tabela

Conforme a especificação oficial da API; não executado aqui, porque exige o token da sua conta.

O que vale entender:

• URL base fixa + RelativePath + Query: é isso que permite a atualização agendada no serviço do Power BI. Se você montar a URL inteira concatenando texto, o serviço reclama de fonte dinâmica e não atualiza.
• {2 .. TotalPaginas}: quando a janela tem uma página só (ou nenhuma), essa lista fica vazia, e a consulta não faz chamada extra.
• Para contas a pagar, troque o RelativePath para v1/financeiro/eventos-financeiros/contas-a-pagar/buscar. O resto é igual.
• Cada item vem com id, descricao, data_vencimento, status, total, pago, nao_pago, data_competencia, data_alteracao, categorias, centros_custo e cliente (ou fornecedor, em contas a pagar). Expanda as colunas que o seu relatório usa.

4 Em Python: carga em janelas com paginação

Para quem carrega um banco próprio (ou usa Pentaho/n8n e quer a lógica pronta), este script faz o mesmo e respeita o limite de requisições:

import datetime as dt
import time

import requests

BASE = "https://api-v2.contaazul.com"
TOKEN = "SEU_TOKEN"
CAMINHO = "/v1/financeiro/eventos-financeiros/contas-a-receber/buscar"
TAMANHOS_VALIDOS = (10, 20, 50, 100, 200, 500, 1000)


def janelas_anuais(inicio, fim):
    """Quebra o período em anos-calendário: 01/01 a 31/12, cortando nas pontas."""
    for ano in range(inicio.year, fim.year + 1):
        yield max(dt.date(ano, 1, 1), inicio), min(dt.date(ano, 12, 31), fim)


def buscar_janela(sessao, de, ate, tamanho_pagina=1000, filtros_extras=None):
    """Percorre todas as páginas de uma janela de vencimento."""
    if tamanho_pagina not in TAMANHOS_VALIDOS:
        raise ValueError(f"tamanho_pagina precisa ser um de {TAMANHOS_VALIDOS}")
    pagina = 1
    while True:
        params = {
            "pagina": pagina,
            "tamanho_pagina": tamanho_pagina,
            "data_vencimento_de": de.isoformat(),
            "data_vencimento_ate": ate.isoformat(),
            **(filtros_extras or {}),
        }
        resp = sessao.get(BASE + CAMINHO, params=params, timeout=60)
        if resp.status_code == 429:  # passou do limite de requisições: espera e repete
            time.sleep(int(resp.headers.get("Retry-After", 5)))
            continue
        if resp.status_code == 400:  # a API diz o que ajustar no corpo do erro
            raise RuntimeError(f"400 na janela {de} a {ate}: {resp.text}")
        resp.raise_for_status()

        corpo = resp.json()
        itens = corpo.get("itens", [])
        yield from itens
        if not itens or pagina * tamanho_pagina >= corpo.get("itens_totais", 0):
            break
        pagina += 1
        time.sleep(0.15)  # ~6 chamadas/s, abaixo do limite de 10/s


def carregar(inicio, fim, filtros_extras=None):
    sessao = requests.Session()
    sessao.headers["Authorization"] = f"Bearer {TOKEN}"
    contas = []
    for de, ate in janelas_anuais(inicio, fim):
        contas.extend(buscar_janela(sessao, de, ate, filtros_extras=filtros_extras))
    return contas


if __name__ == "__main__":
    hoje = dt.date.today()
    contas = carregar(dt.date(2024, 1, 1), dt.date(hoje.year + 1, 12, 31))
    print(f"{len(contas)} contas a receber carregadas")

Este script foi executado contra o mock server oficial da documentação da Conta Azul. A resposta veio no formato documentado (itens_totais, itens, totais), e a paginação parou no lugar certo. Contra a API real, troque SEU_TOKEN pelo seu token.

Por que o time.sleep(0.15)? O limite da API é de 600 chamadas por minuto e até 10 por segundo por conta conectada. Passando disso, vem 429 Too Many Requests. Paginar 1000 em 1000 com uma pausa curta mantém você longe do limite.

5 Sincronize só o que mudou (carga incremental)

Depois da carga inicial, não baixe tudo de novo todo dia. Os filtros data_alteracao_de e data_alteracao_ate trazem criações, edições, baixas, estornos e cancelamentos. Como o vencimento continua obrigatório, combine a data de alteração com as mesmas janelas de 1 ano:

agora = dt.datetime.now()
ontem = agora - dt.timedelta(days=1)
hoje = dt.date.today()

mudancas = carregar(
    dt.date(2024, 1, 1),
    dt.date(hoje.year + 1, 12, 31),
    filtros_extras={
        "data_alteracao_de": ontem.strftime("%Y-%m-%dT%H:%M:%S"),
        "data_alteracao_ate": agora.strftime("%Y-%m-%dT%H:%M:%S"),
    },
)
# atualize no seu banco cada conta de `mudancas` pelo campo `id`

A documentação define a data de alteração em ISO 8601, no horário de São Paulo (GMT-3), e o exemplo oficial usa o formato 2025-10-20T07:00:00, sem fuso. É o formato do código acima. A janela de alteração também é limitada a 365 dias.

Este trecho segue a documentação; não foi executado contra a API real. O mock server da documentação é mais rígido que o exemplo oficial e só aceita a data com fuso (2025-10-20T07:00:00-03:00). Se a API real recusar o formato sem fuso, acrescente -03:00.

ids_api = {conta["id"] for conta in carregar(dt.date(2024, 1, 1), dt.date(hoje.year + 1, 12, 31))}
excluidos = ids_no_seu_banco - ids_api  # ids_no_seu_banco: conjunto de ids que você já tem

No Power BI em modo de importação, cada atualização recarrega a tabela inteira, então exclusão não é problema. A carga incremental e a varredura fazem diferença para quem guarda os dados num banco próprio.