Python

Cliente mínimo viável usando apenas a biblioteca requests. Cobre cache de token, retry em 401 e backoff em 5xx.

Dependência

pip install requests

Configuração via variáveis de ambiente

export BANQER_TENANT=sua-empresa
export BANQER_CLIENT_ID=...
export BANQER_CLIENT_SECRET=...

Cliente completo

"""
banqer_client.py
Cliente minimal para a API Publica Banqer CRM.
"""
import os
import time
import random
import requests

TENANT = os.environ["BANQER_TENANT"]
CLIENT_ID = os.environ["BANQER_CLIENT_ID"]
CLIENT_SECRET = os.environ["BANQER_CLIENT_SECRET"]

API_BASE = f"https://api-{TENANT}.banqer.com.br"
AUTH_URL = f"https://auth.banqer.com.br/realms/{TENANT}/protocol/openid-connect/token"

_session = requests.Session()
_token_cache = {"access_token": None, "exp": 0.0}


def _fetch_token():
    r = _session.post(
        AUTH_URL,
        data={
            "grant_type": "client_credentials",
            "client_id": CLIENT_ID,
            "client_secret": CLIENT_SECRET,
        },
        timeout=10,
    )
    r.raise_for_status()
    body = r.json()
    # Renova em 80% do TTL para evitar expirar em transito.
    _token_cache["access_token"] = body["access_token"]
    _token_cache["exp"] = time.time() + body["expires_in"] * 0.8
    return _token_cache["access_token"]


def _get_token():
    if _token_cache["access_token"] and time.time() < _token_cache["exp"]:
        return _token_cache["access_token"]
    return _fetch_token()


def call(method, path, *, params=None, retry_on_5xx=True, max_attempts=3):
    """Faz uma chamada autenticada na API. Renova token em 401 uma vez."""
    url = f"{API_BASE}{path}"
    attempt = 0
    refreshed = False
    while True:
        attempt += 1
        token = _get_token()
        r = _session.request(
            method, url,
            headers={"Authorization": f"Bearer {token}"},
            params=params,
            timeout=30,
        )

        if r.status_code == 401 and not refreshed:
            # Token expirou em transito. Renova e tenta uma vez.
            _token_cache["access_token"] = None
            refreshed = True
            continue

        if 500 <= r.status_code < 600 and retry_on_5xx and attempt < max_attempts:
            # Backoff exponencial com jitter, teto de 30s.
            wait = min(2 ** attempt + random.random(), 30)
            time.sleep(wait)
            continue

        if r.status_code == 429 and attempt < max_attempts:
            wait = int(r.headers.get("Retry-After", "5"))
            time.sleep(wait)
            continue

        if not r.ok:
            # Erros 4xx (exceto 401 ja tratado) sao bugs no cliente. Nao retry.
            raise BanqerError.from_response(r)
        return r.json()


class BanqerError(Exception):
    @classmethod
    def from_response(cls, r):
        try:
            body = r.json()
            reason = (body.get("details", [{}])[0]
                          .get("reason", "UNKNOWN"))
            msg = body.get("message", r.text[:200])
        except Exception:
            reason, msg = "UNKNOWN", r.text[:200]
        e = cls(f"[{r.status_code} {reason}] {msg}")
        e.status_code = r.status_code
        e.reason = reason
        return e


def paginar(path, *, params=None, page_size=200, key=None):
    """Itera por uma colecao paginada. `key` e o nome do array no JSON
    (ex: 'fichas', 'pagamentos'); se None, infere do primeiro campo."""
    p = dict(params or {})
    p["pageSize"] = page_size
    token = ""
    while True:
        p["pageToken"] = token
        body = call("GET", path, params=p)
        if key is None:
            key = next(k for k in body if isinstance(body[k], list))
        for item in body[key]:
            yield item
        token = body.get("nextPageToken") or ""
        if not token:
            return

Uso

from banqer_client import call, paginar

# Chamada simples
credores = call("GET", "/v1/credores")
print(credores)

# Detalhe de uma ficha
ficha = call("GET", "/v1/fichas/8aa4b425-5fed-4cea-b970-044b36ea9d73")
print(ficha["saldoDevedor"])

# Iterar todas as fichas em atraso > 90 dias
for ficha in paginar("/v1/fichas",
                     params={"atrasoMin": 90},
                     key="fichas"):
    print(ficha["idFichaCobrancaExibicao"], ficha["saldoDevedor"])

Alternativa async com httpx

Se sua aplicação for async (FastAPI, aiohttp), troque requests por httpx. O fluxo é idêntico, basta usar await client.post(...) e await client.request(...).

pip install httpx

Bibliotecas OAuth de alto nível

Para integrações maiores, considere Authlib que fornece OAuth2Session com cache de token integrado e middleware de renovação. O cliente acima é suficiente para a maioria dos casos e tem zero dependências OAuth.

Erros comuns

Sintoma	Causa provável
401 invalid_client no token endpoint	client_id ou client_secret errados
403 MISSING_SCOPE	Sua credencial não tem o escopo deste endpoint. Veja Escopos.
404 NOT_FOUND em recurso que você sabe que existe	O produto da ficha não está na lista de produtos da sua credencial
ConnectionError: SSL	Você está em ambiente corporativo com proxy/MITM. Configure REQUESTS_CA_BUNDLE