"""SAP Business One Service Layer client, embedded in Odoo. Thin adapter over the B1 Service Layer REST/OData API. Handles session reuse, transparent re-auth on expiry (-5002 / HTTP 401), and retry/backoff on object-locked (-2028). Connection settings come from Odoo system parameters (ir.config_parameter) so no credentials live in source. Config keys (set via Settings > SAP B1, or ir.config_parameter): sapb1.base_url e.g. https://192.168.68.16:50000/b1s/v1 sapb1.company_db e.g. XEN sapb1.username e.g. manager sapb1.password (secret) sapb1.verify_ssl "true"/"false" (self-signed on-prem cert -> false) Write-safety guard (added 2026-07-10, after a stale-cache bug let a test write reach production XEN): sapb1.production_company_db defaults to "XEN" if unset sapb1.confirm_production_writes "true"/"false", defaults to false Any code path that WRITES to B1 (push, absence log) must call check_production_write_allowed() before touching the network. If the configured company_db matches the production one and the confirm flag isn't explicitly set, the write is blocked -- interactively with a clear UserError, or silently-skipped-with-a-log-warning for best-effort background paths that must never block an unrelated Odoo action (like approving a leave). """ import logging import time import requests import urllib3 from odoo import _ from odoo.exceptions import UserError urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) _logger = logging.getLogger(__name__) _SESSION_EXPIRED = -5002 _OBJECT_LOCKED = -2028 _TIMEOUT = 30 _MAX_RETRIES = 3 class B1Error(RuntimeError): def __init__(self, code, message, status=None): self.code = code self.status = status super().__init__(f"[B1 {code}] {message}" if code is not None else message) class B1WriteBlocked(UserError): """Raised when a write to B1 is blocked by the production-write guard.""" def check_production_write_allowed(env, *, soft=False): """Return True if a B1 write may proceed, False/raise otherwise. soft=False (interactive actions, e.g. the "Push to SAP B1" button): raises B1WriteBlocked with a clear, actionable message. soft=True (best-effort background paths, e.g. the leave-approval hook and retry crons): logs a warning and returns False so the caller can skip quietly, matching the existing "B1 outage never blocks Odoo" philosophy -- a safety block should behave the same way. """ p = env["ir.config_parameter"].sudo() company_db = (p.get_param("sapb1.company_db") or "").strip() prod_db = (p.get_param("sapb1.production_company_db") or "XEN").strip() confirmed = p.get_param("sapb1.confirm_production_writes", "false").lower() == "true" if company_db.upper() == prod_db.upper() and not confirmed: msg = _( "SAP B1 writes to the production company '%(prod_db)s' are " "blocked by default. If this is intentional, enable " "'Confirm production writes' in Settings -> SAP B1 Sync first.", prod_db=prod_db, ) if soft: _logger.warning("B1 write blocked (safety guard): %s", msg) return False raise B1WriteBlocked(msg) return True class SAPB1Client: def __init__(self, base_url, company_db, username, password, verify_ssl=False): if not base_url or not password: raise UserError(_( "SAP B1 is not configured. Set sapb1.base_url / company_db / " "username / password in Settings first." )) self.base_url = base_url.rstrip("/") self.company_db = company_db self.username = username self.password = password self.verify_ssl = verify_ssl self._s = requests.Session() self._s.verify = verify_ssl self._logged_in = False @classmethod def from_env(cls, env): p = env["ir.config_parameter"].sudo() return cls( base_url=p.get_param("sapb1.base_url"), company_db=p.get_param("sapb1.company_db"), username=p.get_param("sapb1.username"), password=p.get_param("sapb1.password"), verify_ssl=(p.get_param("sapb1.verify_ssl", "false").lower() == "true"), ) # -- session ------------------------------------------------------------ def login(self): resp = self._s.post( f"{self.base_url}/Login", json={ "CompanyDB": self.company_db, "UserName": self.username, "Password": self.password, }, timeout=_TIMEOUT, ) if resp.status_code != 200: code, msg = _parse_error(resp) raise B1Error(code, f"login failed: {msg}", resp.status_code) self._logged_in = True _logger.info("SAP B1 login OK (company=%s)", self.company_db) def logout(self): if self._logged_in: try: self._s.post(f"{self.base_url}/Logout", timeout=_TIMEOUT) except requests.RequestException: pass self._logged_in = False def __enter__(self): self.login() return self def __exit__(self, *exc): self.logout() # -- request core ------------------------------------------------------- def request(self, method, path, **kwargs): if not self._logged_in: self.login() url = f"{self.base_url}/{path.lstrip('/')}" last = None for attempt in range(1, _MAX_RETRIES + 1): try: resp = self._s.request(method, url, timeout=_TIMEOUT, **kwargs) except requests.RequestException as exc: last = exc _backoff(attempt) continue if resp.status_code in (200, 201, 204): return None if resp.status_code == 204 or not resp.content else resp.json() code, msg = _parse_error(resp) if resp.status_code == 401 or code == _SESSION_EXPIRED: self._logged_in = False self.login() continue if code == _OBJECT_LOCKED: _backoff(attempt) continue raise B1Error(code, msg, resp.status_code) raise B1Error(None, f"request failed after {_MAX_RETRIES} retries: {last}") def get(self, path, params=None): return self.request("GET", path, params=params) def get_all(self, path, params=None): """Follow @odata.nextLink pagination and return the full value list.""" params = dict(params or {}) rows = [] data = self.get(path, params=params) while data: rows.extend(data.get("value", [])) nxt = data.get("odata.nextLink") or data.get("@odata.nextLink") if not nxt: break data = self.get(nxt) return rows def post(self, path, json): return self.request("POST", path, json=json) def patch(self, path, json): return self.request("PATCH", path, json=json) def _parse_error(resp): try: err = resp.json().get("error", {}) code = err.get("code") msg = err.get("message", {}) if isinstance(msg, dict): msg = msg.get("value", "") return code, msg or resp.reason except (ValueError, AttributeError): return None, (resp.text[:200] if resp.text else resp.reason) def _backoff(attempt): time.sleep(min(2 ** attempt * 0.25, 5.0))