Appearance
Order watcher
Python SDK
aiotic.watch.OrderWatcher turns polling into events. It reads the first page(s) of GET /order_status/list, compares each order's status with the last one it stored, and calls your callback with a Transition — exactly once per change.
python
from aiotic.watch import OrderWatcher, SqliteWatchState, Transition
def on_transition(t: Transition) -> None:
# t.request_id, t.from_status (None on first sight), t.to, t.order (full OrderStatus)
match t.to:
case "PROCESSED": queue.put(("book", t.request_id))
case "ATTENTION": tickets.create(t.order)
case "SENT": erp.mark_exported(t.order.erp_ref)
watcher = OrderWatcher(client, on_transition=on_transition, state=SqliteWatchState("watch.db"),
interval=20, page_size=200, pages=2, only=None, seed_silently=True)
watcher.run_forever() # or watcher.run_once() from your own scheduler| Option | Meaning |
|---|---|
state | where last statuses live: InMemoryWatchState (lost on restart → replays history) or SqliteWatchState(path) |
interval | seconds between polls |
pages × page_size | how deep to look each poll; deeper than your busiest 2 hours is enough |
only | restrict callbacks to certain target statuses |
seed_silently | on the very first run, learn current statuses without firing (avoid a storm on a fresh state file) |
Callback exceptions are logged and do not stop the loop; the transition is still recorded (it will not fire again), so make callbacks idempotent or push to a queue.
aiotic orders watch runs the watcher and prints one JSON line per transition:
json
{"request_id": "7f0c…", "from": "PROCESSING", "to": "PROCESSED", "order_number": "PO-4711", "erp_ref": null, "timestamp": "2026-03-14T10:31:02"}Cost
Two small requests every 20 s ≈ 8 600 requests/day, independent of order volume. Fine for any tenant.
Limits
Orders that scroll past the pages you look at before a transition happens are missed; raise pages or run a nightly full walk (client.orders.iter_all()) for reconciliation. Signed status webhooks would remove this — see the proposal.