"""The use cases — what the data steward, other modules, and their agents can do with partners.
Each is declared once and served as an API route AND an MCP tool.
WHO (modules/business-partner): every signed-in user may READ organisation partners — sales,
invoicing and purchasing all need the same customer list. Only the steward (bp:partner:write,
bp:family:write) changes them. Other modules create-or-reuse through `promote_partner`
(bp:partner:promote) and NEVER write these tables: DataK3 does not enforce foreign keys, so this
brick being the only writer is the integrity mechanism.
PERSON partners (0.2.0, BP-05..07) are hr's: written ONLY through `sync_person` (bp:person:sync),
name and work contact only, and visible ONLY with bp:person:read — without it a person is absent
from every list and opening one is 404, never 403 (a starter's name must not reach the whole company
before they start). No steward or module permission writes one: "people are written by hr".
"""
from contextlib import contextmanager
from typing import Literal, Optional
from erp_platform import auth
from erp_platform.bricks import Brick, Refused
from erp_platform.repository import Conflict, retry_on_conflict
from erp_platform.types import BigId, Limit, Name, Note, text
from brick_bp.domain.partner import (PERSON, Edge, Partner, PartnerIncomplete, PartnerRuleBroken,
PeopleAreWrittenByHr, bp_id_for, check_link, is_person_key, normalise_key,
person_card, person_key)
# Reads of organisations need no permission: every signed-in user may read them (spec).
WRITE, PROMOTE, FAMILY = "bp:partner:write", "bp:partner:promote", "bp:family:write"
PERSON_SYNC, PERSON_READ = "bp:person:sync", "bp:person:read"
Rel = Literal["subsidiary_of", "branch_of", "framework_governs"] # what the steward links
FamilyRel = Literal["subsidiary_of", "branch_of", "framework_governs", "works_for"]
Status = Literal["active", "blocked", "archived"]
Key, Country, TaxId, Domain, Role = text(200), text(2), text(64), text(253), text(64)
MIGRATION_READ = "bp:migration:read" # machine: other modules' Odoo-migration clients (spec v3; GET /migration-map checks none yet, as the spec says)
BP = Brick("business-partner", version="0.4.0", ids_as_strings=True, id_keys=("parents",), machine_permissions=(MIGRATION_READ,), instructions=(
"The shared list of companies and people this business deals with — one record per real party, "
"used by sales, invoicing and purchasing alike. A partner has roles (customer, supplier, …) and "
"may belong to a corporate family (subsidiary_of, branch_of) or be governed by a framework "
"agreement. bp_id is the partner's id everywhere; bp_key is the business's own identifier for it. "
"A partner of kind person (an employee, a freelancer) is written only by hr and is visible only to "
"callers with bp:person:read. bp_id is a 64-bit integer, sent and returned as a string of digits. A partner "
"carries a `version`: pass the one you read back on a change, and a 409 means someone changed it since."))
@contextmanager
def _repository_default():
from erp_platform import db
from brick_bp.infrastructure.tables import PartnerRepository
with db.engine.begin() as conn:
yield PartnerRepository(conn)
repository = _repository_default
@contextmanager
def _rules():
try:
yield
except PartnerIncomplete as e:
raise Refused(422, str(e)) from None
except PartnerRuleBroken as e:
raise Refused(409, str(e)) from None
def _view(p: Partner) -> dict:
out = {"bp_id": p.bp_id, "bp_key": p.bp_key, "legal_name": p.legal_name, "country": p.country,
"tax_id": p.tax_id, "domain": p.domain, "status": p.status, "roles": sorted(p.roles),
"can_transact": p.can_transact, "kind": p.kind, "version": p.version}
if p.is_person:
out |= {"preferred_name": p.preferred_name, "work_email": p.work_email, "work_phone": p.work_phone,
"anonymised": p.anonymised}
return out
def _sees_persons(caller: dict) -> bool:
return auth.has_permission(caller, PERSON_READ)
def _get(repo, bp_id: int, caller: Optional[dict] = None) -> Partner:
"""The partner, or 404 — also for a person the caller may not see (BP-06: 404, never 403)."""
p = repo.get(bp_id)
if p is None or (p.is_person and not (caller is not None and _sees_persons(caller))):
raise Refused(404, f"no partner {bp_id}")
return p
def _organisation(repo, bp_id: int, caller: dict) -> Partner:
"""A partner a steward or a module may change: an organisation. A person is 404 to whoever may not see
people, and 409 "people are written by hr" to whoever may."""
p = _get(repo, bp_id, caller)
if p.is_person:
raise PeopleAreWrittenByHr()
return p
def _at(p: Partner, version: Optional[int]) -> Partner:
"""The partner, if it is still at the version the caller loaded; a Conflict (409) if not."""
if version is not None and p.version != int(version):
raise Conflict("business_partner", {"bp_id": p.bp_id}, int(version), p.version)
return p
def _changing(version: Optional[int], fn):
"""Run a change. With the caller's version, a race is their 409; with none, it is re-run on the latest
partner, so two stewards at once both land instead of one silently undoing the other."""
return fn() if version is not None else retry_on_conflict(fn)
def _by_key(repo, bp_key: str) -> Optional[Partner]:
p = repo.get(bp_id_for(bp_key))
if p is not None:
p.same_party(bp_key)
return p
# ── reading ────────────────────────────────────────────────────────────────────────────────
@BP.use_case(kind="query", route="GET /partners")
def find_partners(caller: dict, name: Optional[Name] = None, role: Optional[Role] = None,
bp_key: Optional[Key] = None, limit: Limit = 50) -> dict:
"""Find partners by part of their legal name, by role (customer, supplier, …), or exactly by the
business's own key for them. Up to `limit` results, alphabetical."""
with repository() as repo, _rules():
persons = _sees_persons(caller)
if bp_key:
p = _by_key(repo, bp_key)
return {"partners": [_view(p)] if p and (persons or not p.is_person) else []}
return {"partners": repo.search(name=name, role=role, limit=max(1, min(limit, 500)),
include_persons=persons)}
@BP.use_case(kind="query", route="GET /partners/{bp_id}", etag="version")
def get_partner(caller: dict, bp_id: BigId) -> dict:
"""One partner: who they are, their roles, whether new business may be started with them, and its
`version` (send it back with a change)."""
with repository() as repo:
return _view(_get(repo, bp_id, caller))
@BP.use_case(kind="query", route="GET /partners/{bp_id}/family")
def get_family(caller: dict, bp_id: BigId, rel: FamilyRel = "subsidiary_of") -> dict:
"""The partner's parent of this relationship type, and every partner below it through that type
only — a framework agreement never leaks into an ownership tree. `works_for` lists the people
engaged through a company (only to callers who may see people)."""
with repository() as repo:
_get(repo, bp_id, caller)
parents, below = repo.parents(bp_id, rel), repo.below(bp_id, rel)
names = repo.names(parents + below)
if not _sees_persons(caller):
parents = [i for i in parents if names.get(i, {}).get("kind") != PERSON]
below = [i for i in below if names.get(i, {}).get("kind") != PERSON]
return {"bp_id": bp_id, "rel": rel,
"parents": [{"bp_id": i, **names.get(i, {})} for i in parents],
"below": [{"bp_id": i, **names.get(i, {})} for i in below]}
@BP.use_case(kind="query", route="GET /partners/{bp_id}/history")
def partner_history(caller: dict, bp_id: BigId) -> dict:
"""Who changed this partner, what, and when."""
with repository() as repo:
_get(repo, bp_id, caller)
return {"bp_id": bp_id, "events": repo.history(bp_id)}
# ── writing: the steward ───────────────────────────────────────────────────────────────────
@BP.use_case(kind="command", route="PUT /partners", permission=WRITE, idempotent=True, if_match="version")
def upsert_partner(caller: dict, bp_key: Key, legal_name: Name, country: Country,
tax_id: Optional[TaxId] = None, domain: Optional[Domain] = None,
roles: Optional[list[Role]] = None, version: Optional[int] = None) -> dict:
"""Create a partner, or update who an existing one is (BP-02: the same key always lands on the same
partner, so an import run twice changes nothing). A tax id or domain left out is left as it is, not
blanked. Roles listed are added, never removed. Status is never changed here — a re-import must not
unblock a blocked customer. A person partner is hr's ("people are written by hr"). `version` (or If-Match):
the one you read, when you are changing a partner you looked at."""
def run():
with repository() as repo, _rules():
if is_person_key(bp_key):
raise PeopleAreWrittenByHr()
p = _by_key(repo, bp_key)
if p is not None and p.is_person:
raise PeopleAreWrittenByHr()
if p is None:
p = Partner.new(bp_key, legal_name, country, tax_id=tax_id or "", domain=domain or "",
roles=roles or ())
_at(p, version)
repo.save(p)
repo.record(p.bp_id, "created", caller["email"],
f"{p.legal_name} ({p.country}) roles {sorted(p.roles)}")
return _view(p)
_at(p, version)
changed = p.describe(legal_name=legal_name, country=country, tax_id=tax_id, domain=domain)
added = [r for r in (roles or ()) if p.add_role(r)]
if changed or added:
repo.save(p)
detail = "; ".join([f"{k} {a!r} → {b!r}" for k, (a, b) in changed.items()]
+ [f"role +{r}" for r in added])
repo.record(p.bp_id, "updated", caller["email"], detail)
return _view(p)
return _changing(version, run)
@BP.use_case(kind="command", route="PUT /partners/{bp_id}/status", permission=WRITE, idempotent=True,
if_match="version")
def set_partner_status(caller: dict, bp_id: BigId, status: Status, reason: Note,
version: Optional[int] = None) -> dict:
"""Block a partner (no new quotes or orders), archive one, or make it active again. A reason is
required: whoever is refused business with this partner next week will ask why. `version` (or If-Match):
the one you read; a 409 means it changed since."""
def run():
with repository() as repo, _rules():
if not reason.strip():
raise Refused(422, "a status change needs a reason")
p = _at(_organisation(repo, bp_id, caller), version)
before = p.status
if p.set_status(status):
repo.save(p)
repo.record(p.bp_id, f"status {status}", caller["email"], f"{before} → {status}: {reason}")
return _view(p)
return _changing(version, run)
@BP.use_case(kind="command", route="DELETE /partners/{bp_id}/roles/{role}", permission=WRITE,
destructive=True, if_match="version")
def remove_role(caller: dict, bp_id: BigId, role: Role, version: Optional[int] = None) -> dict:
"""Stop a partner being, say, a supplier. The partner and its other roles stay."""
def run():
with repository() as repo, _rules():
p = _at(_organisation(repo, bp_id, caller), version)
p.remove_role(role)
repo.save(p)
repo.record(p.bp_id, "role removed", caller["email"], role)
return _view(p)
return _changing(version, run)
@BP.use_case(kind="command", route="PUT /partners/{bp_id}/parents", permission=FAMILY, idempotent=True)
def link_partners(caller: dict, bp_id: BigId, parent_id: BigId, rel: Rel = "subsidiary_of") -> dict:
"""Record that partner `bp_id` is `rel` partner `parent_id` (BP-03). Refused if it would give a
company two parents of the same kind, or make a cycle."""
with repository() as repo, _rules():
_organisation(repo, bp_id, caller)
_organisation(repo, parent_id, caller)
edge = Edge(int(bp_id), int(parent_id), rel)
parents = repo.parents(bp_id, rel)
if parent_id not in parents:
check_link(edge, parents, set(repo.below(bp_id, rel)))
repo.link(edge)
repo.record(bp_id, "linked", caller["email"], f"{rel} {parent_id}")
return {"bp_id": bp_id, "rel": rel, "parents": repo.parents(bp_id, rel) or [parent_id]}
@BP.use_case(kind="command", route="DELETE /partners/{bp_id}/parents/{parent_id}", permission=FAMILY,
destructive=True, idempotent=True)
def unlink_partners(caller: dict, bp_id: BigId, parent_id: BigId, rel: Rel = "subsidiary_of") -> dict:
"""Remove a family link. Removing one that is not there changes nothing."""
with repository() as repo, _rules():
child = repo.get(bp_id)
if child is not None and child.is_person: # a person's link is hr's; unknown stays a no-op
_organisation(repo, bp_id, caller)
if repo.unlink(Edge(int(bp_id), int(parent_id), rel)):
repo.record(bp_id, "unlinked", caller["email"], f"{rel} {parent_id}")
return {"bp_id": bp_id, "rel": rel}
# ── the seam other modules call ────────────────────────────────────────────────────────────
@BP.use_case(kind="command", route="POST /partners/promote", permission=PROMOTE, idempotent=True)
def promote_partner(caller: dict, bp_key: Key, role: Role = "customer", legal_name: Optional[Name] = None,
country: Optional[Country] = None, tax_id: TaxId = "", domain: Domain = "") -> dict:
"""Create-or-reuse a partner and give it a role (BP-04) — how sales, invoicing and purchasing join
the shared list instead of keeping their own. An existing partner keeps its name and details; a new
one needs legal_name and country. Calling it twice changes nothing. A person is hr's, never promoted.
With bp:partner:promote alone the role is `customer` (a rep finds customers); any other role — supplier,
carrier — is a master-data decision and needs bp:partner:write."""
if (role or "").strip().lower() != "customer" and not auth.has_permission(caller, WRITE):
raise Refused(403, f"missing permission: {WRITE}")
return retry_on_conflict(lambda: _promote(caller, bp_key, role, legal_name, country, tax_id, domain))
def _promote(caller, bp_key, role, legal_name, country, tax_id, domain) -> dict:
with repository() as repo, _rules():
if is_person_key(bp_key):
raise PeopleAreWrittenByHr()
p = _by_key(repo, bp_key)
if p is not None and p.is_person:
raise PeopleAreWrittenByHr()
if p is None:
if not (legal_name and country):
raise Refused(422, f"there is no partner {normalise_key(bp_key)!r} yet — send legal_name "
f"and country to create it")
p = Partner.new(bp_key, legal_name, country, tax_id=tax_id, domain=domain, roles=[role])
repo.save(p)
repo.record(p.bp_id, "created", caller["email"], f"promoted as {role}")
elif p.add_role(role):
repo.save(p)
repo.record(p.bp_id, "role added", caller["email"], f"{role} (promoted)")
return _view(p)
# ── person partners: hr is the only writer (BP-05, BP-07) ──────────────────────────────────
@BP.use_case(kind="command", route="POST /partners/persons/sync", permission=PERSON_SYNC, idempotent=True)
def sync_person(caller: dict, person: dict, anonymise: bool = False) -> dict:
"""hr keeps one person's partner right, as its only writer (BP-05). `person` is the card:
{person_id, legal_name, preferred_name, work_email, work_phone, country, roles: [employee|worker|
supplier], works_for: <bp_id of the company they work through> | null}. The partner becomes exactly
that — its bp_key is 'person:<person_id>' and its bp_id derives from it, so hr knows it without asking;
roles not listed are removed. A private field (a personal email or phone, a home address, a tax, bank
or NI field) is refused: a person partner carries a name and a work contact only. An identical second
call changes nothing. With anonymise=true (BP-07, erasure) only person_id is needed: the name and work
contact become a fixed placeholder; the bp_id, roles and links stay so other modules' records still join."""
return retry_on_conflict(lambda: _sync_person(caller, person, anonymise))
def _sync_person(caller: dict, person: dict, anonymise: bool) -> dict:
with repository() as repo, _rules():
card = person_card(person)
key = person_key(card["person_id"])
p = _by_key(repo, key)
if p is not None and not p.is_person:
raise PartnerRuleBroken(f"partner {p.bp_id} is an organisation; the key {key!r} is not hr's to use")
if anonymise:
if p is None:
raise Refused(404, f"no person partner for {key}")
if p.anonymise():
repo.save(p)
repo.record(p.bp_id, "person anonymised", caller["email"], "name and work contact erased")
return _view(p)
new = p is None
p = p or Partner.person(card["person_id"])
changed = p.apply_card(card)["changed"]
employer = card["works_for"]
if employer is not None:
company = repo.get(employer)
if company is None or company.is_person:
raise Refused(422, f"works_for {employer} is not an organisation partner")
current = repo.parents(p.bp_id, "works_for")
if new or changed:
repo.save(p)
for old in current:
if old != employer:
repo.unlink(Edge(p.bp_id, old, "works_for"))
changed.append("works_for -")
if employer is not None and employer not in current:
repo.link(Edge(p.bp_id, employer, "works_for"))
changed.append("works_for +")
if new:
# Field NAMES and roles only — never values: the history must hold nothing an erasure has to find.
repo.record(p.bp_id, "person created", caller["email"], "; ".join(changed))
elif changed:
repo.record(p.bp_id, "person synced", caller["email"], "; ".join(changed))
return _view(p)
@BP.use_case(kind="query", route="GET /migration-map/{db}/{model}", tool=False)
def get_partner_migration_map(caller: dict, db: Name, model: Name, ids: Optional[text(8000)] = None) -> list:
"""What Odoo records became here: (odoo db, model, ids) -> our natural key, as this brick's Odoo migration wrote it in
bp_migration_map (a res.partner becomes a bp_id). Later bricks resolve Odoo ids through this map, never by matching names (shared.yaml
conventions.odoo_migration). `ids` is comma-separated (at most 500); an id with no mapping is left out, never a 404."""
with repository() as repo:
out = []
for oid in [i.strip() for i in (ids or "").split(",") if i.strip()][:500]:
bp_id = repo.map_get(db, model, oid)
if bp_id is not None:
out.append({"odoo_model": model, "odoo_id": oid, "ours": str(bp_id)})
return out