spb/air Public MIT
AIR — The Language of Accounting.
Python 100%
1# =============================================================================2# Projet : AIR — Accounting Intermediate Representation3# Auteur : Simon-Pierre Boucher4# Contact : contact@spboucher.ai5# Fichier : transport.py6# Description : QBO transports — offline mock (default, no QuickBooks needed) and real HTTP.7# =============================================================================8"""Transports for the QuickBooks backend.910AIR must build and test WITHOUT any QuickBooks access, so the transport is an11interface with two implementations:1213- MockQboTransport (default): a faithful in-memory simulation of the QBO14 behaviors that matter (per docs/research/erp-apis.md): `requestid`15 idempotency replay, duplicate-DocNumber error 6140, server-assigned Ids.16 All tests run against this — no network, no account, ever.17- HttpQboTransport: the real thing (stdlib urllib, OAuth2 bearer token,18 minorversion=75, exact-decimal JSON serialization). It is only constructed19 when the user explicitly provides credentials; nothing in the test suite20 or default CLI paths touches it.21"""22from __future__ import annotations2324import abc25import json26import re27import urllib.request28from decimal import Decimal29from typing import Any3031QBO_MINOR_VERSION = "75" # minorversion baseline since 2025-08-01 (research)323334class QboError(RuntimeError):35 def __init__(self, code: str, message: str):36 self.code = code37 super().__init__(f"QBO error {code}: {message}")383940class QboTransport(abc.ABC):41 """Minimal surface the backend needs: create a JournalEntry."""4243 @abc.abstractmethod44 def create_journal_entry(self, body: dict[str, Any], request_id: str) -> dict[str, Any]:45 """POST a JournalEntry; MUST be idempotent on request_id. Returns the46 created (or replayed) entity, including its server 'Id'."""474849class MockQboTransport(QboTransport):50 """Offline QBO simulator — the default transport.5152 Simulated behaviors:53 - requestid replay: same request_id returns the original entity, creates54 nothing (QBO guarantees this for supported entities);55 - duplicate DocNumber with a NEW request_id raises error 6140;56 - server-assigned incremental Ids and SyncToken 0.57 """5859 def __init__(self) -> None:60 self.store: dict[str, dict[str, Any]] = {} # Id -> entity61 self._by_request: dict[str, str] = {} # request_id -> Id62 self._by_docnumber: dict[str, str] = {} # DocNumber -> Id63 self._next_id = 16465 def create_journal_entry(self, body: dict[str, Any], request_id: str) -> dict[str, Any]:66 if request_id in self._by_request: # idempotent replay67 return self.store[self._by_request[request_id]]68 doc = str(body.get("DocNumber", ""))69 if doc and doc in self._by_docnumber:70 raise QboError(71 "6140",72 f"Duplicate Document Number Error: DocNumber '{doc}' already exists",73 )74 debits = sum(75 Decimal(str(l["Amount"])) for l in body["Line"]76 if l["JournalEntryLineDetail"]["PostingType"] == "Debit"77 )78 credits = sum(79 Decimal(str(l["Amount"])) for l in body["Line"]80 if l["JournalEntryLineDetail"]["PostingType"] == "Credit"81 )82 if debits != credits:83 raise QboError("6000", f"Journal entry must balance: {debits} != {credits}")8485 entity = dict(body)86 entity["Id"] = str(self._next_id)87 entity["SyncToken"] = "0"88 self._next_id += 189 self.store[entity["Id"]] = entity90 self._by_request[request_id] = entity["Id"]91 if doc:92 self._by_docnumber[doc] = entity["Id"]93 return entity949596_DEC_SENTINEL = re.compile(r'"__DEC__(-?\d+(?:\.\d+)?)__"')979899def _dumps_exact(obj: Any) -> str:100 """JSON with Decimals emitted as exact numeric literals (never float).101102 Decimals are encoded as strict sentinel strings, then unquoted. The103 sentinel pattern only matches the exact canonical form generated here,104 so ordinary string values can never be corrupted.105 """106 def encode(o: Any) -> Any:107 if isinstance(o, Decimal):108 return f"__DEC__{o}__"109 raise TypeError(type(o).__name__)110111 return _DEC_SENTINEL.sub(r"\1", json.dumps(obj, default=encode))112113114class HttpQboTransport(QboTransport):115 """Real QuickBooks Online transport. OPTIONAL — requires explicit116 credentials; never used by tests or default flows.117118 Note: token refresh is the caller's concern for now (Phase 3 scope);119 pass a valid OAuth2 access token.120 """121122 SANDBOX_BASE = "https://sandbox-quickbooks.api.intuit.com"123 PRODUCTION_BASE = "https://quickbooks.api.intuit.com"124125 def __init__(self, realm_id: str, access_token: str, *, sandbox: bool = True):126 self.realm_id = realm_id127 self.access_token = access_token128 self.base = self.SANDBOX_BASE if sandbox else self.PRODUCTION_BASE129130 def create_journal_entry(self, body: dict[str, Any], request_id: str) -> dict[str, Any]:131 url = (132 f"{self.base}/v3/company/{self.realm_id}/journalentry"133 f"?minorversion={QBO_MINOR_VERSION}&requestid={request_id}"134 )135 request = urllib.request.Request(136 url,137 data=_dumps_exact(body).encode("utf-8"),138 headers={139 "Authorization": f"Bearer {self.access_token}",140 "Content-Type": "application/json",141 "Accept": "application/json",142 },143 method="POST",144 )145 with urllib.request.urlopen(request) as response: # pragma: no cover146 payload = json.loads(response.read().decode("utf-8"))147 return payload.get("JournalEntry", payload) # pragma: no cover148