dodil/erp-brick-pos
DODIL ERP brick: point of sale
src/brick_pos/domain/session.py
"""The Session aggregate: one cashier's time on one register between opening the drawer and counting it. The rules, and
nothing else: no database, no HTTP, no identity (a caller's email and sub arrive as plain strings).
open -> counted -> closed count_session closes at once when the difference is within the threshold
open -> force_closed (POS-07: a shift lead, with a reason, when the register died)
closed | force_closed -> posted (POS-20: when gl has the session's journal)
A session never goes back: a counted one is not open again, a closed one is not counted again (a correction is a later
document in a later session).
"""
from dataclasses import dataclass, field
from datetime import date, datetime
from decimal import Decimal
from typing import Optional
from brick_pos.domain.errors import OutOfBounds, RuleBroken
from brick_pos.domain.events import CashMoved, SessionClosed, SessionCounted, SessionForceClosed, SessionOpened, SessionPosted
from brick_pos.domain.values import CashMoveKind, SessionState
def same_person(email_a: str, sub_a: str, email_b: str, sub_b: str) -> bool:
"""The SAME person under email OR sub: a second login keeps the sub, a second address can keep the sub, and either
match is the person. A blank on either side is no evidence."""
ea, eb = (email_a or "").strip().lower(), (email_b or "").strip().lower()
sa, sb = (sub_a or "").strip(), (sub_b or "").strip()
return bool((ea and ea == eb) or (sa and sa == sb))
@dataclass
class CashMove:
"""cash put in or taken out of a drawer other than by a sale. Insert-only: a correction is a later move."""
cash_move_id: str
session_id: str
seq: int
kind: str
amount: Decimal
reason_code: str = ""
approved_by: str = ""
approved_sub: str = ""
created_by: str = ""
created_sub: str = ""
def session_id_of(shop_id: str, register_id: str, session_no: int) -> str:
return f"{shop_id}-{register_id}-{session_no:04d}"
@dataclass
class Session:
session_id: str
shop_id: str = ""
register_id: str = ""
session_no: int = 0
business_date: Optional[date] = None
currency: str = ""
opened_by: str = ""
opened_sub: str = ""
opened_at: Optional[datetime] = None
opening_float: Decimal = Decimal("0.00")
float_difference: Optional[Decimal] = None
counted_by: str = ""
counted_sub: str = ""
counted_at: Optional[datetime] = None
counted_cash: Optional[Decimal] = None
expected_cash: Optional[Decimal] = None
difference: Optional[Decimal] = None
closing_float: Optional[Decimal] = None
approved_by: str = ""
approved_sub: str = ""
approved_at: Optional[datetime] = None
reason: str = ""
status: str = SessionState.OPEN.value
last_receipt_no: Optional[int] = None
claimed_gap_from: Optional[int] = None
claimed_gap_to: Optional[int] = None
posted_at: Optional[datetime] = None
journal_id: str = ""
snapshot_no: Optional[int] = None # the register snapshot this session sells with (POS-30), and its digest (the Z report prints it)
snapshot_digest: str = ""
events: list = field(default_factory=list, repr=False, compare=False)
def _record(self, event):
self.events.append(event)
return event
# ── open_session: none -> open ──────────────────────────────────────────────────────────────────────────────
@classmethod
def open_session(cls, *, shop_id: str, register_id: str, session_no: int, business_date: date, currency: str,
opening_float: Decimal, previous_closing_float: Optional[Decimal], at: datetime, by: str,
sub: str, snapshot_no: Optional[int] = None, snapshot_digest: str = "") -> "Session":
"""The float is counted into the drawer; a count that differs from what the previous session left is recorded as
the float difference (its approval has no transition in the signed workflow: see AGENTS.md)."""
it = cls(session_id=session_id_of(shop_id, register_id, session_no), shop_id=shop_id, register_id=register_id,
session_no=session_no, business_date=business_date, currency=currency, opened_by=by, opened_sub=sub,
opened_at=at, opening_float=opening_float,
float_difference=None if previous_closing_float is None else opening_float - previous_closing_float,
snapshot_no=snapshot_no, snapshot_digest=snapshot_digest)
it._record(SessionOpened(session_id=it.session_id, by=by))
return it
# ── record_cash_move: open -> open ──────────────────────────────────────────────────────────────────────────
def record_cash_move(self, *, seq: int, kind: str, amount: Decimal, reason_code: str, paid_out_limit: Decimal,
approver: Optional[tuple], by: str, sub: str) -> CashMove:
"""A safe drop, a paid-in, a paid-out or a float top-up. A paid-out above the limit needs a shift lead who is not the
cashier whose drawer it is: `approver` is (email, sub) of that lead, or None."""
if self.status != SessionState.OPEN.value:
raise RuleBroken(f"session {self.session_id} is {self.status}: cash moves go in an open session",
gate="session_open", fix="open a session; a counted session is never reopened")
try:
CashMoveKind(kind)
except ValueError:
raise RuleBroken(f"{kind!r} is not a cash move: use {', '.join(k.value for k in CashMoveKind)}") from None
reason = (reason_code or "").strip()
if kind in (CashMoveKind.PAID_OUT.value, CashMoveKind.PAID_IN.value) and not reason:
raise RuleBroken(f"a {kind.replace('_', ' ')} names its reason (MILK, CLEANING, ...)", gate="reason_required",
fix="send reason_code")
a_by = a_sub = ""
if kind == CashMoveKind.PAID_OUT.value and amount > paid_out_limit:
if approver is None:
raise RuleBroken(f"a paid-out above {paid_out_limit} needs a shift lead", gate="paid_out_limit",
fix="ask a shift lead to record it")
if same_person(approver[0], approver[1], self.opened_by, self.opened_sub):
raise RuleBroken("you opened this drawer: a paid-out above the limit is approved by another person",
gate="second_person", fix="ask another shift lead")
a_by, a_sub = approver
move = CashMove(cash_move_id=f"{self.session_id}-cm{seq}", session_id=self.session_id, seq=seq, kind=kind,
amount=amount, reason_code=reason, approved_by=a_by, approved_sub=a_sub, created_by=by,
created_sub=sub)
self._record(CashMoved(self.session_id, move.cash_move_id, kind, amount, by))
return move
# ── count_session: open -> counted (or closed at once) ──────────────────────────────────────────────────────
def count_session(self, *, counted: Decimal, expected: Decimal, threshold: Decimal, closing_float: Optional[Decimal],
register_last: Optional[int], server_last: int, at: datetime, by: str, sub: str) -> Optional[SessionCounted]:
"""The cashier's count against what the drawer should hold. Within the threshold the session closes at once (the
difference is still recorded); above it, it waits for a second person. None: the same count again, changing nothing."""
if self.status != SessionState.OPEN.value:
if self.counted_cash == counted and same_person(by, sub, self.counted_by, self.counted_sub) \
and self.status in (SessionState.COUNTED.value, SessionState.CLOSED.value):
return None
raise RuleBroken(f"session {self.session_id} is {self.status}: it was counted and never goes back to open",
gate="session_open", fix="a correction is a later document in a later session")
if register_last is not None and register_last != server_last:
low, high = sorted((server_last + 1, register_last))
raise RuleBroken(f"the register holds receipts the server has not received: {low:04d} to {high:04d}",
gate="all_synced", fix="bring the register online, or a shift lead force-closes")
if closing_float is not None and closing_float > counted:
raise RuleBroken("the float left in the drawer is not more than what was counted", gate="bounds",
fix="send float_left at most counted_cash")
self.counted_cash, self.expected_cash = counted, expected
self.difference = counted - expected
self.closing_float = counted if closing_float is None else closing_float
self.counted_by, self.counted_sub, self.counted_at = by, sub, at
at_once = abs(self.difference) <= threshold
self.status = SessionState.CLOSED.value if at_once else SessionState.COUNTED.value
return self._record(SessionCounted(self.session_id, counted, expected, self.difference, at_once, by))
# ── approve_difference: counted -> closed ───────────────────────────────────────────────────────────────────
def approve_difference(self, *, reason: str, at: datetime, by: str, sub: str) -> Optional[SessionClosed]:
"""A second person approves the difference with a reason. Never the person who counted (email OR sub). None: the
same approver again with the same reason, changing nothing."""
reason = (reason or "").strip()
if self.status == SessionState.CLOSED.value and self.approved_by and same_person(by, sub, self.approved_by, self.approved_sub) \
and reason == self.reason:
return None
if self.status != SessionState.COUNTED.value:
raise RuleBroken(f"session {self.session_id} is {self.status}: only a counted session waits for approval",
gate="session_counted",
fix="open and count a session" if self.status == SessionState.OPEN.value else "it is already settled")
if not reason:
raise RuleBroken("approving a difference needs the reason, in words", gate="reason_required", fix="send reason")
if same_person(by, sub, self.counted_by, self.counted_sub):
raise RuleBroken("you counted this session", gate="second_person", fix="ask another shift lead")
self.approved_by, self.approved_sub, self.approved_at, self.reason = by, sub, at, reason
self.status = SessionState.CLOSED.value
return self._record(SessionClosed(self.session_id, self.difference, by))
# ── force_close_session: open -> force_closed ───────────────────────────────────────────────────────────────
def force_close(self, *, counted: Decimal, expected: Decimal, closing_float: Optional[Decimal], register_last: Optional[int],
server_last: int, reason: str, at: datetime, by: str, sub: str) -> Optional[SessionForceClosed]:
"""A shift lead closes a session whose register died with receipts on it, counting the drawer. The receipts the register
holds and the server does not are a CLAIMED GAP (server_last + 1 .. register_last), recorded, never filled with another
receipt. Never the person whose drawer it is (email OR sub). None: the same force-close again, changing nothing."""
reason = (reason or "").strip()
if self.status == SessionState.FORCE_CLOSED.value and same_person(by, sub, self.approved_by, self.approved_sub) \
and reason == self.reason and counted == self.counted_cash:
return None
if self.status != SessionState.OPEN.value:
raise RuleBroken(f"session {self.session_id} is {self.status}: only an open session is force-closed", gate="session_open",
fix="a counted session is approved, not force-closed; a closed one never goes back")
if not reason:
raise RuleBroken("a force-close needs the reason, in words", gate="reason_required", fix="send reason")
if same_person(by, sub, self.opened_by, self.opened_sub):
raise RuleBroken("you opened this drawer: another shift lead force-closes it", gate="second_person",
fix="ask another shift lead")
if register_last is not None and register_last < server_last:
raise OutOfBounds(f"the server holds receipts up to {server_last:04d}: the register's last number cannot be lower")
if closing_float is not None and closing_float > counted:
raise RuleBroken("the float left in the drawer is not more than what was counted", gate="bounds",
fix="send float_left at most counted_cash")
self.last_receipt_no = server_last
if register_last is not None and register_last > server_last:
self.claimed_gap_from, self.claimed_gap_to = server_last + 1, register_last
self.counted_cash, self.expected_cash, self.difference = counted, expected, counted - expected
self.closing_float = counted if closing_float is None else closing_float
self.counted_by, self.counted_sub, self.counted_at = by, sub, at
self.approved_by, self.approved_sub, self.approved_at, self.reason = by, sub, at, reason
self.status = SessionState.FORCE_CLOSED.value
return self._record(SessionForceClosed(self.session_id, self.claimed_gap_from, self.claimed_gap_to, self.difference, by))
# ── post_session: closed | force_closed -> posted ───────────────────────────────────────────────────────────
def reviewable_by(self, *, by: str, sub: str) -> None:
"""Whoever approved this session's difference (or force-closed it) does not post it: another person reviews."""
if self.approved_by and same_person(by, sub, self.approved_by, self.approved_sub):
raise RuleBroken("you approved this session's difference: another person posts it", gate="reviewer_not_approver",
fix="ask another accountant")
def mark_posted(self, *, journal_id: str, at: datetime) -> Optional[SessionPosted]:
"""Once gl holds the session's journal. None: already posted. A session never goes back from posted."""
if self.status == SessionState.POSTED.value:
return None
if self.status not in (SessionState.CLOSED.value, SessionState.FORCE_CLOSED.value):
raise RuleBroken(f"session {self.session_id} is {self.status}: only a closed session is posted", gate="session_closed",
fix="count and approve it, or force-close it")
self.status, self.journal_id, self.posted_at = SessionState.POSTED.value, journal_id, at
return self._record(SessionPosted(self.session_id, journal_id))
def expected_cash(opening_float: Decimal, cash_in: Decimal, moves: dict) -> Decimal:
"""What the drawer should hold: the float, plus cash taken (sales and tips), less what went out. Derived from the
session's own rows every time it is asked; never stored as a running total and never incremented."""
g = lambda kind: moves.get(kind, Decimal("0.00")) # noqa: E731
return (opening_float + cash_in + g("paid_in") + g("float_top_up") - g("paid_out") - g("safe_drop"))