dodil/erp-brick-gl

DODIL ERP brick: general ledger

src/brick_gl/infrastructure/seams.py

"""The neighbours gl ASKS, behind one adapter (the ones that call gl are routes, not seams).

gl owns the books; the modules that own open items own what is outstanding. So the receivable control account is reconciled
against payments' GET /settlement, the payables account against purchasing's GET /bills/open, the asset accounts against
fixed-assets' register, and intercompany items against ar's and purchasing's GET /intercompany/open. Each is declared HERE
and nowhere else, shaped from the OTHER module's spec route (seams in module.yaml); gl never reads a pay_, pur_, fa_ or
ar_ table. An absent neighbour is a reply (`installed` false), never an error: the gate that needs it is simply not asked.

    payments    payments    GET /settlement?invoice_ids=<csv, at most 200>&as_at=            pay:receipt:read
    purchasing  pur    GET /bills/open?as_of=&bp_id=                                    purchasing:bill:read
                       GET /intercompany/open?company_code=&trading_partner=&as_at=
    ar          ar     GET /intercompany/open?company_code=&trading_partner=&as_at=     ar:invoice:read
    fixed-assets fa    GET /assets/balances?company_code=&ledger_id=&as_of=             fa:asset:read
                       GET /depreciation/status?company_code=&period_id=&ledger_id=     fa:depreciation:read
    business-partner bp GET /migration-map/{db}/res.partner?ids=                        bp:migration:read
    tax         tax    GET /migration-map/{db}/account.tax?ids=                         tax:migration:read
"""
from dataclasses import dataclass, field
from datetime import date
from typing import Optional

from erp_platform.seam import Seam

PAYMENTS = Seam("payments", "PAYMENTS_BASE_URL", roles=["service"], permissions=["pay:receipt:read"])
PURCHASING = Seam("purchasing", "PURCHASING_BASE_URL", roles=["service"], permissions=["purchasing:bill:read"])
AR = Seam("ar", "AR_BASE_URL", roles=["service"], permissions=["ar:invoice:read"])
ASSETS = Seam("fixed-assets", "FIXED_ASSETS_BASE_URL", roles=["service"], permissions=["fa:asset:read", "fa:depreciation:read"])
PARTNERS = Seam("business-partner", "BUSINESS_PARTNER_BASE_URL", roles=["service"], permissions=["bp:migration:read"])
TAX = Seam("tax", "TAX_BASE_URL", roles=["service"], permissions=["tax:migration:read"])

SETTLEMENT_PAGE = 200


@dataclass
class Answer:
    """What a neighbour said: installed or not, ok or not, with its data and its reason."""
    installed: bool
    ok: bool = False
    data: dict = field(default_factory=dict)
    status: int = 0
    reason: str = ""


def _ask(seam: Seam, path: str, params: dict, actor: str) -> Answer:
    if not seam.installed:
        return Answer(False, reason=f"{seam.name} is not installed")
    r = seam.get(path, params=params, actor=actor)
    return Answer(True, r.ok, r.data, r.status, r.reason)


class Neighbours:
    """The questions gl asks. A composition or a test replaces the module attribute `neighbours`."""

    def settlement(self, invoice_ids: list, as_at: date, actor: str = "") -> Answer:
        """payments: what is outstanding on these invoices as at a date: {as_at, invoices: [{invoice_id, currency, total,
        outstanding, state}], unknown: [ids]}, at most 200 ids a call (this pages through them)."""
        if not PAYMENTS.installed:
            return Answer(False, reason="payments is not installed")
        invoices, unknown = [], []
        for i in range(0, len(invoice_ids), SETTLEMENT_PAGE):
            a = _ask(PAYMENTS, "/settlement", {"invoice_ids": ",".join(invoice_ids[i:i + SETTLEMENT_PAGE]), "as_at": as_at.isoformat()}, actor)
            if not a.ok:
                return a
            invoices += a.data.get("invoices", [])
            unknown += a.data.get("unknown", [])
        return Answer(True, True, {"as_at": as_at.isoformat(), "invoices": invoices, "unknown": unknown}, 200)

    def open_bills(self, as_of: date, actor: str = "") -> Answer:
        """purchasing: the bills still owed as at a date: {as_of, items: [{bill_id, bill_no, bp_id, currency, total, outstanding,
        due_date}]}."""
        return _ask(PURCHASING, "/bills/open", {"as_of": as_of.isoformat()}, actor)

    def intercompany_open(self, module: str, company_code: str, trading_partner: str, as_at: date, actor: str = "") -> Answer:
        """ar (documents) or purchasing (bills) to a partner company: {documents: [...]}."""
        seam = AR if module == "ar" else PURCHASING
        return _ask(seam, "/intercompany/open", {"company_code": company_code, "trading_partner": trading_partner,
                                                 "as_at": as_at.isoformat()}, actor)

    def asset_balances(self, company_code: str, ledger_id: str, as_of: date, actor: str = "") -> Answer:
        """fixed-assets: cost and accumulated depreciation per class and ledger."""
        return _ask(ASSETS, "/assets/balances", {"company_code": company_code, "ledger_id": ledger_id, "as_of": as_of.isoformat()}, actor)

    def depreciation_status(self, company_code: str, period_id: str, ledger_id: Optional[str] = None, actor: str = "") -> Answer:
        """fixed-assets: is the period's depreciation posted: {areas, pending_transactions, readiness: [{gate, met, reason, fix}]}."""
        params = {"company_code": company_code, "period_id": period_id}
        if ledger_id:
            params["ledger_id"] = ledger_id
        return _ask(ASSETS, "/depreciation/status", params, actor)

    def partner_map(self, db: str, ids: list, actor: str = "") -> Answer:
        return _ask(PARTNERS, f"/migration-map/{db}/res.partner", {"ids": ",".join(map(str, ids))}, actor)

    def tax_map(self, db: str, model: str, ids: list, actor: str = "") -> Answer:
        return _ask(TAX, f"/migration-map/{db}/{model}", {"ids": ",".join(map(str, ids))}, actor)


neighbours = Neighbours()


def installed(name: str) -> bool:
    return {"payments": PAYMENTS, "purchasing": PURCHASING, "ar": AR, "fixed-assets": ASSETS}[name].installed