dodil/erp-brick-payments
DODIL ERP brick: payments
src/brick_payments/application/cash.py
"""Petty cash and tills (PAY-26): a cash journal per till, a day opened with the last counted balance, entries dated TODAY (a
customer's cash is a receipt, an expense posts to its account with the VAT tax computes, a deposit leaves for the bank through the
transfer clearing account and is matched on the statement), a count by denomination, and a close that posts the difference to cash
over/short - approved first by someone who did not count it when it is above cash_difference_threshold.
ONE claim key per day (scope cashday, key <journal>:<day>) is taken by every operation that changes the day - open, entry, count,
approval, close - so an entry and a close racing for the same open day collide (409) instead of both landing."""
import json
from datetime import date
from decimal import Decimal
from typing import Annotated, Literal, Optional
from erp_platform.bricks import Refused
from erp_platform.types import BigId, Name, text
from pydantic import Field
from brick_payments.application import statements as st
from brick_payments.application import use_cases as uc
from brick_payments.application.inputs import Code, Currency, Day, Money, Money0, Reason, _In
from brick_payments.domain.errors import RuleBroken
from brick_payments.domain.money import ZERO
from brick_payments.domain.receipt import Receipt
from brick_payments.infrastructure import seams
PAY = uc.PAY
WRITE, APPROVE = "pay:cash_journal:write", "pay:cash_journal:approve"
CLEARING = {"purpose": "transfer_clearing"}
OVER_SHORT = {"purpose": "cash_over_short"}
class Denomination(_In):
value: Money
count: Annotated[int, Field(ge=0, le=1_000_000)]
def _journal(repo, cash_journal_id: str) -> dict:
j = repo.cash.journal(cash_journal_id)
if j is None:
raise Refused(404, f"no cash journal {cash_journal_id}")
return j
def _key(cash_journal_id: str, day: date) -> str:
return f"{cash_journal_id}:{day.isoformat()}"
def _claim(repo, cash_journal_id: str, day: date, who: str, what: str) -> int:
return repo.claims.take("cashday", _key(cash_journal_id, day), uc._now(), who, what)
def _expected(day_row: dict, entries: list) -> Decimal:
delta = sum((Decimal(str(e["amount"])) * (1 if e["kind"] == "receipt" else -1) for e in entries), ZERO)
return Decimal(str(day_row["opening_balance"])) + delta
def _today_open(repo, cash_journal_id: str) -> tuple:
today = uc._today(repo)
d = repo.cash.day(cash_journal_id, today)
if d is None:
raise RuleBroken(f"The cash day {today} of {cash_journal_id} is not open. Open it first - entries are dated today and never earlier or later")
if d["lifecycle"] == "closed":
raise RuleBroken(f"The cash day {today} of {cash_journal_id} is closed and refuses entries")
return today, d
def day_view(repo, j: dict, d: dict) -> dict:
entries = repo.cash.entries(j["cash_journal_id"], d["day"])
counts = repo.cash.counts(j["cash_journal_id"], d["day"])
expected = _expected(d, entries)
counted = Decimal(str(counts[-1]["counted_amount"])) if counts else None
threshold = repo.settings.current().cash_difference_threshold
diff = (counted - expected) if counted is not None else None
needs = diff is not None and abs(diff) > threshold
return {"cash_journal_id": j["cash_journal_id"], "day": d["day"].isoformat(), "lifecycle": d["lifecycle"], "currency": j["currency"],
"opening_balance": f"{Decimal(str(d['opening_balance'])):.2f}", "expected_balance": f"{expected:.2f}",
"counted_amount": None if counted is None else f"{counted:.2f}", "difference": None if diff is None else f"{diff:.2f}",
"needs_approval": needs, "approved_by": d["approved_by"] or None, "closed_by": d["closed_by"] or None,
"entries": [{"seq": e["seq"], "kind": e["kind"], "amount": f"{Decimal(str(e['amount'])):.2f}", "description": e["description"],
"payment_id": e["payment_id"] or None, "account_id": e["account_id"] or None, "tax_code": e["tax_code"] or None,
"vat_amount": None if e["vat_amount"] is None else f"{Decimal(str(e['vat_amount'])):.2f}",
"journal_id": e["journal_id"], "statement_line_id": e["statement_line_id"] or None} for e in entries]}
def _post(repo, j: dict, source: str, event: str, on: date, memo: str, lines: list, who: str) -> str:
posted = seams.post_journal(source, event, on, j["currency"], memo[:200], lines, who, j.get("company_code") or "")
if not posted.get("posted"):
raise RuleBroken(f"The ledger did not take the posting ({posted.get('reason', 'no reason given')}). Nothing was recorded; try again")
return posted["journal_id"]
@PAY.use_case(kind="command", route="POST /cash-journals/{cash_journal_id}", permission=WRITE, idempotent=True)
def put_cash_journal(caller: dict, cash_journal_id: Code, name: Name, currency: Currency, gl_account_id: Code,
company_code: text(4, min_length=1), opening_balance: Optional[Money0] = None) -> dict:
"""Set up a till or petty-cash box: its currency, the ledger account it posts to and the company, with the balance it starts from.
The opening balance cannot change once a day has been opened."""
with uc.repository() as repo, uc._decide():
old = repo.cash.journal(cash_journal_id)
if old and repo.cash.days(cash_journal_id) and (opening_balance is not None and Decimal(str(old["opening_balance"])) != opening_balance):
raise RuleBroken(f"Cash journal {cash_journal_id} has days already; its opening balance can no longer change")
company = company_code.strip().upper()
if (old or {}).get("company_code") != company and seams.company(company, uc._who(caller)) is None:
raise Refused(422, f"the ledger has no company {company}")
row = {"cash_journal_id": cash_journal_id, "company_code": company, "currency": uc._cur(currency), "gl_account_id": gl_account_id, "name": name,
"status": "open", "opening_balance": opening_balance if opening_balance is not None else Decimal((old or {}).get("opening_balance") or 0)}
repo.cash.put_journal(row)
return {**{k: row[k] for k in ("cash_journal_id", "company_code", "currency", "gl_account_id", "name", "status")},
"opening_balance": f"{Decimal(str(row['opening_balance'])):.2f}"}
@PAY.use_case(kind="query", route="GET /cash-journals", permission=uc.READ)
def list_cash_journals(caller: dict) -> dict:
"""The tills and petty-cash boxes: currency, company, ledger account and opening balance."""
with uc.repository() as repo:
return {"cash_journals": [{**{k: j[k] for k in ("cash_journal_id", "name", "currency", "company_code", "gl_account_id", "status")},
"opening_balance": f"{Decimal(str(j['opening_balance'])):.2f}"} for j in repo.cash.journals()]}
@PAY.use_case(kind="command", route="POST /cash-journals/{cash_journal_id}/open-day", permission=WRITE, idempotent=True)
def open_cash_day(caller: dict, cash_journal_id: Code) -> dict:
"""Open today's cash day. The previous day must be closed; the opening balance is the last counted balance (the journal's own
opening balance for the first day). Opening again the same day answers the open day."""
with uc.repository() as repo, uc._decide():
j = _journal(repo, cash_journal_id)
today, who = uc._today(repo), uc._who(caller)
days = repo.cash.days(cash_journal_id)
if any(d["day"] == today for d in days):
return day_view(repo, j, next(d for d in days if d["day"] == today))
prev = days[-1] if days else None
if prev and prev["lifecycle"] != "closed":
raise RuleBroken(f"The cash day {prev['day']} is still open. Count and close it before opening {today}")
opening = Decimal(str(prev["counted_amount"])) if prev else Decimal(str(j["opening_balance"]))
row = {"cash_journal_id": cash_journal_id, "day": today, "lifecycle": "open", "opening_balance": opening, "expected_balance": opening,
"counted_amount": None, "difference": None, "counted_by": "", "approved_by": "", "approval_reason": "", "closed_by": "",
"closed_at": None, "difference_journal_id": ""}
_claim(repo, cash_journal_id, today, who, "open")
repo.cash.add_day(row, uc._now())
return day_view(repo, j, row)
@PAY.use_case(kind="command", route="POST /cash-journals/{cash_journal_id}/entries", permission=WRITE)
def record_cash_entry(caller: dict, cash_journal_id: Code, kind: Literal["receipt", "expense", "deposit"], amount: Money,
description: Optional[Reason] = None, bp_id: Optional[BigId] = None, account_id: Optional[Code] = None,
tax_code: Optional[Code] = None, receipt_image: Optional[text(300)] = None) -> dict:
"""An entry of today's open cash day, dated today. receipt: a customer paid cash - a receipt (method cash) allocated like any other.
expense: posts to account_id with tax_code (E for none); the VAT is computed by tax (direction purchase) and reclaimed only with the
receipt image kept. deposit: cash leaving for the bank, through transfer clearing, matched on the statement as a cash_deposit."""
with uc.repository() as repo, uc._decide():
j = _journal(repo, cash_journal_id)
today, d = _today_open(repo, cash_journal_id)
who, now = uc._who(caller), uc._now()
n = _claim(repo, cash_journal_id, today, who, f"entry:{kind}")
seq, doc = n - 1, None
doc = f"{cash_journal_id}:{today.isoformat()}:{seq}"
g, cash = f"{amount:.2f}", seams.account_line(j["gl_account_id"])
entry = {"cash_journal_id": cash_journal_id, "day": today, "seq": seq, "kind": kind, "amount": amount, "currency": j["currency"], "bp_id": None,
"payment_id": "", "account_id": account_id or "", "tax_code": tax_code or "", "description": description or "", "receipt_image": receipt_image or "",
"vat_amount": None, "vat_reclaimed": "", "journal_id": "", "statement_line_id": "", "recorded_by": who, "recorded_at": now}
extra: dict = {}
if kind == "receipt":
if bp_id is None:
raise RuleBroken("A customer's cash is a receipt on that customer. Name the customer (bp_id)")
known = seams.check_customer(bp_id, caller)
if known.status != "known":
raise Refused(known.http, known.note)
settings = repo.settings.current()
r = Receipt.record(payment_id=f"cash:{doc}", bp_id=bp_id, currency=j["currency"], amount=amount, received_on=today, today=today,
tz=settings.business_timezone, by=who, at=now, method="cash", bank_reference=f"cash {doc}",
remittance_text=description or "", origin=f"cash:{doc}")
repo.receipts.add(r)
entry.update(bp_id=bp_id, payment_id=r.payment_id)
lines = [{**cash, "debit": g}, {**seams.receivable_line(bp_id), "credit": g}]
entry["journal_id"] = _post(repo, j, doc, "cash_entry", today, f"Cash receipt {doc}", lines, who)
extra = {"receipt": uc._receipt_view(repo, r)}
elif kind == "expense":
if not (account_id or "").strip() or not (tax_code or "").strip():
raise RuleBroken("A cash expense needs the ledger account it belongs to and a tax code (E for no VAT)")
vat, reclaimed = ZERO, False
if tax_code not in st.NO_VAT:
try:
got = seams.tax_compute(f"pay_cash:{doc}", today, j["currency"], tax_code, amount, description or doc, who)
except seams.Unavailable as down:
raise Refused(down.http, down.reason) from None
vat, reclaimed = got["vat"], bool((receipt_image or "").strip()) and got["vat_reclaimable"] > ZERO
expense = {"account_id": account_id, **({"tax_code": tax_code} if reclaimed else {})}
entry.update(vat_amount=vat, vat_reclaimed="Y" if reclaimed else "")
entry["journal_id"] = _post(repo, j, doc, "cash_entry", today, f"{description or 'Cash expense'} {doc}",
[{**expense, "debit": g}, {**cash, "credit": g}], who)
extra = {"vat": f"{vat:.2f}", "vat_reclaimed": reclaimed}
else:
entry["journal_id"] = _post(repo, j, doc, "transfer", today, f"Cash to bank {doc}", [{**CLEARING, "debit": g}, {**cash, "credit": g}], who)
repo.cash.add_entry(entry, now)
if d.get("approved_by"): # the entry moves the expected balance: the difference the approver accepted is no longer this one
repo.cash.save_day({**d, "approved_by": "", "approval_reason": ""})
return {"entry_id": doc, "seq": seq, "kind": kind, "amount": g, "day": today.isoformat(), "journal_id": entry["journal_id"], **extra}
@PAY.use_case(kind="command", route="POST /cash-journals/{cash_journal_id}/count", permission=WRITE)
def count_cash(caller: dict, cash_journal_id: Code, counted_amount: Optional[Money0] = None, denominations: Optional[list[Denomination]] = None) -> dict:
"""Record the count of today's till: the total, or the notes and coins by denomination (the total is then their sum). A new count
replaces the last one and withdraws any approval given on the old figure. Returns the expected balance and the difference."""
with uc.repository() as repo, uc._decide():
j = _journal(repo, cash_journal_id)
today, d = _today_open(repo, cash_journal_id)
who, now = uc._who(caller), uc._now()
by_value = sum((x.value * x.count for x in denominations or []), ZERO)
if denominations and counted_amount is not None and counted_amount != by_value:
raise RuleBroken(f"The count says {counted_amount} but the notes and coins add up to {by_value:.2f}. Fix one")
total = by_value if denominations else counted_amount
if total is None:
raise Refused(422, "give counted_amount, or the denominations that were counted")
_claim(repo, cash_journal_id, today, who, "count")
seq = len(repo.cash.counts(cash_journal_id, today)) + 1
repo.cash.add_count({"cash_journal_id": cash_journal_id, "day": today, "seq": seq, "counted_amount": total,
"denominations": json.dumps([{"value": f"{x.value:.2f}", "count": x.count} for x in denominations or []]),
"counted_by": who, "counted_by_ids": ",".join(uc._ids(caller)), "counted_at": now}, now)
d = {**d, "approved_by": "", "approval_reason": ""}
repo.cash.save_day(d)
view = day_view(repo, j, d)
expected = Decimal(view["expected_balance"])
diff = total - expected
return {**{k: view[k] for k in ("cash_journal_id", "day", "expected_balance")}, "counted_amount": f"{total:.2f}", "difference": f"{diff:.2f}",
"needs_approval": abs(diff) > repo.settings.current().cash_difference_threshold, "count_no": seq}
@PAY.use_case(kind="command", route="POST /cash-journals/{cash_journal_id}/approve-difference", permission=APPROVE, idempotent=True)
def approve_cash_difference(caller: dict, cash_journal_id: Code, reason: Reason) -> dict:
"""An approver who did not count the till (email AND sub) accepts a difference above cash_difference_threshold, with a reason."""
with uc.repository() as repo, uc._decide():
j = _journal(repo, cash_journal_id)
today, d = _today_open(repo, cash_journal_id)
counts = repo.cash.counts(cash_journal_id, today)
view = day_view(repo, j, d)
if not counts:
raise RuleBroken("Nothing is counted yet, so there is no difference to approve")
if not view["needs_approval"]:
raise RuleBroken(f"The difference {view['difference']} is within the threshold; it needs no approval")
if set(uc._ids(caller)) & {x for x in counts[-1]["counted_by_ids"].split(",") if x}:
raise RuleBroken("The person who counted the till cannot approve its difference. Ask another approver")
who = uc._who(caller)
if d["approved_by"]:
return {**view, "already_approved_by": d["approved_by"]}
_claim(repo, cash_journal_id, today, who, "approve")
d = {**d, "approved_by": who, "approval_reason": reason}
repo.cash.save_day(d)
return day_view(repo, j, d)
@PAY.use_case(kind="command", route="POST /cash-journals/{cash_journal_id}/close-day", permission=WRITE, idempotent=True)
def close_cash_day(caller: dict, cash_journal_id: Code) -> dict:
"""Close today's cash day: it needs a count; a difference above the threshold needs an approver's acceptance first. The difference is
posted to cash over/short (cash_difference), the day is closed once and refuses entries, and the counted amount is tomorrow's opening."""
with uc.repository() as repo, uc._decide():
j = _journal(repo, cash_journal_id)
today = uc._today(repo)
open_days = [d for d in repo.cash.days(cash_journal_id) if d["lifecycle"] == "open"]
if not open_days:
raise RuleBroken("There is no open cash day to close (a day closes once)")
d = open_days[-1]
view = day_view(repo, j, d)
if view["counted_amount"] is None:
raise RuleBroken("Count the cash before closing the day")
diff, who = Decimal(view["difference"]), uc._who(caller)
if view["needs_approval"] and not d["approved_by"]:
raise RuleBroken(f"The difference {view['difference']} is above {repo.settings.current().cash_difference_threshold}. "
"An approver who did not count must accept it first (approve_cash_difference)")
n = _claim(repo, cash_journal_id, d["day"], who, "close")
journal = ""
if diff != ZERO:
g, cash = f"{abs(diff):.2f}", seams.account_line(j["gl_account_id"])
lines = [{**cash, "debit": g}, {**OVER_SHORT, "credit": g}] if diff > 0 else [{**OVER_SHORT, "debit": g}, {**cash, "credit": g}]
journal = _post(repo, j, f"{cash_journal_id}:{d['day'].isoformat()}:{n}", "cash_difference", today, f"Cash {'over' if diff > 0 else 'short'} {d['day']}", lines, who)
row = {**d, "lifecycle": "closed", "expected_balance": Decimal(view["expected_balance"]), "counted_amount": Decimal(view["counted_amount"]),
"difference": diff, "counted_by": repo.cash.counts(cash_journal_id, d["day"])[-1]["counted_by"], "closed_by": who, "closed_at": uc._now(),
"difference_journal_id": journal}
repo.cash.save_day(row)
return {**day_view(repo, j, row), "difference_journal_id": journal}
@PAY.use_case(kind="query", route="GET /cash-journals/{cash_journal_id}/days/{day}", permission=uc.READ)
def get_cash_day(caller: dict, cash_journal_id: Code, day: Day) -> dict:
"""One cash day: opening, entries, expected balance, the count and the difference."""
with uc.repository() as repo:
j = _journal(repo, cash_journal_id)
on = uc._date(day, "day")
d = repo.cash.day(cash_journal_id, on) if on else None
if d is None:
raise Refused(404, f"no cash day {day} for {cash_journal_id}")
return day_view(repo, j, d)