dodil/erp-brick-inventory

DODIL ERP brick: inventory

src/brick_inventory/infrastructure/odoo.py

"""The two ports of the Odoo migration: the READER of the customer's Odoo and the id MAPS the earlier bricks left behind.

Tests put in-memory fakes here; a composition leaves the real adapters, which read Odoo over JSON-RPC (ODOO_URL, ODOO_DB,
ODOO_USER, ODOO_PASSWORD) and ask product and business-partner for their id maps
(GET /migration-map/{db}/{odoo model}?ids=, shared.yaml conventions.odoo_migration.id_map)."""
import os
from typing import Optional

from brick_inventory.infrastructure.seams import BP, PRODUCT, Missing

READER = None          # an object with read(model, fields, domain=None) -> list[dict]
MAPS = None            # an object with product(odoo_id) and partner(odoo_id)


class OdooJsonRpc:
    """search_read over Odoo's /jsonrpc, authenticated once. Never used in tests."""

    def __init__(self, url: str, db: str, user: str, password: str):
        import httpx
        self.url, self.db, self.password, self.http = url.rstrip("/"), db, password, httpx.Client(timeout=60)
        self.uid = self._call("common", "login", db, user, password)

    def _call(self, service: str, method: str, *args):
        r = self.http.post(f"{self.url}/jsonrpc", json={"jsonrpc": "2.0", "method": "call", "params": {"service": service, "method": method, "args": args}})
        r.raise_for_status()
        body = r.json()
        if body.get("error"):
            raise Missing("odoo", body["error"].get("message", "odoo refused"), r.status_code)
        return body["result"]

    def read(self, model: str, fields: list, domain: Optional[list] = None) -> list:
        return self._call("object", "execute_kw", self.db, self.uid, self.password, model, "search_read", [domain or []], {"fields": fields})

    @classmethod
    def from_env(cls):
        try:
            return cls(os.environ["ODOO_URL"], os.environ["ODOO_DB"], os.environ["ODOO_USER"], os.environ["ODOO_PASSWORD"])
        except KeyError as k:
            raise Missing("odoo", f"set {k.args[0]} so the migration can read Odoo", 0, installed=False) from None


class MapsOverSeams:
    def __init__(self, db: str):
        self.db = db

    def _ours(self, seam, model: str, odoo_id: int):
        r = seam.get(f"/migration-map/{self.db}/{model}", params={"ids": str(int(odoo_id))}, actor="inventory")
        if not r.ok:
            raise Missing("the id map", r.reason, r.status)
        return next((x.get("ours") for x in r.data.get("body") or [] if str(x.get("odoo_id")) == str(int(odoo_id))), None)

    def product(self, odoo_id: int):
        return self._ours(PRODUCT, "product.product", odoo_id)

    def partner(self, odoo_id: int):
        return self._ours(BP, "res.partner", odoo_id)


def reader():
    return READER or OdooJsonRpc.from_env()


def maps(db: str):
    return MAPS or MapsOverSeams(db)