"""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"") 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)"