dodil/erp-brick-timesheets

DODIL ERP brick: timesheets

src/brick_timesheets/infrastructure/odoo.py

"""The Odoo reader port (migration.odoo, TS-22): 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
(hr_timesheet's account.analytic.line, sale_timesheet's so_line and timesheet_invoice_id).

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

LINE_FIELDS = ["id", "name", "date", "unit_amount", "amount", "employee_id", "project_id", "task_id", "validated", "validated_status", "so_line",
               "timesheet_invoice_id", "company_id"]


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 company_currency(self, db: str) -> str: ...                              # the company's currency code (amounts are in it)
    def timesheet_lines(self, db: str, years: list) -> list: ...                 # account.analytic.line timesheets (project set) dated 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 company_currency(self, db: str) -> str:
        got = self._call(db, "res.company", "search_read", [], fields=["currency_id"], limit=1)
        return str((got[0]["currency_id"] or [0, ""])[1]) if got else ""

    def timesheet_lines(self, db: str, years: list) -> list:
        dom = [["project_id", "!=", False], ["date", ">=", f"{min(years)}-01-01"], ["date", "<=", f"{max(years)}-12-31"]]
        rows = self._call(db, "account.analytic.line", "search_read", dom, fields=LINE_FIELDS, context={"active_test": False})
        return [r for r in rows if int(str(r.get("date") or "0000")[:4]) in years]


ports = Ports()