dodil/erp-brick-fixed-assets

DODIL ERP brick: fixed assets

src/brick_fixed_assets/infrastructure/odoo.py

"""The Odoo reader port (migration.odoo): read-only access to an Odoo database, behind one small interface so the migration
use case never knows how Odoo is reached and tests hand it fake rows (no live Odoo). `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,
read-only: search_read and read_group, nothing written). NOT verified against a live Odoo here — the map is, by the fake rows."""
import os
from datetime import date
from decimal import Decimal
from typing import Optional, Protocol

from brick_fixed_assets.infrastructure.seams import Unavailable

ASSET_FIELDS = ["id", "name", "state", "asset_type", "model_id", "original_value", "book_value", "value_residual",
                "salvage_value", "acquisition_date", "prorata_date", "prorata_computation_type", "method", "method_number",
                "method_period", "method_progress_factor", "account_asset_id", "account_depreciation_id",
                "account_depreciation_expense_id", "journal_id", "already_depreciated_amount_import", "disposal_date",
                "parent_id", "original_move_line_ids", "currency_id", "company_id", "analytic_distribution"]
MOVE_FIELDS = ["id", "asset_id", "date", "state", "depreciation_value", "asset_remaining_value", "asset_depreciated_value",
               "asset_value_change", "reversed_entry_id"]


class OdooReader(Protocol):
    def assets(self, db: str) -> list: ...                     # account.asset rows (models and assets)
    def moves(self, db: str, asset_ids: list, cutover: date) -> list: ...   # posted depreciation moves up to the cutover
    def trial_balance(self, db: str, odoo_account_id: int, cutover: date) -> Decimal: ...


READER: Optional[OdooReader] = None


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


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):
        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)

    def assets(self, db: str) -> list:
        return self._call(db, "account.asset", "search_read", [["state", "in", ["model", "open", "paused", "close"]]],
                          fields=ASSET_FIELDS, context={"active_test": False})

    def moves(self, db: str, asset_ids: list, cutover: date) -> list:
        return self._call(db, "account.move", "search_read", [["asset_id", "in", asset_ids], ["state", "=", "posted"],
                                                               ["date", "<=", cutover.isoformat()]], fields=MOVE_FIELDS)

    def trial_balance(self, db: str, odoo_account_id: int, cutover: date) -> Decimal:
        rows = self._call(db, "account.move.line", "read_group", [["account_id", "=", odoo_account_id],
                                                                  ["parent_state", "=", "posted"],
                                                                  ["date", "<=", cutover.isoformat()]], ["balance:sum"], [])
        return Decimal(str((rows[0].get("balance") if rows else 0) or 0)).quantize(Decimal("0.01"))