dodil/erp-brick-consolidation
DODIL ERP brick: consolidation
src/brick_consolidation/domain/cons_journal.py
"""The ConsJournal aggregate — the rules, and nothing else.
one consolidation entry in group currency, automatic or manual (its own ledger; never in a company's books)
No database, no HTTP, no identity: a rule here is true whoever calls it and however it is stored.
Generated by `erp-compose scaffold` from module.yaml; the TODO lines are the work left.
"""
from dataclasses import dataclass, field
from decimal import Decimal
from typing import Callable, Optional
from brick_consolidation.domain import consolidation as C
from brick_consolidation.domain.errors import RuleBroken
from brick_consolidation.domain.events import (
ConsJournalPosted,
ConsJournalPostedFromProposed,
ConsJournalProposed,
ConsJournalRejected,
ConsJournalReversed,
)
from brick_consolidation.domain.values import ConsJournalDocumentType, ConsJournalState
@dataclass
class ConsJournalLine:
"""one line of a consolidation entry"""
cons_journal_id: str
line_no: int
group_account_id: str = ""
company_code: str = ""
partner_company: str = ""
movement_type: str = ""
segment: str = ""
amount: Decimal = Decimal("0.00")
@dataclass
class ConsJournal:
"""one consolidation entry in group currency, automatic or manual (its own ledger; never in a company's
books)
"""
cons_journal_id: str
group_id: str = ""
period_id: str = ""
document_type: str = ConsJournalDocumentType.TRANSLATION.value
auto: str = ""
task: str = ""
run_no: int = 0
period_specific: str = ""
entry_no: int = 0
proposed_by: str = ""
proposed_sub: str = ""
approved_by: str = ""
approved_sub: str = ""
reverses_id: str = ""
memo: str = ""
migration_run_id: str = ""
status: str = ConsJournalState.POSTED.value
cons_journal_lines: list = field(default_factory=list)
events: list = field(default_factory=list, repr=False, compare=False)
def _record(self, event):
self.events.append(event)
return event
# ── the lifecycle (module.yaml workflows) ──────────────────────────────────────────────────────────────────
@classmethod
def post_cons_auto_entry(cls, cons_journal_id: str, *, by: str = "", **attrs) -> "ConsJournal":
"""Post an automatic consolidation entry: none → posted. Gate: balanced in group currency; the period
not signed; keyed derived_id('cons', group, period, task, company, partner, run_no). Effect: numbered
in the group's book and year; a replay answers the entry posted; the previous run's entry for the key
reversed. Returns the new aggregate.
"""
it = cls(cons_journal_id=cons_journal_id, **attrs)
# TODO(rule): the gate — balanced in group currency; the period not signed; keyed derived_id('cons', group, period, task, company, partner, run_no)
# TODO(rule): the effect — numbered in the group's book and year; a replay answers the entry posted; the previous run's entry for the key reversed
it.status = ConsJournalState.POSTED.value
it._record(ConsJournalPosted(cons_journal_id=it.cons_journal_id, by=by))
return it
@classmethod
def propose_cons_entry(cls, cons_journal_id: str, *, by: str = "", lines=(), scope=frozenset(), rules=(), period_signed: bool = False,
**attrs) -> "ConsJournal":
"""Propose a manual consolidation entry: none → proposed. Gate: balanced in group currency; every line names a company in scope;
the period not signed. Effect: proposer recorded (email and sub); document type and period-specific flag recorded."""
if period_signed:
raise RuleBroken(f"{attrs.get('period_id')} is signed; an entry cannot be proposed in it until it is reopened")
checked = C.entry_lines(list(lines), set(scope), list(rules), attrs["period_id"])
it = cls(cons_journal_id=cons_journal_id, document_type=ConsJournalDocumentType.MANUAL.value, auto="false", **attrs)
it.cons_journal_lines = [ConsJournalLine(cons_journal_id=cons_journal_id, line_no=str(n), **ln) for n, ln in enumerate(checked, 1)]
it.status = ConsJournalState.PROPOSED.value
it._record(ConsJournalProposed(cons_journal_id=it.cons_journal_id, by=by))
return it
def approve_cons_entry(self, *, by: str = "", sub: str = "", second_person_required: bool = True, period_signed: bool = False,
claim_number: Optional[Callable[[], int]] = None) -> Optional[ConsJournalPostedFromProposed]:
"""Approve the entry: proposed → posted. Gate: the approver is not the proposer (email AND sub) where signoff_second_person is on;
the period not signed. Effect: numbered (the number is claimed only once every gate has passed); a period-specific entry's
reversal due at the next period's start (the books of a later period do not carry it)."""
if self.status == ConsJournalState.POSTED:
return None # already posted: a repeat changes nothing
if self.status != ConsJournalState.PROPOSED:
raise RuleBroken(f"a {self.status} cons journal cannot move to posted")
if period_signed:
raise RuleBroken(f"{self.period_id} is signed; the entry cannot be posted until it is reopened")
if second_person_required and not C.second_person(by, sub, self.proposed_by, self.proposed_sub):
raise RuleBroken("the approver cannot be the person who proposed the entry (matched on email and sub)")
self.entry_no = claim_number() if claim_number else 0
self.approved_by, self.approved_sub = by, sub
self.status = ConsJournalState.POSTED.value
return self._record(ConsJournalPostedFromProposed(cons_journal_id=self.cons_journal_id, by=by or sub))
def reject_cons_entry(self, reason: str, *, by: str = "") -> Optional[ConsJournalRejected]:
"""Reject the entry: proposed → rejected."""
it = self
if it.status == ConsJournalState.REJECTED:
return None # already rejected: a repeat changes nothing
if it.status not in (ConsJournalState.PROPOSED,):
raise RuleBroken(f"a {it.status} cons journal cannot move to rejected")
if not reason:
raise RuleBroken("reason is required")
it.status = ConsJournalState.REJECTED.value
return self._record(ConsJournalRejected(cons_journal_id=it.cons_journal_id, by=by))
def reverse_cons_entry(self, reason: str, *, by: str = "", period_signed: bool = False) -> Optional[ConsJournalReversed]:
"""Reverse the entry: posted → reversed. Gate: the period not signed. Effect: a mirror entry keyed
derived_id(cons_journal_id, 'reversal'); neither deleted.
"""
it = self
if it.status == ConsJournalState.REVERSED:
return None # already reversed: a repeat changes nothing
if it.status not in (ConsJournalState.POSTED,):
raise RuleBroken(f"a {it.status} cons journal cannot move to reversed")
if period_signed:
raise RuleBroken(f"{it.period_id} is signed; a correction is made in a period that is not")
if it.reverses_id:
raise RuleBroken("a reversal is not itself reversed; post a new entry")
if not reason:
raise RuleBroken("reason is required")
it.status = ConsJournalState.REVERSED.value
return self._record(ConsJournalReversed(cons_journal_id=it.cons_journal_id, by=by))