dodil/erp-brick-pos
DODIL ERP brick: point of sale
src/brick_pos/application/use_cases.py
"""The use cases: what a person (or their agent) can do at the till and in the shop. Each is declared ONCE and served as an API
route and as an MCP tool, calling the same function. Two things are decided here and nowhere else: WHO MAY (the `permission`
on the declaration) and WHICH ROWS (a shop's cashier sees their own shop; a row of another shop is a 404, never a 403).
Slice B adds the closing of the day (force-close, post_session, the cash bag), then the rest of the sale (void, refund, voucher,
card timeout, recipe) and the register's snapshot: see the sections below.
Slice A of pos v1 (signed 2026-10-25): the session (open the till with a float, record cash moves, count the drawer blind,
a second person approves a difference above the threshold) and the sale (price from product, VAT from tax with eat-in or
takeaway, tenders with their tips, a final receipt, stock leaving through inventory from a pending outbox row).
- 'today' is the date in business_timezone with the shop's day cutoff, never the UTC date (domain/businessday.py).
- a money operation takes ONE claim key, a plain INSERT in its own transaction: pay:<receipt_id>:<pay_no>.
- a neighbour is written only from a pending outbox row, after the receipt's own commit (dispatch_outbox).
- a state never goes back: a counted session is not open again, a finalised receipt is never edited, a sent row is not resent.
- an amount outside what the books can hold is 422 {gate: bounds}, never a 500."""
import dataclasses
import json
from contextlib import contextmanager
from datetime import date, datetime, timedelta
from decimal import Decimal
from typing import Optional
from erp_platform import auth, work
from erp_platform import repository as repository_
from erp_platform.after import after_response
from erp_platform.bricks import Brick, Refused
from sqlalchemy.exc import IntegrityError
from brick_pos.application.inputs import (
Amt,
Code,
LineIn,
Reason,
RecipeLineIn,
RecipeModifierIn,
RefundLineIn,
Setting,
VoucherSaleIn,
Word,
)
from brick_pos.domain import journal, money, pricing, refund, snapshot
from brick_pos.domain import recipe as recipes_
from brick_pos.domain.businessday import business_date, parse_cutoff
from brick_pos.domain.deposit import CashDeposit, safe_balance
from brick_pos.domain.errors import OutOfBounds, RuleBroken
from brick_pos.domain.outbox import DEPOSIT, GL, RESTOCK, STOCK, OutboxRow, deposit_batch, goods_issue
from brick_pos.domain.outbox import restock as restock_body
from brick_pos.domain.receipt import LineComponent, Receipt, ReceiptLine
from brick_pos.domain.receipt import refund_of as refund_receipt_of
from brick_pos.domain.session import Session, expected_cash, same_person
from brick_pos.domain.values import (
DEVICE_PROFILES,
FISCAL_PROFILES,
SETTING_DEFS,
OutboxStatus,
PaymentMethod,
ReceiptState,
SessionState,
TerminalStatus,
)
from brick_pos.domain.voucher import Voucher
from brick_pos.infrastructure import seams
from brick_pos.infrastructure.seams import Unavailable
from brick_pos.infrastructure.tables import ClaimLost
READ_ALL = "pos:shop:read_all"
CONTROLS = "pos:settings:controls"
OVERRIDE = "pos:override:approve"
MACHINE = ("pos:receipt:ingest", "pos:fiscal:sign", "pos:outbox:dispatch", "pos:day:derive", "pos:number_series:claim",
"pos:migration:load", "pos:migration:reconcile")
POS = Brick("pos", version="0.2.0", machine_permissions=MACHINE, instructions=(
"Point of sale: open a till session with a counted float, ring up products (the price is product's, the VAT is tax's, "
"eat-in or takeaway where the rules distinguish it), take cash or card tenders (a tip rides on its tender and is a "
"liability to staff, never revenue), and finalise the receipt when the payments equal the bill. A finalised receipt is "
"never edited. Stock leaves the shop through inventory from a pending row after the receipt commits; a sale is never "
"refused for stock. At closing count the drawer WITHOUT seeing what it should hold: a difference above the threshold "
"waits for a second person (never the one who counted); a shift lead may force-close a dead register's session, and a closed session is posted "
"to the books as one journal. An open receipt can be voided; a final one is refunded against the original, the way it was paid, once per quantity. "
"Gift vouchers are sold outside VAT and spent through claims; a card that timed out stays pending until the terminal's answer is sent. A drink "
"takes its recipe's components out of stock, not the drink. The register sells from the snapshot it read when the session opened. Money is "
"exact: send amounts as strings."))
# The process view (BRICKS.md): the workflows as BPMN, GET /process/bpmn.
POS.declare_process("process.yaml")
# ── plumbing ────────────────────────────────────────────────────────────────────────────────────────────────
@contextmanager
def _repository_default():
from erp_platform import db # imported late: it needs BUCKET
with _on_engine(db.engine) as repo:
yield repo
def _lost_a_unique_race(exc: IntegrityError) -> bool:
if repository_.as_conflict(exc) is not None:
return False
text = f"{getattr(exc, 'orig', '')} {exc}"
return "23505" in text or "duplicate key" in text or "UNIQUE constraint" in text
@contextmanager
def _on_engine(engine):
"""One transaction. On DataK3 a racer's duplicate key can surface at COMMIT, after the use case returned: it is caught
here, around the commit, and answered as the retry-safe 409 a conflict seen mid-use-case gets, never a 500."""
from brick_pos.infrastructure.tables import Repository
try:
with engine.begin() as conn:
yield Repository(conn)
except IntegrityError as exc:
if not _lost_a_unique_race(exc):
raise
raise Refused(409, "Someone else wrote the same record a moment ago - load it again and retry") from None
repository = _repository_default # tests swap in one on SQLite
@contextmanager
def _decide():
"""A broken rule is a 409 in the rule's own words (422 for an amount out of bounds), a neighbour that cannot answer is what
it did, a lost claim is the retry-safe 409: through both doors."""
try:
yield
except OutOfBounds as e:
raise Refused(422, str(e), gate=e.gate, fix=e.fix) from None
except RuleBroken as e:
raise Refused(409, str(e), gate=e.gate, fix=e.fix) from None
except Unavailable as e:
raise Refused(e.http, e.reason) from None
except ClaimLost as e:
raise Refused(409, f"{e} was claimed a moment ago by someone else - load the record again and retry",
gate="claim", fix="reload and retry") from None
def _who(caller: dict) -> str:
return caller.get("email") or caller.get("sub") or ""
def _sub(caller: dict) -> str:
return caller.get("sub") or ""
def _now() -> datetime:
return work.now()
def _today(repo) -> date:
return _now().astimezone(work.zone(repo.settings.timezone())).date()
def _cutoff(repo, shop: dict) -> str:
return shop.get("day_cutoff_time") or repo.settings.in_force(_today(repo))["day_cutoff_time"]
def _business_date(repo, shop: dict) -> date:
return business_date(_now(), repo.settings.timezone(), _cutoff(repo, shop))
def _same_version(what: str, current: int, asked: Optional[int]) -> None:
if asked is not None and int(asked) != current:
raise Refused(409, f"{what} changed since you loaded it (you have version {asked}, it is at {current}) - load it again")
def _iso(v):
return v.isoformat() if isinstance(v, (date, datetime)) else v
def _scope(repo, caller: dict) -> Optional[list]:
"""None: every shop (retail operations, accountants, the owner hold pos:shop:read_all). Else the caller's own shops."""
return None if auth.has_permission(caller, READ_ALL) else repo.shops.shops_of(_sub(caller))
def _visible_shop(repo, caller: dict, shop_id: str) -> tuple:
found = repo.shops.get(shop_id)
scope = _scope(repo, caller)
if found is None or (scope is not None and shop_id not in scope):
raise Refused(404, f"no shop {shop_id}")
return found
def _visible_session(repo, caller: dict, session_id: str) -> tuple:
found = repo.sessions.get(session_id)
scope = _scope(repo, caller)
if found is None or (scope is not None and found[0].shop_id not in scope):
raise Refused(404, f"no session {session_id}")
return found
def _visible_receipt(repo, caller: dict, receipt_id: str) -> tuple:
found = repo.receipts.get(receipt_id)
scope = _scope(repo, caller)
if found is None or (scope is not None and found[0].shop_id not in scope):
raise Refused(404, f"no receipt {receipt_id}")
return found
def _view(agg, version: Optional[int] = None) -> dict:
out = {}
for f in dataclasses.fields(agg):
if f.name == "events":
continue
v = getattr(agg, f.name)
out[f.name] = [dataclasses.asdict(x) for x in v] if isinstance(v, list) else v
if version is not None:
out["version"] = version
return out
def _check_code(v: str, what: str) -> str:
v = (v or "").strip()
if not v or any(c in v for c in " /?#%"):
raise Refused(422, f"{what} is a code with no spaces or slashes (LON-SOHO, R1)", gate="bounds")
return v
# ── shops, registers and who works where (POS-01, the part a sale needs) ───────────────────────────────────
@POS.use_case(kind="query", route="GET /shops", permission="pos:shop:read")
def list_shops(caller: dict) -> dict:
"""The shops the caller may see (their own; retail operations, accountants and the owner see all). Ids only."""
with repository() as repo:
scope = _scope(repo, caller)
return {"shops": [s for s in repo.shops.ids() if scope is None or s in scope]}
@POS.use_case(kind="query", route="GET /shops/{shop_id}", permission="pos:shop:read")
def get_shop(caller: dict, shop_id: str) -> dict:
"""One shop with its registers and its version, if the caller may see it."""
with repository() as repo:
shop, version = _visible_shop(repo, caller, shop_id)
return {**shop, "version": version}
def _check_setup(before: dict, after: dict) -> list:
"""POS-01: what a shop names must exist where it lives - the company in gl, the warehouse in inventory (and be that company's, and open),
the consumption location one of that warehouse's internal locations. Only what CHANGED is asked, so saving the same shop again reads
nothing. A neighbour that is not installed has nothing to check against (those fields are listed as `unchecked`); one that is down
refuses the change (502/503) and nothing is saved."""
unchecked = []
moved = lambda key: after[key] != before[key] # noqa: E731
if after["company_code"] and moved("company_code"):
if seams.company(after["company_code"]) is None:
unchecked.append("company_code")
if after["warehouse_id"] and (moved("warehouse_id") or moved("company_code")):
found = seams.warehouse(after["warehouse_id"])
if found is None:
unchecked.append("warehouse_id")
else:
if after["company_code"] and found.get("company_code") not in (None, after["company_code"]):
raise RuleBroken(f"warehouse {after['warehouse_id']} belongs to company {found['company_code']}, the shop to {after['company_code']}",
gate="warehouse_company", fix="name a warehouse of the shop's company")
if found.get("status") not in (None, "active"):
raise RuleBroken(f"warehouse {after['warehouse_id']} is {found['status']}: a shop does not draw stock from it", gate="warehouse_active",
fix="name an active warehouse")
if after["consumption_location_id"] and (moved("consumption_location_id") or moved("warehouse_id")):
if not after["warehouse_id"]:
raise RuleBroken("a consumption location is a place in the shop's warehouse: name the warehouse too", gate="warehouse_known",
fix="send warehouse_id")
places = seams.internal_locations(after["warehouse_id"])
if places is None:
unchecked.append("consumption_location_id")
elif after["consumption_location_id"] not in places:
raise RuleBroken(f"{after['consumption_location_id']} is not an internal location of warehouse {after['warehouse_id']}", gate="location_known",
fix=f"name one of: {', '.join(places) or 'none (create one in inventory)'}")
return unchecked
@POS.use_case(kind="command", route="POST /shops", permission="pos:shop:write", idempotent=True, if_match="version")
def keep_shop(caller: dict, shop_id: Code, version: Optional[int] = None, name: Optional[Word] = None,
company_code: Optional[Code] = None, warehouse_id: Optional[Code] = None,
consumption_location_id: Optional[Code] = None, price_list_id: Optional[Code] = None,
fiscal_profile: Optional[Code] = None, country: Optional[Code] = None, cost_centre: Optional[Code] = None,
day_cutoff_time: Optional[Code] = None, status: Optional[Code] = None) -> dict:
"""Create or change a shop: a place that sells, with its company (gl), the warehouse and location its stock leaves from
(inventory), the price list it sells at (product), its country (where VAT is found) and its fiscal profile. Fields left
out keep their value; saving the same values again changes nothing. The id is the customer's code and is fixed once a
receipt names it."""
with repository() as repo, _decide():
shop_id = _check_code(shop_id, "shop_id")
found = repo.shops.get(shop_id)
if found is not None:
_visible_shop(repo, caller, shop_id)
shop, current = found if found else ({"shop_id": shop_id, "name": "", "company_code": "", "warehouse_id": "",
"consumption_location_id": "", "price_list_id": "", "fiscal_profile": "none",
"country": "", "cost_centre": "", "day_cutoff_time": "", "status": "active",
"registers": []}, 0)
_same_version(f"shop {shop_id}", current, version)
change = {k: v for k, v in dict(name=name, company_code=company_code, warehouse_id=warehouse_id,
consumption_location_id=consumption_location_id, price_list_id=price_list_id,
fiscal_profile=fiscal_profile, country=country, cost_centre=cost_centre,
day_cutoff_time=day_cutoff_time, status=status).items() if v is not None}
if "fiscal_profile" in change and change["fiscal_profile"] not in FISCAL_PROFILES:
raise Refused(422, f"fiscal_profile is one of {', '.join(FISCAL_PROFILES)}", gate="bounds")
if "status" in change and change["status"] not in ("active", "closed"):
raise Refused(422, "status is active or closed", gate="bounds")
if change.get("country") and (len(change["country"]) != 2 or not change["country"].isalpha()):
raise Refused(422, "country is an ISO 3166-1 code like GB", gate="bounds")
if change.get("day_cutoff_time"):
parse_cutoff(change["day_cutoff_time"])
merged = {**shop, **change, "country": (change.get("country") or shop["country"]).upper()}
if found is not None and merged == shop:
return {**shop, "version": current}
unchecked = _check_setup(shop, merged)
return {**merged, "version": repo.shops.save(merged, current), "unchecked": unchecked}
@POS.use_case(kind="command", route="POST /shops/{shop_id}/registers", permission="pos:register:write", idempotent=True,
if_match="version")
def keep_register(caller: dict, shop_id: str, register_id: Code, version: Optional[int] = None,
terminal_id: Optional[Code] = None, fiscal_device_serial: Optional[Code] = None,
fiscal_device_status: Optional[Code] = None, status: Optional[Code] = None) -> dict:
"""Add or change one till of a shop: its card terminal and its fiscal device. A register is retired, never deleted, and a
retired register's id is never reused. A register in a shop whose fiscal profile needs a device (de_kassensichv,
at_rksv) cannot open a session until the device is `ready`."""
with repository() as repo, _decide():
shop, current = _visible_shop(repo, caller, shop_id)
_same_version(f"shop {shop_id}", current, version)
register_id = _check_code(register_id, "register_id")
regs = {r["register_id"]: r for r in shop["registers"]}
reg = regs.get(register_id) or {"shop_id": shop_id, "register_id": register_id, "terminal_id": "",
"fiscal_device_serial": "", "fiscal_device_status": "", "status": "active"}
new = {**reg, **{k: v for k, v in dict(terminal_id=terminal_id, fiscal_device_serial=fiscal_device_serial,
fiscal_device_status=fiscal_device_status, status=status).items() if v is not None}}
if new["status"] not in ("active", "retired"):
raise Refused(422, "status is active or retired", gate="bounds")
if reg["status"] == "retired" and new["status"] != "retired":
raise Refused(409, f"register {register_id} is retired and its id is never reused", gate="register_retired",
fix="add a register with a new id")
if register_id in regs and new == reg:
return {**shop, "version": current}
shop = {**shop, "registers": [r for r in shop["registers"] if r["register_id"] != register_id] + [new]}
return {**shop, "version": repo.shops.save(shop, current)}
@POS.use_case(kind="command", route="POST /shops/{shop_id}/staff", permission="pos:shop:write", idempotent=True)
def assign_shop_staff(caller: dict, shop_id: str, sub: Code, email: Optional[Word] = None) -> dict:
"""Say that a signed-in person (their `sub`) works in this shop: a cashier or shift lead sees only the shops they are
assigned to. Assigning again changes nothing."""
with repository() as repo, _decide():
_visible_shop(repo, caller, shop_id)
before = repo.shops.staff(shop_id)
repo.shops.assign(shop_id, sub, email or "", _who(caller))
return {"shop_id": shop_id, "staff": sorted({*before, sub})}
# ── settings (the controls, dated) ───────────────────────────────────────────────────────────────────────
def _check_setting(key: str, value):
kind = SETTING_DEFS[key][0]
if kind == "bool":
if not isinstance(value, bool):
raise Refused(422, f"{key} is true or false", gate="bounds")
return value
if kind == "decimal":
d = money.amount(value, "GBP", key) # a limit is an amount of money, two places
return format(d, "f")
if kind == "int":
if isinstance(value, bool) or not isinstance(value, int) or not 1 <= value <= 120:
raise Refused(422, f"{key} is a whole number of months, 1 to 120", gate="bounds")
return value
if kind == "time":
parse_cutoff(value)
return str(value)
if kind == "zone":
try:
work.zone(str(value))
except ValueError as e:
raise Refused(422, f"{key}: {e}", gate="bounds") from None
return str(value)
if kind.startswith("choice:"):
if value not in kind.split(":", 1)[1].split("|"):
raise Refused(422, f"{key} is one of {kind.split(':', 1)[1].replace('|', ', ')}", gate="bounds")
return value
raise Refused(422, f"{key}: unknown kind")
@POS.use_case(kind="query", route="GET /settings", permission="pos:shop:read")
def list_pos_settings(caller: dict) -> dict:
"""Every setting with the value in force today, whether it is a control (the owner's), and its dated rows."""
with repository() as repo:
today = _today(repo)
now = repo.settings.in_force(today)
return {"settings": [{"setting_key": k, "value": now[k], "default": d[1], "control": d[2], "changes": d[3]}
for k, d in SETTING_DEFS.items()],
"rows": [{**r, "valid_from": _iso(r["valid_from"])} for r in repo.settings.rows()]}
def _add_setting(repo, caller: dict, setting_key: str, value, valid_from: Optional[str]) -> dict:
clean = _check_setting(setting_key, value)
today = _today(repo)
try:
start = date.fromisoformat(valid_from) if valid_from else (today if not repo.settings.has(setting_key) else None)
except ValueError:
raise Refused(422, "valid_from is a date like 2026-12-01", gate="bounds") from None
if start is None:
raise Refused(422, "a changed setting starts after today: send valid_from", gate="valid_from",
fix="a control changes by a dated row, never retroactively")
if start < today or (start == today and repo.settings.has(setting_key)):
raise Refused(422, f"{setting_key} has a row already: the new one starts after {today}", gate="valid_from",
fix="send a later valid_from")
repo.settings.add(setting_key, start, clean, _who(caller), _now())
return {"setting_key": setting_key, "value": clean, "valid_from": start.isoformat()}
@POS.use_case(kind="command", route="POST /settings/{setting_key}", permission="pos:settings:write", idempotent=True)
def set_pos_setting(caller: dict, setting_key: str, value: Setting, valid_from: Optional[str] = None) -> dict:
"""Change one setting that is not a control (day_cutoff_time, tips_treatment) by adding a DATED row: it takes over on
`valid_from`, which must be after today (the first row of a key may start today). The controls on the people who handle
cash (difference_threshold, blind_close, paid_out_limit, business_timezone) are refused here (403): they are the owner's,
through set_pos_control."""
with repository() as repo, _decide():
if setting_key not in SETTING_DEFS:
raise Refused(404, f"no setting {setting_key}")
if SETTING_DEFS[setting_key][2]:
raise Refused(403, f"missing permission: {CONTROLS} ({setting_key} is a control on the people who handle cash: use set_pos_control)")
return _add_setting(repo, caller, setting_key, value, valid_from)
@POS.use_case(kind="command", route="POST /controls/{setting_key}", permission="pos:settings:controls", idempotent=True)
def set_pos_control(caller: dict, setting_key: str, value: Setting, valid_from: Optional[str] = None) -> dict:
"""The owner changes a control on the people who handle cash (difference_threshold, blind_close, paid_out_limit,
business_timezone) by adding a DATED row that takes over on `valid_from` (after today; the first row of a key may start
today). Held by the owner only: never by a cashier, a shift lead or an accountant, whom the controls bind."""
with repository() as repo, _decide():
if setting_key not in SETTING_DEFS or not SETTING_DEFS[setting_key][2]:
raise Refused(404, f"no control {setting_key}: use set_pos_setting")
return _add_setting(repo, caller, setting_key, value, valid_from)
# ── sessions (POS-03, POS-04, POS-05, POS-06) ────────────────────────────────────────────────────────────
def _summary(repo, s: Session, hide_cash: bool) -> dict:
"""The session's takings, derived from its OWN rows (one statement each, however many receipts): never a running total."""
t, moves, sales = repo.sessions.takings(s.session_id), repo.sessions.move_totals(s.session_id), repo.sessions.sales(s.session_id)
z = money.ZERO
cash, card = t.get("cash", {}), t.get("card", {})
cash_in = cash.get("amount", z) + cash.get("tips", z)
tips = cash.get("tips", z) + card.get("tips", z)
treatment = repo.settings.in_force(s.business_date or _today(repo))["tips_treatment"]
out = {"receipts": sales["receipts"], "voids": sales["voids"], "gross": sales["gross"], "net": sales["net"], "vat": sales["vat"],
"card": card.get("amount", z), "tips": tips, "tips_treatment": treatment,
"tips_payable": tips if treatment == "liability" else z}
if hide_cash:
out["cash"], out["expected_cash"] = None, None
else:
out["cash"] = cash.get("amount", z)
out["expected_cash"] = expected_cash(s.opening_float, cash_in, moves)
out["_cash_in"], out["_moves"] = cash_in, moves
return out
def _session_view(repo, caller: dict, s: Session, version: int) -> dict:
blind = repo.settings.in_force(s.business_date or _today(repo))["blind_close"]
hide = bool(blind) and s.status == SessionState.OPEN.value
summary = _summary(repo, s, hide)
summary.pop("_cash_in"), summary.pop("_moves")
view = _view(s, version)
if hide:
view["expected_cash"] = None
view["summary"] = summary
view["server_last_receipt_no"] = repo.receipts.last_no(s.session_id)
row = repo.outbox.get(GL, s.session_id)
view["posting"] = None if row is None else {"status": row.status, "attempts": row.attempts, "last_error": row.last_error}
return view
@POS.use_case(kind="query", route="GET /sessions", permission="pos:shop:read")
def list_sessions(caller: dict, shop_id: Optional[str] = None) -> dict:
"""The sessions the caller may see (their shops), optionally of one shop. Ids only; get one for the rest."""
with repository() as repo:
scope = _scope(repo, caller)
if shop_id:
scope = [shop_id] if scope is None or shop_id in scope else []
return {"sessions": repo.sessions.ids(scope)}
@POS.use_case(kind="query", route="GET /sessions/{session_id}", permission="pos:shop:read")
def get_session(caller: dict, session_id: str) -> dict:
"""One session with its takings. While the session is open and blind_close is on, the cash it should hold is NOT shown:
the cashier counts first and sees the expected amount only after the count is saved."""
with repository() as repo:
s, version = _visible_session(repo, caller, session_id)
return _session_view(repo, caller, s, version)
def _register_snapshot(repo, shop: dict, day: date) -> "snapshot.Snapshot":
"""The register's snapshot for the business date (POS-30): the shop's whole price list, the item master's tax category of each priced
product, and tax's rates for those categories, read ONCE now. Two registers that open on the same answers share one snapshot (same
digest); a change between two openings is the next snapshot_no. A product the item master does not know, or gives no tax category,
is left out (it is not sellable). A neighbour that is down refuses the opening: a till cannot sell without them, but once open it
needs none of them."""
currency, incl, prices, next_change = seams.price_list_all(shop["price_list_id"], day)
if not currency:
raise Unavailable(502, "the price list answered no currency")
info = seams.catalogue([p.product_id for p in prices], day)
rates, valid_until = seams.rate_table(shop["country"], [i["tax_category"] for i in info.values()], day)
ends = [d for d in (next_change, valid_until) if d]
built = snapshot.build(shop_id=shop["shop_id"], business_date=day, price_list_id=shop["price_list_id"], country=shop["country"],
currency=currency, prices_include_tax=incl, valid_until=min(ends) if ends else None, prices=prices, info=info,
rates=rates)
latest = repo.snapshots.latest(shop["shop_id"], day)
if latest is not None and latest.digest == built.digest:
return latest
built.snapshot_no = (latest.snapshot_no if latest else 0) + 1
repo.snapshots.add(built, _now())
return built
@POS.use_case(kind="query", route="GET /sessions/{session_id}/snapshot", permission="pos:shop:read")
def get_register_snapshot(caller: dict, session_id: str) -> dict:
"""The catalogue, prices and tax rates this session's register sells with, as they were read when it opened (POS-30): the dated copy an
offline till downloads, with its snapshot number and digest (the digest is printed on the Z report). `valid_until` is the day a price or
a rate next changes."""
with repository() as repo:
s, _ = _visible_session(repo, caller, session_id)
snap = repo.snapshots.get(s.shop_id, s.business_date, s.snapshot_no) if s.snapshot_no else None
if snap is None:
raise Refused(404, f"session {session_id} has no register snapshot (it opened before snapshots)")
out = dataclasses.asdict(snap)
return {**out, "business_date": snap.business_date.isoformat(), "valid_until": snap.valid_until.isoformat() if snap.valid_until else None}
@POS.use_case(kind="command", route="POST /sessions", permission="pos:session:open", idempotent=True)
def open_session(caller: dict, shop_id: str, register_id: str, opening_float: Amt) -> dict:
"""Open the till with the float counted into the drawer (session: none -> open). Only when: the register has no other open
session; its fiscal device is ready where the shop's profile needs one; the float is a bounded amount. The business date is
fixed now (business_timezone and the shop's day cutoff: 01:30 belongs to the day before). A float that differs from what
the previous session left is recorded as the float difference."""
with repository() as repo, _decide():
shop, _ = _visible_shop(repo, caller, shop_id)
reg = next((r for r in shop["registers"] if r["register_id"] == register_id), None)
if reg is None or reg["status"] != "active" or shop["status"] != "active":
raise Refused(404, f"no active register {register_id} in shop {shop_id}")
if shop["fiscal_profile"] in DEVICE_PROFILES and reg["fiscal_device_status"] != "ready":
raise RuleBroken("the register's TSE or signature device is not registered", gate="fiscal_device_ready",
fix="Shops: register the device")
if not shop["price_list_id"]:
raise RuleBroken(f"shop {shop_id} has no price list: a till cannot sell without one", gate="price_list",
fix="set price_list_id on the shop")
history = repo.sessions.of_register(shop_id, register_id)
open_now = next((sid for sid, _no, st, _cf in history if st == SessionState.OPEN.value), None)
if open_now:
raise RuleBroken(f"the register already has session {open_now} open", gate="one_open_session", fix="close it first")
day = _business_date(repo, shop)
snap = _register_snapshot(repo, shop, day)
currency = snap.currency
flo = money.amount(opening_float, currency, "the opening float")
previous = next((cf for _sid, _no, st, cf in history if st != SessionState.OPEN.value and cf is not None), None)
session = Session.open_session(shop_id=shop_id, register_id=register_id, session_no=(history[0][1] if history else 0) + 1,
business_date=day, currency=currency, opening_float=flo,
previous_closing_float=None if previous is None else Decimal(str(previous)),
at=_now(), by=_who(caller), sub=_sub(caller), snapshot_no=snap.snapshot_no,
snapshot_digest=snap.digest)
return _session_view(repo, caller, session, repo.sessions.save(session, 0))
@POS.use_case(kind="command", route="POST /sessions/{session_id}/cash-moves", permission="pos:cash:move", idempotent=True,
if_match="version")
def record_cash_move(caller: dict, session_id: str, kind: Code, amount: Amt, reason_code: Optional[Code] = None,
version: Optional[int] = None) -> dict:
"""Record cash going in or out of the drawer (session: open -> open): a safe drop, a paid-in, a paid-out (with its reason)
or a float top-up, so the count at close still adds up. A paid-out above paid_out_limit must be recorded by a shift lead
who did not open the drawer. The amount is bounded (more than zero, within NUMERIC(18,2), the currency's decimals)."""
with repository() as repo, _decide():
s, current = _visible_session(repo, caller, session_id)
_same_version(f"session {session_id}", current, version)
amt = money.amount(amount, s.currency, "the amount", positive=True)
limit = Decimal(str(repo.settings.in_force(s.business_date)["paid_out_limit"]))
approver = (_who(caller), _sub(caller)) if auth.has_permission(caller, OVERRIDE) else None
move = s.record_cash_move(seq=repo.sessions.next_move_seq(session_id), kind=kind, amount=amt,
reason_code=reason_code or "", paid_out_limit=limit, approver=approver,
by=_who(caller), sub=_sub(caller))
repo.sessions.add_move(move, _now())
new_version = repo.sessions.save(s, current) # the version claim serialises moves against the count
return {"cash_move": dataclasses.asdict(move), "session_id": session_id, "version": new_version}
@POS.use_case(kind="command", route="POST /sessions/{session_id}/count", permission="pos:session:count", idempotent=True,
if_match="version")
def count_session(caller: dict, session_id: str, counted_cash: Amt, float_left: Optional[Amt] = None,
register_last_receipt_no: Optional[int] = None, version: Optional[int] = None) -> dict:
"""Count the drawer (session: open -> counted). The expected amount is the float plus the cash taken (sales and tips) less
what moved out, derived from this session's own rows. Within difference_threshold the session closes at once and the
difference is still recorded; above it, it waits for a second person (approve_difference). `register_last_receipt_no`
is the register's own last number: when it is not the server's, the count is refused (gate all_synced). A counted
session never goes back to open."""
with repository() as repo, _decide():
s, current = _visible_session(repo, caller, session_id)
_same_version(f"session {session_id}", current, version)
counted = money.amount(counted_cash, s.currency, "the count")
left = None if float_left is None else money.amount(float_left, s.currency, "the float left")
if register_last_receipt_no is not None and not (0 <= int(register_last_receipt_no) <= 1_000_000):
raise OutOfBounds("register_last_receipt_no is a receipt number: 0 to 1,000,000")
now = _now()
threshold = Decimal(str(repo.settings.in_force(business_date(now, repo.settings.timezone(), "00:00"))["difference_threshold"]))
summary = _summary(repo, s, hide_cash=False)
done = s.count_session(counted=counted, expected=summary["expected_cash"], threshold=threshold, closing_float=left,
register_last=register_last_receipt_no, server_last=repo.receipts.last_no(session_id),
at=now, by=_who(caller), sub=_sub(caller))
if done is None:
return _session_view(repo, caller, s, current)
return _session_view(repo, caller, s, repo.sessions.save(s, current))
@POS.use_case(kind="command", route="POST /sessions/{session_id}/approve-difference",
permission="pos:session:approve_difference", idempotent=True, if_match="version")
def approve_difference(caller: dict, session_id: str, reason: Reason, version: Optional[int] = None) -> dict:
"""Approve a counted session's cash difference with a reason (session: counted -> closed), as a SECOND person: never the
one who counted it, whichever login they use (matched on email OR sub). The difference then posts to cash over/short
with the shop's day."""
with repository() as repo, _decide():
s, current = _visible_session(repo, caller, session_id)
_same_version(f"session {session_id}", current, version)
if s.approve_difference(reason=reason, at=_now(), by=_who(caller), sub=_sub(caller)) is None:
return _session_view(repo, caller, s, current)
return _session_view(repo, caller, s, repo.sessions.save(s, current))
@POS.use_case(kind="command", route="POST /sessions/{session_id}/force-close", permission="pos:session:force_close",
idempotent=True, if_match="version")
def force_close_session(caller: dict, session_id: str, counted_cash: Amt, reason: Reason, float_left: Optional[Amt] = None,
register_last_receipt_no: Optional[int] = None, version: Optional[int] = None) -> dict:
"""A shift lead closes a session whose register died with receipts on it (session: open -> force_closed), counting the drawer
and giving the reason. The receipts the register holds and the server does not are recorded as a CLAIMED GAP
(server's last + 1 to `register_last_receipt_no`), never filled with another receipt, so the shop's day can close and the
receipts can still arrive later. Never the person whose drawer it is (another shift lead, matched on email OR sub)."""
with repository() as repo, _decide():
s, current = _visible_session(repo, caller, session_id)
_same_version(f"session {session_id}", current, version)
counted = money.amount(counted_cash, s.currency, "the count")
left = None if float_left is None else money.amount(float_left, s.currency, "the float left")
if register_last_receipt_no is not None and not (0 <= int(register_last_receipt_no) <= 1_000_000):
raise OutOfBounds("register_last_receipt_no is a receipt number: 0 to 1,000,000")
summary = _summary(repo, s, hide_cash=False)
done = s.force_close(counted=counted, expected=summary["expected_cash"], closing_float=left,
register_last=register_last_receipt_no, server_last=repo.receipts.last_no(session_id), reason=reason,
at=_now(), by=_who(caller), sub=_sub(caller))
if done is None:
return _session_view(repo, caller, s, current)
return _session_view(repo, caller, s, repo.sessions.save(s, current))
@POS.use_case(kind="command", route="POST /sessions/{session_id}/post", permission="pos:day:post", idempotent=True)
def post_session(caller: dict, session_id: str) -> dict:
"""Post a closed (or force-closed) session to the books. Its takings become ONE balanced journal for gl (revenue net of VAT, VAT
per tax code, tenders, tips as a liability to staff, the drawer's difference, paid-outs and safe drops), written as a pending
row now and sent after this commit; the session becomes `posted` once gl holds it. A closed gl period refuses it and the row
waits as failed with gl's reason until a person requeues it. Whoever approved the session's difference does not post it. Asking
again changes nothing."""
with repository() as repo, _decide():
s, current = _visible_session(repo, caller, session_id)
if s.status == SessionState.POSTED.value:
return _session_view(repo, caller, s, current)
if s.status not in (SessionState.CLOSED.value, SessionState.FORCE_CLOSED.value):
raise RuleBroken(f"session {session_id} is {s.status}: only a closed session is posted", gate="session_closed",
fix="count and approve it, or force-close it")
s.reviewable_by(by=_who(caller), sub=_sub(caller))
if repo.outbox.get(GL, session_id) is None:
shop, _ = repo.shops.get(s.shop_id)
if not shop["company_code"]:
raise RuleBroken(f"shop {s.shop_id} has no company: the ledger cannot be told where to post", gate="company_code",
fix="set company_code on the shop")
t, moves = repo.sessions.takings(session_id), repo.sessions.move_totals(session_id)
cash, card, vouch = t.get("cash", {}), t.get("card", {}), t.get("voucher", {})
z = money.ZERO
products, vouchers_sold = repo.sessions.sales_by_code(session_id)
body = journal.session_journal(
session_id=session_id, business_date=s.business_date, currency=s.currency, company_code=shop["company_code"],
cost_centre=shop["cost_centre"] or "", cash=cash.get("amount", z), cash_tips=cash.get("tips", z),
card=card.get("amount", z), card_tips=card.get("tips", z), voucher_redeemed=vouch.get("amount", z),
sales_by_code=products, voucher_sold=vouchers_sold,
tips_treatment=repo.settings.in_force(s.business_date)["tips_treatment"], difference=s.difference or z, moves=moves)
repo.outbox.add(OutboxRow.pending(GL, session_id, body), _now())
after_response(_dispatch, session_id, conn=repo.conn)
return _session_view(repo, caller, s, current)
# ── the sale (POS-08, with the tender and the tip of POS-09) ─────────────────────────────────────────────
def _receipt_view(r: Receipt, version: int) -> dict:
view = _view(r, version)
view["paid"], view["remaining"], view["tips"] = r.paid(), r.remaining(), r.tips()
view["readiness"] = [{"gate": "paid_in_full", "met": r.paid() == r.gross and r.status == ReceiptState.OPEN.value,
"reason": f"the payments come to {r.paid()}, the bill is {r.gross}", "fix": f"take {r.remaining()} more"}]
return view
@POS.use_case(kind="query", route="GET /receipts", permission="pos:shop:read")
def list_receipts(caller: dict, session_id: Optional[str] = None) -> dict:
"""The receipts the caller may see (their shops), optionally of one session. Ids only."""
with repository() as repo:
return {"receipts": repo.receipts.ids(_scope(repo, caller), session_id)}
@POS.use_case(kind="query", route="GET /receipts/{receipt_id}", permission="pos:shop:read")
def get_receipt(caller: dict, receipt_id: str) -> dict:
"""One receipt with its lines (each with the price and VAT it was given), its payments, tips and what is still to pay."""
with repository() as repo:
return _receipt_view(*_visible_receipt(repo, caller, receipt_id))
@POS.use_case(kind="command", route="POST /receipts", permission="pos:receipt:sell", idempotent=True)
def ring_up(caller: dict, session_id: str, lines: list[LineIn] = (), eat_in: Optional[bool] = None,
vouchers: Optional[list[VoucherSaleIn]] = None) -> dict:
"""Ring up products in an open session (receipt: none -> open). Each line is a product and a quantity; the unit price is
product's price list for the shop on the business date (never typed), the VAT rate is tax's for the product's tax
category and, where the rules distinguish it, `eat_in` (true: eat in, false: takeaway; ask the customer when the answer
says it matters). VAT is computed per tax code on the total including tax, rounded once. `vouchers` sells gift vouchers on the
same receipt (a line of kind voucher_sale: outside VAT, a liability, no stock; held by pos:voucher:sell): the voucher is live
when the receipt is final. The receipt number is the next one of the session (a plain INSERT: a racer is told 409)."""
with repository() as repo, _decide():
s, _ = _visible_session(repo, caller, session_id)
if s.status != SessionState.OPEN.value:
raise RuleBroken(f"session {session_id} is {s.status}: sales go in an open session", gate="session_open",
fix="open a session on the register")
shop, _ = repo.shops.get(s.shop_id)
lines, vouchers = list(lines or []), list(vouchers or [])
if vouchers and not auth.has_permission(caller, "pos:voucher:sell"):
raise Refused(403, "missing permission: pos:voucher:sell")
if not lines and not vouchers or len(lines) + len(vouchers) > 200:
raise OutOfBounds("a receipt has one to 200 lines")
built, components = [], []
if lines:
ids = [ln.product_id for ln in lines]
snap = repo.snapshots.get(s.shop_id, s.business_date, s.snapshot_no) if s.snapshot_no else None
if snap is not None: # the register sells with what it read when it opened: nothing is read now
currency, incl, prices, info, rates = snap.currency, snap.prices_include_tax, snap.prices(), snap.info(), snap.rate_rows()
missing = sorted(set(ids) - set(info))
if missing:
raise RuleBroken(f"{missing[0]} is not on the shop's price list on {s.business_date}: it cannot be sold here",
gate="price_known", fix="ask retail operations to put it on the price list: it sells from the next snapshot")
else: # a session opened before snapshots: read the neighbours as the sale is rung
currency, incl, prices = seams.price_list(shop["price_list_id"], s.business_date, set(ids))
info = rates = None
if currency != s.currency:
raise RuleBroken(f"the price list is in {currency}, the session in {s.currency}", gate="currency",
fix="open a new session after the price list's currency changed")
if not incl:
raise RuleBroken("the price list holds prices without VAT: a till sells at the price on the shelf, VAT included",
gate="prices_include_tax", fix="ask retail operations for a tax-inclusive price list")
if info is None:
info = seams.products(ids, s.business_date)
rates = seams.rates(shop["country"], [i["tax_category"] for i in info.values()], s.business_date, eat_in)
built = pricing.build_lines([ln.model_dump() for ln in lines], prices, info, rates, currency, eat_in, s.business_date)
components = _resolve_components(repo, built, lines, s.business_date)
seen = set()
for v in vouchers:
code = _check_code(v.voucher_code, "voucher_code")
if code in seen or repo.vouchers.exists(code):
raise RuleBroken(f"voucher {code} was issued already: a code is sold once", gate="voucher_code",
fix="use the next code on the voucher stock")
seen.add(code)
face = money.amount(v.amount, s.currency, "the voucher's value", positive=True)
built.append(ReceiptLine(line_no=len(built) + 1, product_id=code, description=f"Gift voucher {code}", quantity=Decimal(1),
unit_code="C62", unit_price=face, line_gross=face, tax_category="voucher", tax_code="",
rate_pct=Decimal(0), vat=money.ZERO, kind="voucher_sale", voucher_code=code))
if vouchers and not shop["company_code"]:
raise RuleBroken(f"shop {s.shop_id} has no company: a voucher is a debt of a company", gate="company_code",
fix="set company_code on the shop")
receipt = Receipt.ring_up(session=s, receipt_no=repo.receipts.last_no(session_id) + 1, lines=built, eat_in=eat_in,
by=_who(caller), sub=_sub(caller), components=components)
return _receipt_view(receipt, repo.receipts.save(receipt, 0))
def _resolve_components(repo, built: list, requested: list, day: date) -> list:
"""What each product line takes from stock, frozen on the line: the product itself when it has no recipe, else the recipe's
components (pos's own dated recipe, or manufacturing's usage-sale bill of materials when recipe_source says so). A modifier no
recipe defines is refused. Reads only: the recipe is as dated as the price."""
source = repo.settings.in_force(day)["recipe_source"]
mine = repo.recipes.for_products(sorted({ln.product_id for ln in built})) if source == "pos" else {}
out = []
for ln, req in zip(built, requested):
mods = sorted(set(req.modifiers or []))
if ln.line_no and len(mods) > 10:
raise OutOfBounds("a line has at most 10 modifiers")
picked = None
if source == "manufacturing":
bom = seams.explode(ln.product_id, ln.quantity, ln.unit_code, mods, day)
if bom is not None:
picked = ("manufacturing", bom["bom_id"], bom["version"], bom["components"])
else:
rec = recipes_.pick(mine.get(ln.product_id, []), day)
if rec is not None:
picked = ("pos", rec.product_id, rec.version, recipes_.explode(rec, ln.quantity, mods))
if picked is None:
if mods:
raise RuleBroken(f"{ln.product_id} has no recipe on {day.isoformat()}, so it has no modifier {', '.join(mods)}",
gate="modifier_unknown", fix="ring it up without the modifier")
out.append(LineComponent(ln.line_no, 1, ln.product_id, ln.quantity, ln.unit_code))
continue
ln.recipe_source, ln.recipe_ref, ln.recipe_version, ln.modifiers = picked[0], picked[1], picked[2], ",".join(mods)
for no, (pid, qty, unit) in enumerate(picked[3], start=1):
if qty <= 0 or qty.as_tuple().exponent < -6 or qty >= Decimal("1e12"):
raise OutOfBounds(f"the {pid} {ln.product_id} consumes ({format(qty, 'f')}) is not a quantity the stock can hold")
out.append(LineComponent(ln.line_no, no, pid, qty, unit))
return out
@POS.use_case(kind="command", route="POST /receipts/{receipt_id}/payments", permission="pos:receipt:sell", if_match="version")
def take_payment(caller: dict, receipt_id: str, method: Code, amount: Amt, tip_amount: Optional[Amt] = None,
terminal_txn_ref: Optional[Code] = None, terminal_status: Optional[Code] = None, voucher_code: Optional[Code] = None,
version: Optional[int] = None) -> dict:
"""Take one tender on an open receipt (receipt: open -> open): cash (more than the bill handed over gives change), a gift voucher
(`voucher_code`: its balance is claimed, voucher:<code>:<n>; at most the bill, no change, no tip; refused when expired, in another
currency or company, or holding less) or a card the terminal has answered for (terminal_txn_ref and terminal_status authorised |
declined, or `pending` when the terminal timed out: the register then asks the terminal for the outcome by the line's key and sends
it to resolve_card_payment before any other payment; the terminal's idempotency key is the payment line, <receipt_id>:<pay_no>). A
tip rides on its tender, is not part of the bill, and is a liability to staff. The line takes the claim pay:<receipt_id>:<pay_no> (a
plain INSERT: a second person taking the same line is told 409). A declined card is kept and counts for nothing."""
with repository() as repo, _decide():
r, current = _visible_receipt(repo, caller, receipt_id)
_same_version(f"receipt {receipt_id}", current, version)
s, _ = repo.sessions.get(r.session_id)
amt = money.amount(amount, r.currency, "the payment", positive=True)
tip = money.amount(tip_amount if tip_amount is not None else "0", r.currency, "the tip")
line = r.take_payment(method=method, amount=amt, tip=tip, terminal_txn_ref=terminal_txn_ref or "",
terminal_status=terminal_status or "", session_status=s.status, by=_who(caller),
voucher_code=(voucher_code or "").strip())
repo.claims.claim(f"pay:{receipt_id}:{line.pay_no}", "pay", r.session_id, _who(caller), _now())
if line.method == "voucher":
_redeem(repo, r, line)
return _receipt_view(r, repo.receipts.save(r, current))
def _redeem(repo, r: Receipt, line) -> None:
"""Claim the voucher's balance for a payment line: voucher:<code>:<claim_no>, a plain INSERT on (code, claim_no) in the same
transaction as the line. The balance is re-derived from the claims, so a till that spent it a moment ago has already taken it."""
voucher = repo.vouchers.get(line.voucher_code)
if voucher is None:
raise RuleBroken(f"no voucher {line.voucher_code}", gate="voucher_known", fix="check the code on the voucher")
shop, _ = repo.shops.get(r.shop_id)
balance, last = repo.vouchers.balance(line.voucher_code)
claim = voucher.redeem(balance=balance, last_claim=last, amount=line.amount, currency=r.currency, company_code=shop["company_code"],
on=r.business_date, receipt_id=r.receipt_id, pay_no=line.pay_no)
repo.claims.claim(claim.key, "voucher", r.session_id, r.created_by, _now())
repo.vouchers.add_claims([claim], _now())
@POS.use_case(kind="command", route="POST /receipts/{receipt_id}/payments/{pay_no}/outcome", permission="pos:receipt:sell",
idempotent=True, if_match="version")
def resolve_card_payment(caller: dict, receipt_id: str, pay_no: int, terminal_status: Code, terminal_txn_ref: Optional[Code] = None,
version: Optional[int] = None) -> dict:
"""Settle a card line whose terminal timed out (payment line: pending -> authorised | declined). The register asks the terminal for
the outcome by the line's key (<receipt_id>:<pay_no>) and sends what it says; until then no other payment is taken and the receipt
cannot be finalised or voided, so a card is never charged twice. Once settled the answer is final: the same answer again changes
nothing, a different one is refused. A declined line keeps no tip; the customer paying again is a new line."""
with repository() as repo, _decide():
r, current = _visible_receipt(repo, caller, receipt_id)
_same_version(f"receipt {receipt_id}", current, version)
s, _ = repo.sessions.get(r.session_id)
if not 1 <= int(pay_no) <= 1000:
raise OutOfBounds("pay_no is a payment line number: 1 to 1000")
line = r.resolve_payment(pay_no=int(pay_no), terminal_status=terminal_status, terminal_txn_ref=terminal_txn_ref or "",
session_status=s.status, by=_who(caller))
if line is None:
return _receipt_view(r, current)
return _receipt_view(r, repo.receipts.save(r, current))
@POS.use_case(kind="command", route="POST /receipts/{receipt_id}/finalise", permission="pos:receipt:sell", idempotent=True,
if_match="version")
def finalise_receipt(caller: dict, receipt_id: str, version: Optional[int] = None) -> dict:
"""Finalise and print (receipt: open -> finalised) when the payments sum EXACTLY to the bill. From then on the receipt is
never edited: a refund is a new receipt that names it. In the SAME transaction a pending outbox row is written for the
stock (inventory goods issue); it is sent after the commit, so a crash between the two sends it on the next run and a
rolled-back receipt sends nothing. A shop whose fiscal profile signs receipts (de_kassensichv, fr_nf525, at_rksv) cannot
finalise here yet: the signing capability is a later release."""
with repository() as repo, _decide():
r, current = _visible_receipt(repo, caller, receipt_id)
_same_version(f"receipt {receipt_id}", current, version)
s, _ = repo.sessions.get(r.session_id)
shop, _ = repo.shops.get(r.shop_id)
if r.status == ReceiptState.OPEN.value:
if s.status != SessionState.OPEN.value:
raise RuleBroken(f"session {s.session_id} is {s.status}: receipts are finalised in an open session",
gate="session_open", fix="open a session on the register")
if shop["fiscal_profile"] != "none":
raise RuleBroken(f"shop {shop['shop_id']} signs every receipt ({shop['fiscal_profile']}) before it prints, and the "
"signing capability is not part of this release", gate="fiscal_signature",
fix="use a shop with fiscal profile none, or wait for the capability (POS-24)")
if r.finalise(at=_now(), by=_who(caller)) is None:
return _receipt_view(r, current)
if shop["warehouse_id"] and shop["consumption_location_id"] and any(ln.kind == "product" for ln in r.lines):
repo.outbox.add(OutboxRow.pending(STOCK, r.receipt_id, goods_issue(r, company_code=shop["company_code"],
location_id=shop["consumption_location_id"])), _now())
_issue_vouchers(repo, r, shop, s)
new_version = repo.receipts.save(r, current)
after_response(_dispatch, r.receipt_id, conn=repo.conn) # after the commit; safe to lose: the row stays pending
return _receipt_view(r, new_version)
@POS.use_case(kind="command", route="POST /receipts/{receipt_id}/void", permission="pos:receipt:sell", idempotent=True, if_match="version")
def void_receipt(caller: dict, receipt_id: str, reason: Reason, terminal_reversals: Optional[list[Code]] = None,
version: Optional[int] = None) -> dict:
"""Cancel an OPEN receipt before it is final (receipt: open -> voided). It stays as a void record with its lines and keeps its number
(fiscal law counts voids; a number is never reused) and takes no stock, no VAT and no takings. A card line the terminal authorised
must have been reversed on the terminal first: send its references in `terminal_reversals`. A final receipt is never voided: it is
refunded. Voiding twice changes nothing."""
with repository() as repo, _decide():
r, current = _visible_receipt(repo, caller, receipt_id)
_same_version(f"receipt {receipt_id}", current, version)
s, _ = repo.sessions.get(r.session_id)
done = r.void(reason=reason, terminal_reversals=set(terminal_reversals or []), session_status=s.status, at=_now(), by=_who(caller))
if done is None:
return _receipt_view(r, current)
_give_back_vouchers(repo, r)
return _receipt_view(r, repo.receipts.save(r, current))
def _give_back_vouchers(repo, r: Receipt) -> None:
"""A voided receipt spent nothing: what its voucher lines claimed goes back to the vouchers (operation refund_to_voucher)."""
state = {}
for line in r.voucher_lines():
voucher = repo.vouchers.get(line.voucher_code)
balance, last = state.get(line.voucher_code) or repo.vouchers.balance(line.voucher_code)
claim = voucher.give_back(balance=balance, last_claim=last, amount=line.amount, receipt_id=r.receipt_id, pay_no=line.pay_no)
state[line.voucher_code] = (claim.balance_after, claim.claim_no)
repo.claims.claim(claim.key, "voucher", r.session_id, _who({"email": r.created_by}), _now())
repo.vouchers.add_claims([claim], _now())
# ── refunds (POS-11) ─────────────────────────────────────────────────────────────────────────────────────
@POS.use_case(kind="command", route="POST /receipts/{receipt_id}/refund", permission="pos:receipt:refund")
def refund_receipt(caller: dict, receipt_id: str, session_id: str, lines: list[RefundLineIn], method: Code, reason: Reason,
terminal_txn_ref: Optional[Code] = None, terminal_status: Optional[Code] = None) -> dict:
"""Refund lines of a FINAL sale, in the caller's open session (the drawer pays): a new final receipt of kind refund that names the
original and its lines, with negative amounts and the ORIGINAL's tax code and rate (a rate that changed since the sale does not
change it). Each line's quantity is claimed (refund:<original>:<line>:<n>) so a line is never refunded past what was sold, even by
two people at once. The money goes back the way it came: `method` cash or card (a card refund is a referenced refund the terminal
already answered: terminal_txn_ref and terminal_status authorised; its idempotency key is the refund's payment line). A cash refund
above cash_refund_limit needs a shift lead who is not the drawer's cashier. Lines the cashier says go back on the shelf (within what
their category allows) are sent to inventory as receipts after the commit, valued at the cost the sale left at."""
with repository() as repo, _decide():
original, _ = _visible_receipt(repo, caller, receipt_id)
s, _ = _visible_session(repo, caller, session_id)
if s.status != SessionState.OPEN.value:
raise RuleBroken(f"session {session_id} is {s.status}: refunds are given in an open session", gate="session_open",
fix="open a session on the register")
if s.shop_id != original.shop_id:
raise RuleBroken(f"receipt {receipt_id} was sold at {original.shop_id}: it is refunded there", gate="same_shop",
fix="refund it at the shop that sold it")
refund.refundable(original)
if not lines or len(lines) > 200:
raise OutOfBounds("a refund has one to 200 lines")
try:
how = PaymentMethod(method)
except ValueError:
how = None
if how not in (PaymentMethod.CASH, PaymentMethod.CARD):
raise RuleBroken(f"{method!r} is not a way to refund: use cash or card", gate="method", fix="send method cash or card")
picks = [(ln.line_no, money.quantity(ln.quantity, f"line {ln.line_no} quantity"), ln.restock) for ln in lines]
claimed = repo.refunds.claimed(original.receipt_id)
rules = repo.categories.restock_map(sorted({x.tax_category for x in original.lines}))
allowed = {x.line_no: rules.get(x.tax_category, False) and not x.recipe_source for x in original.lines} # a made item is not restocked
built = refund.plan(original, claimed, picks, original.currency, allowed)
amount = -sum((ln.line_gross for ln in built), money.ZERO)
refund.method_of_refund(original, how.value, amount, repo.refunds.refunded_by_method(original.receipt_id, how.value))
approver = None
if how == PaymentMethod.CARD:
if not terminal_txn_ref:
raise RuleBroken("a card refund names the terminal's transaction", gate="terminal_ref", fix="send terminal_txn_ref")
if terminal_status != TerminalStatus.AUTHORISED.value:
raise RuleBroken("the terminal did not authorise the card refund: nothing is refunded", gate="terminal_status",
fix="ask the terminal for the outcome by its key, then send it")
else:
limit = Decimal(str(repo.settings.in_force(s.business_date)["cash_refund_limit"]))
who = (_who(caller), _sub(caller)) if auth.has_permission(caller, OVERRIDE) else None
approver = refund.cash_limit(amount, limit, who, (s.opened_by, s.opened_sub), same_person)
number = repo.receipts.last_no(session_id) + 1
new = refund_receipt_of(original=original, session=s, receipt_no=number, lines=built, method=how.value,
terminal_txn_ref=terminal_txn_ref or "", reason=reason, at=_now(), by=_who(caller), sub=_sub(caller))
by_original = {ln.original_line_no: ln for ln in new.lines}
keys, rows = [], []
for line_no, qty, _r in picks:
done = claimed.get(line_no, refund.Claimed())
no = done.claims + 1
keys.append(f"refund:{original.receipt_id}:{line_no}:{no}")
ln = by_original[line_no]
rows.append({"original_receipt_id": original.receipt_id, "line_no": line_no, "claim_no": no, "refund_receipt_id": new.receipt_id,
"quantity": qty, "gross": -ln.line_gross, "vat": -ln.vat, "refunded_after": done.quantity + qty})
repo.claims.claim_many(keys, "refund", session_id, _who(caller), _now()) # the quantity, claimed: the loser is told 409
repo.refunds.add_claims(rows, _now())
shop, _ = repo.shops.get(original.shop_id)
back = [OutboxRow.pending(RESTOCK, f"pos:{new.receipt_id}:{ln.line_no}:1",
restock_body(new, ln, original_receipt_id=original.receipt_id, company_code=shop["company_code"],
location_id=shop["consumption_location_id"]))
for ln in new.lines if ln.restock and shop["warehouse_id"] and shop["consumption_location_id"]]
repo.outbox.add_many(back, _now())
version = repo.receipts.save(new, 0)
if back:
after_response(_dispatch, None, 100, f"pos:{new.receipt_id}:", conn=repo.conn)
view = _receipt_view(new, version)
view["approved_by"] = approver[0] if approver else ""
return view
@POS.use_case(kind="query", route="GET /category-rules", permission="pos:shop:read")
def list_category_rules(caller: dict) -> dict:
"""Which product categories go back on the shelf when a sale is refunded (a category with no row is not restocked)."""
with repository() as repo:
return {"restock_on_refund": repo.categories.all()}
@POS.use_case(kind="command", route="POST /category-rules/{category_code}", permission="pos:settings:write", idempotent=True)
def set_category_rule(caller: dict, category_code: str, restock_on_refund: Setting) -> dict:
"""Say whether a refunded item of this category goes back on the shelf (true: packaged goods) or never does (false: a made drink).
A cashier restocks only where the category allows it."""
with repository() as repo, _decide():
code = _check_code(category_code, "category_code")
if not isinstance(restock_on_refund, bool):
raise Refused(422, "restock_on_refund is true or false", gate="bounds")
repo.categories.save(code, restock_on_refund, _who(caller), _now())
return {"category_code": code, "restock_on_refund": restock_on_refund}
def _issue_vouchers(repo, r: Receipt, shop: dict, s: Session) -> None:
"""The vouchers a final receipt sold become live: the voucher row and its issue claim (voucher:<code>:1), in the same transaction
as the receipt, so a receipt that does not commit issues nothing."""
sold = [ln for ln in r.lines if ln.kind == "voucher_sale"]
if not sold:
return
months = int(repo.settings.in_force(r.business_date)["voucher_validity_months"])
made = [Voucher.issue(voucher_code=ln.voucher_code, company_code=shop["company_code"], currency=r.currency, face_value=ln.line_gross,
receipt_id=r.receipt_id, on=r.business_date, validity_months=months) for ln in sold]
repo.claims.claim_many([c.key for _v, c in made], "voucher", r.session_id, _who({"email": r.created_by}), _now())
for v, _c in made:
repo.vouchers.add(v, _now())
repo.vouchers.add_claims([c for _v, c in made], _now())
def _voucher_view(repo, v: Voucher, on: date) -> dict:
balance, _last = repo.vouchers.balance(v.voucher_code)
out = dataclasses.asdict(v)
out.pop("events")
return {**out, "balance": balance, "state": v.state(balance, on), "claims": repo.vouchers.claims(v.voucher_code)}
@POS.use_case(kind="query", route="GET /vouchers/{voucher_code}", permission="pos:shop:read")
def get_voucher(caller: dict, voucher_code: str) -> dict:
"""A gift voucher: its face value, the balance (derived from its claims: the issue, each redemption, each balance given back), its
expiry and whether it is active, exhausted or expired today. A caller sees the vouchers of their shops' companies."""
with repository() as repo:
v = repo.vouchers.get(voucher_code)
scope = _scope(repo, caller)
if v is None or (scope is not None and v.company_code not in repo.vouchers.companies(scope)):
raise Refused(404, f"no voucher {voucher_code}")
return _voucher_view(repo, v, _today(repo))
# ── the outbox (POS-16, the stock part) ──────────────────────────────────────────────────────────────────
SENDERS = {STOCK: seams.goods_issue, GL: seams.journal, DEPOSIT: seams.deposit_batch, RESTOCK: seams.goods_receipt}
def _after_sent(repo, row: OutboxRow) -> None:
"""What a sent row means for this brick's own aggregates, in the same transaction that records it as sent: gl holds the session's
journal -> the session is posted; payments has the bag -> the bag is handed over."""
if row.target == GL:
found = repo.sessions.get(row.ref)
if found is not None and found[0].mark_posted(journal_id=row.payload["journal_id"], at=_now()) is not None:
repo.sessions.save(found[0], found[1])
elif row.target == DEPOSIT:
bag = repo.deposits.get(row.ref)
if bag is not None and bag.status != "handed_over":
bag.handed_over(_now())
repo.deposits.save(bag, _now())
def _dispatch(ref: Optional[str] = None, limit: int = 100, prefix: Optional[str] = None) -> dict:
"""Send pending rows to their neighbours, AFTER the local commit that wrote them: read the rows (committed), call the
neighbour with no transaction open, record the outcome. A row is the neighbour's idempotency key: a resend is its replay."""
with repository() as repo:
rows = [x for x in repo.outbox.pending(limit) if (ref is None or x.ref == ref) and (prefix is None or x.ref.startswith(prefix))]
done = {"sent": 0, "pending": 0, "failed": 0}
for row in rows:
send = SENDERS.get(row.target)
if send is None:
continue
ok, permanent, why = send(row.payload)
with repository() as repo:
fresh = repo.outbox.get(row.target, row.ref)
if fresh is None or fresh.status != OutboxStatus.PENDING.value:
continue # another dispatcher settled it: never resent, never reopened
fresh.sent(_now()) if ok else fresh.tried(why, permanent=permanent)
repo.outbox.save(fresh, _now())
if ok:
_after_sent(repo, fresh)
done[fresh.status] += 1
return done
@POS.use_case(kind="command", route="POST /outbox/dispatch", permission="pos:outbox:dispatch", idempotent=True)
def dispatch_outbox(caller: dict, limit: int = 100) -> dict:
"""Send what is waiting for the neighbours (machine: the composition's worker, or a retry after a neighbour was down). A
goods issue to inventory is idempotent on its move_ref, so running this twice sends nothing twice."""
with _decide():
if not 1 <= limit <= 1000:
raise OutOfBounds("limit is 1 to 1000")
return _dispatch(limit=limit)
@POS.use_case(kind="command", route="POST /outbox/requeue", permission="pos:day:post", idempotent=True)
def requeue_outbox_row(caller: dict, target: Code, ref: Code) -> dict:
"""Put a FAILED row back to pending once its cause is fixed (a closed gl period reopened, a bank account created, a product
inventory did not know): it is sent on the next dispatch. A pending row stays pending; a sent row is never sent again."""
with repository() as repo, _decide():
row = repo.outbox.get(target, ref)
if row is None:
raise Refused(404, f"no outbox row {target} {ref}")
if row.status == OutboxStatus.FAILED.value:
row.requeue()
repo.outbox.save(row, _now())
after_response(_dispatch, ref, conn=repo.conn)
elif row.status == OutboxStatus.SENT.value:
row.requeue() # refused: a sent row never goes back
return {"target": row.target, "ref": row.ref, "status": row.status, "attempts": row.attempts, "last_error": row.last_error}
@POS.use_case(kind="query", route="GET /outbox", permission="pos:report:read")
def list_outbox(caller: dict, status: Optional[str] = None) -> dict:
"""The side effects waiting for, sent to, or refused by a neighbour (pending | sent | failed), with the last error."""
with repository() as repo:
return {"rows": [{"target": x.target, "ref": x.ref, "status": x.status, "attempts": x.attempts,
"last_error": x.last_error, "payload": x.payload} for x in repo.outbox.listing(status)]}
# ── recipes (POS-15) ─────────────────────────────────────────────────────────────────────────────────────
def _recipe_view(r) -> dict:
return {"product_id": r.product_id, "version": r.version, "valid_from": r.valid_from.isoformat(), "created_by": r.created_by,
"lines": [{"line_no": ln.line_no, "component_id": ln.component_id, "quantity": ln.quantity, "unit_code": ln.unit_code,
"modifier_id": ln.modifier_id, "replaces": ln.replaces} for ln in r.lines]}
@POS.use_case(kind="query", route="GET /recipes", permission="pos:shop:read")
def list_recipes(caller: dict) -> dict:
"""The products that have a recipe of their own here (what they consume when sold), and where recipes come from today
(recipe_source: pos, or manufacturing's bills of materials)."""
with repository() as repo:
return {"products": repo.recipes.products(), "recipe_source": repo.settings.in_force(_today(repo))["recipe_source"]}
@POS.use_case(kind="query", route="GET /recipes/{product_id}", permission="pos:shop:read")
def get_recipe(caller: dict, product_id: str) -> dict:
"""Every dated version of one product's recipe, oldest first, with its components and its modifiers."""
with repository() as repo:
found = repo.recipes.for_products([product_id]).get(product_id)
if not found:
raise Refused(404, f"no recipe for {product_id}")
return {"product_id": product_id, "versions": [_recipe_view(r) for r in found]}
@POS.use_case(kind="command", route="POST /recipes", permission="pos:recipe:write", idempotent=True)
def keep_recipe(caller: dict, product_id: Code, lines: list[RecipeLineIn], valid_from: Optional[Code] = None,
modifiers: Optional[list[RecipeModifierIn]] = None) -> dict:
"""Write what a product consumes when sold (a latte: 0.018 KGM beans, 0.250 LTR milk, a cup, a lid) as the next DATED version: it
starts after today (the first version of a product may start today) and after the version before it, and never changes a sale
already rung (each line freezes the version it used). Every component must be a stocked product and every quantity must convert
EXACTLY to its base unit (product), or the recipe is refused. A modifier (MOD-OAT) adds a component or replaces a base one.
Read-only while recipe_source is manufacturing (its bill of materials is the recipe). Saving the same content again is the same
version."""
with repository() as repo, _decide():
today = _today(repo)
if repo.settings.in_force(today)["recipe_source"] == "manufacturing":
raise RuleBroken("recipes come from manufacturing's bills of materials (recipe_source): they are read-only here", gate="recipe_source",
fix="change the bill of materials in manufacturing, or set recipe_source to pos")
product_id = _check_code(product_id, "product_id")
mods = list(modifiers or [])
if not 1 <= len(lines) <= 30 or len(mods) > 30:
raise OutOfBounds("a recipe has one to 30 lines and at most 30 modifier lines")
seams.stocked_product(product_id, today) # the product must exist
rows, seen = [], {}
base_components = set()
for no, ln in enumerate(lines, start=1):
rows.append(recipes_.RecipeLine(no, _check_code(ln.component_id, "component_id"), money.quantity(ln.quantity, f"line {no} quantity"),
_check_code(ln.unit_code, "unit_code")))
base_components.add(rows[-1].component_id)
for ln in mods:
replaces = _check_code(ln.replaces, "replaces") if ln.replaces else ""
if replaces and replaces not in base_components:
raise RuleBroken(f"modifier {ln.modifier_id} replaces {replaces}, which is not a component of the recipe", gate="replaces_known",
fix="name a component of the base recipe")
rows.append(recipes_.RecipeLine(len(rows) + 1, _check_code(ln.component_id, "component_id"),
money.quantity(ln.quantity, f"modifier {ln.modifier_id} quantity"), _check_code(ln.unit_code, "unit_code"),
_check_code(ln.modifier_id, "modifier_id"), replaces))
for ln in rows:
if ln.component_id not in seen:
seen[ln.component_id] = seams.stocked_product(ln.component_id, today)
info = seen[ln.component_id]
if not info["stocked"]:
raise RuleBroken(f"{ln.component_id} is not a stocked product: a recipe takes only what inventory holds", gate="component_stocked",
fix="use a stocked product, or set it stocked in product")
if info["base_unit_code"] and ln.unit_code != info["base_unit_code"]:
seams.convert_exactly(ln.component_id, ln.quantity, ln.unit_code, info["base_unit_code"])
history = repo.recipes.for_products([product_id]).get(product_id, [])
new = recipes_.Recipe(product_id=product_id, version=(history[-1].version if history else 0) + 1, valid_from=today, lines=rows,
created_by=_who(caller))
if history and recipes_.same_content(history[-1], new):
return _recipe_view(history[-1])
try:
start = date.fromisoformat(valid_from) if valid_from else (today if not history else None)
except (ValueError, TypeError):
raise Refused(422, "valid_from is a date like 2026-12-01", gate="bounds") from None
if start is None:
raise Refused(422, "a changed recipe starts after today: send valid_from", gate="valid_from",
fix="a recipe changes by a dated version, never retroactively")
if start < today or (history and (start == today or start <= history[-1].valid_from)):
raise Refused(422, f"the new version starts after {max(today, history[-1].valid_from) if history else today}", gate="valid_from",
fix="send a later valid_from")
new.valid_from = start
repo.recipes.add(new, _now())
return _recipe_view(new)
# ── the cash bag for the bank (POS-19, POS-26) ──────────────────────────────────────────────────────────────
def _bag_view(d: CashDeposit) -> dict:
out = dataclasses.asdict(d)
out.pop("events")
return out
@POS.use_case(kind="command", route="POST /shops/{shop_id}/cash-bags", permission="pos:deposit:prepare", idempotent=True)
def seal_cash_bag(caller: dict, shop_id: str, bag_no: Code, amount: Amt, bank_account_id: Code, expected_on: Optional[Code] = None,
denominations: Optional[dict] = None) -> dict:
"""A shift lead seals a bag of the safe's cash for the bank (cash_deposit: none -> sealed). A bag is sealed once and never
unsealed (the same bag again, with the same amount, is the same bag); its amount is positive and at most what the shop's safe
holds (the safe drops, less the float topped up from it, less the bags already sealed). `denominations` ({"20.00": 10}) must add
up to the amount. The bag is handed to payments from a pending row after this commit (an expected deposit batch dep:pos:<shop>:
<bag>), which matches the bank's credit to it; `expected_on` is the day the courier banks it (the next business day)."""
with repository() as repo, _decide():
shop, _ = _visible_shop(repo, caller, shop_id)
bag_no = _check_code(bag_no, "bag_no")
currency, _incl = seams.price_list_header(shop["price_list_id"], _business_date(repo, shop)) if shop["price_list_id"] else ("", True)
if not currency:
raise RuleBroken(f"shop {shop_id} has no price list, so no currency for the bag", gate="price_list",
fix="set price_list_id on the shop")
amt = money.amount(amount, currency, "the bag", positive=True)
later = _business_date(repo, shop) + timedelta(days=1)
try:
day = date.fromisoformat(expected_on) if expected_on else later
except (ValueError, TypeError):
raise Refused(422, "expected_on is a date like 2026-11-15", gate="bounds") from None
counted = ""
if denominations is not None:
if not isinstance(denominations, dict) or not 1 <= len(denominations) <= 30:
raise OutOfBounds("denominations is one to 30 entries like {\"20.00\": 10}")
total = money.ZERO
for face, n in denominations.items():
if isinstance(n, bool) or not isinstance(n, int) or not 1 <= n <= 1_000_000:
raise OutOfBounds("a denomination's count is a whole number, 1 to 1,000,000")
total += money.amount(face, currency, "a denomination", positive=True) * n
if total != amt:
raise OutOfBounds(f"the denominations add up to {total}, the bag is {amt}")
counted = json.dumps({k: denominations[k] for k in sorted(denominations)})
ref = f"dep:pos:{shop_id}:{bag_no}"
existing = repo.deposits.get(ref)
if existing is not None:
if existing.amount != amt:
raise RuleBroken(f"bag {bag_no} was sealed with {existing.amount}: a bag is sealed once", gate="bag_sealed",
fix="seal a new bag with a new number")
return _bag_view(existing)
moves = repo.sessions.safe_moves(shop_id)
sealed_total, sealed_n = repo.deposits.sealed(shop_id)
safe = safe_balance(moves.get("safe_drop", money.ZERO), moves.get("float_top_up", money.ZERO), sealed_total)
bag = CashDeposit.seal(shop_id=shop_id, bag_no=bag_no, amount=amt, currency=currency, company_code=shop["company_code"],
bank_account_id=bank_account_id, expected_on=day, denominations=counted, safe=safe, at=_now(),
by=_who(caller), sub=_sub(caller))
repo.claims.claim(f"bag:{shop_id}:{bag_no}", "bag", "", _who(caller), _now()) # a bag is sealed once
repo.claims.claim(f"safe:{shop_id}:{sealed_n + 1}", "bag", "", _who(caller), _now()) # two bags never drain one safe twice
repo.deposits.add(bag, _now())
repo.outbox.add(OutboxRow.pending(DEPOSIT, ref, deposit_batch(bag)), _now())
after_response(_dispatch, ref, conn=repo.conn)
return _bag_view(bag)
@POS.use_case(kind="query", route="GET /shops/{shop_id}/cash-bags", permission="pos:report:read")
def list_cash_bags(caller: dict, shop_id: str) -> dict:
"""The shop's sealed bags (with where each stands: sealed, handed to payments) and what its safe still holds."""
with repository() as repo:
_visible_shop(repo, caller, shop_id)
moves = repo.sessions.safe_moves(shop_id)
sealed_total, _n = repo.deposits.sealed(shop_id)
safe = safe_balance(moves.get("safe_drop", money.ZERO), moves.get("float_top_up", money.ZERO), sealed_total)
return {"shop_id": shop_id, "safe_balance": safe, "bags": [_bag_view(b) for b in repo.deposits.of_shop(shop_id)]}