"""Policies as ROWS (modules/contracts policies): approval policies, reminder cadences, retention rules, consent
texts, consumers and settings — read into one immutable Configuration per config version (the snapshot the
infrastructure caches until the version row changes; hr reloaded its config on every request, which is what made
it slow on DataK3).
What is decided here, in code, is only the SHAPE every row must have: a policy rule names a mandatory fallback
group; a retention rule has a term; a consumer names at least one principal.
"""
import json
from dataclasses import dataclass, field
from typing import Any, Optional
from brick_contracts.domain.errors import Invalid
DEVIATION_KINDS = ("position", "override", "bound")
def _list(v: Any) -> list:
if v is None or v == "":
return []
if isinstance(v, str):
try:
v = json.loads(v)
except ValueError:
v = [x.strip() for x in v.split(",") if x.strip()]
return list(v)
@dataclass(frozen=True)
class Deviation:
"""One way a draft is away from the standard. `subject` is the clause family or the variable name."""
kind: str # position | override | bound
subject: str
what: str # in words: "liability cap is a Fallback"
level: int
needs: Optional[str] = None # the approver group (None: no approval needed, e.g. a level-0 Alternate)
fallback: Optional[str] = None
timeout_working_days: int = 2
def as_json(self) -> dict:
return {"what": self.what, "kind": self.kind, "subject": self.subject, "level": str(self.level),
"needs": self.needs, "fallback": self.fallback}
@dataclass(frozen=True)
class ApprovalRule:
rule_no: int
applies_to: str # "*", "position:FALLBACK", "position:ALTERNATE", "override", "bound",
# "family:<family>", "variable:<name>"
approver_group: str
fallback_group: str
timeout_working_days: int = 2
def __post_init__(self) -> None:
if not (self.approver_group or "").strip():
raise Invalid(f"approval rule {self.rule_no}: it must name the approver group")
if not (self.fallback_group or "").strip():
raise Invalid(f"approval rule {self.rule_no}: a fallback approver group is mandatory, so nothing stalls")
if self.fallback_group == self.approver_group:
raise Invalid(f"approval rule {self.rule_no}: the fallback group must be another group")
if not (1 <= int(self.timeout_working_days) <= 60):
raise Invalid(f"approval rule {self.rule_no}: the timeout is 1 … 60 working days")
def specificity(self) -> int:
if self.applies_to.startswith(("family:", "variable:")):
return 3
if self.applies_to.startswith("position:"):
return 2
if self.applies_to in ("override", "bound"):
return 1
return 0
def matches(self, kind: str, subject: str, position: str = "") -> bool:
a = self.applies_to
if a == "*":
return True
if a == kind:
return True
if a == f"position:{position}" and kind == "position":
return True
if a == f"family:{subject}" and kind in ("position", "override"):
return True
return a == f"variable:{subject}" and kind == "bound"
@dataclass(frozen=True)
class ApprovalPolicy:
code: str
rules: tuple
def __post_init__(self) -> None:
if not self.rules:
raise Invalid(f"approval policy {self.code}: it needs at least one rule")
if not any(r.applies_to == "*" for r in self.rules):
raise Invalid(f"approval policy {self.code}: it needs a catch-all rule (applies_to \"*\"), so every "
"deviation has someone to approve it")
def rule_for(self, kind: str, subject: str, position: str = "") -> Optional[ApprovalRule]:
hits = [r for r in self.rules if r.matches(kind, subject, position)]
return max(hits, key=lambda r: (r.specificity(), -r.rule_no)) if hits else None
@dataclass(frozen=True)
class ReminderCadence:
code: str
kind: str # SIGNER_CHASE | KEY_DATE
every_days: int = 3
max_count: int = 3
first_after_days: Optional[int] = None
unit: str = "working_days"
days_before: tuple = ()
def __post_init__(self) -> None:
if self.kind not in ("SIGNER_CHASE", "KEY_DATE"):
raise Invalid(f"cadence {self.code}: kind is SIGNER_CHASE or KEY_DATE")
if self.kind == "KEY_DATE" and not self.days_before:
raise Invalid(f"cadence {self.code}: a key-date cadence lists the days before (e.g. 120, 90, 60, 30)")
if any(int(d) < 0 or int(d) > 730 for d in self.days_before):
raise Invalid(f"cadence {self.code}: days before are 0 … 730")
@dataclass(frozen=True)
class RetentionRule:
"""Executed contracts and their evidence: kept `years_after_end` after the term (Limitation Act 1980 s.5: 6 for
a simple contract, s.8: 12 for a deed). Dead drafts (declined, cancelled, expired unsigned): `dead_draft_days`."""
code: str
years_after_end: int
dead_draft_days: int = 30
basis: str = ""
def __post_init__(self) -> None:
if not (1 <= int(self.years_after_end) <= 100):
raise Invalid(f"retention rule {self.code}: years after the end is 1 … 100")
if not (1 <= int(self.dead_draft_days) <= 3650):
raise Invalid(f"retention rule {self.code}: a dead draft lingers 1 … 3650 days")
@dataclass(frozen=True)
class Consumer:
"""Which identities speak for a source_ref namespace (hr:, crm:). The platform's seam does not say which brick
called (in process the caller is the actor string the consumer passes; across apps its service client), so the
composition lists them: 'sub:hr-system', 'role:hr_service', 'email:…'."""
namespace: str
principals: tuple
owner_group: str = ""
label: str = ""
def __post_init__(self) -> None:
if not self.namespace or ":" in self.namespace or self.namespace == "manual":
raise Invalid("a consumer namespace is a word like hr or crm ('manual' is the staff's own)")
if not self.principals:
raise Invalid(f"consumer {self.namespace}: name at least one principal (sub:…, role:… or email:…)")
for p in self.principals:
if not str(p).startswith(("sub:", "role:", "email:")):
raise Invalid(f"consumer {self.namespace}: a principal is sub:<id>, role:<pool role> or email:<address>")
def matches(self, caller: dict) -> bool:
sub = str(caller.get("sub") or "").lower()
email = str(caller.get("email") or "").lower()
roles = {str(r).lower() for r in caller.get("roles") or []}
for p in self.principals:
kind, _, val = str(p).partition(":")
val = val.strip().lower()
if (kind == "sub" and val == sub) or (kind == "email" and val == email) or (kind == "role" and val in roles):
return True
return False
@dataclass(frozen=True)
class Configuration:
"""Everything a use case reads from configuration rows, at one config version. Immutable: a cached snapshot is
shared by every request until the version row changes."""
token: str = ""
types: dict = field(default_factory=dict) # type_code → ContractType
policies: dict = field(default_factory=dict) # code → ApprovalPolicy
cadences: dict = field(default_factory=dict) # code → ReminderCadence
retention: dict = field(default_factory=dict) # code → RetentionRule
consents: dict = field(default_factory=dict) # version → ConsentText
consumers: dict = field(default_factory=dict) # namespace → Consumer
settings: dict = field(default_factory=dict) # key → value (text)
templates: dict = field(default_factory=dict) # type_code → (template_id, current APPROVED version | None)
def setting(self, key: str, default: Any = None) -> Any:
v = self.settings.get(key)
return default if v is None or v == "" else v
def int_setting(self, key: str, default: int) -> int:
try:
return int(self.setting(key, default))
except (TypeError, ValueError):
return default
def bool_setting(self, key: str, default: bool = False) -> bool:
v = self.setting(key)
return default if v is None else str(v).strip().lower() in ("1", "true", "yes", "on")
def json_setting(self, key: str, default: Any = None) -> Any:
v = self.setting(key)
if v is None:
return default
try:
return json.loads(v) if isinstance(v, str) else v
except ValueError:
return default
def consent(self) -> Optional[Any]:
version = self.setting("consent_text_version")
if version and version in self.consents:
return self.consents[version]
return self.consents[sorted(self.consents)[-1]] if self.consents else None
def consumer_for(self, caller: dict) -> list:
return [c for c in self.consumers.values() if c.matches(caller)]
def chase_cadence(self) -> ReminderCadence:
c = self.cadences.get(self.setting("signer_chase_cadence", "signer-chase"))
return c or ReminderCadence("signer-chase", "SIGNER_CHASE", 3, 3)
def key_date_cadence(self, kind: str) -> tuple:
c = self.cadences.get(self.setting(f"key_date_cadence:{kind}", "")) or self.cadences.get(
self.setting("key_date_cadence", "key-dates"))
if c is not None:
return tuple(int(d) for d in c.days_before)
return (120, 90, 60, 30) if kind == "NOTICE_DEADLINE" else (30, 7)
def rule_from_row(r: dict) -> ApprovalRule:
return ApprovalRule(int(r["rule_no"]), str(r.get("applies_to") or "*"), str(r.get("approver_group") or ""),
str(r.get("fallback_group") or ""), int(r.get("timeout_working_days") or 2))
def cadence_from_row(r: dict) -> ReminderCadence:
fa = r.get("first_after_days")
return ReminderCadence(str(r["cadence_code"]), str(r.get("kind") or "SIGNER_CHASE"), int(r.get("every_days") or 3),
int(r.get("max_count") if r.get("max_count") is not None else 3),
int(fa) if fa not in (None, "") else None, str(r.get("unit") or "working_days"),
tuple(int(x) for x in _list(r.get("days_before"))))
def retention_from_row(r: dict) -> RetentionRule:
return RetentionRule(str(r["rule_code"]), int(r.get("years_after_end") or 6), int(r.get("dead_draft_days") or 30),
str(r.get("basis") or ""))
def consumer_from_row(r: dict) -> Consumer:
return Consumer(str(r["namespace"]).strip().lower(), tuple(str(p) for p in _list(r.get("principals"))),
str(r.get("owner_group") or ""), str(r.get("label") or ""))
__all__ = ["Deviation", "ApprovalRule", "ApprovalPolicy", "ReminderCadence", "RetentionRule", "Consumer",
"Configuration", "rule_from_row", "cadence_from_row", "retention_from_row", "consumer_from_row",
"DEVIATION_KINDS"]