Exportar las ventas del día

Es la integración más común y la mejor para empezar: todas las noches, bajar las ventas del día anterior de todos los locales y dejarlas en un archivo que después consume otro proceso.

Lo que vamos a construir maneja las tres cosas que se olvidan siempre: la paginación completa, el vencimiento del token y los reintentos.

Lo que necesitás antes de empezar

• Un usuario de BistroWeb con el rol ApiUser y los locales que vas a consultar.
• El host del país contra el que vas a integrar.
• Los shop_code de esos locales.

Probá primero contra preproducción, que tiene los mismos endpoints que producción pero sin datos reales.

El script completo

"""Exporta a CSV las ventas del día anterior de todos los locales configurados."""

import csv
import os
import time
from datetime import date, datetime, timedelta
from decimal import Decimal

import requests

HOST = os.environ["BISTRO_HOST"]
USERNAME = os.environ["BISTRO_USERNAME"]
PASSWORD = os.environ["BISTRO_PASSWORD"]
SHOP_CODES = [int(code) for code in os.environ["BISTRO_SHOP_CODES"].split(",")]

PAGE_SIZE = 500


def get_token(session: requests.Session) -> str:
    response = session.post(
        f"https://{HOST}/api/v1.0/Token",
        json={"username": USERNAME, "password": PASSWORD},
        timeout=30,
    )
    if response.status_code == 401:
        raise SystemExit("Usuario o contraseña incorrectos.")
    if response.status_code == 403:
        raise SystemExit("Al usuario le falta el rol ApiUser en BistroWeb.")
    response.raise_for_status()
    return response.json()["token"]


def request_with_retry(session, url, params, headers, attempts=5):
    for attempt in range(attempts):
        response = session.get(url, params=params, headers=headers, timeout=60)

        if response.status_code == 429:
            time.sleep(min(60, 2 ** attempt))  # la API no manda Retry-After
            continue
        if response.status_code >= 500:
            time.sleep(2 ** attempt)
            continue

        response.raise_for_status()
        return response

    raise RuntimeError(f"No se pudo obtener {url} después de {attempts} intentos")


def fetch_sales(session, token, shop_codes, window_from, window_to):
    """Recorre TODAS las páginas del período; el cursor marca el final."""
    params = {
        "ShopCode": shop_codes,          # se repite el parámetro por cada local
        "From": window_from.isoformat(timespec="seconds"),
        "To": window_to.isoformat(timespec="seconds"),
        "Limit": PAGE_SIZE,
    }
    headers = {"Authorization": f"Bearer {token}"}

    while True:
        response = request_with_retry(
            session, f"https://{HOST}/api/v1.0/sales", params, headers
        )
        payload = response.json(parse_float=Decimal)

        yield from payload["data"]

        cursor = payload.get("next_page")
        if not cursor:
            return
        params["Page"] = cursor


def main() -> None:
    yesterday = date.today() - timedelta(days=1)
    window_from = datetime.combine(yesterday, datetime.min.time())
    window_to = window_from + timedelta(days=1)   # To es exclusivo

    with requests.Session() as session:
        token = get_token(session)
        sales = list(fetch_sales(session, token, SHOP_CODES, window_from, window_to))

    output = f"sales-{yesterday.isoformat()}.csv"
    with open(output, "w", newline="", encoding="utf-8") as handle:
        writer = csv.writer(handle)
        writer.writerow(["uuid", "shop_code", "created_at", "total", "currency"])
        for sale in sales:
            writer.writerow([
                sale["uuid"],
                sale["shop_code"],
                sale["created_at"],
                sale["total"],
                sale["currency"],
            ])

    print(f"{len(sales)} ventas exportadas a {output}")


if __name__ == "__main__":
    main()

Las decisiones que tomamos y por qué

To exclusivo, y por eso +1 día

La ventana va de las 00:00 de ayer a las 00:00 de hoy. Escribirla como 23:59:59 se come el último segundo del día. Ver Fechas y zonas horarias.

Un solo token para toda la corrida

El token dura dos días; pedirlo una vez por corrida alcanza y sobra. Pedirlo por página sería el error opuesto y el más caro.

Limit=500

Es el máximo. Menos llamadas, menos tiempo y menos exposición a cortes de red.

Todos los locales en una llamada

ShopCode se puede repetir. Pedir los diez locales juntos es mucho más eficiente que diez recorridos separados, y la respuesta ya trae el shop_code en cada fila.

Reintentos distintos para 429 y para 5xx

El 429 es una espera, y como la API no manda Retry-After se estima con backoff exponencial. El 5xx es una falla transitoria: backoff exponencial. Los 4xx restantes no se reintentan porque reintentar no los va a arreglar.

Verificar que salió bien

Tres chequeos que conviene automatizar:

1. Cantidad de filas distinta de cero para los días en que los locales abrieron. Un CSV vacío un lunes es una alarma, no un lunes tranquilo.
2. Todos los locales presentes. Si configuraste diez shop_code y el archivo tiene ocho, faltan dos.
3. Total del día contra el arqueo. Es el chequeo fuerte y está desarrollado en Conciliación contable.

Un paso más

Si en vez de un archivo querés una tabla que se mantenga al día, cambiá la ventana fija por una incremental sobre updated_at con superposición y hacé upsert por uuid. Está explicado en