A wine-club retention dashboard in three files of standard-library Python. Read-only, no dependencies, 2083 lines including its own documentation.
See what it produces. A real report, produced by the real pipeline against a sandbox winery. Invented members, invented wines.
These are the files, not a description of them. This page reads examples/club_pulse/ out of the SipsPOS repository at the moment you load it, and those same files are run against the real API on every build — four paginated reads through the real endpoints, the real serializers and the real cursor paginator, against a sandbox winery. An example that drifts from the API it documents is worse than none, so this one cannot: if it broke, the build would have.
Everything here is standard-library Python. No pip install, nothing to vendor. SipsPOS itself depends on requests, and the client below deliberately does not — a file that can only be copied along with a dependency list is not really copyable.
Taken from the example's own path constants and from the blueprint that enforces the scopes — not from a list typed onto this page.
| Request | Scope | |
|---|---|---|
GET /api/v1/club/members |
customers:read |
|
GET /api/v1/customers |
customers:read |
|
GET /api/v1/club/tiers |
customers:read |
|
GET /api/v1/sales |
sales:read |
|
GET /api/v1/orders |
orders:read |
optional |
Ask the winery for a test key with customers:read and sales:read. Their admin issues one in Admin → API keys: Build my sandbox, then Create test key. A sk_test_ key reaches a sandbox twin of their winery — invented wines, customers and club — and cannot see or touch the real one, so nothing you do while learning is visible to their customers.
orders:read is optional, and worth understanding before you ask for it. orders:read requires the online store, which is an Estate feature — a winery on a smaller plan cannot issue a key carrying it, however much they would like to. The example asks anyway, accepts the refusal, and says on its report that the shop was not read. An integration that works across plans has to do something like this; treating a plan boundary as a crash is the most common way a partner's client works at one winery and not the next.
Then:
Sipsexport SIPSPOS_BASE_URL=https://their-winery.sipspos.com
export SIPSPOS_API_KEY=sk_test_…
python pulse.pyIt writes one self-contained HTML file and prints how many API requests it took. Every response says which winery answered, in X-Sipspos-Environment, and when that is test the report opens with a banner saying the numbers are invented — so a sandbox report can never be mistaken for a real one.
Every file is also served as plain text, so you can fetch the whole thing with curl instead of a copy button.
The client. THIS is the file to copy — bearer auth, cursor pagination, the error envelope, the environment header and 429 backoff, with nothing about wine clubs in it.
"""A SipsPOS public-API client in one file — THE PART YOU COPY.
WHAT THIS IS. Everything a partner needs to talk to the SipsPOS v1 read
surface: bearer auth, cursor pagination, the JSON error envelope turned into
one exception, the `X-Sipspos-Environment` header, and 429 backoff. It knows
nothing about wine clubs. The dashboard that lives beside it (`pulse.py`)
imports this module and could be deleted without touching a line of it — that
split is the point. Copy this file into your own project, delete the rest of
the example, and you have a working client.
WHO READS IT, AND WHAT THAT BUYS. This file is written to be read end to end
by somebody about to paste it into their own repository. So it is longer in
comments than in code, and it never reaches for an abstraction it would then
have to explain. Four decisions a reader will otherwise stop and question:
1. NO `requests`, EVEN THOUGH SIPSPOS ITSELF DEPENDS ON IT. `requests` is in
this repository's `requirements.txt` and would make `_request` about eight
lines shorter. It is deliberately not used here, because the value of a
reference client is that it can be copied — and a copyable file cannot
arrive with a dependency list, a version pin, or an argument about which
HTTP library your team standardised on. `urllib` ships with Python. If you
would rather use `requests`, `httpx` or your own pooled session, replace
`urllib_transport` (below) and change nothing else.
2. THE `transport` SEAM. `Client` takes an optional callable
`(method, url, headers, body) -> (status, headers, bytes)`. That single
argument is how this example is tested in CI: the test drives the whole
dashboard through Flask's test client, in-process, against a seeded
sandbox tenant, with no port open and no live server. It is exactly the
seam you need to test YOUR integration too — record real responses once,
replay them from a function — which is why it is in the copyable half
rather than hidden in a test file.
3. IT ONLY EVER READS. Every request this module can make is a GET. There is
no code path here that changes anything at a winery, and that is a
property worth keeping when you copy it: the first client you write
against a new API should not be one whose bugs can damage somebody's
customer records. Add the writing half once the reading half is boring.
4. THE CURSOR IS FOLLOWED, NOT AN OFFSET INCREMENTED, and the filters go
with it. See `Client.paginate` — this is the single most common way a
partner integration passes in testing and loses rows at a real winery.
REQUIRED READING FOR THE PERSON COPYING THIS: `ApiError.request_id`. Every
SipsPOS API response carries an `X-Request-Id`, and every error envelope
repeats it in the body. Log it. When something goes wrong at 2am the
difference between "your API returned a 500" and "your API returned a 500,
request_id=req_9f3c…" is the difference between a support thread that takes a
week and one that takes ten minutes: that id is what SipsPOS greps its own
logs for. It is the only thing in the envelope you cannot reconstruct.
"""
import json
import time
import urllib.error
import urllib.parse
import urllib.request
#: Page size defaults, matching the API's own (`services/api_pagination.py`).
#: A limit outside 1..MAX_LIMIT is an ERROR at SipsPOS, not a silent clamp —
#: asking for 500 and being handed 200 without being told would make you
#: believe you had seen the whole list. `paginate` checks the range here so
#: you find out at the call site instead of one HTTP round trip later.
DEFAULT_LIMIT = 50
MAX_LIMIT = 200
#: Seconds before a stalled connection gives up. `urlopen` with no timeout
#: waits forever on a socket that was accepted and then never answered, which
#: turns a blip into a hung cron job that nobody notices until the report
#: stops arriving. Every network call in a copyable client needs a number
#: here; this one is deliberately generous rather than tuned.
DEFAULT_TIMEOUT_SECONDS = 30
#: How long to wait after a 429 that arrives with no usable `Retry-After`.
DEFAULT_RETRY_AFTER_SECONDS = 1.0
#: Ceiling on an honoured `Retry-After`. The server is trusted about WHETHER
#: to wait and roughly how long, but not to park a script for an hour: a
#: misconfigured proxy answering `Retry-After: 86400` would otherwise hang
#: this process for a day with no output. Waiting less than asked risks
#: spending another retry on another 429, which is a bounded, visible cost —
#: an unbounded sleep is neither.
MAX_RETRY_AFTER_SECONDS = 60.0
#: Sent on every request. Identify yourself when you copy this: SipsPOS
#: support can see this string, and "which of our partners is hammering
#: /v1/orders" is a question somebody will eventually need answered.
USER_AGENT = "sipspos-club-pulse-example/1.0 (+https://sipspos.com/developers)"
#: `.type` for a failure that never reached SipsPOS at all — DNS, a refused
#: connection, a TLS failure, a timeout. Minted CLIENT-SIDE, and deliberately
#: not one of the API's own types, so code branching on `.type` can tell "the
#: API refused us" from "we never got there". The API will never send it.
CONNECTION_ERROR_TYPE = "connection_error"
#: What `.type` to use when a non-2xx response does NOT carry the API's error
#: envelope — a proxy's 502 HTML page, a load balancer's plain-text 503, a
#: captive portal. Mirrors the API's own status mapping
#: (`routes/api/errors.CODE_TO_TYPE`) so the type you branch on is the same
#: string whether the envelope survived or not.
_TYPE_BY_STATUS = {
400: "invalid_request",
401: "authentication_failed",
402: "subscription_inactive",
403: "permission_denied",
404: "not_found",
405: "method_not_allowed",
409: "conflict",
429: "rate_limited",
}
#: How much of an unparseable body to quote back. Enough to recognise an
#: nginx error page or a login redirect; short enough not to dump a megabyte
#: of HTML into a log line.
_BODY_SNIPPET_CHARS = 200
class ApiError(Exception):
"""Any refusal from the SipsPOS API, and any failure to reach it.
ONE EXCEPTION TYPE, NOT A HIERARCHY. The temptation is a subclass per
error type — `PermissionDenied`, `RateLimited`, and so on — and it is
the wrong shape for a copied file. `type` is a string in the API's
contract; adding a type there is additive and must not require you to
upgrade this file before you can catch the new one. So callers branch on
`.type`, which is data, rather than on a class, which is code:
try:
members = list(client.paginate("/api/v1/club/members"))
except ApiError as e:
if e.type == "permission_denied":
... # the key is missing a scope; tell somebody
elif e.type == "rate_limited":
... # already retried; back off harder
raise
ATTRIBUTES:
type the API's stable machine-readable string — one of
invalid_request, authentication_failed,
subscription_inactive, permission_denied,
feature_not_enabled, not_found, method_not_allowed,
conflict, idempotency_mismatch, rate_limited, internal —
or CONNECTION_ERROR_TYPE when we never reached the API.
message a sentence written for a human. Safe to show an operator;
never parse it, it is not part of the contract.
request_id `req_…`, the one thing SipsPOS support needs. May be None
only when nothing that could carry it came back.
status the HTTP status, or None for a connection failure.
"""
def __init__(self, type_, message, *, request_id=None, status=None):
super().__init__(message)
self.type = type_
self.message = message
self.request_id = request_id
self.status = status
def __str__(self):
"""Written to be the whole of a support ticket.
Whatever a partner's error handling does with this exception, the
odds are high that somewhere it becomes `str(e)` in a log line or on
a terminal — so that one string has to carry everything needed to act
on it, without the reader knowing this class exists. Hence type
(what to do), message (why), status and request id (what to quote).
"""
detail = []
if self.status is not None:
detail.append(f"HTTP {self.status}")
if self.request_id:
detail.append(f"request_id={self.request_id}")
head = f"{self.type}: {self.message}"
return f"{head} ({', '.join(detail)})" if detail else head
def urllib_transport(method, url, headers, body):
"""The default transport: one request, one `(status, headers, bytes)`.
A NON-2XX IS A RETURN VALUE HERE, NOT AN EXCEPTION, and that is the whole
reason this wrapper exists. `urlopen` raises `HTTPError` for a 4xx or
5xx — but `HTTPError` IS the response (it has `.status`, `.headers` and
`.read()`), and the API puts the part you actually need, the error
envelope, in that body. A client that lets the exception escape throws
the envelope away and reports "HTTP Error 403: FORBIDDEN", which names
neither the missing scope nor the request id. So both paths are flattened
to the same triple, and `Client` decides what is an error.
A transport you write yourself must do the same: return the failing
response, do not raise on it. It must only raise when there is genuinely
no response — which is the second half of this function.
"""
request = urllib.request.Request(url, data=body, method=method)
for name, value in headers.items():
request.add_header(name, value)
try:
with urllib.request.urlopen(
request, timeout=DEFAULT_TIMEOUT_SECONDS) as response:
return (response.status, dict(response.headers.items()),
response.read())
except urllib.error.HTTPError as e:
return e.code, dict(e.headers.items()), e.read()
except urllib.error.URLError as e:
# Nothing answered. Raising the raw `URLError` here would hand a
# partner's `except ApiError` block nothing to catch and print a
# traceback at whoever is running the script, so it is translated
# into the same exception every other failure uses — with a type
# that says plainly which side of the wire the problem is on.
raise ApiError(
CONNECTION_ERROR_TYPE,
f"Couldn't reach {urllib.parse.urlsplit(url).netloc}: {e.reason}. "
f"Check the base URL, your network and any proxy — this request "
f"never got as far as SipsPOS.") from e
except TimeoutError as e:
raise ApiError(
CONNECTION_ERROR_TYPE,
f"Timed out after {DEFAULT_TIMEOUT_SECONDS}s waiting for "
f"{urllib.parse.urlsplit(url).netloc}. The request may or may not "
f"have been processed; every endpoint this client calls is a read, "
f"so it is safe to try again.") from e
class Client:
"""A read-only SipsPOS API client bound to one winery's host and key.
client = Client("https://lucky.sipspos.com", os.environ["SIPSPOS_API_KEY"])
tiers = client.get("/api/v1/club/tiers")
for member in client.paginate("/api/v1/club/members"):
...
`base_url` is the winery's own host — SipsPOS is addressed per winery,
not through one shared api.* domain — and `token` is an `sk_test_…` or
`sk_live_…` key. Prefer a test key: it reaches a sandbox twin of the
winery, so nothing you do while learning the API touches real customers.
THIS OBJECT IS NOT THREAD-SAFE and is not trying to be. `environment` and
`request_count` are last-response state; sharing one instance across
threads makes both meaningless. Instances are cheap — make one per
worker. (It holds no connection pool, which is the usual reason to share
one; if you swap in a pooled transport, pool inside the transport.)
"""
def __init__(self, base_url, token, *, transport=None, max_retries=3):
#: Trailing slash stripped so `base_url + path` is unambiguous
#: whether the caller passed "https://x.sipspos.com" or ".../".
self.base_url = (base_url or "").rstrip("/")
#: Private, and named with a leading underscore for one specific
#: reason: this example renders an HTML report that a winery emails
#: to people, and a public `self.token` is one careless `vars(client)`
#: or debug dump away from being in it. See `__repr__`.
self._token = token
self._transport = transport or urllib_transport
self.max_retries = max_retries
#: "test", "live", or None — from the last response's
#: `X-Sipspos-Environment` header. THE SINGLE MOST USEFUL LINE YOUR
#: LOGS CAN CARRY. The worst mistake available to a partner is running
#: a test suite you believe is hitting the sandbox against the real
#: winery, and the only other signal you have is the prefix of a
#: token in a `.env` file you cannot read back. SipsPOS derives this
#: header from the tenant it actually served, not from your token, so
#: it is an answer rather than an echo. None until the first
#: authenticated response (an unauthenticated request ran against no
#: tenant, so there is no environment to report).
self.environment = None
#: HTTP requests made by this client, retries included. Exposed
#: because a cursor loop that never loops is invisible otherwise:
#: assert this is greater than 1 in a test that seeds more rows than
#: one page holds, and you will find out in CI rather than at a
#: winery with 4,000 members.
self.request_count = 0
def __repr__(self):
"""Deliberately overridden so the token cannot leak through it."""
return (f"<sipspos.Client base_url={self.base_url!r} "
f"environment={self.environment!r} "
f"requests={self.request_count}>")
def get(self, path, **params):
"""GET `path` with `params` as the query string; return the parsed body.
`path` is the full path including the API prefix, e.g.
"/api/v1/club/members". Parameters whose value is None are dropped,
so an optional filter can be passed straight through:
client.get("/api/v1/customers", club_status=status_or_none)
"""
return self._request("GET", path, params)
def paginate(self, path, **params):
"""Yield every item of a list endpoint, following the cursor.
WHY A CURSOR AND NOT `?page=` / `?offset=`. SipsPOS list endpoints
are keyset-paginated, and the reason matters enough to repeat here
rather than link: an offset is a COUNT of the rows behind you, so
anything that changes that count between two page fetches slides the
window. A row arriving behind you re-delivers one row (annoying,
detectable). A row LEAVING from behind you — a cancellation, a
correction, an import re-dating a record — silently skips one, with
no error and nothing in the response to notice. You finish the walk
believing you have the whole list. A cursor is a POSITION, not a
count: "the rows after this one". Nothing happening elsewhere in the
table can move it. So follow `next_cursor` until `has_more` is false,
and never try to reconstruct a page number from it.
THE FILTERS ARE RE-SENT WITH THE CURSOR, and this is the part people
get wrong. A SipsPOS cursor is signed against the filter set it was
minted under, so `?club_status=none` (or `?status=`, or
`?created_since=`) must be sent again on EVERY page alongside
`cursor`. Drop it and you get a `400 invalid_request` telling you the
cursor isn't valid for this request — which is the API protecting
you, loudly, from the alternative: resuming a *differently filtered*
list part-way down and never seeing the rows before that point. That
is why this method keeps `params` intact across the loop and only
adds `cursor` to it, and it is why a cursor from one call can never
be spent on another.
`limit` defaults to DEFAULT_LIMIT and is sent on every page. It is
not part of what the cursor is signed against, so it may vary between
pages; there is no reason to make it.
"""
limit = params.pop("limit", DEFAULT_LIMIT)
if not isinstance(limit, int) or isinstance(limit, bool) \
or not 1 <= limit <= MAX_LIMIT:
# Checked here rather than left to the API, because the API's
# answer to an out-of-range limit is a 400 rather than a clamp,
# and a caller that passed 500 by accident should learn that at
# the call site.
raise ValueError(
f"limit must be an int from 1 to {MAX_LIMIT} "
f"(the API refuses anything else rather than clamping it); "
f"got {limit!r}")
params["limit"] = limit
while True:
page = self._request("GET", path, params)
items = page.get("data")
if not isinstance(items, list):
raise ApiError(
"internal",
f"{path} did not return a list envelope — expected a "
f"'data' array and got {type(items).__name__}. Use get() "
f"for endpoints that return a single resource.",
request_id=page.get("request_id"))
for item in items:
yield item
if not page.get("has_more"):
return
cursor = page.get("next_cursor")
if not cursor:
# `has_more` true with no cursor is the API contradicting
# itself, and the tempting recovery — start again from the
# top — is the worst available: the loop would re-yield page
# one forever and never say why. Stop, loudly.
raise ApiError(
"internal",
f"{path} reported has_more but sent no next_cursor, so "
f"there is no way to ask for the next page. Stopping "
f"rather than restarting the walk from the beginning.")
params["cursor"] = cursor
# -- internals ---------------------------------------------------------
def _request(self, method, path, params):
"""One logical request: retries on 429, raises `ApiError` on failure."""
url = self._url(path, params)
headers = {
# Spelled here and only here. RFC 7235 says the scheme match is
# case-insensitive; SipsPOS accepts any casing of "Bearer".
"Authorization": f"Bearer {self._token}",
"Accept": "application/json",
"User-Agent": USER_AGENT,
}
# `max_retries` RETRIES, so `max_retries + 1` ATTEMPTS. Only a 429 is
# retried: it is the one status the API documents as "wait and try
# again", and it arrives with a `Retry-After` saying how long. A 5xx
# is NOT retried here even though every request this client makes is
# a safe, read-only GET — a client that silently triples its load
# against a struggling server is part of the outage. Add it if your
# own operational picture says otherwise; do it with a delay.
for attempt in range(self.max_retries + 1):
self.request_count += 1
status, response_headers, body = self._transport(
method, url, headers, None)
found = _lower_keys(response_headers)
# CASE-INSENSITIVE, because HTTP header names are, and because
# the two transports this example actually runs under disagree in
# practice: Flask's test client and urllib normalise casing
# differently, so `headers["X-Sipspos-Environment"]` works in one
# and raises KeyError in the other. Every header read in this
# file goes through `found`.
self.environment = found.get("x-sipspos-environment")
if 200 <= status < 300:
return _decode_success(status, body, found)
if status == 429 and attempt < self.max_retries:
time.sleep(_retry_after_seconds(found))
continue
raise _error_from(status, body, found)
def _url(self, path, params):
"""`base_url` + `path` + the query string, dropping None values.
None is dropped rather than sent as the string "None" so an optional
filter can be threaded straight through from a caller's own optional
argument. An empty string IS sent — the API reads `?cursor=` as
"start at the beginning", and silently swallowing it here would hide
a template variable that failed to interpolate.
"""
query = [(name, _encode_param(value))
for name, value in params.items() if value is not None]
url = f"{self.base_url}{path}"
return f"{url}?{urllib.parse.urlencode(query)}" if query else url
def _encode_param(value):
"""One query-string value as text.
Booleans are spelled the way JSON spells them. Python's `str(True)` is
`"True"`, which no JSON API recognises, and the resulting `400` names the
parameter without explaining that its capital T is the problem.
"""
if isinstance(value, bool):
return "true" if value else "false"
return str(value)
def _lower_keys(headers):
"""A response's headers keyed by lowercase name (see `_request`)."""
return {str(name).lower(): value for name, value in (headers or {}).items()}
def _retry_after_seconds(headers):
"""How long to wait after a 429, from `Retry-After`.
HONOURED LITERALLY, INCLUDING ZERO. `Retry-After: 0` means "go now" and
is what a limiter says when the window has already rolled over; clamping
it up to a "sensible minimum" would add a second of latency to every
burst for no reason. The ceiling at the other end is MAX_RETRY_AFTER_SECONDS.
The header may also be an HTTP-date rather than a number of seconds
(RFC 9110 allows both). SipsPOS sends seconds; a proxy in between might
not, and rather than grow a date parser for a case that ends in the same
place, an unparseable value falls back to the default delay.
"""
raw = headers.get("retry-after")
try:
seconds = float(str(raw).strip())
except (TypeError, ValueError):
return DEFAULT_RETRY_AFTER_SECONDS
return max(0.0, min(seconds, MAX_RETRY_AFTER_SECONDS))
def _decode_success(status, body, headers):
"""A 2xx body -> dict, or `ApiError` if it is not the JSON we were promised."""
try:
payload = json.loads(body.decode("utf-8"))
except (AttributeError, UnicodeDecodeError, ValueError):
# A 2xx that is not JSON almost always means something answered
# INSTEAD of the API — a captive portal, an SSO redirect that
# returned a login page with a 200, a misrouted proxy. Saying so is
# more useful than a `JSONDecodeError` pointing at character 0.
raise ApiError(
"internal",
f"Expected JSON and got something else (HTTP {status}). "
f"Check that the base URL points at the winery's SipsPOS host "
f"and that nothing is intercepting the request. "
f"Body starts: {_snippet(body)}",
request_id=headers.get("x-request-id"), status=status) from None
if not isinstance(payload, dict):
raise ApiError(
"internal",
f"Expected a JSON object and got {type(payload).__name__} "
f"(HTTP {status}).",
request_id=headers.get("x-request-id"), status=status)
return payload
def _error_from(status, body, headers):
"""A non-2xx response -> the `ApiError` to raise.
THE HAPPY PATH IS THE ENVELOPE. Every error the SipsPOS API itself
produces looks like:
{"error": {"type": "permission_denied",
"message": "This API key doesn't have permission to do that.",
"request_id": "req_9f3c…"}}
THE PATH THAT MATTERS IS THE OTHER ONE. Not every non-2xx a client sees
comes from the API: a load balancer returns 502 with an HTML page, a
gateway returns 503 with plain text, a corporate proxy returns 407 with
something of its own devising. Reaching straight into `body["error"]["type"]`
turns each of those into a `KeyError` or a `TypeError` raised from inside
the HTTP client — an exception a partner's `except ApiError` will not
catch, at a moment (someone else's outage) when clear reporting is worth
the most. So the envelope is treated as the likely shape, never the
guaranteed one, and every fallback still produces the same exception with
a usable `.type`.
`request_id` is recovered from the `X-Request-Id` response header when
the body has none. SipsPOS stamps that header on every response it
serves, so a mangled body does not have to cost you the id.
"""
payload = None
try:
payload = json.loads(body.decode("utf-8"))
except (AttributeError, UnicodeDecodeError, ValueError):
pass
envelope = payload.get("error") if isinstance(payload, dict) else None
header_request_id = headers.get("x-request-id")
if isinstance(envelope, dict):
return ApiError(
envelope.get("type") or _type_for_status(status),
envelope.get("message") or _default_message(status),
request_id=envelope.get("request_id") or header_request_id,
status=status)
return ApiError(
_type_for_status(status),
f"{_default_message(status)} The response wasn't the SipsPOS error "
f"envelope, so something between you and the API most likely answered "
f"instead. Body starts: {_snippet(body)}",
request_id=header_request_id, status=status)
def _type_for_status(status):
"""A `.type` for a response that did not name one of its own.
Unknown 5xx becomes "internal" and unknown 4xx becomes "invalid_request",
which is the same split the API uses: 5xx is "not your fault, retry or
report it", 4xx is "change the request".
"""
if status in _TYPE_BY_STATUS:
return _TYPE_BY_STATUS[status]
return "internal" if status >= 500 else "invalid_request"
def _default_message(status):
return f"SipsPOS returned HTTP {status}."
def _snippet(body):
"""The first few characters of a body, safe to put in a log line."""
try:
text = body.decode("utf-8", "replace")
except AttributeError:
text = str(body)
text = " ".join(text.split())
if len(text) > _BODY_SNIPPET_CHARS:
return f"{text[:_BODY_SNIPPET_CHARS]}…"
return text or "(empty)"
What this particular application does with the client: four reads, a pure `analyse`, and the command line.
"""Club Pulse — a wine-club retention dashboard in one read-only script.
WHAT IT DOES. Four GETs through `sipspos.py`, one pure function over the
payloads, one self-contained HTML file:
fetch() /v1/club/members, /v1/customers?club_status=none,
/v1/club/tiers, /v1/orders?created_since=… — nothing derived
analyse() those four lists in, findings out. No clock, no network.
render_html() findings in, one page out (`report.py`)
main() environment, argument parsing, and the only clock read
WHAT IT IS FOR. `sipspos.py` beside it is the part a partner copies — it
knows nothing about clubs. THIS file is the worked example of using it: which
endpoint answers which question, and which field on which payload produces
which conclusion. Nothing here is clever; the whole point is that a reader can
follow a number on the rendered page back to the field it came from without
guessing. Two things in it are genuinely non-obvious, and they are the reason
this example is worth reading rather than skimming.
FIRST: CHURN IS ONLY VISIBLE THROUGH `/v1/customers?club_status=none`.
`/v1/club/members` is `Customer` narrowed to `club_status != "none"`, so the
moment a member cancels they are GONE from it — a retention dashboard built on
the members endpoint alone can show you everything except a loss. Cancelling
at SipsPOS clears `club_status`, `club_tier_id` and `is_club_member` but
leaves `club_since` alone, and that surviving date is the whole signal: a
customer with `club_status == "none"` AND a `club_since` was in the club once
and is not now. A customer who was never a member has no `club_since` at all.
That is the entire definition of `lost` below, and there is no other way to
ask this API the question.
SECOND: `analyse` TAKES `today` AS A PARAMETER, AND THAT IS NOT PEDANTRY.
"Quiet since" and "which cohort" are both answers about a date, so a function
that reads the clock itself has an invisible input: its result changes
overnight, its tests either freeze time with a patch or go flaky in the
window either side of midnight, and nobody can ask it "what would this report
have said on the first of the month?". Passing the date in costs one argument
and buys a function that is a plain mapping from data to findings — testable
with dictionaries typed by hand, no HTTP, no monkeypatching, no fixtures. The
clock is read in exactly one place in this file (`main`), which is where the
rest of the outside world already lives. Every reference application should
be able to point at one line and say "this is where the impurity is".
READ-ONLY, AND STRUCTURALLY SO. Every request this makes is a GET, the key it
asks for carries `customers:read` and `orders:read`, and there is no code path
anywhere in `examples/` that can change a row at a winery — `scripts/
club_pulse_test.py` walks the AST to keep it that way. A dashboard that reads
somebody's customer list should not be able to damage it.
"""
import argparse
import datetime
import os
import sys
import urllib.parse
# Flat imports, because the three modules sit in one directory and this is how
# a partner who copied that directory runs it: `python pulse.py`. `sipspos` is
# the copyable client; `report` is only the HTML rendering, kept out of here so
# that the analysis below can be read (and tested) without a page of markup in
# the way.
import sipspos
from report import render_html
#: The endpoints, named once. Paths carry the `/api` prefix that the winery's
#: host serves the v1 surface under; `base_url` is the winery's own origin
#: (SipsPOS is addressed per winery, not through a shared api.* domain).
MEMBERS_PATH = "/api/v1/club/members"
CUSTOMERS_PATH = "/api/v1/customers"
TIERS_PATH = "/api/v1/club/tiers"
SALES_PATH = "/api/v1/sales"
ORDERS_PATH = "/api/v1/orders"
#: Endpoints this example can do without. `orders:read` requires the
#: `online_store` capability, which is an Estate feature — so a Growth winery
#: with a thriving wine club CANNOT issue a key carrying it, even for its own
#: sandbox. Refusing to run at all for those wineries would be the wrong
#: trade: the shop is where a fraction of a club's purchases happen and the
#: register is where most of them do.
#:
#: `services/api_examples.py` reads this to split the scope table on
#: /developers into required and optional, so the page cannot tell a Growth
#: winery to ask for a scope it is not allowed to have.
OPTIONAL_PATHS = (ORDERS_PATH,)
#: How far back to look for an order before calling a member quiet. 120 days
#: is roughly one quarterly shipment cycle plus a month of slack, which is the
#: interval at which "they haven't bought anything" starts meaning something
#: at a winery rather than just meaning "it's February".
DEFAULT_QUIET_DAYS = 120
#: Where the report lands when `--out` says nothing else.
DEFAULT_OUT = "club-pulse.html"
#: THE PRECEDENCE, AND IT IS FIXED. A member in trouble is usually in trouble
#: in several ways at once — a failing card is very often also a paused
#: membership — and a dashboard that lists them once per symptom is a dashboard
#: nobody finishes reading. So each member appears EXACTLY ONCE, under the
#: first reason in this tuple that fits, and the tuple is ordered by what the
#: winery should do first: money that is already failing, then money that is
#: about to, then two forms of drifting away, then silence.
REASONS = ("past_due", "no_card", "paused", "skipping", "quiet")
#: `ClubTier.billing_period` -> months per charge. The tier's `price` is one
#: CHARGE, not one month, so a quarterly tier's price divided by three is what
#: belongs in a column headed "monthly". Publishing a quarterly price under a
#: monthly heading would be a number nobody could stand behind — see the note
#: on `_tier_rows` for the rounding, and the README for the caveat.
MONTHS_PER_PERIOD = {"monthly": 1, "quarterly": 3, "annual": 12}
#: What a tier row is called when a member's `club_tier_id` matches nothing in
#: `/v1/club/tiers`. See `_tier_rows`: the member is COUNTED, never dropped.
UNKNOWN_TIER = "Unknown tier"
# ---------------------------------------------------------------------------
# Fetch — the only part that touches the network
# ---------------------------------------------------------------------------
def fetch(client, *, today, quiet_days=DEFAULT_QUIET_DAYS):
"""The four reads, verbatim. Nothing is derived here.
Returns `{"members", "lapsed", "tiers", "orders"}` — exactly what the API
sent, in the order a reader would ask for it. Keeping this function free
of judgement is what lets `analyse` be pure: everything that needs a
network is above this line, everything that needs a decision is below it,
and a partner adapting the example can replace either half alone.
`today` is passed in rather than read, for the same reason `analyse` takes
it: the window this asks the API for and the window the analysis reasons
about must be the SAME window, and the only way to guarantee that is for
one date to flow through both. Two calls to `date.today()` a few
milliseconds apart will almost always agree — and the report that runs at
23:59:59.9 on the last day of the month is the one you will be asked to
explain.
Each `paginate` call is a cursor walk that may be many HTTP requests; the
client re-sends the filters on every page (see its docstring for why that
is not optional). `client.request_count` afterwards is how many requests
that actually took.
"""
since = (today - datetime.timedelta(days=quiet_days)).isoformat()
# The optional read. Only these two types are tolerated: both mean "this
# winery/key does not have the shop", which is a plan fact. Anything else
# — a bad token, a malformed cursor, the API being down — is a real
# failure and must not be swallowed into a quietly wrong report.
try:
orders = list(client.paginate(ORDERS_PATH, created_since=since))
orders_included = True
except sipspos.ApiError as exc:
if exc.type not in ("permission_denied", "feature_not_enabled"):
raise
orders, orders_included = [], False
return {
# Everyone in the club funnel: `club_status` of pending, active or
# past_due. Cancelled members are NOT here — that is the next call.
"members": list(client.paginate(MEMBERS_PATH)),
# The churn read. `club_status=none` is every customer who is not in
# the club, which includes everyone who never was; `analyse` filters
# it down to the ones with a `club_since`. See the module docstring.
"lapsed": list(client.paginate(CUSTOMERS_PATH, club_status="none")),
# One page, deliberately: a winery has a handful of tiers, not
# hundreds, and `paginate` here would spend a cursor walk to prove it.
# If a winery ever did have more tiers than one page holds, the
# members on the tiers past the first page land in the UNKNOWN_TIER
# row rather than vanishing — which is the same safety net that
# catches a deactivated tier, and the reason that row exists at all.
"tiers": client.get(TIERS_PATH)["data"],
# THE PRIMARY PURCHASE SIGNAL, and the one an online-store-shaped
# dashboard gets wrong. A wine club member buys at the cellar door;
# those are SALES, not orders. Reading only `/v1/orders` marks a
# member who bought a case in the tasting room last week as "quiet"
# — telling the winery to chase the customers it should be thanking.
#
# `paid_since` filters a NULLABLE column: an open tab has a NULL
# `paid_at` and is excluded, which is right here. An unpaid sale is
# not evidence that somebody bought something.
"sales": list(client.paginate(SALES_PATH, paid_since=since)),
# `created_since` is an ordinary range filter on an immutable column,
# so this window is stable: an order's `created_at` never moves. Only
# the window matters — a member with one order in it is not quiet,
# and how much they spent is not this dashboard's question.
#
# OPTIONAL. See OPTIONAL_PATHS: `orders:read` needs the `online_store`
# capability, so a Growth winery cannot issue a key with it. A refusal
# here is a fact about the winery's plan, not an error in the
# integration, and the report says which sources fed the quiet list
# rather than quietly narrowing it.
"orders": orders,
"orders_included": orders_included,
}
# ---------------------------------------------------------------------------
# Analyse — pure. No clock, no client, no I/O. `today` is an argument.
# ---------------------------------------------------------------------------
def analyse(members, lapsed, tiers, orders, sales=(), *, today,
quiet_days=DEFAULT_QUIET_DAYS, orders_included=True):
"""The payloads in, the findings dict out. No I/O of any kind.
{"at_risk": [{"name", "email", "reason", "detail"}, …],
"cohorts": [{"month", "joined", "still_active"}, …], # ascending
"tiers": [{"name", "members", "monthly_value_cents", "currency"}, …],
"lost": [{"name", "email", "club_since"}, …],
"totals": {"members", "at_risk", "lost", "monthly_value_cents",
"orders_included"}}
Everything this returns is derived from a field named in the code below,
and every lookup is `.get()` rather than `[...]`: a field the API adds
later must not break this, and a field it omits for a particular row
should produce a missing finding, never a traceback in a nightly job.
`today` and `quiet_days` are the only two things this needs that are not
in the payloads. Hand it a date and it will tell you what the report would
have said on that date — which is exactly what makes it testable.
"""
window_start = today - datetime.timedelta(days=quiet_days)
# One membership test for the quiet check, built once. `customer_id` on an
# order is the customer's opaque public id (`cus_…`), the same string as
# `id` on a member — the API never publishes integer primary keys, so
# joining on these ids is the intended way to connect the two payloads.
# An order placed by somebody who is not a customer (a walk-in, a guest
# checkout) has `customer_id: null`; those simply never match anyone.
# BOTH SOURCES. A sale is a cellar-door purchase and an order is a shop
# purchase; a member who did either is not quiet. Reading only orders is
# how a retention dashboard ends up listing a winery's most loyal
# in-person buyers as the ones to chase.
bought_recently = {row.get("customer_id")
for row in list(sales) + list(orders)
if row.get("customer_id")}
at_risk = []
for member in members:
found = _reason_for(member, bought_recently=bought_recently,
window_start=window_start, quiet_days=quiet_days)
if found is None:
continue # A healthy member. The dashboard is a list of exceptions.
reason, detail = found
at_risk.append({
"name": _display_name(member),
"email": member.get("email"),
"reason": reason,
"detail": detail,
})
# Sorted by urgency, then by name, so the top of the page is the call list
# somebody works down and two runs over unchanged data produce the same
# file. The API's own ordering is not meaningful here.
at_risk.sort(key=lambda row: (REASONS.index(row["reason"]),
row["name"].lower()))
lost = _lost(lapsed)
tier_rows = _tier_rows(members, tiers)
return {
"at_risk": at_risk,
# Cohorts read BOTH lists: a month's joiners include the ones who have
# since left, and leaving them out would make every cohort look like
# it retained everybody.
"cohorts": _cohorts(list(members) + list(lapsed)),
"tiers": tier_rows,
"lost": lost,
"totals": {
"members": len(members),
"at_risk": len(at_risk),
"lost": len(lost),
"monthly_value_cents": sum(row["monthly_value_cents"]
for row in tier_rows),
# Provenance for the quiet list, carried to the report. False
# means the key could not read the shop (a Growth winery has no
# shop to read), so "quiet" was judged on register sales alone.
# A reader deciding whether to phone somebody deserves to know
# which question the page actually asked.
"orders_included": bool(orders_included),
},
}
def _reason_for(member, *, bought_recently, window_start, quiet_days):
"""The one reason this member is at risk, or None if they are fine.
Read this top to bottom: the order of the branches IS the precedence in
`REASONS`, and the first one that matches returns. That is why a member
whose card is failing AND missing appears once, as `past_due` — the
winery cannot fix the second problem without fixing the first.
"""
status = member.get("club_status")
last4 = member.get("club_card_last4")
next_charge = member.get("club_next_charge_date")
# 1. past_due — `club_status`. The API sets this when a dues charge has
# already failed, so this is money that has ALREADY been lost, not
# money at risk. Nothing else on the page outranks it.
if status == "past_due":
detail = "The last dues charge did not go through"
detail += (f", on the card ending {last4}" if last4
else ", and there is no card on file to try again")
detail += (f". The next attempt is due {next_charge}."
if next_charge else ".")
return "past_due", detail
# 2. no_card — `club_status == "active"` and `club_card_last4` empty.
# THE STATE THIS API MAKES UNIQUELY VISIBLE. A member migrated from
# another system arrives active, with a real `club_since` and a real
# tier, and no card: the old processor's card cannot come across, so
# the member sits here until they re-authorize payment at this winery.
# Nothing is failing yet, which is exactly why nobody notices — the
# failure is scheduled for the next charge date. `pending` members are
# deliberately NOT caught here: a signup that has not been completed is
# not the same problem and does not want the same email.
if status == "active" and not last4:
detail = ("Active in the club with no card on file "
"(club_card_last4 is empty). This is where a member "
"imported from another system waits until they "
"re-authorize payment here")
detail += (f", so the charge due {next_charge} will fail unless "
f"somebody asks them for a card."
if next_charge else ", so the next charge will fail.")
return "no_card", detail
# 3. paused — `club_paused`, which is a separate column from
# `club_status`: a paused member is still `active`, and reading only
# the status would show them as perfectly healthy.
if member.get("club_paused"):
resume = member.get("club_resume_on")
detail = ("The membership is paused (club_paused)"
+ (f", due to resume on {resume}." if resume else
" with no club_resume_on date set, so nothing will "
"restart it on its own — an open-ended pause is the "
"shape a quiet cancellation usually takes."))
return "paused", detail
# 4. skipping — `club_skip_next`. One skip is ordinary; it is on this list
# because two in a row rarely are, and this is the only place the
# winery would ever see the first one.
if member.get("club_skip_next"):
detail = "They have chosen to skip the next shipment (club_skip_next)"
detail += (f"; the next charge is due {next_charge}."
if next_charge else ".")
return "skipping", detail
# 5. quiet — no order in the window. Last, because it is the weakest
# signal on the page: it says only that nothing was bought THROUGH
# SipsPOS in `quiet_days`, and the club shipment itself may not be an
# order. BOTH counters are read — a cellar-door sale and a shop order —
# because for a wine club the cellar door is where most of the buying
# happens, and a check that saw only the shop would accuse a winery's
# best in-person customers of going quiet. Restricted to `active` so a `pending` signup, who has no
# reason to have ordered yet, is not accused of going quiet.
if status == "active" and member.get("id") not in bought_recently:
detail = (f"Nothing bought since {window_start.isoformat()} "
f"(the last {quiet_days} days) — no cellar-door sale and "
f"no online order. Paying dues, buying nothing else.")
club_since = member.get("club_since")
if club_since and club_since >= window_start.isoformat():
# Dates from this API are `YYYY-MM-DD`, which sorts correctly as
# text — no parsing needed, and none risked.
detail += (f" They joined on {club_since}, so this may simply be "
f"a new member whose first shipment has not gone out "
f"yet.")
return "quiet", detail
return None
def _cohorts(customers):
"""Joiners per month and how many of them are still in the club.
Grouped on `club_since[:7]` — the first seven characters of a
`YYYY-MM-DD` date are its month, so this needs no date parsing and cannot
throw on a value the API formats slightly differently than expected.
`joined` counts everyone who joined that month whether or not they are
still here; `still_active` counts the ones whose `is_club_member` is still
true. The gap between the two columns is the retention curve, and it only
exists because the lapsed customers were passed in alongside the members.
A customer with no `club_since` has never been in the club and belongs to
no cohort, so they are skipped rather than bucketed under a blank month.
"""
joined = {}
still_active = {}
for customer in customers:
club_since = customer.get("club_since")
if not club_since or len(str(club_since)) < 7:
continue
month = str(club_since)[:7]
joined[month] = joined.get(month, 0) + 1
if customer.get("is_club_member"):
still_active[month] = still_active.get(month, 0) + 1
return [{"month": month,
"joined": joined[month],
"still_active": still_active.get(month, 0)}
for month in sorted(joined)]
def _tier_rows(members, tiers):
"""Members and monthly value per tier, joined on `club_tier_id`.
NOBODY IS DROPPED. A member whose `club_tier_id` matches no tier in the
payload — a deactivated tier, a tier past the first page, a member with no
tier at all — is counted under UNKNOWN_TIER. The alternative, silently
skipping them, is the worst available outcome: the tier table would still
look plausible while the member count quietly disagreed with the club's,
and nothing on the page would say so.
MONTHLY VALUE, FROM A PRICE THAT IS NOT MONTHLY. `tier["price"]` is one
charge, and `tier["billing_period"]` says how often that charge happens
(monthly, quarterly, annual). A quarterly tier's price is three months of
value, so it is divided by three here rather than published as if it
arrived every month. Rounding is done on the tier's TOTAL rather than
per member, so a hundred members on a $59.99/quarter tier do not
accumulate a hundred half-cent errors. A tier whose price is null — money
is `{"amount_cents", "currency"}` or null at this API, never a bare
number, and null means unknown — contributes zero and says nothing it
cannot support.
"""
by_id = {tier.get("id"): tier for tier in tiers if tier.get("id")}
counts = {}
for member in members:
tier = by_id.get(member.get("club_tier_id"))
key = tier.get("id") if tier else None # None -> the unknown row
counts[key] = counts.get(key, 0) + 1
rows = []
for key, count in counts.items():
tier = by_id.get(key)
if tier is None:
rows.append({"name": UNKNOWN_TIER, "members": count,
"monthly_value_cents": 0, "currency": None})
continue
price = tier.get("price") or {}
amount = price.get("amount_cents")
months = MONTHS_PER_PERIOD.get(tier.get("billing_period"), 1)
if amount is None:
monthly = 0
else:
# Integer arithmetic with explicit half-up rounding: cents are
# counted, not measured, and float division would put a
# 0.30000000000000004 in a money column.
total = int(amount) * count
monthly = (total + months // 2) // months
rows.append({
"name": tier.get("name") or UNKNOWN_TIER,
"members": count,
"monthly_value_cents": monthly,
"currency": price.get("currency"),
})
# Biggest contribution first — the tier mix is read to answer "where does
# the club's money come from", and ties broken by name keep it stable.
rows.sort(key=lambda row: (-row["monthly_value_cents"], row["name"]))
return rows
def _lost(lapsed):
"""Customers who WERE in the club and are not now.
`lapsed` is `/v1/customers?club_status=none`, which is every customer
outside the club — overwhelmingly people who were never in it. The filter
is `club_since`: cancelling clears `club_status`, `club_tier_id` and
`is_club_member` but leaves the join date alone, so a `none` customer WITH
a `club_since` is precisely somebody who left. Somebody who never joined
has no join date to keep.
This is the only way to see churn through this API — `/v1/club/members`
excludes `club_status == "none"`, so every member it can show you is by
definition one you have not lost.
"""
lost = [{"name": _display_name(customer),
"email": customer.get("email"),
"club_since": customer.get("club_since")}
for customer in lapsed if customer.get("club_since")]
# Newest join date first: it is a rough proxy for who left most recently
# (the API publishes no cancellation date), and recent losses are the ones
# still worth a phone call.
lost.sort(key=lambda row: (str(row["club_since"] or ""),
row["name"].lower()), reverse=True)
return lost
def _display_name(customer):
"""A name for a row on a page a human reads.
`name` is required at this API, but `""` and whitespace both survive older
imports, and a blank cell in a call list is a row nobody can act on. The
email is the next best handle a winery actually has.
"""
name = (customer.get("name") or "").strip()
if name:
return name
email = (customer.get("email") or "").strip()
return email or "(customer with no name on file)"
# ---------------------------------------------------------------------------
# The command line
# ---------------------------------------------------------------------------
def main(argv=None):
"""`python pulse.py [--out FILE] [--quiet-days N]`; 0 on success.
NO TRACEBACK EVER REACHES THE TERMINAL, and that is a deliberate feature
of a file people copy. A traceback is a bug report about this script; what
somebody running a nightly report needs is a sentence about THEIR problem
— a missing scope, a bad host, an expired key — with the request id
attached so support can find it. `ApiError.__str__` is already written to
be that sentence, so this only has to print it and choose an exit code.
"""
parser = argparse.ArgumentParser(
prog="pulse.py",
description="Build a wine-club retention report from the SipsPOS "
"public API. Read-only: it makes GET requests and "
"nothing else.")
parser.add_argument(
"--out", default=DEFAULT_OUT,
help=f"where to write the HTML report (default: {DEFAULT_OUT})")
parser.add_argument(
"--quiet-days", type=int, default=DEFAULT_QUIET_DAYS,
help=f"how many days without an order makes a member 'quiet', and "
f"how far back to read orders (default: {DEFAULT_QUIET_DAYS})")
args = parser.parse_args(argv)
base_url = os.environ.get("SIPSPOS_BASE_URL")
token = os.environ.get("SIPSPOS_API_KEY")
# Named individually, because "configure your credentials" sends somebody
# to a README and "SIPSPOS_API_KEY is not set" sends them to a shell.
for name, value in (("SIPSPOS_BASE_URL", base_url),
("SIPSPOS_API_KEY", token)):
if not (value or "").strip():
print(f"club-pulse: {name} is not set. Set it and try again:\n"
f" export SIPSPOS_BASE_URL=https://your-winery.sipspos.com\n"
f" export SIPSPOS_API_KEY=sk_test_…\n"
f"A test key (Admin -> API keys) reaches your sandbox twin, "
f"so nothing you do while learning touches real customers.",
file=sys.stderr)
return 2
if args.quiet_days < 1:
print(f"club-pulse: --quiet-days must be at least 1, got "
f"{args.quiet_days}.", file=sys.stderr)
return 2
# THE CLOCK, READ ONCE, HERE. Everything below this line is handed the
# date rather than asking for it — see the module docstring. One read also
# means the window `fetch` asks the API for and the window `analyse`
# reasons about cannot disagree.
today = datetime.date.today()
# A naive `datetime`, not a formatted string: `report.render_html` does
# the wording, and handing it the moment rather than one rendering of the
# moment keeps the choice of format in the one module that draws pages.
generated_at = datetime.datetime.now()
client = sipspos.Client(base_url, token)
try:
payloads = fetch(client, today=today, quiet_days=args.quiet_days)
except sipspos.ApiError as e:
print(f"club-pulse: {e}", file=sys.stderr)
if e.type == "permission_denied":
# The single most common first-run failure, and the fix is two
# checkboxes rather than anything to do with this code.
print("club-pulse: this key needs the customers:read and "
"sales:read scopes (Admin -> API keys). orders:read is "
"optional — the shop is skipped without it.",
file=sys.stderr)
return 1
findings = analyse(payloads["members"], payloads["lapsed"],
payloads["tiers"], payloads["orders"],
payloads["sales"],
today=today, quiet_days=args.quiet_days,
orders_included=payloads["orders_included"])
# The report is handed the winery's HOST and the environment, and never
# the client or the key: this file gets emailed around a winery, and the
# surest way for a token never to appear in it is for the renderer to have
# no way to reach one.
html_text = render_html(
findings,
winery=urllib.parse.urlsplit(base_url).netloc or base_url,
environment=client.environment,
generated_at=generated_at)
try:
with open(args.out, "w", encoding="utf-8") as fh:
fh.write(html_text)
except OSError as e:
print(f"club-pulse: couldn't write {args.out}: {e}", file=sys.stderr)
return 1
totals = findings["totals"]
print(f"club-pulse: {args.out} — {totals['members']} members, "
f"{totals['at_risk']} at risk, {totals['lost']} lost "
f"({client.request_count} API requests)")
if client.environment == "test":
# Said on the terminal as well as on the page: somebody who ran this
# against the sandbox and then quoted the numbers in a meeting is the
# failure this line exists to prevent.
print("club-pulse: this is SANDBOX data — the winery in it is a twin, "
"and the members are invented.")
return 0
if __name__ == "__main__":
sys.exit(main())
The findings rendered as one self-contained HTML page. No script, no external asset, no token.
"""The Club Pulse report: findings in, one self-contained HTML file out.
WHAT THIS IS. `pulse.analyse` turns four API reads into a plain dict of
findings; this module turns that dict into a page a winery can open, read,
print and forward. It is the last third of the example and the least
interesting third on purpose — the subject of Club Pulse is the client code
in `sipspos.py`, not the layout here. If you are copying this example, copy
`sipspos.py`; this file exists so the example has somewhere to put its
answer, and so that "renders a report" is a claim a test can check.
It imports `html` and `datetime` and nothing else. No template engine: a
templating dependency would put a `pip install` between a reader and the one
file they came here to read, and the whole page is about two hundred lines of
string building — small enough that the honest version is the readable one.
TWO DECISIONS A READER WILL STOP AND QUESTION:
1. THERE IS NO JAVASCRIPT, AND NO SORTABLE TABLES OR COLLAPSIBLE SECTIONS
BECAUSE OF IT. This file is not a web page in the usual sense; it is a
document that gets emailed. Somebody at a winery opens it from a `file://`
path, or in Outlook's preview pane, or forwards it to a partner who opens
it on a phone — and every one of those contexts either strips scripts,
blocks them, or runs them against a document with no server to talk to.
A page whose content is assembled by script is blank in exactly the places
this one has to survive. So every number is in the markup, the CSS is
inline in a single `<style>`, and there are no external stylesheets, fonts
or images to fail to load. The cost is real — no sorting, no filtering —
and it is the right trade for a file whose job is to still be legible six
months from now in somebody's mail archive.
2. THE SANDBOX BANNER. When the client reports `environment == "test"`, the
page opens with a loud banner saying the numbers are invented. That is not
decoration and it is not defensive over-engineering: a test key reaches a
sandbox TWIN of the winery, seeded with plausible-looking fake members, and
a sandbox report is visually indistinguishable from a real one — same
winery name, same tier names, same shape of numbers. The failure mode is a
winery reading a sandbox report as real and phoning members who are not in
any trouble, or worse, reading real churn as a test fixture and doing
nothing about it. It is the single worst outcome available to this file, it
costs one `<div>` to prevent, and the environment is a fact the API told us
(`X-Sipspos-Environment`, derived from the tenant actually served) rather
than a guess from a token prefix. The banner is drawn with a heavy border
and the word SANDBOX spelled out, not with colour alone, because a printed
copy is greyscale and a red background is then just a grey one.
THE TOKEN CANNOT REACH THIS PAGE, and the mechanism is that this module never
receives it. `render_html` takes findings, a winery name, an environment
string and a timestamp. There is no parameter it could arrive in, no client
object to read it off, and no `repr()` of anything that holds it. A test
asserts the token is absent from the output; this is the reason that test can
never start failing.
EVERYTHING INTERPOLATED IS ESCAPED. Member names, emails and tier names are
partner data — they came out of a winery's customer records, which is to say
out of a web form somebody else filled in. `_text()` is the only way a value
reaches the page, and it escapes. A name containing `<b>` should print as
`<b>`, not turn the rest of the report bold, and the same discipline is what
stops a crafted name from injecting markup into a document that gets emailed
around.
"""
import datetime
import html
#: The at-risk reasons, MOST URGENT FIRST, and the page is grouped in this
#: order. It matches the precedence `analyse` uses to assign a member their
#: single reason, and the two must not drift: a member is filed under their
#: most urgent problem, so a reader working down the page is working down a
#: list already sorted by "who needs a phone call today". Billing failing is
#: first because it is the only one with a deadline attached — the next
#: retry either succeeds or the member is gone.
REASON_ORDER = ("past_due", "no_card", "paused", "skipping", "quiet")
#: Per reason: the heading a winery reads, the sentence saying what to do
#: about it, and the API field the conclusion came from.
#:
#: The field line is on the page, not just in this source, because Club Pulse
#: is a reference application: somebody evaluating the API wants to know which
#: field produced which claim, and the fastest way to tell them is to print it
#: next to the claim. It also keeps this file honest — a heading that drifts
#: away from the field beneath it is visibly wrong on the page.
REASON_LABELS = {
"past_due": {
"title": "Billing is failing",
"advice": "The last charge was declined. These are the calls worth "
"making today: a card updated this week keeps the member, "
"and one more failed retry usually doesn't.",
"field": "club_status == \"past_due\"",
},
"no_card": {
"title": "No card on file",
"advice": "Active members with nothing to charge — usually somebody "
"migrated from an older system who never re-authorised. "
"They look healthy in every other report you have.",
"field": "club_status is active and club_card_last4 is empty",
},
"paused": {
"title": "Paused",
"advice": "Still members, not being shipped. A note shortly before "
"the resume date is what keeps a pause from quietly "
"becoming a cancellation.",
"field": "club_paused (with club_resume_on when it is set)",
},
"skipping": {
"title": "Skipping the next shipment",
"advice": "One skip is ordinary. A skip from a member who also "
"appears in next month's list is somebody telling you "
"something before they cancel.",
"field": "club_skip_next",
},
"quiet": {
"title": "Quiet",
"advice": "Paying, current, and hasn't bought anything inside the "
"window \u2014 at the cellar door or in the shop. The "
"earliest signal available here, and the cheapest one to "
"act on.",
"field": "no sale and no order in the window, "
"joined on customer_id",
},
}
#: Month names spelled out rather than left to `strftime("%B")`, which is
#: LOCALE-DEPENDENT. This report is generated on a server whose locale is
#: nobody's decision in particular and read by a winery that did not choose
#: it either; a page that says "März" because a cron host was configured in
#: German is a bug that only ever reproduces in production.
MONTH_NAMES = ("January", "February", "March", "April", "May", "June", "July",
"August", "September", "October", "November", "December")
#: Currency symbols for the codes a winery is likely to see, with the ISO code
#: printed for everything else. Deliberately small: guessing a symbol wrongly
#: is worse than showing "1,234.56 CHF", which nobody can misread.
CURRENCY_SYMBOLS = {
"USD": "$", "CAD": "CA$", "AUD": "A$", "NZD": "NZ$",
"EUR": "€", "GBP": "£", "JPY": "¥",
}
#: Shown wherever the API gave us nothing. An em dash rather than an empty
#: cell, so "we have no email for this member" is visibly different from a
#: table that failed to render.
MISSING = "—"
#: The whole stylesheet, inlined (see decision 1 in the module docstring).
#: Dark on light, one serif stack with real fallbacks, and a print block —
#: the plausible thing for a winery to do with this page is print the at-risk
#: list and take it to the phone, so page breaks inside a member row are a
#: real failure and not a nicety.
_STYLE = """
:root { color-scheme: light; }
* { box-sizing: border-box; }
body {
margin: 0;
background: #fbfaf8;
color: #1d1a16;
font-family: "Iowan Old Style", "Palatino Linotype", Palatino, Georgia,
"Times New Roman", serif;
font-size: 17px;
line-height: 1.65;
}
.page { max-width: 50rem; margin: 0 auto; padding: 2.5rem 1.25rem 4rem; }
h1, h2, h3 { line-height: 1.25; font-weight: 600; }
h1 { font-size: 1.9rem; margin: 0 0 .25rem; }
h2 {
font-size: 1.25rem; margin: 3rem 0 .25rem;
padding-bottom: .35rem; border-bottom: 2px solid #cfc7b8;
}
h3 { font-size: 1.05rem; margin: 2rem 0 .1rem; }
p { margin: .5rem 0; }
.muted { color: #5d564c; }
.small { font-size: .85rem; }
.field {
font-family: ui-monospace, "SFMono-Regular", Menlo, Consolas, monospace;
font-size: .78rem; color: #5d564c;
}
.masthead { border-bottom: 3px double #cfc7b8; padding-bottom: 1.25rem; }
.warning {
border: 3px solid #8c2f19; background: #fbeee9;
padding: 1rem 1.15rem; margin: 0 0 2rem;
}
.warning .flag {
display: block; text-transform: uppercase; letter-spacing: .12em;
font-weight: 700; font-size: .85rem; margin-bottom: .3rem;
}
.warning p { margin: 0; }
.totals {
display: flex; flex-wrap: wrap; gap: 1.75rem;
margin: 1.5rem 0 0; padding: 0; list-style: none;
}
.totals li { min-width: 7rem; }
.totals .n {
display: block; font-size: 1.7rem; line-height: 1.2;
font-variant-numeric: tabular-nums;
}
.totals .k {
display: block; font-size: .78rem; text-transform: uppercase;
letter-spacing: .08em; color: #5d564c;
}
nav.contents { margin: 1.75rem 0 0; font-size: .9rem; }
nav.contents a { color: #5b3a2e; }
table { width: 100%; border-collapse: collapse; margin: .75rem 0 0; }
caption { text-align: left; font-size: .85rem; color: #5d564c;
padding-bottom: .35rem; }
th, td {
text-align: left; padding: .5rem .6rem;
border-bottom: 1px solid #e4dfd5; vertical-align: top;
}
thead th {
border-bottom: 2px solid #b9b0a1; font-size: .74rem; font-weight: 600;
text-transform: uppercase; letter-spacing: .07em; color: #4a443c;
}
td.num, th.num { text-align: right; font-variant-numeric: tabular-nums; }
tbody tr:last-child td { border-bottom: 1px solid #cfc7b8; }
.group { margin-bottom: 1.5rem; }
.empty {
border-left: 3px solid #cfc7b8; padding: .35rem 0 .35rem .85rem;
color: #5d564c;
}
footer {
margin-top: 3.5rem; padding-top: 1.25rem;
border-top: 1px solid #cfc7b8; color: #5d564c; font-size: .85rem;
}
@media print {
body { background: #fff; font-size: 11pt; }
.page { max-width: none; padding: 0; }
nav.contents { display: none; }
h2 { margin-top: 1.5rem; }
tr, .group, .warning { break-inside: avoid; page-break-inside: avoid; }
a { color: inherit; text-decoration: none; }
}
"""
def render_html(findings, *, winery, environment, generated_at):
"""Render one findings dict as a complete, self-contained HTML document.
`findings` is what `pulse.analyse` returns: `at_risk`, `cohorts`, `tiers`,
`lost` and `totals`. `winery` is a display name, `environment` is "test",
"live" or None as reported by `Client.environment`, and `generated_at` is
a `datetime` — passed in rather than read from the clock here so the page
is a pure function of its inputs and a test can assert on the whole string.
Returns the document as one `str`, ready to write to a file. There is
deliberately no file-writing in this module: the caller decides where the
bytes go, and a renderer that also touches the disk is a renderer that is
annoying to test.
EVERY SECTION SURVIVES BEING EMPTY. A winery whose club is entirely
healthy has no at-risk members, and the answer to that is a page that says
so — not a blank space a reader has to decide the meaning of, and not a
crash. Same for a brand-new club with no cohorts and no cancellations.
"""
findings = findings or {}
totals = findings.get("totals") or {}
out = ["<!doctype html>",
"<html lang=\"en\">",
"<head>",
"<meta charset=\"utf-8\">",
"<meta name=\"viewport\" "
"content=\"width=device-width, initial-scale=1\">",
f"<title>Club Pulse {MISSING} {_text(winery, 'Wine club')}</title>",
f"<style>{_STYLE}</style>",
"</head>",
"<body>",
"<div class=\"page\">"]
# The banner goes FIRST, above the winery's own name, because its whole
# job is to be read before anything else on the page is believed.
out += _sandbox_banner(environment)
out += _masthead(winery, environment, generated_at, totals)
out += _contents()
out += _at_risk_section(findings.get("at_risk") or [])
out += _cohort_section(findings.get("cohorts") or [])
out += _tier_section(findings.get("tiers") or [])
out += _lost_section(findings.get("lost") or [])
out += _footer(environment)
out += ["</div>", "</body>", "</html>", ""]
return "\n".join(out)
# -- sections --------------------------------------------------------------
def _sandbox_banner(environment):
"""The sandbox warning, or nothing at all when this is real data.
Only "test" produces a banner. A live report gets no banner because a
warning that appears on every page is a warning nobody reads by the third
one — the point of this block is that it is unusual. The environment is
still stated in the masthead either way, so a reader who wants to check
can always check.
"""
if environment != "test":
return []
return [
"<div class=\"warning\" role=\"note\">",
"<span class=\"flag\">Sandbox data — not your winery</span>",
"<p>This report was generated with a <strong>test</strong> API key, "
"so every member, order and number below is invented sandbox data. "
"Nobody named here is a real customer and nothing here is a real "
"problem. Re-run with a live key to see the actual club.</p>",
"</div>",
]
def _masthead(winery, environment, generated_at, totals):
"""Title, provenance line, and the four numbers worth seeing first."""
# The environment is stated on EVERY report, live ones included. The
# banner above answers "is this fake?" when the answer is yes; this line
# answers "which is this?" for a reader who is holding two printouts.
if environment == "test":
env_line = ("Read from the <strong>sandbox</strong> (test API key) "
"— invented data.")
elif environment == "live":
env_line = "Read from <strong>live</strong> data (live API key)."
else:
# None means no authenticated response carried the header. Saying
# "unknown" is the honest answer; assuming "live" would be the
# dangerous one and assuming "test" would be the useless one.
env_line = ("Environment <strong>unknown</strong> — the API did "
"not report one, so treat these numbers with care.")
# A quiet list judged on register sales alone is a DIFFERENT question
# from one that also saw the shop, and the difference decides whether a
# name on this page is worth a phone call. Said in the provenance line
# rather than left for the reader to infer from nothing.
scope_line = ""
if totals.get("orders_included") is False:
scope_line = (" <strong>Online orders were not read</strong> "
"(this key cannot see the shop), so \u201cquiet\u201d "
"here means no <em>register sale</em> in the window.")
return [
"<header class=\"masthead\">",
f"<h1>Club Pulse {MISSING} {_text(winery, 'Wine club')}</h1>",
f"<p class=\"muted small\">{_timestamp(generated_at)}. {env_line}"
f"{scope_line}</p>",
"<ul class=\"totals\">",
_total_item(totals.get("members"), "Club members"),
_total_item(totals.get("at_risk"), "Need attention"),
_total_item(totals.get("lost"), "Lost"),
_total_item(_money(totals.get("monthly_value_cents"),
_dominant_currency(totals)),
"Monthly value", numeric=False),
"</ul>",
"</header>",
]
def _total_item(value, label, *, numeric=True):
"""One headline number.
The counts come from `totals` rather than from `len()` of the lists below
them. They are the analysis's own answer, and if the two ever disagree the
right place to find that out is a test of `analyse`, not a page that
quietly recounts and papers over it.
"""
shown = _count(value) if numeric else value
return (f"<li><span class=\"n\">{shown}</span>"
f"<span class=\"k\">{label}</span></li>")
def _contents():
"""Anchor links between sections.
Plain `#fragment` links, which work from a `file://` path with no script
and no server — which is the entire reason this page has no other
navigation. Hidden when printed, where they are just underlined noise.
"""
return [
"<nav class=\"contents\">",
"<a href=\"#at-risk\">Needs attention</a> · "
"<a href=\"#cohorts\">Retention by cohort</a> · "
"<a href=\"#tiers\">Tier mix</a> · "
"<a href=\"#lost\">Lost members</a>",
"</nav>",
]
def _at_risk_section(at_risk):
"""The one section somebody acts on, grouped by reason, most urgent first.
GROUPED RATHER THAN SORTED. A single table sorted by urgency would carry
the same rows in the same order, and would make a reader compare each row
against the one above it to work out where "call today" ends and "keep an
eye on" begins. Headings do that work once. Each group also carries the
count and the advice, so the first screen of this section answers "how
many calls, and about what".
"""
out = ["<h2 id=\"at-risk\">Needs attention</h2>"]
if not at_risk:
out.append(
"<p class=\"empty\">Nobody is at risk today. Every club member "
"has a working card, an active subscription and a recent "
"order.</p>")
return out
# "1 members" is the kind of small wrongness that makes a reader trust
# the numbers above it less, and it costs one branch to avoid.
plural = "member" if len(at_risk) == 1 else "members"
out.append(f"<p class=\"muted small\">{_count(len(at_risk))} {plural}, "
f"most urgent first.</p>")
# Bucket first, then walk REASON_ORDER — so the page order is fixed by
# this module rather than by whatever order the analysis happened to
# produce, and an unrecognised reason cannot silently disappear.
buckets = {}
for member in at_risk:
buckets.setdefault((member or {}).get("reason"), []).append(member)
for reason in REASON_ORDER:
members = buckets.pop(reason, [])
if members:
out += _at_risk_group(reason, members)
# Anything left is a reason this file has never heard of — one added to
# `analyse` after this page was written. Printing it under its raw key
# is ugly and correct: dropping members off a retention report because a
# label is missing is the one outcome worth being ugly to avoid.
for reason, members in buckets.items():
out += _at_risk_group(reason, members)
return out
def _at_risk_group(reason, members):
"""One reason: heading, count, advice, the field it came from, table."""
labels = REASON_LABELS.get(reason) or {}
title = labels.get("title") or _text(reason, "Other")
rows = [[_text(m.get("name"), "Unnamed member"),
_text(m.get("email")),
_text(m.get("detail"))]
for m in (member or {} for member in members)]
out = ["<div class=\"group\">",
f"<h3>{title} · {_count(len(members))}</h3>"]
if labels.get("advice"):
out.append(f"<p class=\"muted small\">{labels['advice']}</p>")
if labels.get("field"):
out.append(f"<p class=\"field\">from {_text(labels['field'])}</p>")
out += _table(["Member", "Email", "What's happening"], rows)
out.append("</div>")
return out
def _cohort_section(cohorts):
"""Retention by join month: how many of each month's joiners are still in.
Ascending by month, as `analyse` produces it — a reader scanning down is
reading forwards in time, and the most recent cohort (the one with the
least meaningful number, since it has had no time to churn) is at the
bottom rather than leading the section.
"""
out = ["<h2 id=\"cohorts\">Retention by cohort</h2>"]
if not cohorts:
out.append(
"<p class=\"empty\">No cohorts yet. This table fills in once "
"members have a club start date behind them.</p>")
return out
rows = []
for cohort in (c or {} for c in cohorts):
joined = cohort.get("joined")
still = cohort.get("still_active")
rows.append([_month(cohort.get("month")), _count(joined),
_count(still), _retention(joined, still)])
out.append("<p class=\"muted small\">Members grouped by the month they "
"joined, and how many of them are still in the club.</p>")
out.append("<p class=\"field\">from club_since and is_club_member</p>")
out += _table(["Joined in", "Joined", "Still active", "Retained"], rows,
numeric_from=1)
return out
def _tier_section(tiers):
"""Tier mix and what each tier is worth per month."""
out = ["<h2 id=\"tiers\">Tier mix</h2>"]
if not tiers:
out.append("<p class=\"empty\">No tiers to show.</p>")
return out
rows = [[_text(tier.get("name"), "Unknown tier"),
_count(tier.get("members")),
_money(tier.get("monthly_value_cents"), tier.get("currency"))]
for tier in (t or {} for t in tiers)]
out.append("<p class=\"field\">from club_tier_id joined on "
"/v1/club/tiers</p>")
out += _table(["Tier", "Members", "Monthly value"], rows, numeric_from=1)
return out
def _lost_section(lost):
"""Members who cancelled — the only view of churn this API affords.
A cancelled member is not in `/v1/club/members` at all (cancelling clears
`club_status`), so these come from `/v1/customers?club_status=none`
filtered to the ones that still carry a `club_since`. A retention
dashboard that cannot show a loss is not one, which is why the section is
here even though it is the least actionable one on the page.
"""
out = ["<h2 id=\"lost\">Lost members</h2>"]
if not lost:
out.append("<p class=\"empty\">No cancellations on record. Nothing to "
"win back.</p>")
return out
rows = [[_text(member.get("name"), "Unnamed member"),
_text(member.get("email")),
_text(member.get("club_since"))]
for member in (m or {} for m in lost)]
out.append("<p class=\"field\">from club_status == \"none\" with a "
"club_since</p>")
out += _table(["Member", "Email", "Club since"], rows)
return out
def _footer(environment):
"""Provenance, and the sandbox warning repeated for a printed second page.
Somebody who prints this and reads page three has lost sight of the banner
on page one, so the sandbox fact is stated again where the page ends.
"""
out = ["<footer>"]
if environment == "test":
out.append("<p><strong>Sandbox data.</strong> Generated with a test "
"API key: none of the members above are real.</p>")
out.append(
"<p>Club Pulse reads the SipsPOS public API and writes this file. It "
"only ever reads — nothing in this report changed anything at "
"the winery, and there is no way to record here that a member has "
"been contacted.</p>")
out.append("</footer>")
return out
# -- small helpers ---------------------------------------------------------
def _text(value, missing=MISSING):
"""The ONLY way a value reaches the page, and it escapes.
`quote=True` even for values landing between tags: this file has no
attribute interpolation today, and the way that stops being true is
somebody adding one and reusing this helper without noticing which mode it
was in. Escaping the quotes always costs nothing here and removes the
question.
Emails are rendered as text, never as a `mailto:` link, and that is
deliberate. An email address out of a winery's customer records is
partner-supplied text; turning it into an `href` means escaping is no
longer sufficient, because a URL is interpreted as well as displayed and
`javascript:` is a valid-looking scheme. Validating a URL properly is more
code than this example wants to carry and more code than a reader wants to
audit, so the page shows the address and lets the reader copy it.
"""
if value is None:
return missing
text = html.escape(str(value), quote=True)
return text if text.strip() else missing
def _count(value):
"""An integer with thousands separators, or a dash if it isn't one."""
if isinstance(value, bool) or not isinstance(value, int):
# Also catches None. A count the analysis did not produce shows as a
# dash rather than as 0 — "we don't know" and "none" are different
# answers, and only one of them means the club is fine.
return MISSING
return f"{value:,}"
def _retention(joined, still_active):
"""`still_active / joined` as a percentage, guarding the empty cohort.
A cohort with no joiners cannot have a retention rate, and 0 % would read
as a catastrophe rather than as an absence. Divide-by-zero in a report
generator is a crash at 6am on the one morning somebody needed the file.
"""
if not isinstance(joined, int) or isinstance(joined, bool) or joined <= 0:
return MISSING
if not isinstance(still_active, int) or isinstance(still_active, bool):
return MISSING
return f"{round(100 * still_active / joined)}%"
def _money(cents, currency):
"""Integer cents to something a winery reads without decoding it.
TWO DECIMAL PLACES, ALWAYS. SipsPOS stores money in minor units and this
report assumes a hundred of them to the major unit, which is wrong for
JPY and a handful of others. That is a known simplification rather than an
oversight: the alternative is a table of currency exponents in a file
whose subject is HTTP pagination, and a yen figure a hundred times too
small is visibly wrong rather than quietly wrong.
An unknown or missing currency prints the amount with no symbol instead of
guessing dollars, because a euro figure with a dollar sign in front of it
is the kind of error that survives review.
"""
if isinstance(cents, bool) or not isinstance(cents, (int, float)):
return MISSING
sign = "-" if cents < 0 else ""
amount = f"{abs(cents) / 100:,.2f}"
code = str(currency).strip().upper() if currency else ""
if code in CURRENCY_SYMBOLS:
return f"{sign}{CURRENCY_SYMBOLS[code]}{amount}"
if code:
return f"{sign}{amount} {_text(code)}"
return f"{sign}{amount}"
def _dominant_currency(totals):
"""The currency for the headline monthly figure.
`totals` may carry one; if it does not, the figure is rendered bare rather
than in a currency picked from the first tier that happened to have one.
Summing several currencies into one number is already a question this
example does not answer, and labelling that sum with one of their symbols
would be answering it wrongly.
"""
return (totals or {}).get("currency")
def _month(value):
""""2026-03" to "March 2026", falling back to whatever we were given."""
try:
year, month = str(value).split("-")[:2]
year, month = int(year), int(month)
if not 1 <= month <= 12:
# Range-checked BEFORE indexing, because month 0 would otherwise
# index MONTH_NAMES[-1] and confidently print December.
raise ValueError(value)
return f"{MONTH_NAMES[month - 1]} {year}"
except (AttributeError, IndexError, TypeError, ValueError):
# A month key this function cannot parse still names a real cohort
# with real members in it, so it is printed as-is rather than dropped.
return _text(value)
def _timestamp(moment):
"""The generation time, in words, with the zone when there is one.
A naive `datetime` is whatever clock the machine that ran this had, and
this line does not claim otherwise — stamping "UTC" on a value that might
be a laptop's local time is worse than saying nothing, because it invites
somebody to subtract an offset from it.
"""
if not isinstance(moment, (datetime.datetime, datetime.date)):
return f"Generated {_text(moment, 'at an unrecorded time')}"
stamp = (f"{MONTH_NAMES[moment.month - 1]} {moment.day}, {moment.year}")
if isinstance(moment, datetime.datetime):
stamp += f" at {moment.hour:02d}:{moment.minute:02d}"
zone = moment.tzname()
if zone:
stamp += f" {_text(zone)}"
return f"Generated {stamp}"
def _table(headers, rows, *, numeric_from=None):
"""A table. Cell values arrive ALREADY ESCAPED, by `_text` and friends.
That is the one convention in this module worth stating out loud, because
it is the inversion of the safer default. It exists so a cell can hold
` ` or `—` from `_money` and `_count` without those being
escaped into visible ampersands — every producer of a cell in this file
goes through `_text`, `_count`, `_money`, `_month` or `_retention`, and
each of those escapes what it was given. If you add a column, route its
value through one of them; do not interpolate a raw value here.
`numeric_from` right-aligns every column from that index on, which is
where the numbers live in all three tables that pass it.
"""
first_number = len(headers) if numeric_from is None else numeric_from
def cls(index):
return " class=\"num\"" if index >= first_number else ""
out = ["<table>", "<thead>", "<tr>"]
out += [f"<th{cls(i)}>{_text(name)}</th>"
for i, name in enumerate(headers)]
out += ["</tr>", "</thead>", "<tbody>"]
for row in rows:
out.append("<tr>")
out += [f"<td{cls(i)}>{cell}</td>" for i, cell in enumerate(row)]
out.append("</tr>")
out += ["</tbody>", "</table>"]
return out
How to get a test key, what each section of the report means, and the field behind every conclusion.
# Club Pulse — a SipsPOS public API reference application
A read-only wine-club retention dashboard in three files of standard-library
Python. Run it and it writes one self-contained HTML page: who is about to
leave your club, who already has, how each month's joiners have held up, and
what each tier is worth per month.
It exists to be copied. If you are integrating with the SipsPOS API, the file
you want is `sipspos.py` — bearer auth, cursor pagination, the error envelope,
`X-Sipspos-Environment` and 429 backoff, with nothing about wine clubs in it.
The other two show what using it looks like.
| File | What it is |
|---|---|
| `sipspos.py` | The client. **Copy this one.** Knows nothing about clubs. |
| `pulse.py` | This application: `fetch` (four GETs), `analyse` (pure), `main` (the CLI). |
| `report.py` | `render_html(findings, …)` — the findings dict as one HTML page. |
No dependencies. `import requests` never appears; SipsPOS itself depends on
`requests`, and a file that can only be copied along with a dependency list is
not really copyable. Python 3 and `urllib` are the whole toolchain.
## Get a test key first
1. Sign in to your winery and go to **Admin → API keys**.
2. Press **Build my sandbox**. That creates a *sandbox twin* — a second
winery, seeded with demo data, that runs the same code as the real one.
3. Press **Create test key** and tick **`customers:read`** and
**`sales:read`**. Those are the two this example needs. Tick
**`orders:read`** as well if the winery has the online store — see below.
Every request this example makes is a GET.
4. Copy the `sk_test_…` key when it is shown. It is not shown twice.
**`orders:read` is optional, and that is the interesting part.** It maps to
the `online_store` capability, which is on Estate and not on Growth — and a
sandbox twin deliberately inherits its parent's plan, so a Growth winery
*cannot* issue a key carrying it, however much it would like to.
An integration that works across plans has to cope with that rather than treat
it as a crash, which is the most common reason a partner's client works at one
winery and not the next. So `fetch()` attempts the orders read, accepts a
`permission_denied` or `feature_not_enabled` refusal as a fact about the plan,
and the report says on its own masthead that the shop was not read. **Any
other error is re-raised** — a bad token or an API outage must not be
swallowed into a quietly wrong report.
**Why sales and not just orders.** The *quiet member* check asks whether a
member has bought anything recently, and for a wine club the answer usually
happens at the cellar door — which is a **sale**, not an order. Reading only
`/v1/orders` marks a member who bought a case in the tasting room last week as
quiet, and tells the winery to chase the customers it should be thanking. So
`/v1/sales` is the primary signal and the shop is additional. `sales:read` is
core to the register on every plan and needs no capability, which is why the
example still works in full on Growth apart from the shop half of that one
check.
`paid_since` filters a nullable column: an open tab has a NULL `paid_at` and
is excluded. That is right here — an unpaid sale is not evidence that somebody
bought something.
A `sk_test_` key reaches the twin and cannot reach live data — the two are
different tenants, so isolation does not depend on anything this script does.
Every response says which one answered, in `X-Sipspos-Environment`; the client
exposes it as `Client.environment`, and when it is `"test"` the report opens
with a banner saying the numbers are invented.
If the key is missing a **required** scope, the API answers `403` and the
client raises `ApiError` with `type == "permission_denied"`. `pulse.py` prints
that message and a line naming the scopes it needs, then exits `1`.
## Run it
```sh
export SIPSPOS_BASE_URL=https://your-winery.sipspos.com # your winery's own host
export SIPSPOS_API_KEY=sk_test_…
python pulse.py
```
`SIPSPOS_BASE_URL` is the winery's own origin — SipsPOS is addressed per
winery, not through one shared `api.` domain. If either variable is unset,
the script names the missing one and exits `2` without making a request.
```
--out FILE where to write the report (default: club-pulse.html)
--quiet-days N days without an order before a member counts as quiet,
and how far back orders are read (default: 120)
```
On success it writes the file, prints a one-line summary with the number of
API requests it took, and exits `0`. On an API failure it prints the
`ApiError` — type, message, HTTP status and `request_id` — to stderr and exits
non-zero. It never prints a traceback: a traceback is a bug report about this
script, and what you need is a sentence about your problem with the id support
can grep for.
The rendered page never contains the key. `render_html` is handed the findings,
the host name, the environment and a timestamp — there is no parameter the
token could arrive in.
## What the four requests are
```
GET /api/v1/club/members everyone in the club funnel
GET /api/v1/customers?club_status=none everyone who is not in it
GET /api/v1/club/tiers tier names and prices
GET /api/v1/orders?created_since=YYYY-MM-DD the order window
```
The first, second and fourth are cursor-paginated, and `Client.paginate`
follows `next_cursor` until `has_more` is false, re-sending your filters on
every page. That last part is not optional: a SipsPOS cursor is signed against
the filters it was minted under, and a page-two request that drops
`club_status=none` is refused with a `400`.
## What each section of the report means
**Needs attention.** Every member appears **exactly once**, under the first
reason that fits, in this fixed order:
| Reason | Derived from |
|---|---|
| `past_due` | `club_status == "past_due"` |
| `no_card` | `club_status == "active"` and `club_card_last4` is empty |
| `paused` | `club_paused` (with `club_resume_on` when it is set) |
| `skipping` | `club_skip_next` |
| `quiet` | an active member with no order in the window, joined on `customer_id` |
A member whose card is failing *and* missing appears once, as `past_due` — you
cannot fix the second problem without fixing the first. `no_card` is the one
worth knowing about: a member migrated from another system arrives active, with
a real join date and a real tier and no card, and looks perfectly healthy in
every other report until the next charge fails.
**Cohorts.** Joiners grouped by `club_since[:7]`, with how many of that month's
joiners still have `is_club_member` set. Both the current members and the
cancelled ones are counted, or every cohort would look like it retained
everybody. Customers with no `club_since` are in no cohort.
**Tier mix.** Members per `club_tier_id`, joined to `/v1/club/tiers`. A member
whose tier is not in that list — deactivated, or otherwise absent — is counted
under **Unknown tier**, never dropped, so the tier table always adds up to the
club. Money at this API is `{"amount_cents", "currency"}` or `null`, never a
bare number; a tier with a `null` price contributes zero rather than a guess.
`tier["price"]` is **one charge, not one month**, so a `quarterly` tier is
divided by three and an `annual` one by twelve before it is called a monthly
value.
**Lost.** Customers with `club_status == "none"` who still have a `club_since`.
This is the only way to see churn through this API: `/v1/club/members` is
`Customer` narrowed to `club_status != "none"`, so a member is gone from it the
moment they cancel. Cancelling clears `club_status`, `club_tier_id` and
`is_club_member` but leaves `club_since` alone, and that surviving date is what
separates somebody who left from somebody who never joined.
## `analyse` is pure, and that is the part to copy the thinking from
```python
findings = pulse.analyse(members, lapsed, tiers, orders,
today=datetime.date(2026, 8, 25), quiet_days=120)
```
No network, no clock, no client — `today` is an argument. The clock is read in
exactly one place in `pulse.py`, in `main`. That is what lets the analysis be
tested with dictionaries typed by hand (`scripts/club_pulse_test.py` does
exactly that, with no HTTP at all), and what lets you ask what the report would
have said on any other date.
## The gap this example found, and what came of it
**There was no way to write back *"we contacted this member"*.**
Club Pulse produces a call list, and the natural next step is marking the rows
somebody has worked through. When this example was written it could not be
done: `CUSTOMER_FIELDS` in `services/api_writes.py` — the whole writable
surface of a customer — had no way to record either a label or a contact. So
"who have we called" lived outside the report, and re-running it tomorrow
produced the same list minus whoever happened to fix their card.
**It can be done now, and finding it this way is the argument for having a
reference application at all.** Two writes closed it:
```sh
# Mark them, using the winery's own CRM vocabulary.
PATCH /v1/customers/{id} {"tags": ["VIP", "Contacted"]}
# Or record what you actually did, onto the winery's customer timeline.
POST /v1/customers/{id}/notes
{"note": "Called about the failed charge; card updated.",
"channel": "call"}
```
Both need `customers:write`. Two things about them are worth knowing before
you build on them, because both are deliberate and neither is guessable:
**Setting `tags` replaces the whole set.** `{"tags": ["Contacted"]}` on
somebody who was `["VIP"]` leaves them with `Contacted` alone. Read the
customer first and send the list they should end up with. The failure here is
silent, which is why it is said twice.
**A tag this winery has not created is refused**, and the error names the ones
that exist. Tags are not free-text labels: a winery segments an allocated wine
release by tag, so a typo would become a permanent segment matching nobody.
The vocabulary belongs to the winery; this API attaches and detaches.
**The notes endpoint records a `note` or a `call`, and nothing else.** It
cannot write an `email` or `sms` entry, because such a row asserts that a
message was *sent* — and that timeline is what a winery consults when arguing
about whether anybody was contacted. If you sent your own email, log a note
saying what you sent.
`POST …/notes` needs an `Idempotency-Key`; it creates a row, so a retry after
a timeout would otherwise claim a second call. `PATCH` needs none — assigning
the same tags twice is already the same outcome.
**Still not there:** reading the winery's timeline back
(`GET /v1/customers/{id}/communications`). An integration cannot yet see that
the winery emailed somebody yesterday, which is the real way a retention
workflow double-contacts a member. Recorded here the same way the last gap
was.
Fetch all of it:
curl -O https://sipspos.com/developers/examples/club-pulse/sipspos.py
curl -O https://sipspos.com/developers/examples/club-pulse/pulse.py
curl -O https://sipspos.com/developers/examples/club-pulse/report.py
curl -O https://sipspos.com/developers/examples/club-pulse/README.md