"""Read-only Odoo client (CONTRACT §6). STRICTLY READ ONLY. This module issues only ``search_read`` / ``search`` / ``read`` calls — there is deliberately no ``create``, ``write`` or ``unlink`` path anywhere. ``fetch_software_invoices`` pulls posted customer invoices for a month, filtered to "Software" per ``ODOO_SOFTWARE_FILTER``; ``fetch_customers`` lists ``res.partner`` customers for name resolution. Two API styles (CONTRACT open question 2): * ``jsonrpc`` — native Odoo ``/jsonrpc`` (implemented here). * ``oca_rest`` — OCA REST module (clear stub, raises NotImplementedError). """ from __future__ import annotations import calendar from abc import ABC, abstractmethod from datetime import date from typing import Any, Optional import httpx from app.matching.models import Invoice # Read-only JSON-RPC methods we are permitted to call. Any mutation method is # intentionally absent and asserted against before every call. _ALLOWED_METHODS = frozenset({"search_read", "search", "read", "fields_get", "search_count"}) class OdooConfigError(RuntimeError): """Raised when required Odoo credentials/config are missing.""" class OdooClient(ABC): """Abstract read-only Odoo interface used by the runner.""" @abstractmethod def fetch_software_invoices(self, year: int, month: int) -> list[Invoice]: """Return posted Software customer invoices for the given month.""" @abstractmethod def fetch_customers(self) -> list[dict[str, Any]]: """Return the Odoo customer directory (id, name, ...).""" @abstractmethod def check_connection(self, year: int, month: int) -> dict[str, Any]: """Authenticate and run a tiny read, returning a diagnostic summary. Credential-free callers use this as a one-shot self-test once real Odoo credentials are supplied — it proves auth works and that the Software filter resolves to a sensible invoice count for the month. """ def _month_bounds(year: int, month: int) -> tuple[str, str]: last_day = calendar.monthrange(year, month)[1] return f"{year}-{month:02d}-01", f"{year}-{month:02d}-{last_day:02d}" def _parse_software_filter(raw: str) -> tuple[str, str]: """Split ``kind:value`` (journal/category/analytic) from ODOO_SOFTWARE_FILTER.""" if ":" in raw: kind, value = raw.split(":", 1) return kind.strip().lower(), value.strip() return "journal", raw.strip() class JsonRpcOdooClient(OdooClient): """Native Odoo JSON-RPC client (``/jsonrpc`` ``object.execute_kw``).""" def __init__( self, url: str, db: str, username: str, api_key: str, software_filter: str = "journal:Software", timeout: float = 30.0, ) -> None: if not all([url, db, username, api_key]): raise OdooConfigError( "Odoo requires ODOO_URL, ODOO_DB, ODOO_USERNAME and ODOO_API_KEY." ) self._url = url.rstrip("/") self._db = db self._username = username self._api_key = api_key self._filter_kind, self._filter_value = _parse_software_filter(software_filter) self._client = httpx.Client(timeout=timeout) self._uid: Optional[int] = None # -- low-level JSON-RPC ------------------------------------------------- def _rpc(self, service: str, method: str, args: list[Any]) -> Any: payload = { "jsonrpc": "2.0", "method": "call", "params": {"service": service, "method": method, "args": args}, "id": 1, } resp = self._client.post(f"{self._url}/jsonrpc", json=payload) resp.raise_for_status() data = resp.json() if "error" in data: raise RuntimeError(f"Odoo JSON-RPC error: {data['error']}") return data.get("result") def _login(self) -> int: if self._uid is None: uid = self._rpc("common", "login", [self._db, self._username, self._api_key]) if not uid: raise OdooConfigError("Odoo authentication failed — check credentials.") self._uid = int(uid) return self._uid def _execute(self, model: str, method: str, *params: Any) -> Any: if method not in _ALLOWED_METHODS: # Hard guard: this client is read-only by construction. raise PermissionError(f"Refusing non-read Odoo method: {method!r}") uid = self._login() return self._rpc( "object", "execute_kw", [self._db, uid, self._api_key, model, method, *params], ) # -- domain helpers ----------------------------------------------------- def _software_domain(self) -> list[Any]: """Extra domain clause selecting Software invoices per the filter kind.""" if self._filter_kind == "journal": return [["journal_id.name", "=", self._filter_value]] if self._filter_kind == "category": return [["invoice_line_ids.product_id.categ_id.name", "=", self._filter_value]] if self._filter_kind == "analytic": # Analytic tag/account on the invoice lines. return [["invoice_line_ids.analytic_distribution", "ilike", self._filter_value]] raise OdooConfigError(f"Unknown ODOO_SOFTWARE_FILTER kind: {self._filter_kind}") # -- public read-only API ---------------------------------------------- def fetch_software_invoices(self, year: int, month: int) -> list[Invoice]: start, end = _month_bounds(year, month) domain: list[Any] = [ ["move_type", "=", "out_invoice"], ["state", "=", "posted"], ["invoice_date", ">=", start], ["invoice_date", "<=", end], *self._software_domain(), ] fields = [ "id", "name", "partner_id", "invoice_date", "amount_untaxed", "amount_total", "is_move_sent", ] rows = self._execute( "account.move", "search_read", [domain], {"fields": fields} ) invoices: list[Invoice] = [] for row in rows or []: partner = row.get("partner_id") or [None, ""] partner_id = partner[0] if isinstance(partner, (list, tuple)) else None partner_name = partner[1] if isinstance(partner, (list, tuple)) else str(partner) invoices.append( Invoice( odoo_id=int(row["id"]), number=row.get("name") or "", customer=partner_name or "", odoo_customer_id=int(partner_id) if partner_id else None, invoice_date=date.fromisoformat(str(row["invoice_date"])[:10]), amount_untaxed=float(row.get("amount_untaxed") or 0.0), amount_total=float(row.get("amount_total") or 0.0), sent=bool(row.get("is_move_sent")), ) ) return invoices def fetch_customers(self) -> list[dict[str, Any]]: domain = [["customer_rank", ">", 0]] rows = self._execute( "res.partner", "search_read", [domain], {"fields": ["id", "name", "vat", "email"]}, ) return list(rows or []) def check_connection(self, year: int, month: int) -> dict[str, Any]: uid = self._login() start, end = _month_bounds(year, month) base_domain: list[Any] = [ ["move_type", "=", "out_invoice"], ["state", "=", "posted"], ["invoice_date", ">=", start], ["invoice_date", "<=", end], ] posted = self._execute("account.move", "search_count", [base_domain]) software = self._execute( "account.move", "search_count", [[*base_domain, *self._software_domain()]] ) return { "authenticated": True, "uid": uid, "url": self._url, "db": self._db, "month": f"{year}-{month:02d}", "filter_kind": self._filter_kind, "filter_value": self._filter_value, "posted_out_invoices": int(posted or 0), "software_invoices": int(software or 0), } def close(self) -> None: self._client.close() class OcaRestOdooClient(OdooClient): """Stub for the OCA REST API flavour (CONTRACT open question 2). Left as a clear branch to implement if the client runs the OCA ``base_rest`` modules instead of native JSON-RPC. Read-only endpoints only when built. """ def __init__(self, *_: Any, **__: Any) -> None: self._err = NotImplementedError( "ODOO_API_STYLE=oca_rest is not implemented yet. Use jsonrpc, or " "implement the OCA base_rest read-only endpoints here." ) def fetch_software_invoices(self, year: int, month: int) -> list[Invoice]: raise self._err def fetch_customers(self) -> list[dict[str, Any]]: raise self._err def check_connection(self, year: int, month: int) -> dict[str, Any]: raise self._err def build_odoo_client(settings: "Any") -> OdooClient: """Factory selecting the client implementation from settings (ODOO_API_STYLE).""" if settings.odoo_api_style == "oca_rest": return OcaRestOdooClient() return JsonRpcOdooClient( url=settings.odoo_url or "", db=settings.odoo_db or "", username=settings.odoo_username or "", api_key=settings.odoo_api_key or "", software_filter=settings.odoo_software_filter, ) if __name__ == "__main__": # One-shot read-only connection self-test. Once real Odoo credentials are in # the environment (ODOO_URL/ODOO_DB/ODOO_USERNAME/ODOO_API_KEY), run: # python -m app.odoo.client --check --month 2026-10 import argparse import json as _json from datetime import date as _date parser = argparse.ArgumentParser(description="Read-only Odoo connection self-check") parser.add_argument("--check", action="store_true", help="authenticate and run a tiny read") parser.add_argument("--month", default=_date.today().strftime("%Y-%m"), help="YYYY-MM") args = parser.parse_args() from app.config import get_settings settings = get_settings() year_s, month_s = args.month.split("-") client = None try: client = build_odoo_client(settings) summary = client.check_connection(int(year_s), int(month_s)) print(_json.dumps(summary, indent=2)) except Exception as exc: # noqa: BLE001 — surface any failure cleanly to the operator print(_json.dumps( {"authenticated": False, "error": str(exc), "type": type(exc).__name__}, indent=2, )) raise SystemExit(1) finally: closer = getattr(client, "close", None) if callable(closer): closer()