"""Anti-corruption layer: the business-partner brick (context map: hr is UPSTREAM for person fields).
hr is the single writer of a person's partner, through bp's PUBLISHED write path — POST /partners/persons/sync
(bp:person:sync), body {"person": <card>, "anonymise": false} — and never writes business_partner, bp_role or
bp_edges itself. A freelancer's company is created or reused through POST /partners/promote with the SUPPLIER
role — which bp 0.3.0 counts as a master-data decision and gates on bp:partner:write (promote alone gives only
`customer`), so the seam carries both.
The card is built by the domain (services.person_card): ONLY the legal and preferred name, the work email and
phone, the roles and the works_for company. A guard here refuses to send anything else — the last line before a
private field could reach the record every module reads.
bp_id is DERIVED, never asked for: blake2b("person:<person_id>", 7 bytes), bp's own derivation (bp_id_for), so
hr knows the partner's id without a round trip. With bp not installed every call is a no-op that says so.
"""
import hashlib
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:sync", "bp:person:read", "bp:partner:promote", "bp:partner:write"])
CARD_FIELDS = ("person_id", "legal_name", "preferred_name", "work_email", "work_phone", "country", "roles", "works_for")
def bp_id_for(bp_key: str) -> int:
return int.from_bytes(hashlib.blake2b(bp_key.strip().lower().encode(), digest_size=7).digest(), "big")
def person_bp_id(person_id: int) -> int:
return bp_id_for(f"person:{int(person_id)}")
def company_key(company: dict) -> str:
"""The business key a freelancer's company is promoted under: its company number if it has one."""
no = (company.get("company_number") or "").strip().lower().replace(" ", "")
if no:
return f"{(company.get('country') or 'GB').strip().lower()}-company:{no}"
name = (company.get("legal_name") or company.get("name") or "").strip().lower()
return f"company:{' '.join(name.split())}"
@dataclass
class Reply:
ok: bool
status: str # synced | absent | refused | unreachable
reason: str = ""
bp_id: Optional[int] = None
data: Optional[dict] = None
def installed() -> bool:
return BP.installed
def sync(card: dict, actor: str = "") -> Reply:
extra = sorted(set(card) - set(CARD_FIELDS))
if extra: # never: the card is the domain's, and this is the last guard
raise ValueError(f"the business partner carries a name and a work contact only — not {', '.join(extra)}")
if not BP.installed:
return Reply(False, "absent", "business-partner is not installed")
r = BP.post("/partners/persons/sync", {"person": card, "anonymise": False}, actor)
if not r:
return Reply(False, "refused" if r.status else "unreachable", r.reason)
return Reply(True, "synced", "", int(r.data.get("bp_id")), r.data)
def anonymise(person_id: int, actor: str = "") -> Reply:
if not BP.installed:
return Reply(False, "absent", "business-partner is not installed")
r = BP.post("/partners/persons/sync", {"person": {"person_id": str(int(person_id))}, "anonymise": True}, actor)
if not r:
if r.status == 404:
return Reply(True, "synced", "no partner to anonymise")
return Reply(False, "refused" if r.status else "unreachable", r.reason)
return Reply(True, "synced", "", int(r.data.get("bp_id")), r.data)
def promote_company(company: dict, actor: str = "") -> Reply:
"""Create or reuse the freelancer's company as an organisation partner with the supplier role."""
key = company_key(company)
if not BP.installed:
return Reply(False, "absent", "business-partner is not installed", bp_id_for(key))
body = {"bp_key": key, "role": "supplier", "legal_name": company.get("legal_name") or company.get("name"),
"country": (company.get("country") or "GB").upper(), "tax_id": company.get("vat_number") or "",
"domain": company.get("domain") or ""}
r = BP.post("/partners/promote", body, actor)
if not r:
return Reply(False, "refused" if r.status else "unreachable", r.reason, bp_id_for(key))
return Reply(True, "synced", "", int(r.data.get("bp_id")), r.data)
def read(bp_id: int, actor: str = "") -> Optional[dict]:
"""The partner's fields (for the subject-access export) — what bp holds, as bp shows it to hr."""
if not BP.installed:
return None
r = BP.get(f"/partners/{int(bp_id)}", actor=actor)
return r.data if r else None
__all__ = ["BP", "sync", "anonymise", "promote_company", "read", "person_bp_id", "bp_id_for", "company_key",
"installed", "Reply"]