"""Small standard-library EBM client example. No automatic fiscal retries. Persist the body and request key before calling. After an uncertain transport failure, use those exact values to recover the result; never replace the order. """ import json from urllib.error import HTTPError from urllib.parse import urlsplit from urllib.request import Request, HTTPRedirectHandler, build_opener class _NoRedirect(HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None class EbmError(Exception): def __init__(self, status, response, message=None, *, retry_after=None): self.status = status self.response = response self.code = response.get("code") if isinstance(response.get("code"), str) else None self.retry_after = retry_after self.retry_after_seconds = None # The service publishes integer delay-seconds. Keep other header forms # available as text without guessing or scheduling a retry automatically. if isinstance(retry_after, str): delay = retry_after.strip() if delay.isascii() and delay.isdecimal() and len(delay) <= 10: seconds = int(delay) if seconds <= 2_147_483_647: self.retry_after_seconds = seconds message = message or response.get("resultMsg") or response.get("message") or "Request failed" super().__init__(f"HTTP {status}: {message}") class EbmClient: def __init__(self, base_url, api_key, timeout=30): url = urlsplit(base_url) if url.scheme != "https" and not (url.scheme == "http" and url.hostname in ("127.0.0.1", "localhost", "::1")): raise ValueError("Use HTTPS, or HTTP only for a loopback fixture") if not url.hostname or url.username or url.password or url.query or url.fragment: raise ValueError("Use the assigned base URL without embedded credentials, query or fragment") self.base_url = base_url.rstrip("/") self.api_key = api_key self.timeout = timeout self._opener = build_opener(_NoRedirect()) def request(self, method, path, body=None, request_key=None, *, if_match=None, if_none_match=None): if not path.startswith("/") or path.startswith("//"): raise ValueError("Path must be relative to the assigned service base URL") headers = {"X-API-Key": self.api_key, "Accept": "application/json"} if if_match is not None and if_none_match is not None: raise ValueError("Use only one catalogue precondition") if if_match is not None: from uuid import UUID if not isinstance(if_match, str) or not (if_match.startswith('"') and if_match.endswith('"')): raise ValueError("If-Match must be one quoted catalogue version") UUID(if_match[1:-1]) headers["If-Match"] = if_match if if_none_match is not None: if if_none_match != "*": raise ValueError("If-None-Match must be *") headers["If-None-Match"] = if_none_match data = None if body is not None: headers["Content-Type"] = "application/json" data = json.dumps(body, ensure_ascii=False, allow_nan=False).encode("utf-8") if request_key is not None: if not 1 <= len(request_key) <= 128 or any(not 33 <= ord(c) <= 126 for c in request_key): raise ValueError("Request key must contain 1–128 visible ASCII characters") headers["X-Idempotency-Key"] = request_key request = Request(self.base_url + path, data=data, headers=headers, method=method) # HTTPError has a response body; transport failures propagate unchanged # so the caller cannot mistake an unknown outcome for a rejected sale. try: response = self._opener.open(request, timeout=self.timeout) except HTTPError as error: response = error with response: status = response.status retry_after = response.headers.get("Retry-After") raw = response.read() try: value = json.loads(raw) except (ValueError, UnicodeDecodeError) as error: raise EbmError(status, {"message": "Non-JSON response; preserve the original request for reconciliation"}, retry_after=retry_after) from error if not isinstance(value, dict): raise EbmError(status, {"message": "Unexpected response shape; preserve the original request"}, retry_after=retry_after) if not 200 <= status < 300 or value.get("success") is False or value.get("resultCd", "000") != "000": raise EbmError(status, value, retry_after=retry_after) return value def issue_invoice(self, body, request_key): from uuid import UUID result = self.request("POST", "/api/portal/fiscal/invoices", body, request_key) try: UUID(result["cisReceiptId"]) if result["resultCd"] != "000" or not isinstance(result["data"], dict): raise ValueError("Incomplete receipt envelope") except (KeyError, ValueError, TypeError, AssertionError) as error: raise EbmError(200, {"message": "Incomplete issuance response; retain the original body and key"}) from error return result def save_item(self, body, *, version=None, create_only=False): """Conditionally save a reviewed item. A stale version raises HTTP 412. Supply a version from iter_items(), or create_only=True for a new code. No automatic replay: after a lost response, read and review the item. """ from uuid import UUID if not isinstance(body, dict) or not isinstance(body.get("itemCd"), str) or not body["itemCd"]: raise ValueError("Conditional saves require an explicit itemCd") if not isinstance(create_only, bool) or (version is not None) == create_only: raise ValueError("Supply exactly one of version or create_only=True") condition = {"if_none_match": "*"} if create_only else {"if_match": json.dumps(str(UUID(str(version))))} result = self.request("POST", "/api/portal/items", body, **condition) try: UUID(result["data"]["id"]) UUID(result["data"]["version"]) except (KeyError, TypeError, ValueError, AttributeError) as error: raise EbmError(200, result, "Incomplete catalogue save response; reload and review the current item") from error return result["data"] def iter_items(self): """Read catalogue pages; concurrent edits do not form a frozen snapshot.""" from urllib.parse import urlencode after = None while True: result = self.request("GET", "/api/portal/items" + ("?" + urlencode({"after": after}) if after is not None else "")) yield from result["data"] following = result.get("next_after") if following is None: return if not isinstance(following, str) or following == after: raise EbmError(200, result, "Catalogue pagination did not advance") after = following def get_item(self, item_id): """Read the current body, version and registration status of one item.""" from uuid import UUID return self.request("GET", f"/api/portal/items/{UUID(item_id)}")["data"] def get_receipt(self, receipt_id): from uuid import UUID return self.request("GET", f"/api/portal/receipts/{UUID(receipt_id)}")["data"] def resolve_inbound(self, inbound_id, body, request_key): """Approve/reject imports or purchases; retain nested fiscal outcomes. Does not retry or clear UNKNOWN outcomes. A successful decision does not establish completion of its separately queued stock-delivery messages. """ from uuid import UUID result = self.request("POST", f"/api/portal/cis/inbound/{UUID(inbound_id)}", body, request_key) data = result.get("data") fiscal = data.get("response") if isinstance(data, dict) else None code = fiscal.get("resultCd") if isinstance(fiscal, dict) else None if not isinstance(code, str) or not code: raise EbmError(200, result, "Incomplete inbound outcome; retain the original input and reconcile") if code != "000": raise EbmError(200, result, f"Inbound fiscal result {code}: {fiscal.get('resultMsg', 'Decision not confirmed')}") return data def generate_report(self, body, request_key): return self.request("POST", "/api/portal/cis/reports", body, request_key)["data"] if __name__ == "__main__": import argparse import os from pathlib import Path parser = argparse.ArgumentParser(description="Submit a persisted managed CIS invoice using environment credentials") parser.add_argument("body", type=Path, help="Original invoice JSON file; keep unchanged for retries") parser.add_argument("--request-key", required=True, help="Persistent order identifier; reuse after a lost response") args = parser.parse_args() client = EbmClient(os.environ["EBM_BASE_URL"], os.environ["EBM_API_KEY"]) result = client.issue_invoice(json.loads(args.body.read_text()), args.request_key) print(json.dumps(result, ensure_ascii=False, indent=2))