dodil/erp-brick-pos
DODIL ERP brick: point of sale
src/brick_pos/infrastructure/tables.py
"""The tables, and the repositories that load and save them.
Every table is prefixed `pos_` and has a natural primary key (never SERIAL). Money is NUMERIC(18,2), a unit price
NUMERIC(19,4). schema.sql and migrations/ are generated from this column model; tests/test_schema.py keeps them equal.
aggregates (loaded whole, saved at the version loaded): shop (+ its registers), session, receipt (+ lines, payments)
facts (plain INSERTs, never edited): cash_move, claim
rows that move forward by their own rule: outbox (pending -> sent | failed), settings (dated; a new row, never an edit)
Nothing here updates a row with a join (UPDATE ... FROM), and a claim is a plain INSERT: the loser of a race is told 23505."""
import json
from datetime import date, datetime
from decimal import Decimal
from typing import Optional
from zoneinfo import ZoneInfo
from erp_platform import work
from erp_platform.repository import Aggregate, Part, upsert
from sqlalchemy import (
TIMESTAMP,
Boolean,
Column,
Date,
Integer,
MetaData,
Numeric,
String,
Table,
and_,
case,
func,
insert,
select,
)
from sqlalchemy.exc import IntegrityError
from brick_pos.domain.deposit import CashDeposit
from brick_pos.domain.outbox import OutboxRow
from brick_pos.domain.receipt import LineComponent, PaymentLine, Receipt, ReceiptLine
from brick_pos.domain.recipe import Recipe, RecipeLine
from brick_pos.domain.refund import Claimed
from brick_pos.domain.session import CashMove, Session
from brick_pos.domain.snapshot import Snapshot
from brick_pos.domain.values import SETTING_DEFS, OutboxStatus, ReceiptState
from brick_pos.domain.voucher import Voucher
md = MetaData()
MONEY, PRICE, QTY, PCT = Numeric(18, 2), Numeric(19, 4), Numeric(18, 3), Numeric(9, 4)
# What a release after the first adds (scripts/gen_schema.py writes 00001_init without these and a migration with them):
# {release: {table: [columns]}} and {release: [tables]}. Additive only.
ADDED: dict = {2: {"pos_session": ["last_receipt_no", "claimed_gap_from", "claimed_gap_to", "posted_at", "journal_id", "snapshot_no",
"snapshot_digest"],
"pos_receipt": ["original_receipt_id", "void_reason", "voided_by", "voided_at"],
"pos_receipt_line": ["original_line_no", "restock", "kind", "voucher_code", "recipe_source", "recipe_ref", "recipe_version",
"modifiers"],
"pos_payment_line": ["voucher_code"]}}
NEW_TABLES: dict = {2: ["pos_cash_deposit", "pos_refund_claim", "pos_category_rule", "pos_voucher", "pos_voucher_claim", "pos_line_component",
"pos_recipe", "pos_recipe_line", "pos_register_snapshot"]}
pos_shop = Table("pos_shop", md,
Column("shop_id", String, primary_key=True),
Column("name", String),
Column("company_code", String),
Column("warehouse_id", String),
Column("consumption_location_id", String),
Column("price_list_id", String),
Column("fiscal_profile", String),
Column("country", String),
Column("cost_centre", String),
Column("day_cutoff_time", String),
Column("status", String),
Column("version", Integer),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
SHOP_COLUMNS = ("shop_id", "name", "company_code", "warehouse_id", "consumption_location_id", "price_list_id", "fiscal_profile",
"country", "cost_centre", "day_cutoff_time", "status")
pos_register = Table("pos_register", md,
Column("shop_id", String, primary_key=True),
Column("register_id", String, primary_key=True),
Column("terminal_id", String),
Column("fiscal_device_serial", String),
Column("fiscal_device_status", String),
Column("status", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
REGISTER_COLUMNS = ("shop_id", "register_id", "terminal_id", "fiscal_device_serial", "fiscal_device_status", "status")
SHOP = Aggregate(Part(pos_shop, ("shop_id",)), Part(pos_register, ("shop_id", "register_id")), version="version")
pos_shop_user = Table("pos_shop_user", md,
Column("shop_id", String, primary_key=True),
Column("sub", String, primary_key=True),
Column("email", String),
Column("assigned_by", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pos_session = Table("pos_session", md,
Column("session_id", String, primary_key=True),
Column("shop_id", String),
Column("register_id", String),
Column("session_no", Integer),
Column("business_date", Date),
Column("currency", String),
Column("opened_by", String),
Column("opened_sub", String),
Column("opened_at", TIMESTAMP),
Column("opening_float", MONEY),
Column("float_difference", MONEY),
Column("counted_by", String),
Column("counted_sub", String),
Column("counted_at", TIMESTAMP),
Column("counted_cash", MONEY),
Column("expected_cash", MONEY),
Column("difference", MONEY),
Column("closing_float", MONEY),
Column("approved_by", String),
Column("approved_sub", String),
Column("approved_at", TIMESTAMP),
Column("reason", String),
Column("status", String),
Column("version", Integer),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP),
Column("last_receipt_no", Integer),
Column("claimed_gap_from", Integer),
Column("claimed_gap_to", Integer),
Column("posted_at", TIMESTAMP),
Column("journal_id", String),
Column("snapshot_no", Integer),
Column("snapshot_digest", String))
SESSION_COLUMNS = ("session_id", "shop_id", "register_id", "session_no", "business_date", "currency", "opened_by", "opened_sub",
"opened_at", "opening_float", "float_difference", "counted_by", "counted_sub", "counted_at", "counted_cash",
"expected_cash", "difference", "closing_float", "approved_by", "approved_sub", "approved_at", "reason", "status",
"last_receipt_no", "claimed_gap_from", "claimed_gap_to", "posted_at", "journal_id", "snapshot_no", "snapshot_digest")
SESSION = Aggregate(Part(pos_session, ("session_id",)), version="version")
pos_cash_move = Table("pos_cash_move", md,
Column("cash_move_id", String, primary_key=True),
Column("session_id", String),
Column("seq", Integer),
Column("kind", String),
Column("amount", MONEY),
Column("reason_code", String),
Column("approved_by", String),
Column("approved_sub", String),
Column("created_by", String),
Column("created_sub", String),
Column("created_at", TIMESTAMP))
# a sealed bag of cash on its way to the bank: a plain INSERT (the key is the claim: a bag is sealed once), then its status moves
# forward by its own rule (sealed -> handed_over) when payments has the expected batch.
pos_cash_deposit = Table("pos_cash_deposit", md,
Column("deposit_ref", String, primary_key=True),
Column("shop_id", String),
Column("bag_no", String),
Column("amount", MONEY),
Column("currency", String),
Column("company_code", String),
Column("bank_account_id", String),
Column("expected_on", Date),
Column("denominations", String),
Column("sealed_by", String),
Column("sealed_sub", String),
Column("sealed_at", TIMESTAMP),
Column("status", String),
Column("handed_over_at", TIMESTAMP),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
DEPOSIT_COLUMNS = ("deposit_ref", "shop_id", "bag_no", "amount", "currency", "company_code", "bank_account_id", "expected_on",
"denominations", "sealed_by", "sealed_sub", "sealed_at", "status", "handed_over_at")
pos_receipt = Table("pos_receipt", md,
Column("receipt_id", String, primary_key=True),
Column("session_id", String),
Column("shop_id", String),
Column("register_id", String),
Column("receipt_no", Integer),
Column("kind", String),
Column("business_date", Date),
Column("currency", String),
Column("eat_in", Boolean),
Column("gross", MONEY),
Column("net", MONEY),
Column("vat", MONEY),
Column("created_by", String),
Column("created_sub", String),
Column("finalised_at", TIMESTAMP),
Column("status", String),
Column("version", Integer),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP),
Column("original_receipt_id", String),
Column("void_reason", String),
Column("voided_by", String),
Column("voided_at", TIMESTAMP))
RECEIPT_COLUMNS = ("receipt_id", "session_id", "shop_id", "register_id", "receipt_no", "kind", "business_date", "currency", "eat_in",
"gross", "net", "vat", "created_by", "created_sub", "finalised_at", "status", "original_receipt_id", "void_reason",
"voided_by", "voided_at")
pos_receipt_line = Table("pos_receipt_line", md,
Column("receipt_id", String, primary_key=True),
Column("line_no", Integer, primary_key=True),
Column("product_id", String),
Column("description", String),
Column("quantity", QTY),
Column("unit_code", String),
Column("unit_price", PRICE),
Column("line_gross", MONEY),
Column("tax_category", String),
Column("tax_code", String),
Column("rate_pct", PCT),
Column("vat", MONEY),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP),
Column("original_line_no", Integer),
Column("restock", Boolean),
Column("kind", String),
Column("voucher_code", String),
Column("recipe_source", String),
Column("recipe_ref", String),
Column("recipe_version", Integer),
Column("modifiers", String))
LINE_COLUMNS = ("receipt_id", "line_no", "product_id", "description", "quantity", "unit_code", "unit_price", "line_gross",
"tax_category", "tax_code", "rate_pct", "vat", "original_line_no", "restock", "kind", "voucher_code", "recipe_source", "recipe_ref", "recipe_version", "modifiers")
# What a receipt line took from stock, frozen when it was rung (the recipe version used, as quantities): the move_refs inventory was
# given are pos:<receipt_id>:<line_no>:<component_no>. A part of the receipt aggregate; inserted with the receipt, never edited.
pos_line_component = Table("pos_line_component", md,
Column("receipt_id", String, primary_key=True),
Column("line_no", Integer, primary_key=True),
Column("component_no", Integer, primary_key=True),
Column("product_id", String),
Column("quantity", Numeric(19, 6)),
Column("unit_code", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
COMPONENT_COLUMNS = ("receipt_id", "line_no", "component_no", "product_id", "quantity", "unit_code")
pos_payment_line = Table("pos_payment_line", md,
Column("receipt_id", String, primary_key=True),
Column("pay_no", Integer, primary_key=True),
Column("method", String),
Column("amount", MONEY),
Column("tip_amount", MONEY),
Column("change_amount", MONEY),
Column("terminal_txn_ref", String),
Column("terminal_status", String),
Column("idempotency_key", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP),
Column("voucher_code", String))
PAYMENT_COLUMNS = ("receipt_id", "pay_no", "method", "amount", "tip_amount", "change_amount", "terminal_txn_ref", "terminal_status",
"idempotency_key", "voucher_code")
RECEIPT = Aggregate(Part(pos_receipt, ("receipt_id",)), Part(pos_receipt_line, ("receipt_id", "line_no")),
Part(pos_payment_line, ("receipt_id", "pay_no")), Part(pos_line_component, ("receipt_id", "line_no", "component_no")),
version="version")
# A recipe: what one product consumes when sold, as DATED versions (a new version, never an edit). Lines with modifier_id "" are the base
# recipe; a modifier's lines add a component or (replaces) take the place of one.
pos_recipe = Table("pos_recipe", md,
Column("product_id", String, primary_key=True),
Column("version", Integer, primary_key=True),
Column("valid_from", Date),
Column("created_by", String),
Column("created_at", TIMESTAMP))
pos_recipe_line = Table("pos_recipe_line", md,
Column("product_id", String, primary_key=True),
Column("version", Integer, primary_key=True),
Column("line_no", Integer, primary_key=True),
Column("component_id", String),
Column("quantity", QTY),
Column("unit_code", String),
Column("modifier_id", String),
Column("replaces", String))
# One claim per money operation, a plain INSERT in the operation's own transaction: whoever commits first owns the key and
# the loser is told 23505 (a 409 through the doors). Keys: pay:<receipt_id>:<pay_no>. A refund's quantity and a voucher's
# redemption have their own keys (refund:..., voucher:...) when those slices land; they never share one.
pos_claim = Table("pos_claim", md,
Column("claim_key", String, primary_key=True),
Column("kind", String),
Column("session_id", String),
Column("claimed_by", String),
Column("claimed_at", TIMESTAMP))
# The row two refunds race for: one per refund of an original line, a plain INSERT on (original, line, claim_no) - the loser is told 23505,
# re-reads and finds nothing left (spec POS-11). `gross` and `vat` are what that refund gave back, so the last refund of a line returns
# exactly what is left and the cents add up; `refunded_after` is the line's quantity given back in all, after this one.
pos_refund_claim = Table("pos_refund_claim", md,
Column("original_receipt_id", String, primary_key=True),
Column("line_no", Integer, primary_key=True),
Column("claim_no", Integer, primary_key=True),
Column("refund_receipt_id", String),
Column("quantity", QTY),
Column("gross", MONEY),
Column("vat", MONEY),
Column("refunded_after", QTY),
Column("created_at", TIMESTAMP))
REFUND_CLAIM_COLUMNS = ("original_receipt_id", "line_no", "claim_no", "refund_receipt_id", "quantity", "gross", "vat", "refunded_after")
# Per product category: does a refunded item go back on the shelf (module.yaml config_rows: pos_category_rule). No row: no restock.
pos_category_rule = Table("pos_category_rule", md,
Column("category_code", String, primary_key=True),
Column("restock_on_refund", Boolean),
Column("set_by", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# A gift voucher: live when the receipt that sold it is final. Its balance is NOT a column: it is derived from its claims every time
# (a stored balance is a running total two tills could both read). `status` is `active` for as long as the voucher exists; expired and
# exhausted are derived from expires_on and the balance.
pos_voucher = Table("pos_voucher", md,
Column("voucher_code", String, primary_key=True),
Column("company_code", String),
Column("currency", String),
Column("face_value", MONEY),
Column("profile", String),
Column("issued_receipt_id", String),
Column("expires_on", Date),
Column("status", String),
Column("created_at", TIMESTAMP))
VOUCHER_COLUMNS = ("voucher_code", "company_code", "currency", "face_value", "profile", "issued_receipt_id", "expires_on", "status")
# One row per movement of a voucher's balance, a plain INSERT on (voucher_code, claim_no): the operation is part of what is claimed.
pos_voucher_claim = Table("pos_voucher_claim", md,
Column("voucher_code", String, primary_key=True),
Column("claim_no", Integer, primary_key=True),
Column("operation", String),
Column("receipt_id", String),
Column("pay_no", Integer),
Column("amount", MONEY),
Column("balance_after", MONEY),
Column("created_at", TIMESTAMP))
VOUCHER_CLAIM_COLUMNS = ("voucher_code", "claim_no", "operation", "receipt_id", "pay_no", "amount", "balance_after")
# The register's dated catalogue, prices and rates (POS-30): a copy of product's and tax's answers on a business date, never edited. A plain
# INSERT on (shop, business date, snapshot_no): two registers opening at once cannot both make the same number.
pos_register_snapshot = Table("pos_register_snapshot", md,
Column("shop_id", String, primary_key=True),
Column("business_date", Date, primary_key=True),
Column("snapshot_no", Integer, primary_key=True),
Column("price_list_id", String),
Column("country", String),
Column("currency", String),
Column("prices_include_tax", Boolean),
Column("valid_until", Date),
Column("digest", String),
Column("products_json", String),
Column("rates_json", String),
Column("built_at", TIMESTAMP))
pos_outbox = Table("pos_outbox", md,
Column("target", String, primary_key=True),
Column("ref", String, primary_key=True),
Column("payload", String),
Column("status", String),
Column("attempts", Integer),
Column("last_error", String),
Column("sent_at", TIMESTAMP),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pos_settings = Table("pos_settings", md,
Column("setting_key", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("value", String),
Column("set_by", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
class ClaimLost(Exception):
"""Another transaction owns this claim key: re-read and retry or be refused."""
def _pick(obj, names: tuple) -> dict:
return {n: getattr(obj, n) for n in names}
def _make(cls, row, names: tuple):
return cls(**{n: row[n] for n in names})
def _remember(agg, loaded):
"""Keep what was loaded so a save writes the DIFF (one statement per table and column set, however many lines)."""
object.__setattr__(agg, "_loaded", loaded)
return agg
def _dec(v) -> Optional[Decimal]:
return None if v is None else Decimal(str(v))
class ShopRepository:
def __init__(self, conn):
self.conn = conn
def get(self, shop_id) -> Optional[tuple]:
"""(shop dict with its registers, version), or None."""
rows = SHOP.load(self.conn, {"shop_id": shop_id})
if rows is None:
return None
shop = {c: rows["pos_shop"][0][c] for c in SHOP_COLUMNS}
shop["registers"] = sorted(({c: r[c] for c in REGISTER_COLUMNS} for r in rows["pos_register"]), key=lambda r: r["register_id"])
return shop, SHOP.version_of(rows)
def save(self, shop: dict, expected_version: int) -> int:
return SHOP.save(self.conn, {"pos_shop": [{c: shop[c] for c in SHOP_COLUMNS}],
"pos_register": [{c: r[c] for c in REGISTER_COLUMNS} for r in shop["registers"]]},
expected_version=expected_version)
def ids(self) -> list:
return sorted(r[0] for r in self.conn.execute(select(pos_shop.c.shop_id)))
def staff(self, shop_id: str) -> list:
return sorted(r[0] for r in self.conn.execute(select(pos_shop_user.c.sub).where(pos_shop_user.c.shop_id == shop_id)))
def shops_of(self, sub: str) -> list:
return sorted(r[0] for r in self.conn.execute(select(pos_shop_user.c.shop_id).where(pos_shop_user.c.sub == sub)))
def assign(self, shop_id: str, sub: str, email: str, by: str) -> None:
upsert(self.conn, pos_shop_user, [{"shop_id": shop_id, "sub": sub, "email": email, "assigned_by": by}], ("shop_id", "sub"))
class SessionRepository:
def __init__(self, conn):
self.conn = conn
def get(self, session_id) -> Optional[tuple]:
rows = SESSION.load(self.conn, {"session_id": session_id})
if rows is None:
return None
agg = _make(Session, rows["pos_session"][0], SESSION_COLUMNS)
for f in ("opening_float", "float_difference", "counted_cash", "expected_cash", "difference", "closing_float"):
setattr(agg, f, _dec(getattr(agg, f)))
for f in ("counted_by", "counted_sub", "approved_by", "approved_sub", "reason"):
setattr(agg, f, getattr(agg, f) or "")
return _remember(agg, rows), SESSION.version_of(rows)
def save(self, agg: Session, expected_version: int) -> int:
return SESSION.save(self.conn, {"pos_session": [_pick(agg, SESSION_COLUMNS)]}, expected_version=expected_version,
snapshot=getattr(agg, "_loaded", None) or SESSION.nothing_stored({"session_id": agg.session_id}))
def ids(self, shops: Optional[list] = None) -> list:
stmt = select(pos_session.c.session_id)
if shops is not None:
stmt = stmt.where(pos_session.c.shop_id.in_(shops))
return sorted(r[0] for r in self.conn.execute(stmt))
def of_register(self, shop_id: str, register_id: str) -> list:
"""The register's sessions, newest first: (session_id, session_no, status, closing_float). One statement."""
stmt = (select(pos_session.c.session_id, pos_session.c.session_no, pos_session.c.status, pos_session.c.closing_float)
.where(and_(pos_session.c.shop_id == shop_id, pos_session.c.register_id == register_id))
.order_by(pos_session.c.session_no.desc()))
return [tuple(r) for r in self.conn.execute(stmt)]
def add_move(self, move: CashMove, at: datetime) -> None:
self.conn.execute(insert(pos_cash_move).values(**_pick(move, ("cash_move_id", "session_id", "seq", "kind", "amount", "reason_code",
"approved_by", "approved_sub", "created_by", "created_sub")), created_at=at))
def moves(self, session_id: str) -> list:
stmt = select(pos_cash_move).where(pos_cash_move.c.session_id == session_id).order_by(pos_cash_move.c.seq)
return [dict(r._mapping) for r in self.conn.execute(stmt)]
def next_move_seq(self, session_id: str) -> int:
n = self.conn.execute(select(func.max(pos_cash_move.c.seq)).where(pos_cash_move.c.session_id == session_id)).scalar()
return int(n or 0) + 1
def move_totals(self, session_id: str) -> dict:
"""kind -> total, from this session's own rows (one statement)."""
stmt = (select(pos_cash_move.c.kind, func.sum(pos_cash_move.c.amount))
.where(pos_cash_move.c.session_id == session_id).group_by(pos_cash_move.c.kind))
return {k: Decimal(str(v)) for k, v in self.conn.execute(stmt)}
def takings(self, session_id: str) -> dict:
"""Tender totals of the session's FINALISED receipts, by method (one statement): amount, tips, change, count of lines.
A declined card, or a line on a receipt still open, counts for nothing."""
j = pos_payment_line.join(pos_receipt, pos_payment_line.c.receipt_id == pos_receipt.c.receipt_id)
stmt = (select(pos_payment_line.c.method, func.sum(pos_payment_line.c.amount), func.sum(pos_payment_line.c.tip_amount),
func.sum(pos_payment_line.c.change_amount), func.count())
.select_from(j)
.where(and_(pos_receipt.c.session_id == session_id, pos_receipt.c.status == ReceiptState.FINALISED.value,
pos_payment_line.c.terminal_status.in_(("", "authorised"))))
.group_by(pos_payment_line.c.method))
return {m: {"amount": Decimal(str(a)), "tips": Decimal(str(t)), "change": Decimal(str(c)), "lines": int(n)}
for m, a, t, c, n in self.conn.execute(stmt)}
def sales_by_code(self, session_id: str) -> tuple:
"""([(tax_code, gross, vat)] of the session's finalised PRODUCT lines, the gross of the gift vouchers it sold), in one statement
however many lines or receipts there are. A voucher sold is a liability, not a sale: it has no tax code and no revenue."""
j = pos_receipt_line.join(pos_receipt, pos_receipt_line.c.receipt_id == pos_receipt.c.receipt_id)
stmt = (select(pos_receipt_line.c.kind, pos_receipt_line.c.tax_code, func.sum(pos_receipt_line.c.line_gross), func.sum(pos_receipt_line.c.vat))
.select_from(j)
.where(and_(pos_receipt.c.session_id == session_id, pos_receipt.c.status == ReceiptState.FINALISED.value))
.group_by(pos_receipt_line.c.kind, pos_receipt_line.c.tax_code))
products, vouchers = [], Decimal("0.00")
for kind, code, g, v in self.conn.execute(stmt):
if (kind or "product") == "voucher_sale":
vouchers += Decimal(str(g))
else:
products.append((code, Decimal(str(g)), Decimal(str(v))))
return products, vouchers
def safe_moves(self, shop_id: str) -> dict:
"""kind -> total of the shop's safe drops and float top-ups, over every session of the shop (one statement)."""
j = pos_cash_move.join(pos_session, pos_cash_move.c.session_id == pos_session.c.session_id)
stmt = (select(pos_cash_move.c.kind, func.sum(pos_cash_move.c.amount)).select_from(j)
.where(and_(pos_session.c.shop_id == shop_id, pos_cash_move.c.kind.in_(("safe_drop", "float_top_up"))))
.group_by(pos_cash_move.c.kind))
return {k: Decimal(str(v)) for k, v in self.conn.execute(stmt)}
def sales(self, session_id: str) -> dict:
"""Receipt totals of the session's finalised receipts, and how many receipts were voided (one statement)."""
final = pos_receipt.c.status == ReceiptState.FINALISED.value
stmt = (select(func.sum(case((final, 1), else_=0)), func.sum(case((final, pos_receipt.c.gross), else_=0)),
func.sum(case((final, pos_receipt.c.net), else_=0)), func.sum(case((final, pos_receipt.c.vat), else_=0)),
func.sum(case((pos_receipt.c.status == ReceiptState.VOIDED.value, 1), else_=0)))
.where(pos_receipt.c.session_id == session_id))
n, g, ne, v, voids = self.conn.execute(stmt).one()
z = Decimal("0.00")
return {"receipts": int(n or 0), "gross": Decimal(str(g)) if g is not None else z,
"net": Decimal(str(ne)) if ne is not None else z, "vat": Decimal(str(v)) if v is not None else z,
"voids": int(voids or 0)}
class ReceiptRepository:
def __init__(self, conn):
self.conn = conn
def get(self, receipt_id) -> Optional[tuple]:
rows = RECEIPT.load(self.conn, {"receipt_id": receipt_id})
if rows is None:
return None
agg = _make(Receipt, rows["pos_receipt"][0], RECEIPT_COLUMNS)
for f in ("gross", "net", "vat"):
setattr(agg, f, _dec(getattr(agg, f)))
for f in ("original_receipt_id", "void_reason", "voided_by"):
setattr(agg, f, getattr(agg, f) or "")
agg.lines = [_line(r) for r in sorted(rows["pos_receipt_line"], key=lambda r: r["line_no"])]
agg.payments = [_payment(r) for r in sorted(rows["pos_payment_line"], key=lambda r: r["pay_no"])]
agg.components = [_component(r) for r in sorted(rows["pos_line_component"], key=lambda r: (r["line_no"], r["component_no"]))]
return _remember(agg, rows), RECEIPT.version_of(rows)
def save(self, agg: Receipt, expected_version: int) -> int:
rid = agg.receipt_id
return RECEIPT.save(self.conn, {
"pos_receipt": [_pick(agg, RECEIPT_COLUMNS)],
"pos_receipt_line": [{"receipt_id": rid, **_pick(x, LINE_COLUMNS[1:])} for x in agg.lines],
"pos_payment_line": [{"receipt_id": rid, **_pick(x, PAYMENT_COLUMNS[1:])} for x in agg.payments],
"pos_line_component": [{"receipt_id": rid, **_pick(x, COMPONENT_COLUMNS[1:])} for x in agg.components]},
expected_version=expected_version,
snapshot=getattr(agg, "_loaded", None) or RECEIPT.nothing_stored({"receipt_id": rid}))
def ids(self, shops: Optional[list] = None, session_id: Optional[str] = None) -> list:
stmt = select(pos_receipt.c.receipt_id)
if shops is not None:
stmt = stmt.where(pos_receipt.c.shop_id.in_(shops))
if session_id:
stmt = stmt.where(pos_receipt.c.session_id == session_id)
return sorted(r[0] for r in self.conn.execute(stmt))
def last_no(self, session_id: str) -> int:
"""The server's last receipt number of the session (committed rows)."""
n = self.conn.execute(select(func.max(pos_receipt.c.receipt_no)).where(pos_receipt.c.session_id == session_id)).scalar()
return int(n or 0)
def _line(r) -> ReceiptLine:
return ReceiptLine(line_no=r["line_no"], product_id=r["product_id"], description=r["description"] or "",
quantity=Decimal(format(Decimal(str(r["quantity"])).normalize(), "f")), unit_code=r["unit_code"],
unit_price=Decimal(str(r["unit_price"])), line_gross=Decimal(str(r["line_gross"])),
tax_category=r["tax_category"], tax_code=r["tax_code"], rate_pct=Decimal(str(r["rate_pct"])),
vat=Decimal(str(r["vat"])), original_line_no=r["original_line_no"], restock=r["restock"],
kind=r["kind"] or "product", voucher_code=r["voucher_code"] or "", recipe_source=r["recipe_source"] or "",
recipe_ref=r["recipe_ref"] or "", recipe_version=r["recipe_version"], modifiers=r["modifiers"] or "")
def _component(r) -> LineComponent:
return LineComponent(line_no=r["line_no"], component_no=r["component_no"], product_id=r["product_id"],
quantity=Decimal(format(Decimal(str(r["quantity"])).normalize(), "f")), unit_code=r["unit_code"])
def _payment(r) -> PaymentLine:
return PaymentLine(pay_no=r["pay_no"], method=r["method"], amount=Decimal(str(r["amount"])),
tip_amount=Decimal(str(r["tip_amount"])), change_amount=Decimal(str(r["change_amount"])),
terminal_txn_ref=r["terminal_txn_ref"] or "", terminal_status=r["terminal_status"] or "",
idempotency_key=r["idempotency_key"] or "", voucher_code=r["voucher_code"] or "")
class ClaimRepository:
def __init__(self, conn):
self.conn = conn
def claim(self, key: str, kind: str, session_id: str, by: str, at: datetime) -> None:
"""A plain INSERT. A key already claimed (committed, or by a racer that commits first) raises ClaimLost."""
try:
self.conn.execute(insert(pos_claim).values(claim_key=key, kind=kind, session_id=session_id, claimed_by=by, claimed_at=at))
except IntegrityError as exc:
text = f"{getattr(exc, 'orig', '')} {exc}"
if "23505" in text or "duplicate key" in text or "UNIQUE constraint" in text:
raise ClaimLost(key) from None
raise
def claim_many(self, keys: list, kind: str, session_id: str, by: str, at: datetime) -> None:
"""Several claims in ONE statement (a refund of 20 lines is not 20 round trips). Any key already claimed loses them all."""
if not keys:
return
try:
self.conn.execute(insert(pos_claim).values([dict(claim_key=k, kind=kind, session_id=session_id, claimed_by=by, claimed_at=at)
for k in keys]))
except IntegrityError as exc:
text = f"{getattr(exc, 'orig', '')} {exc}"
if "23505" in text or "duplicate key" in text or "UNIQUE constraint" in text:
raise ClaimLost(", ".join(keys)) from None
raise
def exists(self, key: str) -> bool:
return self.conn.execute(select(pos_claim.c.claim_key).where(pos_claim.c.claim_key == key)).first() is not None
class OutboxRepository:
def __init__(self, conn):
self.conn = conn
def add(self, row: OutboxRow, at: datetime) -> None:
"""A new pending row: a plain INSERT, in the transaction of the receipt it belongs to."""
self.conn.execute(insert(pos_outbox).values(target=row.target, ref=row.ref, payload=row.dump(), status=row.status,
attempts=row.attempts, last_error=row.last_error, sent_at=None,
created_at=at, updated_at=at))
def save(self, row: OutboxRow, at: datetime) -> None:
"""The row's new state. Every column is in VALUES (ON CONFLICT DO UPDATE writes them all on DataK3)."""
upsert(self.conn, pos_outbox, [{"target": row.target, "ref": row.ref, "payload": row.dump(), "status": row.status,
"attempts": row.attempts, "last_error": row.last_error, "sent_at": row.sent_at,
"updated_at": at}], ("target", "ref"))
def get(self, target: str, ref: str) -> Optional[OutboxRow]:
r = self.conn.execute(select(pos_outbox).where(and_(pos_outbox.c.target == target, pos_outbox.c.ref == ref))).mappings().first()
return _outbox(r) if r else None
def pending(self, limit: int = 100) -> list:
stmt = (select(pos_outbox).where(pos_outbox.c.status == OutboxStatus.PENDING.value)
.order_by(pos_outbox.c.created_at, pos_outbox.c.ref).limit(limit))
return [_outbox(r) for r in self.conn.execute(stmt).mappings()]
def add_many(self, rows: list, at: datetime) -> None:
"""New pending rows in ONE statement, in the transaction of the receipt they belong to."""
if rows:
self.conn.execute(insert(pos_outbox).values([dict(target=r.target, ref=r.ref, payload=r.dump(), status=r.status, attempts=r.attempts,
last_error=r.last_error, sent_at=None, created_at=at, updated_at=at) for r in rows]))
def listing(self, status: Optional[str] = None, limit: int = 200) -> list:
stmt = select(pos_outbox).order_by(pos_outbox.c.created_at, pos_outbox.c.ref).limit(limit)
if status:
stmt = stmt.where(pos_outbox.c.status == status)
return [_outbox(r) for r in self.conn.execute(stmt).mappings()]
def _outbox(r) -> OutboxRow:
return OutboxRow(target=r["target"], ref=r["ref"], payload=json.loads(r["payload"]), status=r["status"],
attempts=int(r["attempts"] or 0), last_error=r["last_error"] or "", sent_at=r["sent_at"])
class SettingsRepository:
"""The controls and defaults, dated: a setting changes by a NEW row that takes over on its valid_from, never by an edit."""
def __init__(self, conn):
self.conn = conn
def _rows(self) -> list:
return [dict(r) for r in self.conn.execute(select(pos_settings).order_by(pos_settings.c.setting_key, pos_settings.c.valid_from)).mappings()]
def timezone(self) -> str:
"""THE port for business_timezone: 'today' in every brick is counted in ONE zone per bucket, the one in force NOW. A
row takes over once the business date, counted in the zone BEFORE it, has reached its valid_from (never by the UTC
date, which could put a row dated tomorrow in force at once)."""
name = SETTING_DEFS["business_timezone"][1]
now = work.now()
for r in self._rows():
if r["setting_key"] != "business_timezone":
continue
if r["valid_from"] <= now.astimezone(ZoneInfo(name)).date():
name = json.loads(r["value"])
else:
break
return name
def in_force(self, on: date) -> dict:
"""Every setting's value on a business date: the latest row at or before it, else the default."""
out = {k: d[1] for k, d in SETTING_DEFS.items()}
for r in self._rows():
if r["valid_from"] <= on:
out[r["setting_key"]] = json.loads(r["value"])
return out
def rows(self) -> list:
return [{"setting_key": r["setting_key"], "valid_from": r["valid_from"], "value": json.loads(r["value"]),
"set_by": r["set_by"]} for r in self._rows()]
def add(self, key: str, valid_from: date, value, by: str, at: datetime) -> None:
"""A plain INSERT on (key, valid_from): a second row for the same day is a conflict, not a silent rewrite."""
self.conn.execute(insert(pos_settings).values(setting_key=key, valid_from=valid_from, value=json.dumps(value), set_by=by,
created_at=at, updated_at=at))
def has(self, key: str) -> bool:
return self.conn.execute(select(pos_settings.c.setting_key).where(pos_settings.c.setting_key == key)).first() is not None
class RefundRepository:
"""The refund claims of an original receipt, and what its refunds paid back by each method."""
def __init__(self, conn):
self.conn = conn
def claimed(self, original_receipt_id: str) -> dict:
"""line_no -> Claimed (quantity, gross, vat, number of claims), from the original's own claim rows: one statement."""
stmt = (select(pos_refund_claim.c.line_no, func.sum(pos_refund_claim.c.quantity), func.sum(pos_refund_claim.c.gross),
func.sum(pos_refund_claim.c.vat), func.count())
.where(pos_refund_claim.c.original_receipt_id == original_receipt_id).group_by(pos_refund_claim.c.line_no))
return {n: Claimed(quantity=Decimal(format(Decimal(str(q)).normalize(), "f")), gross=Decimal(str(g)), vat=Decimal(str(v)), claims=int(c))
for n, q, g, v, c in self.conn.execute(stmt)}
def add_claims(self, rows: list, at: datetime) -> None:
"""One statement for every claimed line. A plain INSERT: a racer that committed claim n first makes this one a duplicate key."""
if rows:
try:
self.conn.execute(insert(pos_refund_claim).values([{**r, "created_at": at} for r in rows]))
except IntegrityError as exc:
text = f"{getattr(exc, 'orig', '')} {exc}"
if "23505" in text or "duplicate key" in text or "UNIQUE constraint" in text:
raise ClaimLost(f"refund of {rows[0]['original_receipt_id']}") from None
raise
def refunded_by_method(self, original_receipt_id: str, method: str) -> Decimal:
"""What the original's finalised refunds paid back by one method (a positive number), one statement."""
j = pos_payment_line.join(pos_receipt, pos_payment_line.c.receipt_id == pos_receipt.c.receipt_id)
total = self.conn.execute(select(func.sum(pos_payment_line.c.amount)).select_from(j)
.where(and_(pos_receipt.c.original_receipt_id == original_receipt_id, pos_receipt.c.kind == "refund",
pos_payment_line.c.method == method))).scalar()
return -Decimal(str(total)) if total is not None else Decimal("0.00")
class CategoryRuleRepository:
def __init__(self, conn):
self.conn = conn
def restock_map(self, categories: list) -> dict:
"""category_code -> restock on refund, for the categories asked about (one statement); an unknown category is no restock."""
if not categories:
return {}
stmt = select(pos_category_rule.c.category_code, pos_category_rule.c.restock_on_refund).where(pos_category_rule.c.category_code.in_(categories))
found = {c: bool(r) for c, r in self.conn.execute(stmt)}
return {c: found.get(c, False) for c in categories}
def all(self) -> dict:
return {c: bool(r) for c, r in self.conn.execute(select(pos_category_rule.c.category_code, pos_category_rule.c.restock_on_refund))}
def save(self, category_code: str, restock: bool, by: str, at: datetime) -> None:
upsert(self.conn, pos_category_rule, [{"category_code": category_code, "restock_on_refund": restock, "set_by": by, "updated_at": at}],
("category_code",))
class VoucherRepository:
"""Vouchers and their claims. The balance is derived from the claims: one statement gives the last claim number and the sum."""
def __init__(self, conn):
self.conn = conn
def get(self, voucher_code: str) -> Optional[Voucher]:
r = self.conn.execute(select(pos_voucher).where(pos_voucher.c.voucher_code == voucher_code)).mappings().first()
if r is None:
return None
v = Voucher(**{c: r[c] for c in VOUCHER_COLUMNS})
v.face_value = Decimal(str(v.face_value))
return v
def exists(self, voucher_code: str) -> bool:
return self.conn.execute(select(pos_voucher.c.voucher_code).where(pos_voucher.c.voucher_code == voucher_code)).first() is not None
def balance(self, voucher_code: str) -> tuple:
"""(balance, last claim number) from the voucher's own claims: one statement."""
total, last = self.conn.execute(select(func.sum(pos_voucher_claim.c.amount), func.max(pos_voucher_claim.c.claim_no))
.where(pos_voucher_claim.c.voucher_code == voucher_code)).one()
return (Decimal(str(total)) if total is not None else Decimal("0.00")), int(last or 0)
def claims(self, voucher_code: str) -> list:
stmt = select(pos_voucher_claim).where(pos_voucher_claim.c.voucher_code == voucher_code).order_by(pos_voucher_claim.c.claim_no)
return [{**{k: r[k] for k in VOUCHER_CLAIM_COLUMNS}, "amount": Decimal(str(r["amount"])), "balance_after": Decimal(str(r["balance_after"]))}
for r in self.conn.execute(stmt).mappings()]
def add(self, v: Voucher, at: datetime) -> None:
"""A new voucher: a plain INSERT (a code is issued once)."""
self.conn.execute(insert(pos_voucher).values(**_pick(v, VOUCHER_COLUMNS), created_at=at))
def add_claims(self, claims: list, at: datetime) -> None:
"""Claim rows in ONE statement. A plain INSERT: a racer that committed claim n first makes this one a duplicate key."""
if claims:
try:
self.conn.execute(insert(pos_voucher_claim).values([{**_pick(c, VOUCHER_CLAIM_COLUMNS), "created_at": at} for c in claims]))
except IntegrityError as exc:
text = f"{getattr(exc, 'orig', '')} {exc}"
if "23505" in text or "duplicate key" in text or "UNIQUE constraint" in text:
raise ClaimLost(f"voucher {claims[0].voucher_code}") from None
raise
def companies(self, shop_ids: list) -> list:
"""The companies of some shops (for who may see a voucher): one statement."""
if not shop_ids:
return []
return sorted({r[0] for r in self.conn.execute(select(pos_shop.c.company_code).where(pos_shop.c.shop_id.in_(shop_ids)))})
class RecipeRepository:
"""Recipe versions: two statements to read the recipes of any number of products, plain INSERTs to write one (a version is never edited)."""
def __init__(self, conn):
self.conn = conn
def for_products(self, product_ids: list) -> dict:
"""product_id -> [Recipe] (every version, oldest first) for the products asked about: ONE statement for the roots, ONE for the lines."""
if not product_ids:
return {}
roots = self.conn.execute(select(pos_recipe).where(pos_recipe.c.product_id.in_(product_ids))).mappings().all()
if not roots:
return {}
lines = self.conn.execute(select(pos_recipe_line).where(pos_recipe_line.c.product_id.in_(product_ids))).mappings().all()
by = {}
for r in roots:
by[(r["product_id"], r["version"])] = Recipe(product_id=r["product_id"], version=r["version"], valid_from=r["valid_from"],
created_by=r["created_by"] or "")
for ln in sorted(lines, key=lambda x: x["line_no"]):
by[(ln["product_id"], ln["version"])].lines.append(RecipeLine(
line_no=ln["line_no"], component_id=ln["component_id"], quantity=Decimal(format(Decimal(str(ln["quantity"])).normalize(), "f")),
unit_code=ln["unit_code"], modifier_id=ln["modifier_id"] or "", replaces=ln["replaces"] or ""))
out: dict = {}
for (pid, _v), rec in sorted(by.items(), key=lambda kv: (kv[0][0], kv[0][1])):
out.setdefault(pid, []).append(rec)
return out
def products(self) -> list:
return sorted({r[0] for r in self.conn.execute(select(pos_recipe.c.product_id))})
def add(self, recipe: Recipe, at: datetime) -> None:
"""A new version: plain INSERTs (the root, then all its lines in one statement). A racer for the same version loses on the key."""
self.conn.execute(insert(pos_recipe).values(product_id=recipe.product_id, version=recipe.version, valid_from=recipe.valid_from,
created_by=recipe.created_by, created_at=at))
self.conn.execute(insert(pos_recipe_line).values([dict(product_id=recipe.product_id, version=recipe.version, 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 recipe.lines]))
class SnapshotRepository:
def __init__(self, conn):
self.conn = conn
def latest(self, shop_id: str, business_date: date) -> Optional[Snapshot]:
"""The shop's newest snapshot of that business date (one statement), or None."""
r = self.conn.execute(select(pos_register_snapshot).where(and_(pos_register_snapshot.c.shop_id == shop_id,
pos_register_snapshot.c.business_date == business_date))
.order_by(pos_register_snapshot.c.snapshot_no.desc()).limit(1)).mappings().first()
return _snapshot(r) if r else None
def get(self, shop_id: str, business_date: date, snapshot_no: int) -> Optional[Snapshot]:
r = self.conn.execute(select(pos_register_snapshot).where(and_(
pos_register_snapshot.c.shop_id == shop_id, pos_register_snapshot.c.business_date == business_date,
pos_register_snapshot.c.snapshot_no == snapshot_no))).mappings().first()
return _snapshot(r) if r else None
def add(self, snap: Snapshot, at: datetime) -> None:
self.conn.execute(insert(pos_register_snapshot).values(
shop_id=snap.shop_id, business_date=snap.business_date, snapshot_no=snap.snapshot_no, price_list_id=snap.price_list_id,
country=snap.country, currency=snap.currency, prices_include_tax=snap.prices_include_tax, valid_until=snap.valid_until,
digest=snap.digest, products_json=json.dumps(snap.products, sort_keys=True), rates_json=json.dumps(snap.rates, sort_keys=True),
built_at=at))
def _snapshot(r) -> Snapshot:
return Snapshot(shop_id=r["shop_id"], business_date=r["business_date"], snapshot_no=r["snapshot_no"], price_list_id=r["price_list_id"] or "",
country=r["country"] or "", currency=r["currency"] or "", prices_include_tax=bool(r["prices_include_tax"]),
valid_until=r["valid_until"], digest=r["digest"] or "", products=json.loads(r["products_json"] or "[]"),
rates=json.loads(r["rates_json"] or "[]"), built_at=r["built_at"])
class DepositRepository:
"""Sealed bags: a plain INSERT (the key is the claim), then the status moves forward by its own rule."""
def __init__(self, conn):
self.conn = conn
def add(self, d: CashDeposit, at: datetime) -> None:
self.conn.execute(insert(pos_cash_deposit).values(**_pick(d, DEPOSIT_COLUMNS), created_at=at, updated_at=at))
def save(self, d: CashDeposit, at: datetime) -> None:
upsert(self.conn, pos_cash_deposit, [{**_pick(d, DEPOSIT_COLUMNS), "updated_at": at}], ("deposit_ref",))
def get(self, ref: str) -> Optional[CashDeposit]:
r = self.conn.execute(select(pos_cash_deposit).where(pos_cash_deposit.c.deposit_ref == ref)).mappings().first()
return _deposit(r) if r else None
def of_shop(self, shop_id: str) -> list:
stmt = select(pos_cash_deposit).where(pos_cash_deposit.c.shop_id == shop_id).order_by(pos_cash_deposit.c.deposit_ref)
return [_deposit(r) for r in self.conn.execute(stmt).mappings()]
def sealed(self, shop_id: str) -> tuple:
"""(total, count) of the shop's sealed bags, one statement."""
total, n = self.conn.execute(select(func.sum(pos_cash_deposit.c.amount), func.count())
.where(pos_cash_deposit.c.shop_id == shop_id)).one()
return Decimal(str(total)) if total is not None else Decimal("0.00"), int(n or 0)
def _deposit(r) -> CashDeposit:
d = CashDeposit(**{c: r[c] for c in DEPOSIT_COLUMNS})
d.amount, d.denominations = Decimal(str(d.amount)), d.denominations or ""
return d
class Repository:
"""Every repository on one connection: what `use_cases.repository()` yields. One transaction."""
def __init__(self, conn):
self.conn = conn
self.shops = ShopRepository(conn)
self.sessions = SessionRepository(conn)
self.receipts = ReceiptRepository(conn)
self.claims = ClaimRepository(conn)
self.outbox = OutboxRepository(conn)
self.settings = SettingsRepository(conn)
self.deposits = DepositRepository(conn)
self.refunds = RefundRepository(conn)
self.vouchers = VoucherRepository(conn)
self.recipes = RecipeRepository(conn)
self.snapshots = SnapshotRepository(conn)
self.categories = CategoryRuleRepository(conn)
def seed(self) -> None:
"""Nothing to seed: a setting's default is in code, and a row exists only once the customer changed it."""