dodil/erp-brick-contracts
src/brick_contracts/infrastructure/tables.py
"""The tables, the aggregates over them, and the unit of work (ContractsRepository) every use case opens.
Every table is prefixed ct_ (modules/shared.yaml prefixes: contracts → ct_). Natural keys only, never SERIAL; every
child table's key starts with its aggregate root's key — how a save finds stale rows at any depth. schema.sql is
generated from this file (scripts/gen_schema.py) plus the capability claim tables erp_platform prints for a
provider (capabilities.ddl("contracts")), and tests/test_schema.py fails when they drift.
Aggregates (erp_platform.repository):
Contract ct_contract (+ variable, section, version, signer, executed_copy, approval, delivery, key_date,
amendment) with an optimistic `version` — two people saving one contract: one gets 409. Its
signature events (ct_signature_event) are APPEND-ONLY: loaded with the aggregate, written only by a
plain INSERT of the next seq, never updated or deleted (within retention).
Clause ct_clause + ct_clause_version — a new version is claimed by a plain INSERT of (clause_id, n).
Template ct_template + ct_template_version + ct_template_slot + ct_variable_def — likewise.
ContractType ct_type, one row.
Configuration rows are small one-table aggregates; ct_config_version holds the token a cached snapshot is checked
against (one tiny read per request instead of reloading every catalogue).
"""
from sqlalchemy import BigInteger, Boolean, Column, Date, Integer, MetaData, String, Table, TIMESTAMP
from erp_platform.repository import Aggregate, Part
md = MetaData()
ID = BigInteger
def _pk(name, type_=String):
return Column(name, type_, primary_key=True, autoincrement=False, nullable=False)
# ── configuration ──────────────────────────────────────────────────────────────────────────
ctype = Table("ct_type", md, _pk("type_code"), Column("label", String), Column("allowed_bp_roles", String),
Column("execution", String), Column("esign_allowed", Boolean), Column("sensitivity", String),
Column("retention_rule", String), Column("approval_policy", String), Column("pack", String),
Column("pack_checks", String), Column("allow_roleless_person", Boolean), Column("status", String),
Column("updated_at", TIMESTAMP), Column("updated_by", String))
approval_policy = Table("ct_approval_policy", md, _pk("policy_code"), _pk("rule_no", Integer),
Column("applies_to", String), Column("approver_group", String), Column("fallback_group", String),
Column("timeout_working_days", Integer), Column("updated_at", TIMESTAMP))
reminder_cadence = Table("ct_reminder_cadence", md, _pk("cadence_code"), Column("kind", String),
Column("every_days", Integer), Column("max_count", Integer), Column("first_after_days", Integer),
Column("unit", String), Column("days_before", String), Column("updated_at", TIMESTAMP))
consent_text = Table("ct_consent_text", md, _pk("version"), Column("text", String), Column("created_at", TIMESTAMP))
retention_rule = Table("ct_retention_rule", md, _pk("rule_code"), Column("years_after_end", Integer),
Column("dead_draft_days", Integer), Column("basis", String), Column("updated_at", TIMESTAMP))
settings = Table("ct_settings", md, _pk("setting_key"), Column("value", String), Column("updated_at", TIMESTAMP),
Column("updated_by", String))
consumer = Table("ct_consumer", md, _pk("namespace"), Column("principals", String), Column("owner_group", String),
Column("label", String), Column("updated_at", TIMESTAMP))
config_version = Table("ct_config_version", md, _pk("scope"), Column("token", String), Column("changed_at", TIMESTAMP))
# ── the library ────────────────────────────────────────────────────────────────────────────
clause = Table("ct_clause", md, _pk("clause_id"), Column("family", String), Column("position", String),
Column("deviation_level", Integer), Column("category", String), Column("status", String),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
clause_version = Table("ct_clause_version", md, _pk("clause_id"), _pk("version", Integer), Column("title", String),
Column("body", String), Column("created_by", String), Column("created_at", TIMESTAMP))
template = Table("ct_template", md, _pk("template_id"), Column("type_code", String), Column("name", String),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
template_version = Table("ct_template_version", md, _pk("template_id"), _pk("version", Integer),
Column("status", String), Column("authored_by", String), Column("authored_at", TIMESTAMP),
Column("approved_by", String), Column("approved_at", TIMESTAMP),
Column("retired_at", TIMESTAMP), Column("note", String))
template_slot = Table("ct_template_slot", md, _pk("template_id"), _pk("version", Integer), _pk("slot_no", Integer),
Column("family", String), Column("kind", String), Column("alternative_of_slot", Integer),
Column("clause_id", String), Column("clause_version", Integer), Column("condition_json", String))
variable_def = Table("ct_variable_def", md, _pk("template_id"), _pk("version", Integer), _pk("name"),
Column("type", String), Column("source", String), Column("required", Boolean),
Column("label", String), Column("validation_json", String), Column("enum_values_json", String),
Column("aliases_json", String), Column("condition_json", String), Column("default_value", String),
Column("currency_var", String))
# ── the contract ───────────────────────────────────────────────────────────────────────────
contract = Table("ct_contract", md, _pk("contract_id"), Column("type_code", String), Column("template_id", String),
Column("template_version", Integer), Column("bp_id", ID), Column("context_ref", String),
Column("source_ref", String), Column("origin", String), Column("owner", String),
Column("owner_group", String), Column("title", String), Column("state", String),
Column("sensitivity", String), Column("execution", String), Column("starts_on", Date),
Column("ends_on", Date), Column("notice_days", Integer), Column("amends_contract_id", String),
Column("amendment_kind", String), Column("draft_fingerprint", String), Column("request_hash", String),
Column("missing_json", String), Column("deviations_json", String), Column("reference", String),
Column("current_version", Integer), Column("sent_at", TIMESTAMP), Column("link_expires_at", TIMESTAMP),
Column("signed_at", TIMESTAMP), Column("principal_statement_issued_at", TIMESTAMP),
Column("last_note", String), Column("retention_until", Date), Column("next_due_at", TIMESTAMP),
Column("created_at", TIMESTAMP), Column("created_by", String), Column("updated_at", TIMESTAMP),
Column("version", Integer))
contract_variable = Table("ct_contract_variable", md, _pk("contract_id"), _pk("name"), Column("type", String),
Column("value", String), Column("currency", String), Column("source", String),
Column("resolved_at", String), Column("frozen_in_version", Integer))
contract_section = Table("ct_contract_section", md, _pk("contract_id"), _pk("position", Integer),
Column("slot_no", Integer), Column("family", String), Column("clause_id", String),
Column("clause_version", Integer), Column("position_kind", String),
Column("deviation_level", Integer), Column("overridden", Boolean), Column("required", Boolean),
Column("title", String), Column("body", String))
contract_version = Table("ct_contract_version", md, _pk("contract_id"), _pk("version", Integer),
Column("object_key", String), Column("sha256", String), Column("size_bytes", BigInteger),
Column("frozen_at", TIMESTAMP), Column("frozen_by", String), Column("draft_fingerprint", String),
Column("status", String), Column("text_body", String))
signer = Table("ct_signer", md, _pk("contract_id"), _pk("signer_no", Integer), Column("kind", String),
Column("side", String), Column("role", String), Column("signing_order", Integer), Column("email", String),
Column("full_name", String), Column("title", String), Column("authority_basis", String),
Column("bp_id", ID), Column("pool_sub", String), Column("status", String), Column("link_id", String),
Column("copy_link_id", String), Column("signed_at", TIMESTAMP), Column("signed_version", Integer),
Column("first_opened_at", TIMESTAMP), Column("last_opened_at", TIMESTAMP),
Column("delivered_at", TIMESTAMP), Column("chase_count", Integer), Column("last_chased_at", TIMESTAMP))
signature_event = Table("ct_signature_event", md, _pk("contract_id"), _pk("seq", Integer), Column("kind", String),
Column("signer_no", Integer), Column("signer_kind", String), Column("version", Integer),
Column("version_sha256_shown", String), Column("email", String), Column("token_ref", String),
Column("pool_sub", String), Column("session_id", String), Column("ip", String),
Column("user_agent", String), Column("typed_name", String),
Column("consent_text_version", String), Column("consent_text", String),
Column("auth_method", String), Column("occurred_at", TIMESTAMP), Column("detail", String),
Column("prev_hash", String), Column("row_hash", String))
executed_copy = Table("ct_executed_copy", md, _pk("contract_id"), Column("object_key", String), Column("sha256", String),
Column("size_bytes", BigInteger), Column("provenance", String), Column("signed_version", Integer),
Column("signed_where", String), Column("filed_by", String), Column("filed_at", TIMESTAMP),
Column("filename", String), Column("content_type", String))
approval = Table("ct_approval", md, _pk("contract_id"), _pk("seq", Integer), Column("draft_fingerprint", String),
Column("deviations_json", String), Column("approver_group", String), Column("fallback_group", String),
Column("original_group", String), Column("due_at", TIMESTAMP), Column("state", String),
Column("decided_by", String), Column("decided_at", TIMESTAMP), Column("reason", String),
Column("requested_by", String), Column("requested_at", TIMESTAMP), Column("escalated_at", TIMESTAMP))
delivery = Table("ct_delivery", md, _pk("contract_id"), _pk("signer_no", Integer), _pk("kind"), _pk("seq", Integer),
Column("channel", String), Column("capability", String), Column("idempotency_key", String),
Column("recipient", String), Column("subject", String), Column("message", String),
Column("plan", String), Column("reason", String), Column("status", String),
Column("provider_ref", String), Column("done_by", String), Column("done_at", TIMESTAMP),
Column("created_at", TIMESTAMP), Column("retry_at", TIMESTAMP))
key_date = Table("ct_key_date", md, _pk("contract_id"), _pk("kind"), _pk("seq", Integer), Column("due_on", Date),
Column("owner_group", String), Column("days_before_json", String), Column("reminded_json", String),
Column("next_remind_on", Date), Column("done_at", TIMESTAMP), Column("done_by", String),
Column("note", String))
amendment = Table("ct_amendment", md, _pk("contract_id"), _pk("seq", Integer), Column("kind", String),
Column("child_contract_id", String), Column("note", String), Column("recorded_by", String),
Column("recorded_at", TIMESTAMP))
# ── bytes, logs, claims ────────────────────────────────────────────────────────────────────
# The bytes of a frozen version or an executed copy, in the bucket, in parts — read only through a use case that
# checked the caller may see it, and streamed; never an object link (a platform object-store helper does not exist
# yet; brick-hr keeps its documents the same way).
object_chunk = Table("ct_object_chunk", md, _pk("object_key"), _pk("part_no", Integer), Column("content_b64", String),
Column("sha256", String), Column("size_bytes", BigInteger), Column("content_type", String),
Column("stored_at", TIMESTAMP))
retention_log = Table("ct_retention_log", md, _pk("run_date", Date), _pk("contract_id"), _pk("category"),
Column("purged_at", TIMESTAMP), Column("detail", String))
tick_claim = Table("ct_tick_claim", md, _pk("job"), _pk("slot"), Column("claimed_at", TIMESTAMP))
CONTRACT_PARTS = (Part(contract_variable, ("contract_id", "name")),
Part(contract_section, ("contract_id", "position")),
Part(contract_version, ("contract_id", "version")),
Part(signer, ("contract_id", "signer_no")),
Part(executed_copy, ("contract_id",)),
Part(approval, ("contract_id", "seq")),
Part(delivery, ("contract_id", "signer_no", "kind", "seq")),
Part(key_date, ("contract_id", "kind", "seq")),
Part(amendment, ("contract_id", "seq")))
CONTRACT_ROOT = Part(contract, ("contract_id",))
CONTRACT = Aggregate(CONTRACT_ROOT, *CONTRACT_PARTS, version="version")
EVENTS = Part(signature_event, ("contract_id", "seq")) # loaded with the contract; appended, never saved
CLAUSE = Aggregate(Part(clause, ("clause_id",)), Part(clause_version, ("clause_id", "version")))
TEMPLATE = Aggregate(Part(template, ("template_id",)), Part(template_version, ("template_id", "version")),
Part(template_slot, ("template_id", "version", "slot_no")),
Part(variable_def, ("template_id", "version", "name")))
CONFIG_TABLES = {"contract_type": (ctype, ("type_code",)), "approval_policy": (approval_policy, ("policy_code", "rule_no")),
"reminder_cadence": (reminder_cadence, ("cadence_code",)), "consent_text": (consent_text, ("version",)),
"retention_rule": (retention_rule, ("rule_code",)), "setting": (settings, ("setting_key",)),
"consumer": (consumer, ("namespace",))}
class ContractsRepository:
"""The unit of work a use case opens: ONE connection (one transaction), with a repository per aggregate. A use
case loads and saves ONE aggregate per unit of work (BRICKS.md); what a commit causes runs in a later one.
(This is the class the scenario harness wires to its storage.)"""
def __init__(self, conn):
from brick_contracts.infrastructure import repositories as r
self.conn = conn
self.contracts = r.ContractRepository(conn)
self.templates = r.TemplateRepository(conn)
self.clauses = r.ClauseRepository(conn)
self.types = r.ContractTypeRepository(conn)
self.config = r.ConfigRepository(conn)
self.objects = r.ObjectStore(conn)
self.retention = r.RetentionLog(conn)
self.claims = r.TickClaims(conn)
__all__ = ["md", "ContractsRepository", "CONTRACT", "CLAUSE", "TEMPLATE", "EVENTS", "CONFIG_TABLES"]