dodil/erp-brick-payments
DODIL ERP brick: payments
src/brick_payments/application/accounts.py
"""Bank accounts (PAY-27): the accounts we hold per company, their purposes, the default receiving account per company and
currency (a dated row from tomorrow), a close date after today, and the ONE route that serves an account UNMASKED - to
purchasing's machine identity only, every read audited in the same transaction."""
import re
from datetime import timedelta, timezone
from typing import Optional
from erp_platform.bricks import Refused
from erp_platform.types import Name, text
from sqlalchemy.exc import IntegrityError
from brick_payments.application import use_cases as uc
from brick_payments.application.inputs import Code, Currency, Day, Ref
from brick_payments.domain.errors import RuleBroken
from brick_payments.infrastructure import seams
PAY = uc.PAY
WRITE = "pay:bank_account:write"
PURPOSES = ("receive", "pay", "dd_collect")
IBAN = re.compile(r"^[A-Z]{2}\d{2}[A-Z0-9]{8,30}$")
def purposes_of(bank: dict) -> list:
return [p for p in (bank.get("purposes") or "receive").replace(" ", "").split(",") if p]
def _purposes(v: Optional[str]) -> Optional[str]:
if v is None or not v.strip():
return None
got = [p.strip().lower() for p in v.replace("|", ",").split(",") if p.strip()]
bad = [p for p in got if p not in PURPOSES]
if bad or not got:
raise Refused(422, f"purposes are any of {', '.join(PURPOSES)} (comma separated); {', '.join(bad) or v!r} is not one")
return ",".join(dict.fromkeys(got))
def _mask(v: str) -> str:
return ("*" * max(0, len(v) - 4) + v[-4:]) if v else ""
@PAY.use_case(kind="command", route="POST /bank-accounts/{bank_account_id}", permission=WRITE, idempotent=True)
def put_bank_account(caller: dict, bank_account_id: Code, name: Name, currency: Currency, gl_account_id: Optional[Code] = None,
account_number: Optional[text(40)] = None, company_code: Optional[text(4)] = None,
house_bank_id: Optional[Code] = None, purposes: Optional[text(60)] = None, iban: Optional[text(40)] = None,
account_name: Optional[Name] = None) -> dict:
"""Set up (or change) an account: the business's own code, name, currency, the ledger account it posts to, its company (which
must exist in the ledger), the house bank, its purposes (receive | pay | dd_collect) and the account number or IBAN - shown
masked to its last four characters everywhere except to purchasing's machine identity. An account never moves to another company."""
with uc.repository() as repo, uc._decide():
digits = "".join(ch for ch in (account_number or "") if ch.isalnum()).upper()
ib = "".join(ch for ch in (iban or "").upper() if ch.isalnum())
if ib and not IBAN.match(ib):
raise Refused(422, "an IBAN is two letters, two digits and up to thirty letters or digits")
if not ib and IBAN.match(digits):
ib = digits
shown = ib or digits
masked = _mask(shown)
company = (company_code or "").strip().upper()
if company and not (company.isalnum() and company.isascii()):
raise Refused(422, "company_code is one to four letters or digits, like UK01")
if house_bank_id and repo.bank.house_bank(house_bank_id) is None:
raise Refused(422, f"there is no house bank {house_bank_id} - set it up first")
old = repo.config.bank(bank_account_id) or {}
if company and old.get("company_code") and old["company_code"] != company:
raise RuleBroken(f"Account {bank_account_id} belongs to company {old['company_code']} and keeps its history there; "
f"it cannot move to {company}. Set up a new account")
if company and company != old.get("company_code"):
known = seams.company(company, uc._who(caller))
if known is None:
raise Refused(422, f"the ledger has no company {company}, so an account cannot be set up for it")
cur, kept = uc._cur(currency), _purposes(purposes)
row = {"bank_account_id": bank_account_id, "name": name, "currency": cur, "iban_or_sort_code_masked": masked or old.get("iban_or_sort_code_masked") or "",
"gl_account_id": gl_account_id or old.get("gl_account_id") or "", "origin": old.get("origin") or "",
"company_code": company or old.get("company_code") or "", "house_bank_id": house_bank_id or old.get("house_bank_id") or "",
"purposes": kept or old.get("purposes") or "receive", "account_name": account_name or old.get("account_name") or name,
"status": old.get("status") or "open", "closed_from": old.get("closed_from"),
"account_ref": digits if digits else (old.get("account_ref") or ""), "iban": ib or old.get("iban") or ""}
repo.config.put_bank(row)
return {k: row[k] for k in ("bank_account_id", "name", "currency", "iban_or_sort_code_masked", "gl_account_id", "company_code",
"house_bank_id", "purposes", "account_name", "status")}
@PAY.use_case(kind="command", route="POST /bank-accounts/{bank_account_id}/close", permission=WRITE, idempotent=True)
def close_bank_account(caller: dict, bank_account_id: Code, closed_from: Day) -> dict:
"""Close an account from a date AFTER today: it keeps its history, takes no new statement from that day and stops being offered as
a default. Closing again with the same date answers the same."""
with uc.repository() as repo, uc._decide():
on, today = uc._date(closed_from, "closed_from"), uc._today(repo)
b = repo.config.bank(bank_account_id)
if b is None:
raise Refused(404, f"no bank account {bank_account_id}")
if on is None or on <= today:
raise RuleBroken(f"An account is closed from a date after today ({today}); {closed_from} is not. Pick tomorrow or later")
if b.get("closed_from") and b["closed_from"] != on and b["closed_from"] <= today:
raise RuleBroken(f"Account {bank_account_id} is already closed from {b['closed_from']}")
if b.get("closed_from") == on:
return {"bank_account_id": bank_account_id, "closed_from": on.isoformat(), "status": b["status"]}
repo.claims.take("bank_account", bank_account_id, uc._now(), uc._who(caller), "close")
repo.config.put_bank({**b, "closed_from": on, "status": "closed"})
return {"bank_account_id": bank_account_id, "closed_from": on.isoformat(), "status": "closed"}
@PAY.use_case(kind="command", route="POST /default-bank-accounts", permission=WRITE)
def set_default_bank_account(caller: dict, company_code: text(4, min_length=1), currency: Currency, bank_account_id: Code,
valid_from: Optional[Day] = None) -> dict:
"""Make an account the one invoices print for a company and currency, as a dated row from tomorrow (never today or back:
an invoice already printed shows the account it showed). The account must belong to that company, be in that currency, take
receipts and not be closed by then. One default per (company, currency, date)."""
with uc.repository() as repo, uc._decide():
today = uc._today(repo)
start = uc._date(valid_from, "valid_from") or today + timedelta(days=1)
if start <= today:
raise RuleBroken(f"A default account starts from tomorrow ({today + timedelta(days=1)}) or later; {start} is not after today")
cur, company = uc._cur(currency), company_code.strip().upper()
b = repo.config.bank(bank_account_id)
if b is None:
raise Refused(404, f"no bank account {bank_account_id}")
if (b.get("company_code") or "") != company:
raise RuleBroken(f"Account {bank_account_id} belongs to company {b.get('company_code') or '(none)'}, not {company}")
if b["currency"] != cur:
raise RuleBroken(f"Account {bank_account_id} is in {b['currency']}, not {cur}")
if "receive" not in purposes_of(b):
raise RuleBroken(f"Account {bank_account_id} does not take receipts (purposes: {','.join(purposes_of(b))})")
if b.get("closed_from") and b["closed_from"] <= start:
raise RuleBroken(f"Account {bank_account_id} is closed from {b['closed_from']}, so it cannot be the default from {start}")
repo.claims.take("default_account", f"{company}:{cur}:{start}", uc._now(), uc._who(caller), bank_account_id)
repo.accounts.add_default({"company_code": company, "currency": cur, "valid_from": start, "bank_account_id": bank_account_id,
"set_by": uc._who(caller)}, uc._now())
return {"company_code": company, "currency": cur, "valid_from": start.isoformat(), "bank_account_id": bank_account_id}
@PAY.use_case(kind="query", route="GET /default-bank-accounts", permission=uc.READ)
def list_default_bank_accounts(caller: dict, company_code: Optional[text(4)] = None, currency: Optional[Currency] = None,
on: Optional[Day] = None) -> dict:
"""The default receiving account per company and currency on a day (today when none is given), with the dated rows still to come."""
with uc.repository() as repo:
day = uc._date(on, "on") or uc._today(repo)
rows = repo.accounts.defaults((company_code or "").upper() or None, uc._cur(currency) if currency else None)
current: dict = {}
for r in rows:
if r["valid_from"] <= day:
current[(r["company_code"], r["currency"])] = r
view = lambda r: {"company_code": r["company_code"], "currency": r["currency"], "bank_account_id": r["bank_account_id"], # noqa: E731
"valid_from": r["valid_from"].isoformat()}
return {"on": day.isoformat(), "defaults": [view(r) for r in current.values()],
"upcoming": [view(r) for r in rows if r["valid_from"] > day]}
@PAY.use_case(kind="query", route="GET /bank-accounts/{bank_account_id}/account-details", permission=uc.READ_UNMASKED, tool=False)
def get_account_details(caller: dict, bank_account_id: Code, purpose: Optional[Ref] = None) -> dict:
"""PAY-27, purchasing's seam: the UNMASKED details of one of OUR accounts with purpose pay, for its bank file (Bacs Standard 18
carries the originator's sort code and account; pain.001 the IBAN). Served only to a machine identity holding
pay:bank_account:read_unmasked - no person role holds it and no wildcard grant reaches it. Every answered read writes a
pay_bank_account_read row in the same transaction; a read that cannot be logged is refused (503) and nothing is returned."""
with uc.repository() as repo, uc._decide():
b = repo.config.bank(bank_account_id)
if b is None or "pay" not in purposes_of(b):
raise Refused(404, f"no bank account {bank_account_id} with purpose pay")
who, now = (caller.get("sub") or caller.get("email") or "").strip().lower(), uc._now()
try:
n = repo.claims.take("bank_account_read", bank_account_id, now, who)
repo.accounts.log_read({"read_id": f"{bank_account_id}:{who}:{now.isoformat()}:{n}", "bank_account_id": bank_account_id, "caller": who,
"requested_at": now.replace(tzinfo=None) if now.tzinfo is None else now.astimezone(timezone.utc).replace(tzinfo=None),
"purpose": (purpose or "bank file").strip()[:140]}, now)
except IntegrityError: # two reads claimed the same number a moment apart: a collision, not an outage
raise Refused(409, "Another read of this account was logged at the same moment. Nothing is returned; ask again") from None
except Exception: # noqa: BLE001 - any failure to log is a refusal to answer
raise Refused(503, "The read could not be logged, so the account details are not given. Try again") from None
hb = repo.bank.house_bank(b["house_bank_id"]) if b.get("house_bank_id") else None
ib = b.get("iban") or ""
return {"bank_account_id": bank_account_id, "company_code": b.get("company_code") or "", "account_name": b.get("account_name") or b["name"],
"sort_code": (hb or {}).get("sort_code") or "", "account_number": "" if ib else (b.get("account_ref") or ""),
"iban": ib, "bic": (hb or {}).get("bic") or "", "currency": b["currency"]}
@PAY.use_case(kind="query", route="GET /bank-accounts/{bank_account_id}/reads", permission=WRITE)
def list_account_reads(caller: dict, bank_account_id: Code) -> dict:
"""Who read this account's unmasked details, when and why: the audit trail (the number itself is never shown)."""
with uc.repository() as repo:
return {"reads": [{"read_id": r["read_id"], "caller": r["caller"], "requested_at": r["requested_at"].isoformat(), "purpose": r["purpose"]}
for r in repo.accounts.reads(bank_account_id)]}