"""The business partner — one record per company or person you do business with, shared by every
module, so sales, invoicing and purchasing mean the same customer (modules/business-partner, BP-01).
No database, no HTTP, no identity. The rules here are about what a partner IS and how partners
relate; who may change them is the application layer's business.
TWO AGGREGATES, deliberately:
Partner the party and its roles. Its roles change with it: "Acme is a customer and a supplier".
Of KIND organisation (a company — what every module had until 0.2.0) or person (BP-05..07):
a person partner is written only by hr, carries a name and a work contact only, and is keyed
'person:<person_id>' so hr knows its bp_id without asking.
Family the typed edges between partners (subsidiary_of, branch_of, framework_governs). An
edge joins two partners, so it belongs to neither one's aggregate; its rules — no
cycles, one parent per relationship — are about the whole graph of one type.
Rebuilt from the reference package (blogs/code/erp-business-partner/v1), fixing what it got wrong:
a re-import that silently unblocked a blocked customer, an id collision that would overwrite
another partner, a key stored in whatever case it arrived in, and ownership cycles.
"""
from __future__ import annotations
import hashlib
from dataclasses import dataclass, field
from typing import Iterable, Optional
ACTIVE, BLOCKED, ARCHIVED = "active", "blocked", "archived"
STATUSES = (ACTIVE, BLOCKED, ARCHIVED)
ORGANISATION, PERSON = "organisation", "person"
KINDS = (ORGANISATION, PERSON)
PERSON_PREFIX = "person:"
# What hr may make a person to us. A freelancer through their own company is a worker, and the company is the
# supplier (its own organisation partner, which the person works_for).
PERSON_ROLES = ("employee", "worker", "supplier")
ERASED_NAME = "Erased person" # BP-07: the fixed placeholder an anonymised person carries
# The relationship types, and how many parents each allows. Ownership is a tree: a company has ONE
# parent company. A framework agreement can govern a partner alongside others, so it is not a tree.
# works_for joins a PERSON to the one company they are engaged through; only hr writes it (sync_person).
RELATIONSHIPS = {"subsidiary_of": 1, "branch_of": 1, "framework_governs": None, "works_for": 1}
STEWARD_RELATIONSHIPS = ("subsidiary_of", "branch_of", "framework_governs")
# The person card: everything a person partner may carry. Anything else is refused, not dropped — a
# shared record readable by contracts and finance is the wrong place for a home address, and a field
# silently ignored is a field someone believes was stored.
PERSON_FIELDS = ("person_id", "legal_name", "preferred_name", "work_email", "work_phone", "country", "roles",
"works_for")
_PRIVATE_HINTS = ("personal", "private", "home", "address", "postcode", "birth", "dob", "tax", "vat", "bank",
"iban", "bic", "swift", "sort_code", "account", "ni_", "national_insurance", "ssn", "passport",
"nationality", "emergency", "salary", "pay")
class PartnerRuleBroken(Exception):
"""A rule about partners said no. → 409."""
class PartnerIncomplete(PartnerRuleBroken):
"""Something has to be supplied first — a name for a new partner, say. → 422."""
def normalise_key(bp_key: str) -> str:
"""The form a business key is compared AND stored in. Stored as given, `Acme.com` and `acme.com`
derived the same id while the stored key flipped with whichever import ran last."""
key = (bp_key or "").strip().lower()
if not key:
raise PartnerIncomplete("a partner needs a business key — a customer number, a tax id or a domain")
return key
def bp_id_for(bp_key: str) -> int:
"""The partner's integer id, derived from its business key (BP-02).
An integer, because the graph functions take an integer start node; derived, because DataK3 has
no sequences and a re-import must land on the same row. 56 bits of blake2b fits a signed BIGINT.
The same derivation as the reference package, so ids already in a bucket stay valid.
"""
return int.from_bytes(hashlib.blake2b(normalise_key(bp_key).encode(), digest_size=7).digest(), "big")
class PeopleAreWrittenByHr(PartnerRuleBroken):
"""A steward's or a module's write reached a person partner. → 409."""
def __init__(self):
super().__init__("people are written by hr — a person partner changes only through hr (bp:person:sync)")
def person_key(person_id: str) -> str:
pid = (person_id or "").strip().lower()
if not pid:
raise PartnerIncomplete("a person partner needs hr's person_id")
return PERSON_PREFIX + pid
def is_person_key(bp_key: str) -> bool:
return (bp_key or "").strip().lower().startswith(PERSON_PREFIX)
def person_card(payload: dict) -> dict:
"""The person card hr sends, checked: only the fields a person partner carries (BP-05).
A private field — a personal email or phone, a home address, a tax, bank or NI field — is refused in the
rule's own words; any other unknown field is refused too, by name."""
if not isinstance(payload, dict):
raise PartnerIncomplete("person is an object: {person_id, legal_name, preferred_name, work_email, ...}")
extra = sorted(k for k in payload if k not in PERSON_FIELDS)
private = [k for k in extra if any(h in k.lower() for h in _PRIVATE_HINTS)]
if private:
raise PartnerIncomplete(f"a person partner carries a name and a work contact only — refused: "
f"{', '.join(private)}")
if extra:
raise PartnerIncomplete(f"unknown field(s) for a person partner: {', '.join(extra)} "
f"(it carries {', '.join(PERSON_FIELDS)})")
roles = payload.get("roles") or []
if not isinstance(roles, list) or any(not isinstance(r, str) for r in roles):
raise PartnerIncomplete("roles is a list of words")
wrong = sorted({r.strip().lower() for r in roles} - set(PERSON_ROLES))
if wrong:
raise PartnerIncomplete(f"a person partner's roles are {', '.join(PERSON_ROLES)}, not {', '.join(wrong)}")
works_for = payload.get("works_for")
if isinstance(works_for, str) and works_for.strip().isdigit():
works_for = int(works_for.strip()) # 0.3: a bp_id arrives as digits — bp returns it as a string
if works_for is not None and (isinstance(works_for, bool) or not isinstance(works_for, int)):
raise PartnerIncomplete("works_for is the bp_id (an integer, or its digits) of the company the person "
"works through")
email = (payload.get("work_email") or "").strip().lower()
if email and ("@" not in email or " " in email):
raise PartnerIncomplete(f"work_email is an address, not {payload.get('work_email')!r}")
return {"person_id": str(payload.get("person_id") or "").strip(),
"legal_name": (payload.get("legal_name") or "").strip(),
"preferred_name": (payload.get("preferred_name") or "").strip(),
"work_email": email, "work_phone": (payload.get("work_phone") or "").strip(),
"country": (payload.get("country") or "").strip().upper(),
"roles": sorted({r.strip().lower() for r in roles}), "works_for": works_for}
@dataclass(slots=True)
class Partner:
bp_id: int
bp_key: str
legal_name: str
country: str
tax_id: str = ""
domain: str = "" # an attribute, NOT the key: domains move on a rename
status: str = ACTIVE
roles: set = field(default_factory=set)
kind: str = ORGANISATION
preferred_name: str = "" # persons: the name they go by
work_email: str = ""
work_phone: str = ""
anonymised: bool = False # BP-07: erased in hr; the id and its joins stay, the person does not
version: int = 0 # 0.3: how many times it was saved — two people at once, one wins (409)
@property
def is_person(self) -> bool:
return self.kind == PERSON
@classmethod
def person(cls, person_id: str) -> "Partner":
"""An empty person partner for hr's person_id; `apply_card` fills it."""
key = person_key(person_id)
return cls(bp_id=bp_id_for(key), bp_key=key, legal_name="", country="", kind=PERSON)
def apply_card(self, card: dict) -> dict:
"""Make the person exactly what hr says (BP-05). Returns what changed — field NAMES and roles, never
the values, so the history holds nothing an erasure would have to find."""
if not self.is_person:
raise PartnerRuleBroken(f"partner {self.bp_id} is an organisation, not a person")
if self.anonymised:
raise PartnerRuleBroken("this person was erased (BP-07); a new engagement is a new person_id")
if not card["legal_name"]:
raise PartnerIncomplete("a person partner needs a legal name")
c = card["country"]
if c and (len(c) != 2 or not c.isalpha()):
raise PartnerIncomplete(f"country is a two-letter code (ISO 3166), not {c!r}")
changed = []
for f in ("legal_name", "preferred_name", "work_email", "work_phone", "country"):
if getattr(self, f) != card[f]:
setattr(self, f, card[f])
changed.append(f)
roles = set(card["roles"])
changed += [f"role +{r}" for r in sorted(roles - self.roles)] + [f"role -{r}" for r in sorted(self.roles - roles)]
self.roles = roles
return {"changed": changed}
def anonymise(self) -> bool:
"""BP-07: replace the name and work contact with a fixed placeholder; the id and the roles stay, so
other modules' records still join. True when it changed something."""
if not self.is_person:
raise PartnerRuleBroken(f"partner {self.bp_id} is an organisation; only a person is anonymised")
if self.anonymised:
return False
self.legal_name, self.preferred_name, self.work_email, self.work_phone = ERASED_NAME, "", "", ""
self.anonymised = True
return True
@classmethod
def new(cls, bp_key: str, legal_name: str, country: str, *, tax_id: str = "", domain: str = "",
roles: Iterable[str] = ()) -> "Partner":
"""A new ORGANISATION. A 'person:' key is hr's: refused here."""
if is_person_key(bp_key):
raise PeopleAreWrittenByHr()
key = normalise_key(bp_key)
p = cls(bp_id=bp_id_for(key), bp_key=key, legal_name="", country="")
p.describe(legal_name=legal_name, country=country, tax_id=tax_id, domain=domain)
for r in roles:
p.add_role(r)
return p
def same_party(self, bp_key: str) -> None:
"""Refuse when a key derives this partner's id but is a different key — a 56-bit collision.
Improbable, and the reference package would have overwritten the other partner in silence."""
if normalise_key(bp_key) != self.bp_key:
raise PartnerRuleBroken(f"the key {bp_key!r} collides with partner {self.bp_id} "
f"({self.bp_key}). Use a longer or different business key.")
def describe(self, *, legal_name: Optional[str] = None, country: Optional[str] = None,
tax_id: Optional[str] = None, domain: Optional[str] = None) -> dict:
"""Set who the partner is. Returns what changed, for the history. Status is NOT here: an
import describes a partner, and must never unblock one."""
before = {"legal_name": self.legal_name, "country": self.country, "tax_id": self.tax_id,
"domain": self.domain}
if legal_name is not None:
if not legal_name.strip():
raise PartnerIncomplete("a partner needs a legal name — the name on the contract")
self.legal_name = legal_name.strip()
if country is not None:
c = country.strip().upper()
if len(c) != 2 or not c.isalpha():
raise PartnerIncomplete(f"country is a two-letter code (ISO 3166), not {country!r}")
self.country = c
if tax_id is not None:
self.tax_id = tax_id.strip().upper().replace(" ", "")
if domain is not None:
self.domain = domain.strip().lower()
after = {"legal_name": self.legal_name, "country": self.country, "tax_id": self.tax_id,
"domain": self.domain}
return {k: (before[k], after[k]) for k in after if before[k] != after[k]}
def add_role(self, role: str) -> bool:
"""True when the role is new. Roles are lower-case identifiers: customer, supplier, carrier."""
r = (role or "").strip().lower()
if not r.replace("_", "").isalnum():
raise PartnerIncomplete(f"a role is a word like customer or supplier, not {role!r}")
if r in self.roles:
return False
self.roles.add(r)
return True
def remove_role(self, role: str) -> None:
r = (role or "").strip().lower()
if r not in self.roles:
raise PartnerRuleBroken(f"{self.legal_name} has no role {r!r}")
self.roles.discard(r)
def set_status(self, status: str) -> bool:
"""Block (no new business), archive (gone), or reactivate. True when it changed."""
if status not in STATUSES:
raise PartnerRuleBroken(f"status is one of {', '.join(STATUSES)}, not {status!r}")
if status == self.status:
return False
self.status = status
return True
@property
def can_transact(self) -> bool:
"""Whether a module may start NEW business with this partner — a quote, an order. Existing
documents are not affected; that is each module's own rule."""
return self.status == ACTIVE
# ── the family: typed edges ────────────────────────────────────────────────────────────────
@dataclass(frozen=True, slots=True)
class Edge:
src: int # the child
dst: int # the parent
rel: str
def check_link(new: Edge, parents_of_child: list, below_child: set) -> None:
"""Whether `new` (child → parent, of one relationship type) may be added.
The caller supplies what the rule needs and no more — the child's current parents of this type,
and every partner already below the child through this type — so a family of any size is checked
without loading the whole graph.
* not to itself;
* at most one parent for a tree relationship (a company has one parent company);
* no cycle: if the new parent is already below the child, A would end up its own ancestor and a
family walk would report every member twice over.
"""
if new.rel not in RELATIONSHIPS:
raise PartnerRuleBroken(f"the relationship is one of {', '.join(RELATIONSHIPS)}, not {new.rel!r}")
if new.src == new.dst:
raise PartnerRuleBroken("a partner cannot be its own parent")
if new.dst in parents_of_child:
return # already linked: nothing to do
limit = RELATIONSHIPS[new.rel]
if limit is not None and len(parents_of_child) >= limit:
raise PartnerRuleBroken(f"partner {new.src} is already {new.rel} {parents_of_child[0]}. "
f"Unlink it first — a company has one parent of each kind.")
if new.dst in below_child:
raise PartnerRuleBroken(f"partner {new.dst} is already below {new.src} as {new.rel}; "
f"linking it as the parent would make a cycle")