dodil/erp-brick-product

src/brick_product/infrastructure/seams.py

"""The neighbours this brick asks, behind one adapter: business-partner (a customer price list names a bp_id).

PRODUCT is the owner of the price book and of the product list; it asks nobody for a price. The one neighbour it
reads is the partner register, to check that the customer a list is assigned to exists. Declared HERE and
nowhere else, so the domain never knows how it is reached. The answer is one of:
  - known: the partner exists (as far as the CALLER may see it);
  - unknown: the register says there is no such partner — the assignment is refused;
  - unchecked: business-partner is not part of this installation, refused the question (403), failed (502) or did
    not answer (503). The assignment is REFUSED, with that status and its real reason: the signed spec says that without the register a customer
    price list "cannot be assigned" — every customer is priced from the currency's default list until it can.

The OTHER direction (quote asks find_price, PRD-20; orders and invoices copy a product's fields, PRD-21) is not
here: those bricks call THIS brick's routes (`find_price`, `get_product`); this brick never reaches into them.
"""
from dataclasses import dataclass

from erp_platform.seam import Seam

# Asked AS THE CALLER (caller=): the question is "may THIS person name that partner?", not a fact about the register.
PARTNERS = Seam("business-partner", "BUSINESS_PARTNER_BASE_URL", roles=["service"], permissions=[])


@dataclass(frozen=True)
class Checked:
    status: str                 # known | unknown | unchecked
    note: str = ""
    http: int = 200             # what this brick answers when the customer cannot be confirmed


def check_customer(bp_id: int, caller: dict) -> Checked:
    """Does the partner register have this customer, as the caller sees it? Every answer is a reply, never an exception,
    and says what the register really did: not installed (409), refused the question (403), failed (502), did not answer
    at all (503), or has no such partner (422)."""
    if not PARTNERS.installed:
        return Checked("unchecked", "the partner register (business-partner) is not part of this installation, so a "
                                    "customer cannot be assigned to a price list: they are priced from the currency's "
                                    "default list", 409)
    r = PARTNERS.get(f"/partners/{int(bp_id)}", actor=caller.get("email", ""), caller=caller)
    if r.status == 404:
        return Checked("unknown", f"there is no business partner {bp_id} — add the customer to the partner register first", 422)
    if r.ok:
        return Checked("known")
    why = r.reason or "no reason given"
    if r.status in (401, 403):
        return Checked("unchecked", f"the partner register refused the question ({r.status}: {why}): you may not look "
                                    "that customer up, so the customer was not assigned", 403)
    if r.status and r.status >= 500:
        return Checked("unchecked", f"the partner register failed ({r.status}: {why}), so the customer cannot be checked "
                                    "and was not assigned — try again", 502)
    if r.status:
        return Checked("unchecked", f"the partner register answered {r.status} ({why}); the customer was not assigned", 502)
    return Checked("unchecked", f"the partner register did not answer ({why}), so the customer cannot be checked and "
                                "was not assigned — try again", 503)


# v6: manufacturing in the composition (procurement make and planning mrp are accepted only then). Only its presence is asked.
MANUFACTURING = Seam("manufacturing", "MANUFACTURING_BASE_URL", roles=["service"], permissions=["mfg:bom:explode"])


# ── inventory: a site is one of its warehouses (PRD-37) ──────────────────────────────────────────

# Asked AS THE SERVICE (inv:stock:read is inventory's permission; a planner holds none of it): "does this warehouse exist, and in which company?"
INVENTORY = Seam("inventory", "INVENTORY_BASE_URL", roles=["service"], permissions=["inv:stock:read"])


@dataclass(frozen=True)
class Warehouse:
    status: str                 # known | unknown | unvalidated | unchecked
    company_code: str = ""
    warehouse_status: str = ""
    note: str = ""
    http: int = 200


def check_warehouse(site_id: str, actor: str = "") -> Warehouse:
    """inventory's GET /warehouses/{warehouse_id} -> {warehouse_id, company_code, status}. Without inventory installed the site is
    validated against nothing (the signed spec). Every answer is a reply, never an exception: a warehouse that is not there (422), one
    that cannot be confirmed because inventory refused (403), failed (502) or did not answer (503)."""
    if not INVENTORY.installed:
        return Warehouse("unvalidated", note="inventory is not part of this installation: the site code is validated against nothing")
    r = INVENTORY.get(f"/warehouses/{site_id}", actor=actor)
    if r.status == 404:
        return Warehouse("unknown", note=f"there is no warehouse {site_id} in inventory — add it there first", http=422)
    if r.ok:
        d = r.data or {}
        return Warehouse("known", str(d.get("company_code") or ""), str(d.get("status") or ""))
    why = r.reason or "no reason given"
    if r.status in (401, 403):
        return Warehouse("unchecked", note=f"inventory refused the question ({r.status}: {why}), so {site_id} could not be checked", http=403)
    if r.status and r.status >= 500:
        return Warehouse("unchecked", note=f"inventory failed ({r.status}: {why}), so {site_id} cannot be checked — try again", http=502)
    if r.status:
        return Warehouse("unchecked", note=f"inventory answered {r.status} ({why}) for {site_id}", http=502)
    return Warehouse("unchecked", note=f"inventory did not answer ({why}), so {site_id} cannot be checked — try again", http=503)