dodil/erp-brick-time-off

DODIL ERP brick: time off

src/brick_time_off/infrastructure/odoo.py

"""The Odoo reader port (TOF-23): read-only access to an Odoo 17 database behind one small interface, so the migration never knows how Odoo is
reached and tests hand it fake rows (no live Odoo). `ports.reader` is what the use cases use; the default is None, which refuses with a plain message
until a composition wires one - `XmlRpcReader` reads Odoo's external API (search_read only, nothing written) when ODOO_URL is set.
NOT verified against a live Odoo here: the mapping is verified by the fake rows (tests/test_migration.py), the reader's field lists by Odoo 17's models -
in particular `number_of_hours` (Odoo 17 names it number_of_hours_display on some builds: the loader falls back to days x the contracted day).

Rows are Odoo's own: many2one fields come as [id, name], dates as ISO strings, quantities as numbers."""
import os
from typing import Optional, Protocol

LEAVE_FIELDS = ["id", "employee_id", "holiday_status_id", "request_date_from", "request_date_to", "request_unit_half", "request_date_from_period",
                "request_unit_hours", "number_of_days", "number_of_hours", "state"]
ALLOCATION_FIELDS = ["id", "employee_id", "holiday_status_id", "number_of_days", "date_from", "date_to", "state"]
TYPE_FIELDS = ["id", "name", "requires_allocation", "leave_validation_type", "time_type", "request_unit"]


class OdooUnavailable(Exception):
    """No Odoo connection (409) or the call failed (502): a reply, never a half-loaded run."""

    def __init__(self, http: int, reason: str):
        super().__init__(reason)
        self.http, self.reason = http, reason


class OdooReader(Protocol):
    def leave_types(self, db: str) -> list: ...                                  # hr.leave.type, all active
    def leaves(self, db: str, years: list) -> list: ...                          # hr.leave in state validate whose request_date_from is in those years
    def allocations(self, db: str, years: list) -> list: ...                     # hr.leave.allocation in state validate whose date_from is in those years
    def waiting(self, db: str, years: list) -> list: ...                         # hr.leave in states confirm and validate1 (waiting for approval) in those years


class Ports:
    def __init__(self):
        self.reader: Optional[OdooReader] = None

    def get(self) -> OdooReader:
        if self.reader is not None:
            return self.reader
        if os.environ.get("ODOO_URL"):
            return XmlRpcReader()
        raise OdooUnavailable(409, "no Odoo connection is configured (ODOO_URL, ODOO_USER, ODOO_PASSWORD): the migration reads Odoo read-only through a reader "
                                   "the composition wires")


class XmlRpcReader:
    """Odoo's external API over XML-RPC, read-only."""

    def __init__(self):
        import xmlrpc.client
        self.url = os.environ["ODOO_URL"].rstrip("/")
        self.user, self.password = os.environ.get("ODOO_USER", ""), os.environ.get("ODOO_PASSWORD", "")
        self.common = xmlrpc.client.ServerProxy(f"{self.url}/xmlrpc/2/common")
        self.models = xmlrpc.client.ServerProxy(f"{self.url}/xmlrpc/2/object")
        self._uid: dict = {}

    def _call(self, db: str, model: str, method: str, *args, **kw):
        try:
            if db not in self._uid:
                self._uid[db] = self.common.authenticate(db, self.user, self.password, {})
            return self.models.execute_kw(db, self._uid[db], self.password, model, method, list(args), kw)
        except OSError as e:
            raise OdooUnavailable(502, f"Odoo did not answer ({e.__class__.__name__})") from None

    def leave_types(self, db: str) -> list:
        return self._call(db, "hr.leave.type", "search_read", [], fields=TYPE_FIELDS)

    def _in_years(self, rows: list, field: str, years: list) -> list:
        return [r for r in rows if int(str(r.get(field) or "0000")[:4]) in years]

    def leaves(self, db: str, years: list) -> list:
        rows = self._call(db, "hr.leave", "search_read", [["state", "=", "validate"]], fields=LEAVE_FIELDS, context={"active_test": False})
        return self._in_years(rows, "request_date_from", years)

    def allocations(self, db: str, years: list) -> list:
        rows = self._call(db, "hr.leave.allocation", "search_read", [["state", "=", "validate"]], fields=ALLOCATION_FIELDS, context={"active_test": False})
        return self._in_years(rows, "date_from", years)

    def waiting(self, db: str, years: list) -> list:
        rows = self._call(db, "hr.leave", "search_read", [["state", "in", ["confirm", "validate1"]]], fields=LEAVE_FIELDS, context={"active_test": False})
        return self._in_years(rows, "request_date_from", years)


ports = Ports()