dodil/erp-brick-contracts

src/brick_contracts/infrastructure/business_partner.py

"""Anti-corruption layer: business-partner (context map: contracts is a CONFORMIST READER).

The counterparty is ALWAYS a business partner. contracts reads it through bp's route GET /partners/{bp_id} as its
own service identity holding bp:person:read (a person partner is invisible without it, and contracts needs the
starter's name) — never bp's tables, and it never writes bp. What it reads is translated here into the one shape the
domain uses, and FROZEN into the contract's COUNTERPARTY variables at send (CT-51): a later rename never alters a
sent version.

bp is a REQUIRED neighbour (modules/contracts depends_on): with it not installed a draft is refused in words, because
a contract with a counterparty nobody can check is the bug the spec warns about.
"""
from dataclasses import dataclass
from typing import Optional

from erp_platform.seam import Seam

BP = Seam("business-partner", "BUSINESS_PARTNER_BASE_URL", roles=["service"], permissions=["bp:person:read"])
FIELDS = ("bp_id", "bp_key", "legal_name", "preferred_name", "kind", "roles", "status", "can_transact", "country",
          "tax_id", "work_email", "domain")


@dataclass
class PartnerReply:
    ok: bool
    status: str                     # found | absent | not_found | unreachable
    partner: Optional[dict] = None
    reason: str = ""


def _translate(data: dict) -> dict:
    out = {k: data.get(k) for k in FIELDS}
    out["bp_id"] = int(data["bp_id"])
    out["roles"] = sorted(str(r).lower() for r in data.get("roles") or [])
    out["kind"] = data.get("kind") or "organisation"
    return out


def read(bp_id: int, actor: str = "") -> PartnerReply:
    if not BP.installed:
        return PartnerReply(False, "absent", reason="business-partner is not installed: contracts needs it to know who "
                                                    "the counterparty is")
    r = BP.get(f"/partners/{int(bp_id)}", actor=actor)
    if r:
        return PartnerReply(True, "found", _translate(r.data))
    if r.status == 404:
        return PartnerReply(False, "not_found", reason=f"no business partner {bp_id}")
    return PartnerReply(False, "unreachable", reason=r.reason or "business-partner did not answer")


def installed() -> bool:
    return BP.installed


__all__ = ["BP", "read", "installed", "PartnerReply"]