"""The Library: clauses and templates — each its own small aggregate (ct_clause, ct_template).
CLAUSE INVARIANTS
* a clause belongs to a FAMILY at a POSITION (Preferred | Alternate | Fallback) with a deviation level;
* a version a contract has taken — or an approved template pins — is frozen: changing the words is a NEW version
("version 1 is on 3 contracts, amend it into version 2"); old versions stay readable (a general rule);
* a clause a non-retired template version REQUIRES cannot be retired.
TEMPLATE INVARIANTS
* a template version is DRAFT → APPROVED → RETIRED; only an APPROVED version drafts contracts, and an approved
version is immutable (an edit is a new version);
* each slot pins clause_id@version, so approving a template approves those exact words;
* approval needs every slot to pin an existing, active clause version, and an approver other than the author
unless the setting `author_may_approve` says otherwise.
"""
import json
import re
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Callable, Optional
from brick_contracts.domain.errors import Invalid, RuleBroken
from brick_contracts.domain.values import (POSITIONS, SLOT_KINDS, SOURCES, T_APPROVED, T_DRAFT, T_RETIRED, VAR_TYPES,
ClausePosition)
_CODE = re.compile(r"[a-z0-9][a-z0-9_.-]{0,63}")
_VAR = re.compile(r"[a-z][a-z0-9_]{0,63}")
def code(value: Any, what: str) -> str:
v = str(value or "").strip().lower()
if not _CODE.fullmatch(v):
raise Invalid(f"{what} is a short code of letters, digits, '.', '-' and '_' (got {value!r})")
return v
# ── conditions: which slots and variables apply ─────────────────────────────────────────────
def holds(cond: Optional[dict], values: dict) -> bool:
"""A condition over resolved variable values (canonical text): {"var": n, "eq": v} | {"var": n, "in": [..]} |
{"var": n, "set": true} | {"all": [..]} | {"any": [..]} | {"not": c}. None holds."""
if not cond:
return True
if "all" in cond:
return all(holds(c, values) for c in cond["all"])
if "any" in cond:
return any(holds(c, values) for c in cond["any"])
if "not" in cond:
return not holds(cond["not"], values)
v = values.get(cond.get("var"))
if "eq" in cond:
return v is not None and str(v).upper() == str(cond["eq"]).upper()
if "in" in cond:
return v is not None and str(v).upper() in {str(x).upper() for x in cond["in"]}
if "set" in cond:
return (v not in (None, "")) == bool(cond["set"])
raise Invalid(f"a condition is var+eq, var+in, var+set, all, any or not — not {cond}")
def check_condition(cond: Optional[dict]) -> Optional[dict]:
if cond is None or cond == {}:
return None
if not isinstance(cond, dict):
raise Invalid("a condition is an object")
holds(cond, {}) # raises on a malformed condition
return cond
# ── clauses ─────────────────────────────────────────────────────────────────────────────────
@dataclass
class ClauseVersion:
version: int
title: str
body: str
created_by: str = ""
created_at: Optional[datetime] = None
@dataclass
class Clause:
clause_id: str
family: str
position: str
deviation_level: int
category: str = ""
status: str = "active"
versions: dict = field(default_factory=dict) # n → ClauseVersion
new_versions: list = field(default_factory=list) # versions cut in this unit of work (claimed by plain INSERT)
@classmethod
def new(cls, clause_id: str, family: str, position: str, deviation_level: int, title: str, body: str, *,
category: str = "", by: str = "", at: Optional[datetime] = None) -> "Clause":
position = str(position or "").upper()
ClausePosition(position, int(deviation_level)) # the level must suit the position
c = cls(code(clause_id, "clause_id"), code(family, "family"), position, int(deviation_level), category or "")
c._cut(title, body, by, at)
return c
def _cut(self, title: str, body: str, by: str, at) -> int:
title, body = (title or "").strip(), (body or "").strip()
if not title or not body:
raise Invalid("a clause version needs a title and its words")
if len(body) > 20000 or len(title) > 200:
raise Invalid("a clause is at most 20 000 characters, its title 200")
n = max(self.versions, default=0) + 1
self.versions[n] = ClauseVersion(n, title, body, by, at)
self.new_versions.append(n)
return n
@property
def current_version(self) -> int:
return max(self.versions)
@property
def current(self) -> ClauseVersion:
return self.versions[self.current_version]
@property
def place(self) -> ClausePosition:
return ClausePosition(self.position, self.deviation_level)
def amend(self, title: Optional[str], body: str, *, by: str, at) -> int:
"""Cut a new version (CT-01). Contracts holding older versions are untouched — they hold copies."""
if self.status != "active":
raise RuleBroken(f"clause {self.clause_id} is retired; keep a new clause instead")
cur = self.current
if (title or cur.title).strip() == cur.title and body.strip() == cur.body:
return cur.version # the same words: nothing to cut
return self._cut(title or cur.title, body, by, at)
def edit_version(self, version: int, title: Optional[str], body: str, *, contracts_holding: int,
templates_pinning: list) -> None:
"""Change the words of a version IN PLACE — only while nothing holds it."""
v = self.versions.get(int(version))
if v is None:
raise Invalid(f"clause {self.clause_id} has no version {version}")
if contracts_holding:
raise RuleBroken(f"version {version} is on {contracts_holding} contract{'s' if contracts_holding != 1 else ''}, "
f"amend it into version {self.current_version + 1}")
if templates_pinning:
raise RuleBroken(f"version {version} is pinned by the approved template {templates_pinning[0]}; amend it "
f"into version {self.current_version + 1}")
v.title, v.body = (title or v.title).strip(), body.strip()
def retire(self, required_by: list) -> None:
if required_by:
raise RuleBroken(f"clause {self.clause_id} is required by template {required_by[0]} — make the slot "
"optional or point it at another clause first")
self.status = "retired"
def set_place(self, position: str, deviation_level: int) -> None:
ClausePosition(str(position).upper(), int(deviation_level))
self.position, self.deviation_level = str(position).upper(), int(deviation_level)
# ── templates ───────────────────────────────────────────────────────────────────────────────
@dataclass
class Slot:
slot_no: int
family: str
kind: str # required | optional | alternative_of
clause_id: str
clause_version: int
alternative_of_slot: Optional[int] = None
condition: Optional[dict] = None
@classmethod
def from_input(cls, d: dict) -> "Slot":
kind = str(d.get("kind") or "required").lower()
if kind not in SLOT_KINDS:
raise Invalid(f"a slot is {' | '.join(SLOT_KINDS)}, not {kind!r}")
alt = d.get("alternative_of_slot")
if kind == "alternative_of" and alt in (None, ""):
raise Invalid(f"slot {d.get('slot_no')}: an alternative names the slot it replaces (alternative_of_slot)")
if kind == "alternative_of" and not d.get("condition"):
raise Invalid(f"slot {d.get('slot_no')}: an alternative says when it replaces the other (a condition)")
try:
n, cv = int(d["slot_no"]), int(d["clause_version"])
except (KeyError, TypeError, ValueError):
raise Invalid("a slot has a slot_no and pins clause_id@clause_version") from None
if not 1 <= n <= 500:
raise Invalid("slot numbers are 1 … 500")
return cls(n, code(d.get("family") or "", "family"), kind, code(d.get("clause_id"), "clause_id"), cv,
int(alt) if alt not in (None, "") else None, check_condition(d.get("condition")))
def as_json(self) -> dict:
return {"slot_no": self.slot_no, "family": self.family, "kind": self.kind, "clause_id": self.clause_id,
"clause_version": self.clause_version, "alternative_of_slot": self.alternative_of_slot,
"condition": self.condition}
@dataclass
class VariableDef:
name: str
type: str
source: str
required: bool = True
label: str = ""
validation: dict = field(default_factory=dict) # min, max, currency, bp_field, term, default_from, max_length
enum_values: tuple = ()
aliases: dict = field(default_factory=dict) # "medium" → "MEDIUM_LARGE": the caller's words → ours
condition: Optional[dict] = None # the definition applies only when this holds
default_value: Optional[str] = None # the template's standard ("28 days including bank holidays")
currency_var: str = "currency" # money / rate: the variable carrying the currency
@classmethod
def from_input(cls, d: dict) -> "VariableDef":
name = str(d.get("name") or "").strip().lower()
if not _VAR.fullmatch(name):
raise Invalid(f"a variable name is lower-case letters, digits and _ (got {d.get('name')!r})")
t = str(d.get("type") or "text").lower()
if t not in VAR_TYPES:
raise Invalid(f"{name}: a type is one of {', '.join(VAR_TYPES)}")
src = str(d.get("source") or "CALLER").upper()
if src not in SOURCES:
raise Invalid(f"{name}: a source is one of {', '.join(SOURCES)}")
val = d.get("validation") or {}
if not isinstance(val, dict):
raise Invalid(f"{name}: validation is an object (min, max, currency, bp_field, term, default_from)")
enum_values = tuple(str(x).upper() for x in (d.get("enum_values") or []))
if t == "enum" and not enum_values:
raise Invalid(f"{name}: an enum lists its values")
aliases = {str(k).strip().lower(): str(v).upper() for k, v in (d.get("aliases") or {}).items()}
if any(v not in enum_values for v in aliases.values()):
raise Invalid(f"{name}: an alias maps to one of the enum values")
from decimal import Decimal, InvalidOperation
for bound in ("min", "max"):
if bound in val:
try:
Decimal(str(val[bound]))
except InvalidOperation:
raise Invalid(f"{name}: {bound} is a number") from None
if val.get("term") not in (None, "start", "end", "notice_days", "notice_weeks", "probation_end"):
raise Invalid(f"{name}: term is start, end, notice_days, notice_weeks or probation_end")
return cls(name, t, src, bool(d.get("required", True)), str(d.get("label") or name), dict(val), enum_values,
aliases, check_condition(d.get("condition")),
None if d.get("default_value") in (None, "") else str(d.get("default_value")),
str(d.get("currency_var") or "currency"))
def as_json(self) -> dict:
return {"name": self.name, "type": self.type, "source": self.source, "required": self.required,
"label": self.label, "validation": self.validation, "enum_values": list(self.enum_values),
"aliases": self.aliases, "condition": self.condition, "default_value": self.default_value,
"currency_var": self.currency_var}
@dataclass
class TemplateVersion:
version: int
status: str = T_DRAFT
authored_by: str = ""
authored_at: Optional[datetime] = None
approved_by: Optional[str] = None
approved_at: Optional[datetime] = None
retired_at: Optional[datetime] = None
note: str = ""
slots: list = field(default_factory=list) # [Slot], by slot_no
variables: dict = field(default_factory=dict) # name → VariableDef
def ordered_slots(self) -> list:
return sorted(self.slots, key=lambda s: s.slot_no)
@dataclass
class Template:
template_id: str
type_code: str
name: str
versions: dict = field(default_factory=dict) # n → TemplateVersion
new_versions: list = field(default_factory=list)
@classmethod
def new(cls, template_id: str, type_code: str, name: str) -> "Template":
return cls(code(template_id, "template_id"), code(type_code, "type_code"), (name or template_id).strip()[:200])
def keep_draft(self, slots: list, variables: list, *, by: str, at, note: str = "") -> int:
"""Write the template's DRAFT version: the open draft if there is one, else a new version N+1. An
approved version is never changed (CT-03, CT-04)."""
parsed_slots = [s if isinstance(s, Slot) else Slot.from_input(s) for s in slots]
defs = [v if isinstance(v, VariableDef) else VariableDef.from_input(v) for v in variables]
nos = [s.slot_no for s in parsed_slots]
if len(nos) != len(set(nos)):
raise Invalid("slot numbers are unique within a template version")
if not parsed_slots:
raise Invalid("a template has at least one slot")
for s in parsed_slots:
if s.kind == "alternative_of" and s.alternative_of_slot not in nos:
raise Invalid(f"slot {s.slot_no} is an alternative of slot {s.alternative_of_slot}, which is not here")
names = [d.name for d in defs]
if len(names) != len(set(names)):
raise Invalid("variable names are unique within a template version")
open_draft = next((v for v in self.versions.values() if v.status == T_DRAFT), None)
latest = self.current_approved()
if open_draft is None and latest is not None and _same(latest, parsed_slots, defs):
return latest.version # the approved version already says exactly this: nothing to write
if open_draft is None:
n = max(self.versions, default=0) + 1
open_draft = TemplateVersion(n, T_DRAFT, by, at)
self.versions[n] = open_draft
self.new_versions.append(n)
open_draft.slots, open_draft.variables = parsed_slots, {d.name: d for d in defs}
open_draft.authored_by, open_draft.authored_at, open_draft.note = by, at, note or ""
return open_draft.version
def approve(self, version: int, *, by: str, at, clause_lookup: Callable[[str], Optional[Clause]],
author_may_approve: bool = False) -> None:
v = self.versions.get(int(version))
if v is None:
raise Invalid(f"template {self.template_id} has no version {version}")
if v.status != T_DRAFT:
raise RuleBroken(f"version {version} is {v.status.lower()}; only a draft version is approved")
if v.authored_by and by and v.authored_by.lower() == by.lower() and not author_may_approve:
raise RuleBroken(f"you wrote version {version}; someone else approves it (or allow authors to approve "
"their own in settings)")
for s in v.ordered_slots():
c = clause_lookup(s.clause_id)
if c is None or s.clause_version not in c.versions:
raise RuleBroken(f"slot {s.slot_no} pins {s.clause_id}@{s.clause_version}, which does not exist")
if c.status != "active":
raise RuleBroken(f"slot {s.slot_no} pins {s.clause_id}, which is retired")
if c.family != s.family:
raise RuleBroken(f"slot {s.slot_no} is for the {s.family} family but {s.clause_id} is {c.family}")
v.status, v.approved_by, v.approved_at = T_APPROVED, by, at
def retire(self, version: int, *, at) -> None:
v = self.versions.get(int(version))
if v is None:
raise Invalid(f"template {self.template_id} has no version {version}")
if v.status == T_RETIRED:
return
if v.status != T_APPROVED:
raise RuleBroken(f"version {version} is a draft; only an approved version is retired")
v.status, v.retired_at = T_RETIRED, at
def current_approved(self) -> Optional[TemplateVersion]:
ok = [v for v in self.versions.values() if v.status == T_APPROVED]
return max(ok, key=lambda v: v.version) if ok else None
def requires(self, clause_id: str) -> list:
"""'template_id@n' for every non-retired version whose REQUIRED slot pins this clause."""
return [f"{self.template_id}@{v.version}" for v in sorted(self.versions.values(), key=lambda v: v.version)
if v.status != T_RETIRED and any(s.clause_id == clause_id and s.kind == "required" for s in v.slots)]
def pins(self, clause_id: str, clause_version: int) -> list:
return [f"{self.template_id}@{v.version}" for v in self.versions.values()
if v.status == T_APPROVED and any(s.clause_id == clause_id and s.clause_version == clause_version
for s in v.slots)]
def _same(v: TemplateVersion, slots: list, defs: list) -> bool:
return (sorted((s.as_json() for s in v.slots), key=lambda s: s["slot_no"]) ==
sorted((s.as_json() for s in slots), key=lambda s: s["slot_no"]) and
sorted((d.as_json() for d in v.variables.values()), key=lambda d: d["name"]) ==
sorted((d.as_json() for d in defs), key=lambda d: d["name"]))
def slots_for(version: TemplateVersion, values: dict) -> list:
"""The slots a draft takes, in order: optional and required slots whose condition holds, an alternative
replacing the slot it is an alternative of when ITS condition holds (engagement_via = PSC → the PSC warranty)."""
chosen = {}
for s in version.ordered_slots():
if s.kind == "alternative_of":
continue
if holds(s.condition, values):
chosen[s.slot_no] = s
for s in version.ordered_slots():
if s.kind == "alternative_of" and holds(s.condition, values):
base = s.alternative_of_slot
chosen.pop(base, None)
chosen[base] = s # takes the place of the one it replaces
return [chosen[k] for k in sorted(chosen)]
def dumps(v: Any) -> str:
return json.dumps(v, sort_keys=True, ensure_ascii=False)
__all__ = ["Clause", "ClauseVersion", "Template", "TemplateVersion", "Slot", "VariableDef", "holds", "slots_for",
"check_condition", "code", "POSITIONS"]