dodil/erp-brick-tax

DODIL ERP brick: tax

src/brick_tax/infrastructure/odoo.py

"""The reader port for the one-off Odoo migration (module.yaml migration.odoo). READ-ONLY by construction: the port has no
write method, so history can never be changed through it. The reader returns plain rows: a many2one as its id (or None), a
one2many/many2many as a list of ids, a boolean False as None where the field is a reference.

`reader` is what the migration use case calls. Tests (and a composition that migrates from a file export) replace it with
an in-memory fake; a deployed migration job configures ODOO_URL / ODOO_DB / ODOO_USER / ODOO_PASSWORD for the XML-RPC reader.
"""
import os
import xmlrpc.client
from typing import Optional, Protocol


class OdooUnavailable(Exception):
    """No reader is configured, or Odoo did not answer. The message says which."""


class OdooReader(Protocol):
    def read(self, model: str, fields: list, where: Optional[dict] = None, include_archived: bool = False) -> list:
        """Rows of `model` with `fields`. `where` is {field: value} (equality) or {field: ("in", [values])}; a reader may
        ignore it and return more, because the migration filters again. `include_archived`: archived rows too."""


class UnconfiguredReader:
    def read(self, model, fields, where=None, include_archived=False):
        raise OdooUnavailable("no Odoo reader is configured: set ODOO_URL, ODOO_DB, ODOO_USER and ODOO_PASSWORD, or hand the "
                              "migration a file export through a composition's reader")


class XmlRpcReader:
    """Odoo's external API (execute_kw search_read). Only read calls are ever made."""

    def __init__(self, url: str, db: str, user: str, password: str):
        self.url, self.db, self.user, self.password, self._uid = url.rstrip("/"), db, user, password, None

    def _uid_(self) -> int:
        if self._uid is None:
            self._uid = xmlrpc.client.ServerProxy(f"{self.url}/xmlrpc/2/common").authenticate(self.db, self.user, self.password, {})
            if not self._uid:
                raise OdooUnavailable("Odoo refused the migration user's login")
        return self._uid

    @staticmethod
    def _domain(where: Optional[dict]) -> list:
        out = []
        for field, cond in (where or {}).items():
            out.append((field, cond[0], cond[1]) if isinstance(cond, tuple) else (field, "=", cond))
        return out

    @staticmethod
    def _flat(row: dict) -> dict:
        return {k: (v[0] if isinstance(v, list) and len(v) == 2 and isinstance(v[0], int) and isinstance(v[1], str) else (None if v is False else v))
                for k, v in row.items()}

    def read(self, model, fields, where=None, include_archived=False):
        models = xmlrpc.client.ServerProxy(f"{self.url}/xmlrpc/2/object")
        ctx = {"context": {"active_test": not include_archived}, "fields": list(fields)}
        try:
            rows = models.execute_kw(self.db, self._uid_(), self.password, model, "search_read", [self._domain(where)], ctx)
        except (OSError, xmlrpc.client.Fault, xmlrpc.client.ProtocolError) as e:
            raise OdooUnavailable(f"Odoo did not answer {model}: {e}") from None
        return [self._flat(r) for r in rows]


def from_env() -> OdooReader:
    if os.environ.get("ODOO_URL"):
        return XmlRpcReader(os.environ["ODOO_URL"], os.environ.get("ODOO_DB", ""), os.environ.get("ODOO_USER", ""),
                            os.environ.get("ODOO_PASSWORD", ""))
    return UnconfiguredReader()


reader: OdooReader = from_env()