"""Factories — how a Contract comes into being (modules/contracts factories).
ContractFactory from contracts.draft (or draft_contract): type + approved template version + counterparty +
values → a Contract in DRAFT with sections COPIED from the pinned clause versions, variables
resolved, missing[] and deviations[] computed and the approval state set
AmendmentFactory from a signed contract + kind → a new Contract (amends = the original) in DRAFT, prefilled
from the original's FROZEN variables
"""
from dataclasses import dataclass
from datetime import datetime
from typing import Callable, Optional
from brick_contracts.domain.contract import Contract, Section, Signer
from brick_contracts.domain.contract_type import ContractType
from brick_contracts.domain.errors import Invalid, RuleBroken
from brick_contracts.domain.library import Template, TemplateVersion, slots_for
from brick_contracts.domain.policies import Configuration
from brick_contracts.domain.services import DeviationCheck, Resolver, unfilled
from brick_contracts.domain.values import (AMENDMENT_KINDS, EXECUTED, EXTERNAL, INTERNAL, T_APPROVED, THEM, US,
Fingerprint, canonical, contract_id_for)
TERM_NAMES = {"start": ("start_date", "starts_on"), "end": ("end_date", "ends_on", "last_day"),
"notice_days": ("notice_days",), "notice_weeks": ("notice_weeks",)}
def term_of(version: TemplateVersion, variables: dict) -> tuple:
"""(starts_on, ends_on, notice_days) from the variables a definition marks as the term, or by their usual name."""
marked = {d.validation.get("term"): d.name for d in version.variables.values() if d.validation.get("term")}
def pick(role):
names = ([marked[role]] if role in marked else []) + list(TERM_NAMES.get(role, ()))
for n in names:
if n in variables:
return variables[n]
return None
s, e = pick("start"), pick("end")
nd, nw = pick("notice_days"), pick("notice_weeks")
notice = int(nd.value) if nd is not None and nd.type == "int" else (int(nw.value) * 7 if nw is not None and nw.type == "int" else None)
return (s.as_date if s is not None and s.type == "date" else None,
e.as_date if e is not None and e.type == "date" else None, notice)
def request_hash(type_code: str, bp_id: int, context_ref, values: dict, signers) -> str:
"""What makes two contracts.draft calls 'the same request': a retry replays, a different one is 409."""
return Fingerprint.of({"type_code": type_code, "bp_id": int(bp_id), "context_ref": context_ref,
"values": {k: (None if v is None else str(v)) for k, v in sorted((values or {}).items())},
"signers": signers or None}).hex
@dataclass
class Drafting:
"""Everything the factory needs, gathered by the application from the repositories and the seams."""
ctype: ContractType
template: Template
version: TemplateVersion
clause_words: Callable[[str, int], Optional[tuple]] # (clause_id, version) → (title, body, position, level)
partner: dict
config: Configuration
policy: object
pack: object = None # the localisation pack module (packs.gb), or None
class ContractFactory:
@staticmethod
def draft(ctx: Drafting, *, source_ref: str, values: dict, signers: Optional[list], owner: str, owner_group: str,
context_ref: Optional[str], now: datetime, by: str, contract_id: Optional[str] = None) -> Contract:
ct, tv = ctx.ctype, ctx.version
if ct.status != "active":
raise RuleBroken(f"{ct.label} is retired; no new contracts of this type")
if tv.status != T_APPROVED:
raise RuleBroken(f"template {ctx.template.template_id} version {tv.version} is not approved")
ct.check_counterparty(ctx.partner) # invariant 1
extra = {"our_name": ctx.config.setting("our_legal_name", "")}
res = Resolver(tv, ctx.partner, values, at=now, extra=extra).resolve()
variables = dict(res.variables)
if extra["our_name"] and "our_name" not in variables:
from brick_contracts.domain.values import VariableValue
variables["our_name"] = VariableValue.parse("our_name", "text", extra["our_name"], "TEMPLATE")
cond_values = {k: v.value for k, v in variables.items()} | {str(k).lower(): str(v) for k, v in
(values or {}).items() if v is not None}
sections = []
for i, slot in enumerate(slots_for(tv, cond_values), start=1):
words = ctx.clause_words(slot.clause_id, slot.clause_version)
if words is None:
raise RuleBroken(f"slot {slot.slot_no} pins {slot.clause_id}@{slot.clause_version}, which is gone")
title, body, position, level = words
sections.append(Section(i, slot.family, slot.clause_id, slot.clause_version, position, int(level), title,
body, slot.slot_no, slot.kind == "required"))
missing = set(res.missing) | set(unfilled(sections, variables))
if ctx.pack is not None:
missing |= set(ctx.pack.draft_missing(ct, variables, ctx.config))
cid = contract_id or contract_id_for(source_ref, ct.type_code)
c = Contract(cid, ct.type_code, ctx.template.template_id, tv.version, int(ctx.partner["bp_id"]), source_ref,
(owner or "").lower(), by, now, title=f"{ctx.template.name} — {ctx.partner.get('legal_name') or ''}".strip(" —"),
owner_group=owner_group or "", context_ref=context_ref, sensitivity=ct.sensitivity,
execution=ct.execution)
c.variables, c.sections = variables, sections
c.starts_on, c.ends_on, c.notice_days = term_of(tv, variables)
for s in ContractFactory._signers(ctx, signers):
c.signers.append(s)
c.request_hash = request_hash(ct.type_code, c.bp_id, context_ref, values, signers)
deviations = DeviationCheck(ctx.policy, tv).check(sections, variables)
c.review(sorted(missing), deviations, now=now, by=by)
if ctx.pack is not None:
ctx.pack.draft_key_dates(ct, c, ctx.config)
return c
@staticmethod
def _signers(ctx: Drafting, given: Optional[list]) -> list:
out = []
if given:
for i, g in enumerate(given, start=1):
side = str(g.get("side") or "").upper()
if side not in (US, THEM):
raise Invalid(f"signer {i}: side is US or THEM")
email = str(g.get("email") or "").strip().lower()
name = str(g.get("full_name") or "").strip()
if not name:
raise Invalid(f"signer {i}: a signer has a name")
kind = INTERNAL if side == US else EXTERNAL
if kind == EXTERNAL and "@" not in email:
raise Invalid(f"signer {i} ({name}): the counterparty's signer needs an email address")
order = int(g.get("signing_order") or i)
out.append(Signer(i, kind, side, order, email, name, str(g.get("title") or ""),
str(g.get("authority_basis") or ""),
bp_id=int(ctx.partner["bp_id"]) if side == THEM else None,
pool_sub=g.get("pool_sub")))
return out
# No signers given: our default signatory for this type (a settings row), and the partner when it is a person
# with an address. An organisation's signer is a person the owner names — readiness says so until then.
default = ctx.config.json_setting(f"default_signer:{ctx.ctype.type_code}") or ctx.config.json_setting("default_signer")
n = 0
if isinstance(default, dict) and (default.get("email") or default.get("pool_sub")):
n += 1
out.append(Signer(n, INTERNAL, US, int(default.get("signing_order") or 2),
str(default.get("email") or "").lower(), str(default.get("full_name") or "Our signatory"),
str(default.get("title") or ""), str(default.get("authority_basis") or ""),
pool_sub=default.get("pool_sub")))
p = ctx.partner
email = str(p.get("signer_email") or p.get("work_email") or "").strip().lower()
if p.get("kind") == "person" and "@" in email:
n += 1
out.append(Signer(n, EXTERNAL, THEM, 1, email, str(p.get("legal_name") or ""), bp_id=int(p["bp_id"])))
return out
class AmendmentFactory:
@staticmethod
def draft(parent: Contract, kind: str, ctx: Drafting, *, changes: dict, now: datetime, by: str, owner: str,
seq: int) -> Contract:
kind = (kind or "").upper()
if kind not in AMENDMENT_KINDS or kind == "ADMINISTRATIVE":
raise Invalid("an amendment that is signed is AMENDMENT, RENEWAL or TERMINATION")
if parent.state not in EXECUTED:
raise RuleBroken("only a signed contract is amended, renewed or terminated")
values = {}
for name, v in parent.variables.items():
values[name] = v.value
if v.currency:
values.setdefault("currency", v.currency)
values.update({str(k).lower(): v for k, v in (changes or {}).items()})
signers = [{"side": s.side, "full_name": s.full_name, "email": s.email, "title": s.title,
"authority_basis": s.authority_basis, "signing_order": s.signing_order, "pool_sub": s.pool_sub}
for s in sorted(parent.signers, key=lambda s: s.signer_no)]
src = f"manual:{kind.lower()}/{parent.contract_id}/{seq}"
child = ContractFactory.draft(ctx, source_ref=src, values=values, signers=signers, owner=owner,
owner_group=parent.owner_group, context_ref=parent.context_ref, now=now, by=by)
child.amends_contract_id, child.amendment_kind = parent.contract_id, kind
child.title = f"{kind.title()} of {parent.title}"
return child
__all__ = ["ContractFactory", "AmendmentFactory", "Drafting", "term_of", "request_hash"]