dodil/erp-brick-gl

DODIL ERP brick: general ledger

src/brick_gl/infrastructure/odoo.py

"""The reader of an Odoo database, behind one port (module.yaml migration.odoo.reads). The migration NEVER writes to Odoo.

Rows are plain dicts with the fields the spec names. The aggregates are Odoo's OWN (a read_group), because the reconciliation
compares what Odoo says with what we loaded - computing "Odoo's side" from what we wrote would prove nothing. A production reader
speaks XML-RPC/JSON-RPC to the customer's Odoo; tests install a fake over rows. Dates are `date`, amounts `Decimal`.
"""
from datetime import date
from typing import Optional


class OdooUnavailable(Exception):
    pass


class OdooReader:
    def companies(self) -> list: raise OdooUnavailable("no Odoo reader is wired")
    def accounts(self, company_id: int) -> list: raise OdooUnavailable("no Odoo reader is wired")
    def journals(self, company_id: int) -> list: raise OdooUnavailable("no Odoo reader is wired")
    def moves(self, company_id: int, date_from: Optional[date], date_to: date) -> list: raise OdooUnavailable("no Odoo reader is wired")
    def lines(self, move_ids: list) -> list: raise OdooUnavailable("no Odoo reader is wired")
    def fiscal_years(self, company_id: int) -> list: raise OdooUnavailable("no Odoo reader is wired")
    def rates(self, company_id: int, date_from: Optional[date], date_to: date) -> list: raise OdooUnavailable("no Odoo reader is wired")
    def analytic_accounts(self) -> list: raise OdooUnavailable("no Odoo reader is wired")
    def recurring_moves(self, company_id: int, after: date) -> list: raise OdooUnavailable("no Odoo reader is wired")
    def deferral_lines(self, company_id: int, after: date) -> list: raise OdooUnavailable("no Odoo reader is wired")
    def budgets(self, company_id: int) -> list: raise OdooUnavailable("no Odoo reader is wired")
    def consolidation_accounts(self) -> list: raise OdooUnavailable("no Odoo reader is wired")
    def repartition_lines(self, tax_ids: list) -> list: raise OdooUnavailable("no Odoo reader is wired")
    def company_defaults(self, company_id: int) -> dict: raise OdooUnavailable("no Odoo reader is wired")
    # Odoo's own aggregates (read_group), for the reconciliation
    def opening_balances(self, company_id: int, before: date) -> list: raise OdooUnavailable("no Odoo reader is wired")
    def tb_by_account_pc(self, company_id: int, cutover: date) -> list: raise OdooUnavailable("no Odoo reader is wired")
    def tb_by_account_period(self, company_id: int, cutover: date) -> list: raise OdooUnavailable("no Odoo reader is wired")
    def partner_balances(self, company_id: int, cutover: date) -> list: raise OdooUnavailable("no Odoo reader is wired")
    def foreign_balances(self, company_id: int, cutover: date) -> list: raise OdooUnavailable("no Odoo reader is wired")
    def move_counts(self, company_id: int, cutover: date) -> list: raise OdooUnavailable("no Odoo reader is wired")
    def tax_by_quarter(self, company_id: int, cutover: date) -> list: raise OdooUnavailable("no Odoo reader is wired")
    def account_counts(self, company_id: int) -> list: raise OdooUnavailable("no Odoo reader is wired")
    def rate_counts(self, company_id: int, cutover: date) -> list: raise OdooUnavailable("no Odoo reader is wired")
    def receivable_balance(self, company_id: int, cutover: date): raise OdooUnavailable("no Odoo reader is wired")
    def draft_move_count(self, company_id: int, cutover: date) -> int: raise OdooUnavailable("no Odoo reader is wired")


class Ports:
    """Replace `reader` to wire a reader (a composition, or a test)."""

    def __init__(self):
        self.reader: OdooReader = OdooReader()


ports = Ports()