dodil/erp-brick-expenses

DODIL ERP brick: expenses

src/brick_expenses/infrastructure/odoo.py

"""The Odoo reader port (migration.odoo, EXP-24): 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.

Rows are Odoo's own: many2one fields come as [id, name], dates as ISO strings, amounts as numbers; attachments carry `datas` (base64)."""
import os
from typing import Optional, Protocol

SHEET_FIELDS = ["id", "name", "employee_id", "company_id", "state", "payment_state", "user_id", "total_amount", "currency_id", "accounting_date", "approval_date"]
EXPENSE_FIELDS = ["id", "name", "sheet_id", "employee_id", "product_id", "date", "total_amount_currency", "currency_id", "total_amount", "tax_ids", "payment_mode",
                  "description", "state"]
PRODUCT_FIELDS = ["id", "default_code", "name"]


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 sheets(self, db: str, years: list) -> list: ...                          # hr.expense.sheet in states approve, post, done dated in those years
    def expenses(self, db: str, sheet_ids: list) -> list: ...                    # hr.expense of those sheets
    def products(self, db: str) -> list: ...                                     # product.product with can_be_expensed
    def attachments(self, db: str, expense_ids: list) -> list: ...               # ir.attachment of those expenses, with datas
    def attachment_count(self, db: str, expense_ids: list) -> int: ...


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 sheets(self, db: str, years: list) -> list:
        dom = [["state", "in", ["approve", "post", "done"]]]
        rows = self._call(db, "hr.expense.sheet", "search_read", dom, fields=SHEET_FIELDS, context={"active_test": False})
        return [r for r in rows if int(str(r.get("accounting_date") or r.get("approval_date") or "0000")[:4]) in years]

    def expenses(self, db: str, sheet_ids: list) -> list:
        return self._call(db, "hr.expense", "search_read", [["sheet_id", "in", sheet_ids]], fields=EXPENSE_FIELDS) if sheet_ids else []

    def products(self, db: str) -> list:
        return self._call(db, "product.product", "search_read", [["can_be_expensed", "=", True]], fields=PRODUCT_FIELDS, context={"active_test": False})

    def attachments(self, db: str, expense_ids: list) -> list:
        return self._call(db, "ir.attachment", "search_read", [["res_model", "=", "hr.expense"], ["res_id", "in", expense_ids]],
                          fields=["id", "res_id", "name", "mimetype", "datas"]) if expense_ids else []

    def attachment_count(self, db: str, expense_ids: list) -> int:
        return self._call(db, "ir.attachment", "search_count", [["res_model", "=", "hr.expense"], ["res_id", "in", expense_ids]]) if expense_ids else 0


ports = Ports()