dodil/erp-brick-purchasing
DODIL ERP brick: purchasing
src/brick_purchasing/application/down_payments.py
"""Down payments (PUR-47 to PUR-49): requested against an order, approved by a second person, paid in a run like a bill, cleared against the
final bills. A requested down payment is a noted item, not a liability: nothing posts until it is PAID (debit supplier down payments, credit bank).
Clearing is one row per (down payment, bill, seq) that claims part of what the down payment holds; it is written together with an allocation of
source_kind down_payment that lowers the bill's outstanding, and it bumps the down payment's version in the same transaction, so two clerks clearing
the same down payment at once collide and the loser re-reads what is left."""
from datetime import date
from decimal import Decimal
from typing import Optional
from erp_platform import auth
from erp_platform.bricks import Refused
from erp_platform.seam import derived_id
from erp_platform.types import IsoDate
from brick_purchasing.application import base, bills, payments
from brick_purchasing.application.base import (
DP_APPROVE,
DP_CLEAR,
DP_REQUEST,
PURCHASING,
_decide,
_ids,
_same_version,
_sub,
_today,
_who,
unit,
view,
)
from brick_purchasing.application.inputs import BillLineIn, Money, OptCode, Pct, Reason, Ref, Version, _In
from brick_purchasing.domain import bill as bl
from brick_purchasing.domain import rows
from brick_purchasing.domain.approval import identities, same_person
from brick_purchasing.domain.errors import RuleBroken
from brick_purchasing.domain.payment import Allocation, RunItem, SupplierPayment
from brick_purchasing.domain.values import ZERO, money_round
from brick_purchasing.infrastructure.seams import Missing, ports
repository = lambda: base.repository() # noqa: E731
def cleared(d: rows.DownPayment) -> Decimal:
return sum((c.amount for c in d.clearings), ZERO)
def remaining(d: rows.DownPayment) -> Decimal:
return d.gross_amount - cleared(d) if d.status == "paid" else ZERO
def dp_view(d: rows.DownPayment, version: int) -> dict:
out = view(d)
out["version"], out["cleared"], out["remaining"] = version, cleared(d), remaining(d)
out["paid_in_advance"] = d.status == "paid"
if d.status == "paid" and remaining(d) > 0:
out["fix"] = "clear it against the final bill, or record the supplier's refund"
return out
def _load(repo, caller: dict, did: str) -> tuple:
got = repo.down_payments.get(did)
if got is None or not (auth.has_permission(caller, DP_REQUEST) or auth.has_permission(caller, DP_APPROVE) or auth.has_permission(caller, DP_CLEAR)):
raise Refused(404, f"no down payment {did}")
return got
def _dp_tax(o, did: str, net: Decimal, tax_code, tax_amount, today: date) -> tuple:
"""The VAT on a down payment is the tax module's (direction purchase, one line of the net); the supplier's stated amount is only the fallback
when tax is not installed."""
if not ports.installed("tax"):
return (tax_amount or ZERO), (tax_code or "")
body = {"document_ref": f"pur_dp:{did}", "direction": "purchase", "tax_point": today.isoformat(), "bp_id": str(o.bp_id), "currency": o.currency,
"lines": [{"line_no": 1, "description": "down payment", "category_code": tax_code or None, "supply": "goods", "amount": str(net),
"quantity": "1", "unit_price": str(net)}]}
try:
got = ports.tax.compute(body)
except Missing as m:
raise base._refuse(m, "tax") from None
x = (got.get("lines") or [{}])[0]
return Decimal(str(x.get("vat") or 0)), (x.get("tax_code") or tax_code or "")
@PURCHASING.use_case(kind="command", route="POST /purchase-orders/{purchase_order_id}/down-payments", permission=DP_REQUEST)
def request_down_payment(caller: dict, purchase_order_id: Ref, due_date: IsoDate, percent: Optional[Pct] = None, net_amount: Optional[Money] = None,
tax_code: Optional[OptCode] = None, tax_amount: Optional[Money] = None) -> dict:
"""Ask for the down payment a supplier wants on an order: a percentage of its net total, or an amount, with the VAT the supplier states. Finance
approves it. The VAT is computed by the tax module (tax_amount counts only when tax is not installed). Requests against one order may not together exceed its net total (refused, naming what remains); a cancelled or closed order takes
none. It is a noted item, not a liability, until it is paid."""
def run():
with repository() as repo, _decide():
got = repo.orders.get(purchase_order_id)
if got is None or not auth.has_permission(caller, base.ORD_READ):
raise Refused(404, f"no purchase order {purchase_order_id}")
o = got[0]
if o.status in ("cancelled", "closed", "draft", "to_approve"):
raise RuleBroken(f"an order that is {o.status} takes no down payment request" + (": confirm it first" if o.status in ("draft", "to_approve") else ""))
if (percent is None) == (net_amount is None):
raise Refused(422, "give a percent or a net_amount, not both and not neither")
net = money_round(o.net_total * Decimal(str(percent)) / 100, o.currency) if percent is not None else net_amount
existing = [d for d in (repo.down_payments.all_heads() if hasattr(repo.down_payments, "all_heads") else []) if d.purchase_order_id == purchase_order_id and d.status != "cancelled"]
left = o.net_total - sum((d.net_amount for d in existing), ZERO)
if net <= 0 or net > left:
raise RuleBroken(f"the requests would exceed the order: {left} of {o.net_total} remains to ask for")
today = _today(repo)
did = f"dp:{purchase_order_id}:{len(existing) + 1}"
tax, code = _dp_tax(o, did, net, tax_code, tax_amount, today)
d = rows.DownPayment(did, base.stamp_number(repo, "down_payment", "down_payment_number_format", today, did), purchase_order_id, o.bp_id, net,
code, tax, net + tax, o.currency, base._day(due_date, "due_date"), "requested", _who(caller), _sub(caller))
return dp_view(d, repo.down_payments.save(d, 0))
return unit(None, run)
@PURCHASING.use_case(kind="command", route="POST /down-payments/{down_payment_id}/approve", permission=DP_APPROVE, idempotent=True)
def approve_down_payment(caller: dict, down_payment_id: Ref, version: Optional[Version] = None) -> dict:
"""Finance approves the request: not the buyer who asked (same email OR same sign-in), recorded with who and when."""
def run():
with repository() as repo, _decide():
d, v = _load(repo, caller, down_payment_id)
if d.status == "approved":
return dp_view(d, v)
_same_version("the down payment", v, version)
if not _ids(caller):
raise RuleBroken("second_person: the approver cannot be identified (no email or sign-in); fix: sign in")
if d.status != "requested":
raise RuleBroken(f"a down payment that is {d.status} is not waiting for approval")
if same_person(_ids(caller), identities(d.requested_by_email, d.requested_by_sub)):
raise RuleBroken("second_person: you requested this down payment (same email or same sign-in); fix: another approver")
d.status, d.approved_on = "approved", _today(repo)
d.approved_by_email, d.approved_by_sub = base._email(caller), _sub(caller)
return dp_view(d, repo.down_payments.save(d, v))
return unit(version, run)
@PURCHASING.use_case(kind="query", route="GET /down-payments/{down_payment_id}", permission=DP_REQUEST)
def get_down_payment(caller: dict, down_payment_id: Ref) -> dict:
"""The down payment with what it holds, what is cleared and what remains."""
with repository() as repo:
return dp_view(*_load(repo, caller, down_payment_id))
class ClearIn(_In):
bill_id: Ref
amount: Money
reason: Optional[Reason] = None
@PURCHASING.use_case(kind="command", route="POST /down-payments/{down_payment_id}/clear", permission=DP_CLEAR)
def clear_down_payment(caller: dict, down_payment_id: Ref, bill_id: Ref, amount: Money, reason: Optional[Reason] = None, version: Optional[Version] = None) -> dict:
"""Clear (part of) a PAID down payment against a final bill: the bill's outstanding falls by it and only the balance is paid. More than remains
on the down payment, or more than the bill's outstanding, is refused. A zero clearing needs a reason ('a later bill takes it') and releases the
bill into a run. Two clerks clearing at once clear it once."""
def run():
with repository() as repo, _decide():
d, v = _load(repo, caller, down_payment_id)
_same_version("the down payment", v, version)
if d.status != "paid":
raise RuleBroken(f"a down payment that is {d.status} cannot be cleared: it must be paid first")
got = repo.bills.get(bill_id)
if got is None:
raise Refused(404, f"no bill {bill_id}")
b = got[0]
if b.purchase_order_id != d.purchase_order_id or b.status not in ("approved", "posted"):
raise RuleBroken("the bill must be an approved bill of the same order")
if amount == 0 and not (reason or "").strip():
raise Refused(422, "a zero clearing needs a reason, such as 'a later bill takes it'")
if amount > remaining(d):
raise RuleBroken(f"only {remaining(d)} remains on the down payment")
if amount > bills.outstanding(repo, b):
raise RuleBroken(f"the bill's outstanding is {bills.outstanding(repo, b)}")
today = _today(repo)
seq = 1 + max([c.seq for c in d.clearings] or [0])
d.clearings.append(rows.DownPaymentClearing(d.down_payment_id, bill_id, seq, amount, d.tax_code, base._email(caller), today, (reason or "").strip()))
nv = repo.down_payments.save(d, v)
if amount > 0:
aseq = repo.claims.take(f"alloc:{bill_id}", d.down_payment_id) # the claim payments and credits take: a concurrent allocation collides
repo.allocations.put(Allocation("down_payment", d.down_payment_id, bill_id, aseq, amount, today, _who(caller)))
return dp_view(d, nv), (clearing_journal(repo, d, bill_id, seq, amount, today) if amount > 0 else None)
out, body = unit(version, run)
out["posted"], out["post_note"] = _post(body)
return out
def _vat_share(repo, d: rows.DownPayment, amount: Decimal) -> Decimal:
"""The VAT already reclaimed on the down payment's VAT invoice that this clearing hands back: its share of the gross, to the cent."""
inv = repo.bills.get(d.invoice_bill_id) if d.invoice_bill_id else None
if not inv or inv[0].status != "approved" or d.gross_amount == 0:
return ZERO
reclaimed = sum((ln.vat_reclaimable for ln in inv[0].lines), ZERO)
return money_round(reclaimed * amount / d.gross_amount, d.currency)
def clearing_journal(repo, d: rows.DownPayment, bill_id: str, seq: int, amount: Decimal, on: date) -> dict:
"""Clearing moves the prepayment onto the bill: payables debited, supplier down payments credited for the net share and input VAT credited for
the VAT reclaimed on the down payment's invoice (so the bill's own VAT counts once). Keyed on (down payment, bill, seq)."""
vat = _vat_share(repo, d, amount)
lines = [{"debit": str(amount), "account_id": bills.account_for(repo, "payables", "*", on), "memo": f"clear {d.dp_no}", "bp_id": str(d.bp_id)},
{"credit": str(amount - vat), "account_id": bills.account_for(repo, "supplier_down_payment", "*", on), "memo": f"clear {d.dp_no}", "bp_id": str(d.bp_id)}]
if vat:
lines.append({"credit": str(vat), "account_id": bills.account_for(repo, "input_vat", d.tax_code or "*", on), "memo": f"VAT reversed {d.tax_code}", "bp_id": str(d.bp_id)})
# gl's purchasing seam takes event approve | pay | reverse, journal_id = derived_id(source_document, event): a clearing settles the payable
# ('pay'); each clearing is its own source_document so its id is unique and derivable.
doc = f"{d.down_payment_id}:{bill_id}:{seq}"
return {"journal_id": str(derived_id(doc, "pay")), "source_module": "purchasing", "source_document": doc,
"event": "pay", "journal_date": on.isoformat(), "currency": d.currency, "memo": f"down payment {d.dp_no} cleared against {bill_id}", "lines": lines}
def _post(body: Optional[dict]) -> tuple:
if body is None or not ports.installed("gl"):
return False, ("" if body is None else "not posted: the ledger is not installed")
try:
ports.gl.post_journal(body)
return True, ""
except Missing as m:
return False, f"the ledger refused ({m.status}): " + ("; ".join(f"{g.get('gate')}: {g.get('reason')}" for g in m.gates) or m.reason)
@PURCHASING.use_case(kind="command", route="POST /down-payments/{down_payment_id}/invoice", permission=DP_CLEAR, idempotent=True)
def record_down_payment_invoice(caller: dict, down_payment_id: Ref, supplier_invoice_no: Ref, invoice_date: IsoDate) -> dict:
"""Record the supplier's VAT invoice for a PAID down payment: a bill of kind down_payment_invoice (keyed like any bill), its tax computed by the
tax module. At its approval the VAT moves from supplier down payments to input VAT; it is never paid again. Without it nothing is reclaimed."""
def run():
with repository() as repo, _decide():
d, v = _load(repo, caller, down_payment_id)
if d.status != "paid":
raise RuleBroken(f"a down payment that is {d.status} has no VAT invoice to record: it must be paid first")
if d.invoice_bill_id:
return dp_view(d, v)
o = repo.orders.get(d.purchase_order_id)[0]
idate = base._day(invoice_date, "invoice_date")
line = BillLineIn(quantity=1, unit_code="C62", unit_price=d.net_amount, net_amount=d.net_amount, description=f"down payment {d.dp_no}",
tax_category=d.tax_code or None)
built = bills.build_bill_lines(repo, caller, None, [line], bl.bill_id_of(d.bp_id, supplier_invoice_no, idate), d.bp_id, d.currency)
b, bv, created = bills.record_core(repo, caller, bp_id=d.bp_id, supplier_invoice_no=supplier_invoice_no, invoice_date=idate, currency=d.currency,
lines=built, purchase_order_id=d.purchase_order_id, kind="down_payment_invoice", supplier_tax_total=d.tax_amount,
down_payment_id=d.down_payment_id, company_code=o.company_code)
if created:
repo.bills.save(b, 0)
d.invoice_bill_id = b.bill_id
return dp_view(d, repo.down_payments.save(d, v))
return unit(None, run)
@PURCHASING.use_case(kind="command", route="POST /down-payments/{down_payment_id}/refund", permission=DP_CLEAR, idempotent=True)
def record_down_payment_refund(caller: dict, down_payment_id: Ref, refund_reference: Ref, amount: Optional[Money] = None, version: Optional[Version] = None) -> dict:
"""Record the supplier's refund of what a paid down payment still holds ('clear it, or record the supplier's refund'): bank debited, supplier
down payments credited (and the VAT reclaimed on it handed back). The refund is the whole remainder when no amount is given; more than remains
is refused. A second recording of the same reference changes nothing."""
def run():
with repository() as repo, _decide():
d, v = _load(repo, caller, down_payment_id)
if d.refund_reference == refund_reference:
return dp_view(d, v), None
_same_version("the down payment", v, version)
if d.status != "paid":
raise RuleBroken(f"a down payment that is {d.status} has nothing to refund")
left = remaining(d)
amt = left if amount is None else amount
if amt <= 0 or amt > left:
raise RuleBroken(f"only {left} remains on the down payment")
today = _today(repo)
seq = 1 + max([c.seq for c in d.clearings] or [0])
d.clearings.append(rows.DownPaymentClearing(d.down_payment_id, "refund", seq, amt, d.tax_code, base._email(caller), today, f"refund {refund_reference}"))
d.refund_reference = refund_reference
vat = _vat_share(repo, d, amt)
lines = [{"debit": str(amt), "account_id": bills.account_for(repo, "bank", "*", today), "memo": f"refund {refund_reference}", "bp_id": str(d.bp_id)},
{"credit": str(amt - vat), "account_id": bills.account_for(repo, "supplier_down_payment", "*", today), "memo": f"refund {d.dp_no}", "bp_id": str(d.bp_id)}]
if vat:
lines.append({"credit": str(vat), "account_id": bills.account_for(repo, "input_vat", d.tax_code or "*", today), "memo": "VAT handed back", "bp_id": str(d.bp_id)})
doc = f"{d.down_payment_id}:refund:{refund_reference}" # gl accepts approve | pay | reverse: a refund reverses the prepayment
body = {"journal_id": str(derived_id(doc, "reverse")), "source_module": "purchasing", "source_document": doc,
"event": "reverse", "journal_date": today.isoformat(), "currency": d.currency, "memo": f"down payment {d.dp_no} refunded", "lines": lines}
return dp_view(d, repo.down_payments.save(d, v)), body
out, body = unit(version, run)
out["posted"], out["post_note"] = _post(body)
return out
@PURCHASING.use_case(kind="query", route="GET /down-payments", permission=DP_REQUEST)
def list_down_payments(caller: dict, purchase_order_id: Optional[Ref] = None, left_on_closed_orders: bool = False) -> dict:
"""Down payments, optionally of one order; with left_on_closed_orders, only the paid ones still holding money on an order that closed, each with
the fix 'clear it, or record the supplier's refund'."""
with repository() as repo:
out = []
for h in repo.down_payments.all_heads():
if purchase_order_id and h.purchase_order_id != purchase_order_id:
continue
d, v = repo.down_payments.get(h.down_payment_id)
o = repo.orders.get(d.purchase_order_id)
closed = bool(o and o[0].status in ("closed", "cancelled"))
if left_on_closed_orders and not (closed and d.status == "paid" and remaining(d) > 0):
continue
x = dp_view(d, v)
x["order_closed"] = closed
out.append(x)
return {"items": out}
# ── the hooks payments calls ─────────────────────────────────────────────────────────────────────
def due_down_payments(repo, run, when: date, today: date):
"""Approved down payments due by the pay date go in the run like a bill: one payment each, to the supplier's approved bank details."""
items, left, pays = [], [], []
open_ = {b for b, _ in repo.runs.open_items()}
for d in repo.down_payments.all_heads():
if d.status != "approved" or d.due_date > when:
continue
if d.down_payment_id in open_:
left.append({"bill_id": d.down_payment_id, "reason": "already in an open payment run"})
continue
if payments.bank_in_force(repo, d.bp_id, when) is None and payments.profile_on(repo, d.bp_id, when).get("payment_method") != "outside":
left.append({"bill_id": d.down_payment_id, "reason": "no approved bank details in force on the pay date"})
continue
pid = f"{run.run_id}:dp:{d.down_payment_id}"
items.append(RunItem(run.run_id, pid, d.down_payment_id, d.gross_amount, "down_payment", ZERO, ZERO, "|".join(sorted(identities(d.requested_by_email, d.requested_by_sub)))))
pays.append(SupplierPayment(pid, run.run_id, d.bp_id, d.currency, d.gross_amount, d.gross_amount, ZERO, ZERO, "", "", "", run.bank_account_id, "", f"{d.dp_no}",
None, "planned", "down_payment", (payments.bank_in_force(repo, d.bp_id, when) or {}).get("valid_from")))
return items, left, pays
def on_paid(repo, caller, item, payment, paid_on) -> None:
"""A confirmed payment of a down payment: it is paid in advance, with when and by which payment."""
got = repo.down_payments.get(item.bill_id)
if got is None:
return
d, v = got
if d.status != "paid":
d.status, d.paid_on, d.payment_id = "paid", paid_on, payment.payment_id
repo.down_payments.save(d, v)
def held_for_clearing(repo, b) -> Optional[str]:
"""A bill whose order has a PAID down payment still holding money stays out of a run until a clearing (or a zero clearing with a reason) exists."""
if b.kind != "bill" or not b.purchase_order_id:
return None
memo = repo.__dict__.setdefault("_held_memo", {}) # one request, one repo: the heads (and each paid one's full row) are read once
if "heads" not in memo:
memo["heads"] = repo.down_payments.all_heads()
for d in memo["heads"]:
if d.purchase_order_id == b.purchase_order_id and d.status == "paid":
if d.down_payment_id not in memo:
memo[d.down_payment_id] = repo.down_payments.get(d.down_payment_id)[0]
full = memo[d.down_payment_id]
if remaining(full) > 0 and not any(c.bill_id == b.bill_id for c in full.clearings):
return f"an uncleared down payment {full.dp_no} on its order: clear it against this bill first (or a zero clearing with the reason 'a later bill takes it')"
return None
payments.HOOK_DOWN_PAYMENTS = due_down_payments
payments.HOOK_BILL_HELD = held_for_clearing