dodil/erp-brick-purchasing
DODIL ERP brick: purchasing
src/brick_purchasing/application/master.py
"""Settings, controls and master data: approval limits, release strategies, tolerance keys, terms, the account map, supplier prices, supplier
profiles and bank details (PUR-13, PUR-23, PUR-30, PUR-31, PUR-52).
Every row is DATED: it starts AFTER today (the first row of a key may start today); a correction is a new row, never an edit. `today` is the
date in the shared business_timezone. A control (purchasing:settings:controls) is refused to purchasing:settings:write."""
import json
from datetime import date
from decimal import Decimal
from typing import Annotated, Literal, Optional, Union
from erp_platform import auth
from erp_platform.bricks import Refused
from erp_platform.types import BigId, IsoDate, Limit, Name, text
from pydantic import Field
from brick_purchasing.application import base
from brick_purchasing.application.base import (
BANK_APPROVE,
BANK_WRITE,
CONTROLS,
PRICE_WRITE,
PURCHASING,
SETTINGS_WRITE,
SUP_READ,
SUP_WRITE,
_decide,
_ids,
_stamp,
_today,
_who,
unit,
)
from brick_purchasing.application.inputs import Code, Currency, Money, OptCode, Pct, Price, Qty, Reason, Whole
from brick_purchasing.domain import approval
from brick_purchasing.domain.errors import Invalid, RuleBroken
from brick_purchasing.domain.payment import mask
from brick_purchasing.domain.values import EVERY, SETTING_DEFS, check_setting
from brick_purchasing.infrastructure import ai_steps, vault
from brick_purchasing.infrastructure.seams import Missing, ports
Levels = Annotated[int, Field(ge=1, le=3)]
Days = Annotated[int, Field(ge=0, le=3650)]
repository = lambda: base.repository() # noqa: E731 — resolved live, so a test's swap is seen by every module
def check_start(rows: list, valid_from: date, today: date, what: str) -> None:
"""A dated row starts after today; the first row of its key may start today (module.yaml config_rows)."""
real = [r for r in rows if r.get("set_by") != "default"]
if valid_from <= today and real:
raise Refused(409, f"{what}: a new row starts after today ({today}); {valid_from} is today or earlier. Dated rows are never back-dated: "
"use a later date")
if valid_from < today:
raise Refused(409, f"{what}: the first row of a key may start today ({today}), not before")
# ═══ settings ═══════════════════════════════════════════════════════════════════════════════════
@PURCHASING.use_case(kind="query", route="GET /settings")
def list_purchasing_settings(caller: dict) -> dict:
"""Every setting with its dated rows (the one in force today marked). The controls — tolerances, thresholds, number formats, the payment-file
format — are shown to the holder of purchasing:settings:controls only."""
base._require_any(caller, SETTINGS_WRITE, CONTROLS)
sees_controls = auth.has_permission(caller, CONTROLS)
with repository() as repo:
today = _today(repo)
out = []
for d in SETTING_DEFS.values():
if d.control and not sees_controls:
continue
rows = repo.settings.rows(d.key)
current = repo.settings.value_on(d.key, today)
out.append({"setting_key": d.key, "kind": d.kind, "control": d.control, "changes": d.changes, "value": current,
"rows": [{"valid_from": r["valid_from"], "value": r["value"], "set_by": r["set_by"]} for r in rows]})
return {"today": today, "settings": out}
def _set(repo, caller, key: str, value, valid_from: Optional[str], control: bool) -> dict:
d = SETTING_DEFS.get(key)
if d is None:
raise Refused(422, f"there is no setting {key!r}")
if d.control != control:
raise Refused(403 if d.control else 409,
f"{key} is a control: set it with set_purchasing_control (purchasing:settings:controls)" if d.control
else f"{key} is a day-to-day setting: set it with set_purchasing_setting (purchasing:settings:write)")
try:
value = check_setting(key, value)
except Invalid as e:
raise Refused(422, str(e)) from None
today = _today(repo)
start = base._day(valid_from, "valid_from") or date.fromordinal(today.toordinal() + 1)
check_start(repo.settings.rows(key), start, today, key)
repo.settings.put(key, start, value, _who(caller), _stamp(), d.changes)
return {"setting_key": key, "valid_from": start, "value": value}
@PURCHASING.use_case(kind="command", route="POST /settings", permission=SETTINGS_WRITE)
def set_purchasing_setting(caller: dict, setting_key: Code, value: Union[bool, int, text(200)], valid_from: Optional[IsoDate] = None) -> dict:
"""Change a day-to-day setting (terms, AI step modes, ERS grouping, firm zone) from a date AFTER today. A control — tolerances, the order
threshold, number formats, the payment-file format — is refused here: it is the owner's (set_purchasing_control)."""
def run():
with repository() as repo:
return _set(repo, caller, setting_key, value, valid_from, control=False)
return unit(None, run)
@PURCHASING.use_case(kind="command", route="POST /settings/controls", permission=CONTROLS)
def set_purchasing_control(caller: dict, setting_key: Code, value: Union[bool, int, text(200)], valid_from: Optional[IsoDate] = None) -> dict:
"""The owner sets a control from a date AFTER today (the first row of a key may start today): price and quantity tolerance, the order approval
threshold, number formats, the payment-file format and BACS service user number, the self-billing maximum. The match uses the value in force
on each bill's date."""
def run():
with repository() as repo:
return _set(repo, caller, setting_key, value, valid_from, control=True)
return unit(None, run)
# ═══ approval limits and release strategies (PUR-13, PUR-38) ═════════════════════════════════════
@PURCHASING.use_case(kind="command", route="POST /approval-limits", permission=CONTROLS)
def set_approval_limit(caller: dict, approver_email: text(254, min_length=3), limit_amount: Money, currency: Currency = "GBP",
category: OptCode = EVERY, approver_sub: Optional[text(254)] = None, valid_from: Optional[IsoDate] = None) -> dict:
"""The owner sets how much this person may commit, to what category (or every category), from a date AFTER today — a promotion changes a limit
without a release and nobody can back-date one. The first row of a person and category may start today. Today's approvals use today's limit."""
def run():
with repository() as repo, _decide():
today = _today(repo)
email = approver_email.strip().lower()
if limit_amount < 0:
raise Invalid("a limit is zero or more")
start = base._day(valid_from, "valid_from") or date.fromordinal(today.toordinal() + 1)
mine = [r for r in repo.limits.all(approver_email=email, category=category or EVERY)]
check_start(mine, start, today, "approval limit")
row = {"approver_email": email, "category": category or EVERY, "valid_from": start, "approver_sub": (approver_sub or "").lower(),
"limit_amount": limit_amount, "currency": base._currency(currency), "set_by_email": _who(caller)}
repo.limits.put(row)
return row
return unit(None, run)
@PURCHASING.use_case(kind="query", route="GET /approval-limits", permission=CONTROLS)
def list_approval_limits(caller: dict) -> dict:
"""Every approval limit row, with the one in force today for each person and category marked."""
with repository() as repo:
today = _today(repo)
rows = repo.limits.all()
return {"today": today, "limits": rows}
@PURCHASING.use_case(kind="command", route="POST /release-strategies", permission=CONTROLS)
def set_release_strategy(caller: dict, levels: Levels, level_thresholds: Optional[list[Money]] = None,
min_amount: Money = Decimal(0), currency: Currency = "GBP", category: OptCode = EVERY,
valid_from: Optional[IsoDate] = None) -> dict:
"""The owner says how many people must approve spend of a category from an amount (up to three levels), and the minimum limit an approver of
each level holds. The strategy in force on the day a document is SUBMITTED applies to it, so a strategy changed later moves no pending document."""
def run():
with repository() as repo, _decide():
today = _today(repo)
thresholds = level_thresholds or []
if thresholds and len(thresholds) != levels:
raise Invalid("give one threshold per level, or none")
start = base._day(valid_from, "valid_from") or date.fromordinal(today.toordinal() + 1)
key_rows = [r for r in repo.strategies.all(category=category or EVERY) if r["min_amount"] == min_amount]
check_start(key_rows, start, today, "release strategy")
row = {"category": category or EVERY, "min_amount": min_amount, "valid_from": start, "levels": levels,
"level_thresholds": json.dumps([str(t) for t in thresholds]), "currency": base._currency(currency),
"set_by_email": _who(caller)}
repo.strategies.put(row)
return row
return unit(None, run)
@PURCHASING.use_case(kind="command", route="POST /tolerances", permission=CONTROLS)
def set_tolerance(caller: dict, key: Literal["PP", "DQ", "AN", "AP", "ST", "KW", "LA", "LD", "BD", "TX"], tolerance_group: OptCode = "STANDARD",
upper_pct: Optional[Pct] = None, upper_amount: Optional[Money] = None, lower_pct: Optional[Pct] = None,
lower_amount: Optional[Money] = None, valid_from: Optional[IsoDate] = None) -> dict:
"""The owner sets how much of one kind of difference (PP price, DQ quantity, AN a line with no order, KW delivery cost, BD small difference,
TX tax) a group of suppliers' bills may carry, with upper and lower limits as a percent and an amount — the LARGER applies — from a date AFTER
today. The STANDARD group's PP, DQ and TX rows are the settings price_tolerance_pct, qty_tolerance_pct and tax_tolerance_amount."""
def run():
with repository() as repo, _decide():
today = _today(repo)
group = (tolerance_group or "STANDARD").strip().upper()
start = base._day(valid_from, "valid_from") or date.fromordinal(today.toordinal() + 1)
check_start(repo.tolerances.all(tolerance_group=group, key=key), start, today, f"tolerance {group}/{key}")
row = {"tolerance_group": group, "key": key, "valid_from": start, "lower_amount": lower_amount, "upper_amount": upper_amount,
"lower_pct": lower_pct, "upper_pct": upper_pct, "set_by_email": _who(caller)}
repo.tolerances.put(row)
return row
return unit(None, run)
@PURCHASING.use_case(kind="query", route="GET /tolerances", permission=CONTROLS)
def list_tolerances(caller: dict) -> dict:
"""Every tolerance row (group, key, limits, valid_from)."""
with repository() as repo:
return {"today": _today(repo), "tolerances": repo.tolerances.all()}
@PURCHASING.use_case(kind="command", route="POST /terms", permission=SETTINGS_WRITE, idempotent=True)
def set_payment_terms(caller: dict, terms_id: Code, name: Name, days: Days, end_of_month: bool = False,
discount_pct: Optional[Pct] = None, discount_days: Optional[Days] = None) -> dict:
"""Payment terms: 30 days, end of month following, an early-payment discount. A bill's due date comes from its supplier's terms."""
with repository() as repo:
row = {"terms_id": terms_id, "name": name, "days": days, "end_of_month": end_of_month, "discount_pct": discount_pct,
"discount_days": discount_days}
repo.terms.put(row)
return row
@PURCHASING.use_case(kind="command", route="POST /account-map", permission=SETTINGS_WRITE)
def set_purchasing_account_map(caller: dict, purpose: Literal["payables", "expense", "grni", "input_vat", "price_variance", "asset_clearing",
"supplier_down_payment", "consignment_clearing", "delivery_cost_clearing", "small_difference",
"cis_liability"], account_id: Code, category: OptCode = EVERY,
valid_from: Optional[IsoDate] = None) -> dict:
"""Which ledger account a purpose posts to (payables, an expense category, goods received not invoiced, input VAT per tax code, price
variance, asset clearing, supplier down payments, consignment clearing, delivery-cost clearing per kind, small differences, CIS liability),
from a date AFTER today. The six clearing purposes are sent to the ledger by purpose; the rest by account."""
def run():
with repository() as repo:
today = _today(repo)
start = base._day(valid_from, "valid_from") or date.fromordinal(today.toordinal() + 1)
check_start(repo.account_map.all(purpose=purpose, category=category or EVERY), start, today, f"account map {purpose}")
row = {"purpose": purpose, "category": category or EVERY, "valid_from": start, "account_id": account_id, "set_by_email": _who(caller)}
repo.account_map.put(row)
return row
return unit(None, run)
@PURCHASING.use_case(kind="query", route="GET /account-map", permission=SETTINGS_WRITE)
def list_purchasing_account_map(caller: dict) -> dict:
"""The dated account map."""
with repository() as repo:
return {"today": _today(repo), "account_map": repo.account_map.all()}
# ═══ suppliers ═══════════════════════════════════════════════════════════════════════════════════
@PURCHASING.use_case(kind="command", route="POST /suppliers/promote", permission=SUP_WRITE, idempotent=True)
def promote_supplier(caller: dict, bp_key: text(200, min_length=1), legal_name: Name, country: Optional[text(2)] = None,
tax_id: Optional[text(64)] = None, domain: Optional[text(200)] = None) -> dict:
"""Ask the party master for the supplier, so the company we buy from and the one we sell to are one record. Calling again returns the same
party; purchasing never writes the partner tables."""
try:
got = ports.bp.promote(caller, bp_key=bp_key, legal_name=legal_name, country=country or "", tax_id=tax_id or "", domain=domain or "")
except Missing as m:
raise base._refuse(m, "the party master") from None
got.pop("_status", None)
return got
@PURCHASING.use_case(kind="command", route="POST /suppliers/{bp_id}/profile", permission=SUP_WRITE)
def set_supplier_profile(caller: dict, bp_id: BigId, terms_id: OptCode = "NET30", currency: Currency = "GBP",
payment_method: Literal["pain.001", "bacs", "csv", "outside"] = "pain.001", remittance_email: Optional[text(254)] = None,
reclaims_vat: bool = True, tolerance_group: OptCode = "STANDARD", cis_subcontractor: bool = False, ers: bool = False,
legal_name: Optional[Name] = None, country: Optional[text(2)] = None, valid_from: Optional[IsoDate] = None) -> dict:
"""How we buy from and pay one supplier, from a date AFTER today (the first row may start today): terms, currency, payment method, whether
VAT is reclaimed, the tolerance group, whether they are a CIS subcontractor, whether they are self-billed (ERS needs an agreement in force)."""
def run():
with repository() as repo:
today = _today(repo)
start = base._day(valid_from, "valid_from") or today
check_start(repo.suppliers.all(bp_id=bp_id), start, today, "supplier profile")
if repo.terms.get(terms_id or "NET30") is None:
raise Refused(422, f"there are no payment terms {terms_id!r}: set them first (set_payment_terms)")
row = {"bp_id": bp_id, "valid_from": start, "terms_id": terms_id or "NET30", "currency": base._currency(currency),
"payment_method": payment_method, "remittance_email": remittance_email or "", "reclaims_vat": reclaims_vat,
"tolerance_group": (tolerance_group or "STANDARD").upper(), "cis_subcontractor": cis_subcontractor, "ers": ers,
"legal_name": legal_name or "", "country": country or "", "set_by_email": _who(caller)}
repo.suppliers.put(row)
return row
return unit(None, run)
def profile_on(repo, bp_id: int, on: date) -> dict:
"""The supplier's profile in force on a date: its latest row valid_from <= on, else the defaults."""
rows = [r for r in repo.suppliers.all(bp_id=int(bp_id)) if r["valid_from"] <= on]
if rows:
return max(rows, key=lambda r: r["valid_from"])
return {"bp_id": int(bp_id), "terms_id": "NET30", "currency": "GBP", "payment_method": "pain.001", "reclaims_vat": True,
"tolerance_group": "STANDARD", "cis_subcontractor": False, "ers": False, "legal_name": "", "country": ""}
@PURCHASING.use_case(kind="command", route="POST /supplier-prices", permission=PRICE_WRITE)
def record_supplier_price(caller: dict, bp_id: BigId, product_id: Code, unit_code: Code, unit_price: Price,
min_quantity: Qty = Decimal(1), currency: Currency = "GBP", valid_from: Optional[IsoDate] = None,
valid_to: Optional[IsoDate] = None, pricing_unit_code: Optional[Code] = None, pricing_quantity: Optional[Whole] = None,
pricing_numerator: Optional[Whole] = None, pricing_denominator: Optional[Whole] = None,
order_multiple: Optional[Qty] = None, lead_days: Optional[Days] = None,
supplier_product_code: Optional[text(100)] = None, supplier_product_name: Optional[Name] = None,
consignment: bool = False) -> dict:
"""Keep what one supplier charges for one product, in their unit, from a quantity, from a date: price (per pricing_quantity of the pricing
unit), the supplier's own conversion of that unit where it differs from the product's (2.48 KGM a bar), an own order multiple, lead time and
their product code. The product master is unchanged. The row starts AFTER today (the first row may start today)."""
def run():
with repository() as repo, _decide():
today = _today(repo)
if (pricing_numerator is None) != (pricing_denominator is None):
raise Invalid("give both pricing_numerator and pricing_denominator, or neither")
start = base._day(valid_from, "valid_from") or date.fromordinal(today.toordinal() + 1)
same = [r for r in repo.supplier_prices.all(bp_id=bp_id, product_id=product_id, unit_code=unit_code)
if r["min_quantity"] == min_quantity and bool(r.get("consignment")) == consignment]
check_start(same, start, today, "supplier price")
row = {"bp_id": bp_id, "product_id": product_id, "unit_code": unit_code, "min_quantity": min_quantity, "valid_from": start,
"unit_price": unit_price, "pricing_unit_code": pricing_unit_code or unit_code, "pricing_quantity": pricing_quantity or 1,
"pricing_numerator": pricing_numerator, "pricing_denominator": pricing_denominator, "order_multiple": order_multiple,
"currency": base._currency(currency), "lead_days": lead_days, "supplier_product_code": supplier_product_code or "",
"supplier_product_name": supplier_product_name or "", "valid_to": base._day(valid_to, "valid_to"), "consignment": consignment,
"set_by_email": _who(caller)}
repo.supplier_prices.put(row)
return row
return unit(None, run)
@PURCHASING.use_case(kind="query", route="GET /supplier-prices", permission=PRICE_WRITE)
def list_supplier_prices(caller: dict, bp_id: Optional[int] = None, product_id: Optional[Code] = None, limit: Limit = 200) -> dict:
"""The supplier price rows, with the one in force today per supplier and product marked."""
with repository() as repo:
today = _today(repo)
where = {k: v for k, v in (("bp_id", bp_id), ("product_id", product_id)) if v is not None}
rows = repo.supplier_prices.all(**where)[:limit]
for r in rows:
r["in_force"] = r["valid_from"] <= today and (r["valid_to"] is None or r["valid_to"] >= today)
return {"today": today, "prices": rows}
def price_in_force(repo, bp_id: int, product_id: str, unit_code: Optional[str], quantity: Decimal, on: date,
consignment: bool = False) -> Optional[dict]:
"""The supplier's price row for this product on `on`: of the rows in force whose min_quantity <= quantity, the highest min_quantity (a
quantity break), then the latest valid_from. In `unit_code` when given; else the row with the lowest unit code."""
rows = [r for r in repo.supplier_prices.all(bp_id=int(bp_id), product_id=product_id)
if r["valid_from"] <= on and (r["valid_to"] is None or r["valid_to"] >= on) and bool(r.get("consignment")) == consignment
and (unit_code is None or r["unit_code"] == unit_code) and (r["min_quantity"] or 0) <= quantity]
if not rows:
return None
best = max(r["min_quantity"] or 0 for r in rows)
rows = [r for r in rows if (r["min_quantity"] or 0) == best]
latest = max(r["valid_from"] for r in rows)
return sorted((r for r in rows if r["valid_from"] == latest), key=lambda r: r["unit_code"])[0]
# ═══ supplier bank details (PUR-30): a second person, a call-back, a dated row ═══════════════════
def _bank_view(row: dict, reveal: bool) -> dict:
out = {k: v for k, v in row.items() if k not in ("iban_or_account",)}
try:
plain = vault.unseal(row.get("iban_or_account") or "")
except Exception: # noqa: BLE001 — no key here: show it masked
plain = ""
out["iban_or_account"] = plain if reveal and plain else mask(plain)
return out
@PURCHASING.use_case(kind="command", route="POST /suppliers/{bp_id}/bank", permission=BANK_WRITE)
def propose_bank(caller: dict, bp_id: BigId, iban_or_account: text(64, min_length=4), valid_from: IsoDate,
bic_or_sort_code: Optional[text(32)] = None, account_name: Optional[Name] = None, country: Optional[text(2)] = None,
requested_by: Optional[text(64)] = None) -> dict:
"""Enter new bank details for a supplier. They start AFTER today at the earliest (a run proposed today uses today's details), are stored
ENCRYPTED and shown masked, and need a second person (approve_bank) after a call-back. `requested_by` — email, phone, letter — is shown to the
approver; the AI step flags a risky change (a change asked for by email, a new country, a large run due)."""
def run():
with repository() as repo, _decide():
today = _today(repo)
start = base._day(valid_from, "valid_from")
if start <= today:
raise Refused(409, f"starts_after_today: new bank details start tomorrow at the earliest ({today} is today): pick a later date")
try:
sealed = vault.seal(iban_or_account)
except vault.NoKey as e:
raise Refused(503, str(e)) from None
earlier = [r for r in repo.banks.all(bp_id=bp_id) if r["status"] == "approved"]
changed = bool(earlier) and (country or "").upper() != (earlier and (max(earlier, key=lambda r: r["valid_from"]).get("country") or "")).upper()
row = {"bp_id": bp_id, "valid_from": start, "iban_or_account": sealed, "bic_or_sort_code": bic_or_sort_code or "",
"account_name": account_name or "", "status": "proposed", "entered_by_email": _who(caller), "entered_by_sub": base._sub(caller),
"country": (country or "").upper(), "requested_by": (requested_by or "").lower()}
if repo.banks.get(bp_id, start) is not None:
raise Refused(409, f"there are already bank details for this supplier from {start}: pick another date")
repo.banks.put(row)
answer = ai_steps.ask("ai_flag_bank_change", base.Settings(repo, today).mode("ai_flag_bank_change"), f"pur_supplier_bank:{bp_id}:{start}", 1,
{"requested_by": requested_by or "", "country_changed": changed})
signals = ai_steps.bank_signals(requested_by or "", changed, Decimal(0)) if answer.kind != "off" else []
for n, s in enumerate(signals, 1):
repo.findings.put({"subject_ref": f"pur_supplier_bank:{bp_id}:{start}", "step": "ai_flag_bank_change", "run_no": 1, "finding_no": n,
"model": "baseline", "inputs_digest": "", "finding": json.dumps(s), "evidence": s["why"]})
if answer.task is not None:
repo.tasks.put({"task_id": f"bank:{bp_id}:{start}", "capability": "ai.suggest", "subject_ref": f"pur_supplier_bank:{bp_id}:{start}",
"plan": json.dumps(answer.task.plan) if getattr(answer.task, "plan", None) else "", "reason": answer.message, "status": "open"})
return {**_bank_view(row, reveal=True), "signals": signals, "ai": answer.as_dict()}
return unit(None, run)
def _bank_row(repo, bp_id: int, valid_from: str) -> dict:
row = repo.banks.get(bp_id, base._day(valid_from, "valid_from"))
if row is None:
raise Refused(404, f"no bank details for supplier {bp_id} from {valid_from}")
return row
@PURCHASING.use_case(kind="command", route="POST /suppliers/{bp_id}/bank/{valid_from}/approve", permission=BANK_APPROVE, idempotent=True)
def approve_bank(caller: dict, bp_id: BigId, valid_from: IsoDate, call_back_note: text(500, min_length=1)) -> dict:
"""Approve a change of supplier bank details. The approver is not the person who entered it (email AND sub), and the change has been confirmed
with the supplier on a number we already held — recorded in `call_back_note` (who confirmed). Used by runs from its start date."""
def run():
with repository() as repo, _decide():
row = _bank_row(repo, bp_id, valid_from)
if row["status"] == "approved":
return _bank_view(row, reveal=True)
if row["status"] != "proposed":
raise RuleBroken(f"the change is {row['status']}")
if approval.same_person(_ids(caller), approval.identities(row["entered_by_email"], row["entered_by_sub"])) or not _ids(caller):
raise RuleBroken("second_person: you entered these details (same email or same sign-in): another approver")
if not (call_back_note or "").strip():
raise RuleBroken("call_back: the change has not been confirmed with the supplier on a number we already held: call them and record who confirmed")
row.update(status="approved", approved_by_email=_who(caller), approved_by_sub=base._sub(caller), call_back_note=call_back_note.strip(),
approved_on=_today(repo))
repo.banks.put(row)
return _bank_view(row, reveal=True)
return unit(None, run)
@PURCHASING.use_case(kind="command", route="POST /suppliers/{bp_id}/bank/{valid_from}/reject", permission=BANK_APPROVE, idempotent=True)
def reject_bank(caller: dict, bp_id: BigId, valid_from: IsoDate, reason: Reason) -> dict:
"""Reject a proposed change of supplier bank details, with a reason."""
def run():
with repository() as repo, _decide():
row = _bank_row(repo, bp_id, valid_from)
if row["status"] == "rejected":
return _bank_view(row, reveal=True)
if row["status"] != "proposed":
raise RuleBroken(f"the change is {row['status']}")
if not reason.strip():
raise RuleBroken("rejecting a change needs a reason")
row.update(status="rejected", reject_reason=reason.strip())
repo.banks.put(row)
return _bank_view(row, reveal=True)
return unit(None, run)
@PURCHASING.use_case(kind="query", route="GET /suppliers/{bp_id}/bank", permission=SUP_READ)
def list_supplier_bank(caller: dict, bp_id: BigId) -> dict:
"""A supplier's bank-detail rows. The account is MASKED unless the caller may enter or approve bank details (its two roles)."""
reveal = auth.has_permission(caller, BANK_WRITE) or auth.has_permission(caller, BANK_APPROVE)
with repository() as repo:
return {"bp_id": bp_id, "banks": [_bank_view(r, reveal) for r in sorted(repo.banks.all(bp_id=bp_id), key=lambda r: r["valid_from"])]}
def bank_in_force(repo, bp_id: int, on: date) -> Optional[dict]:
"""The APPROVED bank row in force on a date, or None."""
rows = [r for r in repo.banks.all(bp_id=int(bp_id)) if r["status"] == "approved" and r["valid_from"] <= on]
return max(rows, key=lambda r: r["valid_from"]) if rows else None