"""What every use case shares: permissions, the refusal mapping, the business clock, the configuration snapshot, who
the caller is — and ROW VISIBILITY, decided here from `caller` for both doors and the capabilities alike.
VISIBILITY (modules/contracts permissions.row_visibility):
* a contract's EXISTENCE, state and key dates: its owner, its owner group (a pool role), contract:contract:read_all,
an approver with an open approval of it in their queue, our signatory on it — and contract:employment:read for
an employment contract;
* its BODY (sections, variables, versions, executed copy, evidence): the same people, except that an EMPLOYMENT
contract (it holds pay) shows its body only to contract:employment:read — its owner and a read_all manager
without it get 404 on every body route (SC-CT-20);
* a calling module sees what its source_ref namespace caused, through the capabilities only, and only the
ORIGINATOR is given the executed copy; an external signer sees one version through the door.
A row outside visibility is 404, never 403.
"""
import functools
from contextlib import contextmanager
from datetime import date, datetime, timedelta, timezone
from typing import Annotated, Optional
from pydantic import Field
from erp_platform import auth, work
from erp_platform.bricks import Refused
from erp_platform.types import text
from brick_contracts.domain import packs
from brick_contracts.domain.errors import Incomplete, Invalid, NotFound, RuleBroken
from brick_contracts.domain.factories import Drafting
from brick_contracts.domain.services import retention_statement
from brick_contracts.domain.values import EMPLOYMENT, EXECUTED, utc_iso
from brick_contracts.infrastructure import business_partner as bp
# ── bounded inputs ─────────────────────────────────────────────────────────────────────────
Code = text(64)
ContractId = text(40)
Short = text(200)
Words = text(20000)
Reason = text(2000)
Level = Annotated[int, Field(ge=0, le=2)]
VersionNo = Annotated[int, Field(ge=1, le=100000)]
SignerNo = Annotated[int, Field(ge=1, le=50)]
SeqNo = Annotated[int, Field(ge=1, le=100000)]
Order = Annotated[int, Field(ge=1, le=20)]
Rows = Annotated[list[dict], Field(max_length=200)]
# A date a person types, 1900 … 2199 (a `date`: both doors validate it and the probes can make one).
Day = Annotated[date, Field(ge=date(1900, 1, 1), le=date(2199, 12, 31))]
B64 = text(4_000_000) # ~3 MB of file, inside the doors' 5 MiB body limit
# ── permissions (namespace contract) ───────────────────────────────────────────────────────
TYPE_WRITE, CLAUSE_WRITE, TEMPLATE_WRITE, TEMPLATE_APPROVE = (
"contract:type:write", "contract:clause:write", "contract:template:write", "contract:template:approve")
READ, READ_ALL, EMPLOYMENT_READ = "contract:contract:read", "contract:contract:read_all", "contract:employment:read"
WRITE, SEND, RECALL, UPLOAD, SIGN = ("contract:contract:write", "contract:contract:send", "contract:contract:recall",
"contract:contract:upload_executed", "contract:contract:sign")
DECIDE, EVIDENCE, AMEND, SETTINGS = ("contract:approval:decide", "contract:evidence:read", "contract:amendment:write",
"contract:settings:write")
CAPABILITY, TICK = "contract:capability:invoke", "contract:run:tick"
CATALOG = (TYPE_WRITE, CLAUSE_WRITE, TEMPLATE_WRITE, TEMPLATE_APPROVE, READ, READ_ALL, EMPLOYMENT_READ, WRITE, SEND,
RECALL, UPLOAD, SIGN, DECIDE, EVIDENCE, AMEND, SETTINGS, CAPABILITY, TICK)
def has(caller: dict, perm: str) -> bool:
return auth.has_permission(caller, perm)
def actor(caller: dict) -> str:
return (caller.get("email") or caller.get("sub") or "unknown").lower()
def roles(caller: dict) -> set:
return {str(r).strip().lower() for r in caller.get("roles") or [] if str(r).strip()}
@contextmanager
def rules():
"""A domain refusal in its own words: RuleBroken 409, Incomplete / Invalid 422, NotFound 404."""
try:
yield
except Incomplete as e:
raise Refused(422, str(e)) from None
except RuleBroken as e:
raise Refused(409, str(e)) from None
except NotFound as e:
raise Refused(404, str(e)) from None
except Invalid as e:
raise Refused(422, str(e)) from None
def not_found(what: str) -> Refused:
return Refused(404, f"no {what}")
# ── the business clock ─────────────────────────────────────────────────────────────────────
def now() -> datetime:
"""Business time (erp_platform.work.now): a demo may move it; under DODIL_ENV=prod|live it is the real clock."""
return work.aware(work.now()).astimezone(timezone.utc)
def today(at: Optional[datetime] = None, tz: str = "Europe/London") -> date:
return work.aware(at or now()).astimezone(work.zone(tz)).date()
def iso(t) -> Optional[str]:
if t is None:
return None
if isinstance(t, datetime):
return utc_iso(t)
return t.isoformat()
# ── who sees what ──────────────────────────────────────────────────────────────────────────
class View:
"""The caller in contracts, computed once per unit of work: the configuration snapshot, their groups, and what
they may see."""
def __init__(self, uow, caller: dict, at: Optional[datetime] = None):
self.uow, self.caller = uow, caller
self.now = at or now()
self.config = uow.config.load()
self.tz = self.config.setting("timezone", "Europe/London")
self.today = today(self.now, self.tz)
self.me = actor(caller)
self.groups = roles(caller)
@functools.cached_property
def approving(self) -> set:
if not has(self.caller, DECIDE):
return set()
return {r["contract_id"] for r in self.uow.contracts.approvals_for(self.groups)}
@functools.cached_property
def signing(self) -> set:
if not has(self.caller, SIGN):
return set()
return set(self.uow.contracts.signed_by(email=self.caller.get("email") or "", sub=self.caller.get("sub") or ""))
def _mine(self, owner: str, owner_group: str) -> bool:
return (bool(owner) and owner.lower() == self.me) or (bool(owner_group) and owner_group.lower() in self.groups)
def sees(self, head) -> bool:
"""Existence, state and key dates."""
g = (lambda k: getattr(head, k)) if not isinstance(head, dict) else head.get
cid, owner, group, sens = g("contract_id"), g("owner") or "", g("owner_group") or "", g("sensitivity")
if self.caller.get("auth_source") == "public":
return False
if has(self.caller, READ_ALL) or (sens == EMPLOYMENT and has(self.caller, EMPLOYMENT_READ)):
return True
if (has(self.caller, READ) or has(self.caller, WRITE)) and self._mine(owner, group):
return True
return cid in self.approving or cid in self.signing
def body(self, head) -> bool:
"""Sections, variables, versions, executed copy, evidence."""
sens = head.get("sensitivity") if isinstance(head, dict) else head.sensitivity
if not self.sees(head):
return False
return sens != EMPLOYMENT or has(self.caller, EMPLOYMENT_READ)
def manages(self, c) -> bool:
"""May change it (with the use case's own permission): its owner or owner group, or an admin who reads all."""
if not self.body(c):
return False
return self._mine(c.owner, c.owner_group) or has(self.caller, READ_ALL)
def contract(self, contract_id: str, *, body: bool = True, manage: bool = False):
"""The contract, or 404 — also when it exists and the caller may not see it (or its body)."""
c = self.uow.contracts.get(contract_id)
if c is None:
raise not_found(f"contract {contract_id}")
ok = self.manages(c) if manage else (self.body(c) if body else self.sees(c))
if not ok:
raise not_found(f"contract {contract_id}")
return c
# ── the neighbours, as the domain needs them ───────────────────────────────────────────────
def partner(bp_id: int, by: str = "") -> dict:
r = bp.read(int(bp_id), actor=by)
if r.ok:
p = r.partner
if p.get("can_transact") is False or (p.get("status") or "active") not in ("active",):
raise Refused(409, f"partner {bp_id} is {p.get('status') or 'blocked'}: no new contracts with them")
return p
if r.status == "not_found":
raise Refused(422, f"no business partner {bp_id} — the counterparty is always a business partner")
raise Refused(409 if r.status == "absent" else 503, r.reason)
def drafting(uow, cfg, type_code: str, partner_row: dict, template_id: Optional[str] = None) -> Drafting:
ct = cfg.types.get((type_code or "").lower())
if ct is None:
raise Refused(422, f"there is no contract type {type_code!r} (have: {', '.join(sorted(cfg.types)) or 'none'})")
pointers = cfg.templates.get(ct.type_code) or []
if template_id:
pointers = [p for p in pointers if p[0] == template_id]
usable = [p for p in pointers if p[1]]
if not usable:
raise Refused(409, f"no approved template for {ct.label} yet — approve one before drafting")
tid, version = usable[0]
got = uow.templates.version(tid, version, cfg.token)
if got is None:
raise Refused(409, f"template {tid} version {version} is missing")
tmpl, tv = got
pairs = [(s.clause_id, s.clause_version) for s in tv.slots]
words = uow.clauses.words(pairs, cfg.token)
policy = cfg.policies.get(ct.approval_policy)
if policy is None:
raise Refused(409, f"{ct.label}: its approval policy {ct.approval_policy!r} is missing")
return Drafting(ct, tmpl, tv, lambda cid, v: words.get((cid, int(v))), partner_row, cfg, policy, packs.for_type(ct))
def drafting_for(uow, cfg, c, partner_row: Optional[dict] = None) -> Drafting:
"""The drafting context of an EXISTING contract: its own template version (it never moves to a newer one)."""
ct = cfg.types.get(c.type_code)
got = uow.templates.version(c.template_id, c.template_version, cfg.token)
if ct is None or got is None:
raise Refused(409, f"the type or template version of {c.contract_id} is gone")
tmpl, tv = got
words = uow.clauses.words([(s.clause_id, s.clause_version) for s in tv.slots], cfg.token)
policy = cfg.policies.get(ct.approval_policy)
if policy is None:
raise Refused(409, f"{ct.label}: its approval policy {ct.approval_policy!r} is missing")
return Drafting(ct, tmpl, tv, lambda cid, v: words.get((cid, int(v))), partner_row or counterparty_of(c, tv), cfg,
policy, packs.for_type(ct))
def counterparty_of(c, tv) -> dict:
"""The partner as the contract already holds it (its COUNTERPARTY variables), for an edit: bp is read at draft
and refreshed once at send, not on every keystroke."""
from brick_contracts.domain.services import BP_FIELDS
out = {"bp_id": c.bp_id, "roles": [], "kind": "person"}
for d in tv.variables.values():
if d.source == "COUNTERPARTY" and d.name in c.variables:
out[d.validation.get("bp_field") or BP_FIELDS.get(d.name) or d.name] = c.variables[d.name].value
return out
def settle(c, cfg, at: Optional[datetime] = None) -> None:
"""Before every save: when the clock next has something to do with this contract, and how long it is kept."""
from brick_contracts.domain.services import next_chase
from brick_contracts.domain.values import OUT_FOR_SIGNATURE
at = at or now()
chase_at = None
if c.state == OUT_FOR_SIGNATURE:
cadence = cfg.chase_cadence()
for s in c.whose_turn():
started = c.sent_at
done_before = [x.signed_at for x in c.signers if x.signing_order < s.signing_order and x.signed_at]
if done_before:
started = max(done_before)
st = next_chase(cadence, started or at, s.chase_count, s.last_chased_at)
if st.next_at is not None:
chase_at = st.next_at if chase_at is None else min(chase_at, st.next_at)
c.compute_next_due(chase_at)
rule = retention_rule(cfg, c)
if rule is not None:
c.retention_until = c.retention_from(rule, today(at))
def at_version(c, expected: Optional[int]) -> None:
"""An edit made on what the caller saw: a stale version is 409 'changed since you opened it'."""
from erp_platform.repository import Conflict
if expected is not None and int(expected) != int(c.version):
raise Conflict("ct_contract", {"contract_id": c.contract_id}, int(expected), int(c.version))
def retry_unless_versioned(fn):
"""A caller who sent NO version asked for "this change on whatever is there": a lost race is re-run on the
latest. One who sent the version they loaded gets the 409 — they decided on what they saw."""
from erp_platform.repository import retry_on_conflict
@functools.wraps(fn)
def run(caller, *a, **kw):
if kw.get("expected_version") is not None:
return fn(caller, *a, **kw)
return retry_on_conflict(lambda: fn(caller, *a, **kw))
return run
def pack_refusals(cfg, c) -> list:
ct = cfg.types.get(c.type_code)
pk = packs.for_type(ct)
return pk.send_refusals(ct, c) if pk is not None else []
def warnings(cfg, c, day: date) -> list:
ct = cfg.types.get(c.type_code)
pk = packs.for_type(ct)
out = pk.status_warnings(ct, c, day) if pk is not None else []
if c.state == "OUT_FOR_SIGNATURE" and c.link_expires_at and c.link_expires_at - timedelta(days=2) <= now():
out.append("the signing links expire within two days")
return out
def link_days(cfg) -> int:
return max(1, min(90, cfg.int_setting("link_lifetime_days", 14)))
def copy_link_days(cfg) -> int:
return max(1, min(365, cfg.int_setting("signed_copy_link_days", 30)))
def retention_rule(cfg, c):
ct = cfg.types.get(c.type_code)
return cfg.retention.get(ct.retention_rule) if ct else None
def retention_note(cfg, c) -> str:
r = retention_rule(cfg, c)
return retention_statement(r.years_after_end, c.execution == "DEED") if r else ""
# ── presenting a contract ──────────────────────────────────────────────────────────────────
def key_dates_json(c) -> list:
return [{"kind": k.kind, "seq": k.seq, "due_on": k.due_on.isoformat(), "owner_group": k.owner_group,
"reminders_days_before": list(k.days_before), "done": k.done_at is not None, "done_at": iso(k.done_at),
"done_by": k.done_by, "next_reminder_on": iso(k.next_remind_on)}
for k in sorted(c.key_dates, key=lambda k: (k.due_on, k.kind))]
def summary(c, day: date) -> dict:
"""What anyone who may see the contract sees: existence, state and key dates — never its body."""
return {"contract_id": c.contract_id, "type_code": c.type_code, "title": c.title, "state": c.derived_state(day),
"stored_state": c.state, "bp_id": c.bp_id, "owner": c.owner, "owner_group": c.owner_group,
"sensitivity": c.sensitivity, "starts_on": iso(c.starts_on), "ends_on": iso(c.ends_on),
"version": c.current_version or None, "signed_at": iso(c.signed_at), "key_dates": key_dates_json(c),
"amends_contract_id": c.amends_contract_id, "amendment_kind": c.amendment_kind,
"row_version": c.version, "body": "withheld"}
def signer_json(s, link_state: Optional[dict] = None) -> dict:
out = {"signer_no": s.signer_no, "kind": s.kind, "side": s.side, "signing_order": s.signing_order,
"email": s.email, "full_name": s.full_name, "title": s.title, "authority_basis": s.authority_basis,
"status": s.status, "signed_at": iso(s.signed_at), "signed_version": s.signed_version,
"seen": s.first_opened_at is not None, "first_opened_at": iso(s.first_opened_at),
"last_opened_at": iso(s.last_opened_at), "delivered_at": iso(s.delivered_at), "chased": s.chase_count,
"has_link": bool(s.link_id)}
if link_state is not None:
out["link"] = link_state
return out
def full(c, view: View, *, links: Optional[dict] = None) -> dict:
turn = {s.signer_no for s in c.whose_turn()}
out = summary(c, view.today) | {
"body": "shown", "template_id": c.template_id, "template_version": c.template_version,
"source_ref": c.source_ref, "context_ref": c.context_ref, "reference": c.reference,
"draft_fingerprint": c.draft_fingerprint, "missing": list(c.missing), "deviations": list(c.deviations),
"approval_required": c.approval_required, "note": c.last_note or None,
"sections": [{"position": s.position, "family": s.family, "clause_id": s.clause_id,
"clause_version": s.clause_version, "position_kind": s.position_kind,
"deviation_level": s.deviation_level, "overridden": s.overridden, "title": s.title,
"body": s.body} for s in c.sections],
"variables": [{**v.as_json(), "frozen_in_version": c.frozen_in.get(n)} for n, v in sorted(c.variables.items())],
"signers": [signer_json(s, (links or {}).get(s.signer_no)) | {"whose_turn": s.signer_no in turn}
for s in c.signers],
"approvals": [{"seq": a.seq, "state": a.state, "group": a.deciding_group, "approver_group": a.approver_group,
"fallback_group": a.fallback_group, "draft_fingerprint": a.draft_fingerprint,
"current": a.draft_fingerprint == c.draft_fingerprint, "due_at": iso(a.due_at),
"decided_by": a.decided_by, "decided_at": iso(a.decided_at), "reason": a.reason,
"escalated_at": iso(a.escalated_at), "deviations": a.deviations} for a in c.approvals],
"readiness": c.readiness(view.today),
"versions": [{"version": v.version, "status": v.status, "sha256": v.sha256, "frozen_at": iso(v.frozen_at),
"frozen_by": v.frozen_by, "size_bytes": v.size_bytes} for v in sorted(c.versions.values(),
key=lambda v: v.version)],
"executed": ({"sha256": c.executed.sha256, "provenance": c.executed.provenance,
"signed_version": c.executed.signed_version, "filed_at": iso(c.executed.filed_at),
"signed_where": c.executed.signed_where or None, "filename": c.executed.filename}
if c.executed else None),
"tasks": [delivery_json(d) for d in c.deliveries if d.channel == "manual" and d.status == "pending"],
"deliveries": [delivery_json(d) for d in c.deliveries],
"amendments": [{"seq": a.seq, "kind": a.kind, "child_contract_id": a.child_contract_id, "note": a.note,
"recorded_by": a.recorded_by, "recorded_at": iso(a.recorded_at)} for a in c.amendments],
"warnings": warnings(view.config, c, view.today),
"claim": "signed electronically" if c.state in EXECUTED and c.executed and c.executed.provenance == "GENERATED"
else None}
return out
def delivery_json(d) -> dict:
return {"signer_no": d.signer_no, "kind": d.kind, "seq": d.seq, "channel": d.channel, "recipient": d.recipient,
"subject": d.subject, "message": d.message, "status": d.status, "provider_ref": d.provider_ref,
"reason": d.reason or None, "done_by": d.done_by, "done_at": iso(d.done_at)}
__all__ = ["View", "rules", "has", "actor", "roles", "now", "today", "iso", "partner", "drafting", "summary", "full",
"signer_json", "delivery_json", "key_dates_json", "pack_refusals", "warnings", "link_days",
"copy_link_days", "retention_rule", "retention_note", "CATALOG"]