"""The tables, and the code that turns rows into partners and back.
The Partner aggregate is `business_partner` + `bp_role`, saved whole. Edges are their own rows,
written one at a time with the checks the domain asks for; a family walk is a recursive CTE filtered
on the relationship type (verified on DataK3 2026-10-16), so the graph is never loaded into memory.
"""
import uuid
from datetime import datetime, timezone
from typing import Optional
from sqlalchemy import (BigInteger, Boolean, Column, Integer, MetaData, String, Table, TIMESTAMP, and_, cast,
delete, func, literal, or_, select)
from erp_platform.repository import Aggregate, Part, upsert
from brick_bp.domain.partner import ERASED_NAME, ORGANISATION, PERSON, Edge, Partner
md = MetaData()
partners = Table("business_partner", md,
Column("bp_id", BigInteger, primary_key=True, autoincrement=False),
Column("bp_key", String), Column("legal_name", String), Column("country", String),
Column("tax_id", String), Column("domain", String), Column("status", String),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP),
# 0.2.0, person partners (BP-05..07). NULL kind on a row written before = organisation.
Column("kind", String), Column("preferred_name", String), Column("work_email", String),
Column("work_phone", String), Column("anonymised_at", TIMESTAMP),
# 0.3.0: the optimistic version (erp_platform.repository) — at the end, as its ALTER says.
Column("version", Integer))
roles = Table("bp_role", md,
Column("bp_id", BigInteger, primary_key=True, autoincrement=False),
Column("role", String, primary_key=True), Column("since", TIMESTAMP),
Column("active", Boolean))
edges = Table("bp_edges", md,
Column("src", BigInteger, primary_key=True, autoincrement=False),
Column("dst", BigInteger, primary_key=True, autoincrement=False),
Column("rel", String, primary_key=True), Column("created_at", TIMESTAMP))
events = Table("bp_events", md,
Column("bp_id", BigInteger, primary_key=True, autoincrement=False),
Column("event_id", String, primary_key=True), Column("action", String),
Column("detail", String), Column("actor", String), Column("event_at", TIMESTAMP))
# The Odoo id map (shared.yaml conventions.odoo_migration.id_map): what an Odoo res.partner became here, written by this
# brick's migration and read by every later brick through GET /migration/map/{model}/{odoo_id}?db=, never by matching names.
migration_map = Table("bp_migration_map", md,
Column("odoo_db", String, primary_key=True), Column("odoo_model", String, primary_key=True),
Column("odoo_id", String, primary_key=True), Column("bp_id", BigInteger),
Column("created_at", TIMESTAMP))
# `since` is when the role began: written once, kept on every later save.
PARTNER = Aggregate(Part(partners, ("bp_id",)),
Part(roles, ("bp_id", "role"), insert_only=("created_at", "created_by", "since")),
version="version")
def _now() -> datetime:
return datetime.now(timezone.utc)
class PartnerRepository:
def __init__(self, conn):
self.conn = conn
# ── the Partner aggregate ────────────────────────────────────────────────────────────
def get(self, bp_id: int) -> Optional[Partner]:
rows = PARTNER.load(self.conn, {"bp_id": int(bp_id)})
if rows is None:
return None
h = rows["business_partner"][0]
return Partner(bp_id=int(h["bp_id"]), bp_key=h["bp_key"], legal_name=h["legal_name"],
country=h["country"], tax_id=h["tax_id"] or "", domain=h["domain"] or "",
status=h["status"],
roles={r["role"] for r in rows["bp_role"] if r["active"]},
kind=h.get("kind") or ORGANISATION, preferred_name=h.get("preferred_name") or "",
work_email=h.get("work_email") or "", work_phone=h.get("work_phone") or "",
anonymised=h.get("anonymised_at") is not None, version=int(h.get("version") or 0))
def save(self, p: Partner) -> None:
now = _now()
root = {"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}
if p.is_person:
# Only a person row names the 0.2.0 columns, so an organisation's save writes exactly what it always
# did (its kind stays NULL, which reads as organisation). The first erasure's time is kept.
root |= {"kind": p.kind, "preferred_name": p.preferred_name, "work_email": p.work_email,
"work_phone": p.work_phone,
"anonymised_at": (self._anonymised_at(p.bp_id) or now) if p.anonymised else None}
# Claims version n+1 against the version this partner was LOADED at (0 for a new one): a racer that
# saved in between makes this a Conflict, and nothing of it is written (erp_platform.repository).
p.version = PARTNER.save(self.conn, {
"business_partner": [root],
"bp_role": [{"bp_id": p.bp_id, "role": r, "since": now, "active": True}
for r in sorted(p.roles)]}, now=now, expected_version=p.version)
def _anonymised_at(self, bp_id: int):
r = self.conn.execute(select(partners.c.anonymised_at).where(partners.c.bp_id == int(bp_id))).first()
return r[0] if r else None
def search(self, name: Optional[str] = None, role: Optional[str] = None,
limit: int = 50, include_persons: bool = False) -> list:
stmt = select(partners.c.bp_id, partners.c.bp_key, partners.c.legal_name, partners.c.country,
partners.c.status, partners.c.kind)
if not include_persons:
# BP-06: without bp:person:read a person partner is not in any list. NULL kind = an organisation
# written before 0.2.0.
stmt = stmt.where(or_(partners.c.kind.is_(None), partners.c.kind != PERSON))
if name:
stmt = stmt.where(func.lower(partners.c.legal_name).contains(name.strip().lower()))
if role:
stmt = stmt.where(partners.c.bp_id.in_(
select(roles.c.bp_id).where(and_(roles.c.role == role.strip().lower(),
roles.c.active.is_(True)))))
rows = [dict(r._mapping) for r in self.conn.execute(stmt)]
rows.sort(key=lambda r: (r["legal_name"].lower(), r["bp_id"]))
return [{**r, "bp_id": int(r["bp_id"]), "kind": r["kind"] or ORGANISATION} for r in rows[:limit]]
# ── the family ───────────────────────────────────────────────────────────────────────
def parents(self, child: int, rel: str) -> list:
rows = self.conn.execute(select(edges.c.dst).where(and_(edges.c.src == int(child),
edges.c.rel == rel)))
return sorted(int(r[0]) for r in rows)
def below(self, root: int, rel: str) -> list:
"""Every partner below `root` through `rel` edges only — a recursive CTE, typed (BP-03)."""
fam = select(cast(literal(int(root)), BigInteger).label("bp_id")).cte("fam", recursive=True)
fam = fam.union(select(edges.c.src).join(fam, and_(edges.c.dst == fam.c.bp_id,
edges.c.rel == rel)))
return sorted(int(r[0]) for r in self.conn.execute(select(fam.c.bp_id))
if int(r[0]) != int(root))
def link(self, e: Edge) -> None:
upsert(self.conn, edges, [{"src": e.src, "dst": e.dst, "rel": e.rel, "created_at": _now()}],
("src", "dst", "rel"))
def unlink(self, e: Edge) -> bool:
existed = e.dst in self.parents(e.src, e.rel)
if existed:
self.conn.execute(delete(edges).where(and_(edges.c.src == e.src, edges.c.dst == e.dst,
edges.c.rel == e.rel)))
return existed
def names(self, ids: list) -> dict:
if not ids:
return {}
rows = self.conn.execute(select(partners.c.bp_id, partners.c.legal_name, partners.c.country,
partners.c.kind)
.where(partners.c.bp_id.in_([int(i) for i in ids])))
return {int(r[0]): {"legal_name": r[1], "country": r[2], "kind": r[3] or ORGANISATION} for r in rows}
def kinds(self, ids: list) -> dict:
"""bp_id → kind, for filtering a family walk's members by what the caller may see."""
return {i: v["kind"] for i, v in self.names(ids).items()}
# ── history ──────────────────────────────────────────────────────────────────────────
def record(self, bp_id: int, action: str, actor: str, detail: str = "") -> None:
upsert(self.conn, events, [{"bp_id": int(bp_id), "event_id": uuid.uuid4().hex,
"action": action, "detail": detail, "actor": actor,
"event_at": _now()}], ("bp_id", "event_id"))
def map_get(self, db: str, model: str, odoo_id) -> Optional[int]:
return self.conn.execute(select(migration_map.c.bp_id).where(and_(
migration_map.c.odoo_db == db, migration_map.c.odoo_model == model,
migration_map.c.odoo_id == str(odoo_id)))).scalar()
def map_put(self, db: str, model: str, odoo_id, bp_id: int) -> None:
upsert(self.conn, migration_map, [{"odoo_db": db, "odoo_model": model, "odoo_id": str(odoo_id), "bp_id": int(bp_id),
"created_at": _now()}], ("odoo_db", "odoo_model", "odoo_id"))
def history(self, bp_id: int) -> list:
rows = self.conn.execute(select(events.c.action, events.c.detail, events.c.actor,
events.c.event_at).where(events.c.bp_id == int(bp_id)))
out = [{"action": r[0], "detail": r[1], "actor": r[2],
"at": r[3].isoformat() if hasattr(r[3], "isoformat") else r[3]} for r in rows]
return sorted(out, key=lambda e: e["at"] or "")