Configurando o Loop de Polling de Comandos

O loop de polling de comandos é o coração da integração. A cada ciclo, o ERP pergunta à Maggu: "há algum comando novo para mim?" — e executa o que receber.

Endpoint

GET /v3/comandos/pendentes
Authorization: Bearer <token_da_loja>

Response 200:

[
  {
    "id": 5001,
    "tipo": "ENVIAR_TODOS_USUARIOS",
    "argumentos": {},
    "criadoEm": "2026-05-13T10:00:00Z"
  },
  {
    "id": 5002,
    "tipo": "ENVIAR_VENDAS_INTERVALO",
    "argumentos": {
      "de": "2026-05-01T00:00:00Z",
      "ate": "2026-05-13T23:59:59Z"
    },
    "criadoEm": "2026-05-13T10:05:00Z"
  }
]

Retorna [] se não houver comandos pendentes para a loja.

Loop principal

import time

while True:
    executar()
    time.sleep(1)

def executar():
    comandos = session.get("api/v3/comandos/pendentes").json()
    for cmd in sorted(comandos, key=lambda c: c["criadoEm"]):
        processar(cmd)

Notificando o resultado

Após executar cada comando, informe a Maggu:

# Sucesso
session.post("api/v3/comandos/completou", json={"id": cmd["id"]})

# Falha
session.post("api/v3/comandos/falhou", json={"id": cmd["id"], "motivo": "Descrição do erro"})

Endpoint opcional: POST /v3/comandos/iniciou pode ser chamado logo antes de executar o comando para registrar que o ERP começou o processamento. Útil para diagnóstico de comandos travados.

Prevenindo execução duplicada

O mesmo comando pode aparecer em ciclos consecutivos se a confirmação ainda não chegou à Maggu. Use um dicionário em memória para rastrear o estado:

estado_comandos = {}  # id -> "pendente" | "concluido" | "falhou"
motivos = {}          # id -> str

Antes de executar, verifique:

status = estado_comandos.get(cmd["id"])
if status == "pendente":
    return  # já em andamento
if status == "concluido":
    session.post("api/v3/comandos/completou", json={"id": cmd["id"]})
    return
if status == "falhou":
    session.post("api/v3/comandos/falhou", json={"id": cmd["id"], "motivo": motivos[cmd["id"]]})
    return

estado_comandos[cmd["id"]] = "pendente"

Resumo das respostas

Endpoint	Sucesso	Erro
GET /v3/comandos/pendentes	200 OK + lista	403 não autorizado
POST /v3/comandos/completou	204 No Content	404 comando não encontrado
POST /v3/comandos/falhou	204 No Content	404 comando não encontrado

Implementação completa do processar

Junta o controle de duplicatas, execução e notificação em uma única função:

def processar(cmd):
    status = estado_comandos.get(cmd["id"])
    if status == "pendente":
        return
    if status == "concluido":
        session.post("api/v3/comandos/completou", json={"id": cmd["id"]})
        return
    if status == "falhou":
        session.post("api/v3/comandos/falhou", json={"id": cmd["id"], "motivo": motivos[cmd["id"]]})
        return

    estado_comandos[cmd["id"]] = "pendente"

    # Sinalizar início (opcional — útil para diagnóstico)
    session.post("api/v3/comandos/iniciou", json={"id": cmd["id"], "total": total_registros})

    try:
        processar(cmd)
        estado_comandos[cmd["id"]] = "concluido"
        session.post("api/v3/comandos/completou", json={"id": cmd["id"]})
    except Exception as e:
        motivos[cmd["id"]] = str(e)
        estado_comandos[cmd["id"]] = "falhou"
        session.post("api/v3/comandos/falhou", json={"id": cmd["id"], "motivo": str(e)})