dodil/erp-brick-ar
DODIL ERP brick: accounts receivable
src/brick_ar/application/use_cases_v4b.py
"""v4 slice B: the manual recurring invoice (AR-42), the Peppol e-invoice and its statuses (AR-43, AR-44), the doubtful-debt provision
(AR-45), deferred revenue (AR-46) and the cost centre on revenue (AR-47, which lives in the journals: services.invoice_journal).
Same conventions as use_cases.py / use_cases_v4.py: one declaration per route AND tool, the permission on the declaration, the rules in
domain/, the neighbours behind infrastructure/seams.py. Every call reads `uc.repository` at call time."""
import json
import uuid
from datetime import datetime, timezone
from decimal import Decimal
from typing import Literal, Optional
from erp_platform import work
from erp_platform.bricks import Refused
from erp_platform.repository import retry_on_conflict
from erp_platform.types import BigId, IsoDate, Name, Note
from sqlalchemy.exc import IntegrityError
from brick_ar.application import services as sv
from brick_ar.application import use_cases as uc
from brick_ar.application.inputs import Code, Currency, LineNo, Pct, Qty, Rate, Reason
from brick_ar.application.services import who
from brick_ar.domain import customer as customer_mod
from brick_ar.domain import deferral, einvoice, provision
from brick_ar.domain.dates import check_dated
from brick_ar.domain.errors import RuleBroken
from brick_ar.domain.money import ZERO
from brick_ar.domain.provision import ProvisionRun
from brick_ar.domain.readiness import Gate, GatesUnmet, met, unmet
from brick_ar.domain.schedule import Schedule
from brick_ar.infrastructure import seams
AR = uc.AR
def _to_409(unit):
"""A plain-INSERT claim lost for good is a 409 with its reason, never a swallowed error."""
try:
return uc._claiming(unit)
except IntegrityError:
raise Refused(409, "another request claimed the same key at the same moment: ask again") from None
# ═══ AR-42: a recurring invoice with no order ═══════════════════════════════════════════════════
@AR.use_case(kind="command", route="POST /schedules/manual", permission=sv.SCHEDULE_WRITE, idempotent=True)
def create_recurring_invoice(caller: dict, bp_id: BigId, currency: Currency, description: Note, unit_price: Rate, starts_on: IsoDate,
periods: LineNo, quantity: Optional[Qty] = None,
billing_plan: Literal["monthly", "quarterly", "annual", "periodic"] = "monthly",
discount_pct: Optional[Pct] = None, unit_code: Optional[Code] = None, product_id: Optional[Code] = None,
tax_category: Optional[Code] = None, in_advance: Optional[bool] = None) -> dict:
"""A customer rents a desk at 450.00 a month from 2026-12-01 for 12 months and there is no order: the biller creates the recurring
invoice by hand. It is a schedule keyed "man:<uuid>|1" (source_kind manual) on the same engine as an order's (AR-18): each period
raised as a draft by the daily pass, reviewed unless the owner's auto_issue_sources names manual, finished after the last period.
It starts today or later (there is no source document to date it earlier). A new price is a NEW schedule from a later period; an
issued invoice is never edited. Stop it with stop_billing (source_id "man:<uuid>")."""
if unit_price < 0:
raise Refused(422, "a recurring invoice's unit_price is not negative: a credit is a credit note")
def unit():
with uc.repository() as repo, uc._decide():
today = uc._today(repo)
start = uc._date(starts_on, "starts_on")
gates = [met("starts_on") if start >= today else unmet(
"starts_on", f"a manual schedule starts today or later; {start} is before today ({today}) and there is no source document to date it",
"start it today, or later")]
gates.append(customer_mod.profile_gate(sv.profile_for(repo, bp_id, today), today))
if not all(g.met for g in gates):
raise GatesUnmet("create the recurring invoice", gates)
uc._check_customer(repo, caller, bp_id)
s = repo.settings.in_force(today)
source_id = f"man:{uuid.uuid4().hex}"
sch = Schedule.plan(f"{source_id}|1", by=who(caller), bp_id=bp_id, source_kind="manual", source_id=source_id, line_id="1",
product_id=product_id or "", description=description, billing_kind="recurring", billing_plan=billing_plan,
starts_on=start, periods=periods, quantity=quantity if quantity is not None else Decimal(1),
unit_price=unit_price, discount_pct=discount_pct or Decimal(0), unit_code=unit_code or "",
tax_category=tax_category or "standard", currency=currency.upper(),
in_advance=s.bill_in_advance if in_advance is None else in_advance, billing_day=s.billing_day,
prorate_first=s.prorate_first_period)
return uc._schedule_view(repo, sch, repo.schedules.save(sch, 0))
return retry_on_conflict(unit)
# ═══ AR-43: the Peppol e-invoice ════════════════════════════════════════════════════════════════
def _refuse_invalid(found: list) -> None:
raise GatesUnmet("send the e-invoice", [Gate(**f.as_gate()) for f in found[:1]])
def _hand_to_access_point(repo, caller: dict, *, doc, kind: str, number: str, bill_to: dict, issue_date, due_date, key: str) -> tuple:
"""Render, validate, hand over: (message_id, provider, xml). A failed rule refuses the send naming the rule; nothing leaves."""
today = uc._today(repo)
s = repo.settings.in_force(today)
provider = (s.einvoice_provider or "").strip()
if not provider:
raise GatesUnmet("send the e-invoice", [unmet("einvoice_provider", "no access-point provider is composed for Peppol",
"Settings: einvoice_provider (the access-point capability)")])
participant = bill_to.get("peppol_participant_id", "")
m = sv.einvoice_model(doc, kind=kind, number=number, issue_date=issue_date, due_date=due_date, settings=s, bill_to=bill_to,
participant=participant)
found = einvoice.validate(m)
if found:
_refuse_invalid(found)
xml = einvoice.render(m)
sent = seams.einvoice_send(provider, key, document_ref=getattr(doc, "invoice_id", None) or doc.credit_id,
document_type="credit_note" if kind == "credit_note" else "invoice", receiver_id=participant, xml=xml,
actor=who(caller))
if not sent.ok:
raise Refused(502, sent.reason + (" (try again: the provider claims the key, so a retry sends once)" if sent.retryable else ""))
return sent.message_id, provider, xml
def send_einvoice(repo, caller: dict, inv, current: int) -> dict:
"""The invoice goes out as UBL 2.1 to the access point (called by send_invoice when the frozen bill-to says peppol). The send row
carries the message id; the PDF-for-our-records is still produced."""
bill = json.loads(inv.bill_to_json or "{}")
n = len(inv.sends) + 1
mid, provider, _xml = _hand_to_access_point(repo, caller, doc=inv, kind="invoice", number=inv.invoice_no, bill_to=bill,
issue_date=inv.issue_date, due_date=inv.due_date, key=f"einv:{inv.invoice_id}:{n}")
seen = repo.einvoices.message(mid)
if seen is None:
repo.einvoices.add_message(mid, inv.invoice_id, "invoice", provider, work.now()) # plain INSERT keyed on the message
elif seen["document_id"] != inv.invoice_id:
raise Refused(409, f"the access point returned message {mid}, which already belongs to {seen['document_id']}")
s = inv.record_send(at=work.now(), address=bill.get("peppol_participant_id", ""), channel="peppol", by=who(caller))
s.format, s.result, s.message_id, s.statuses_json = "ubl", f"handed to the access point {provider}", mid, "[]"
inv.einvoice_message_id = mid
body = uc._full_view(inv, repo.invoices.save(inv, current), repo)
body["send"] = {"channel": "peppol", "format": "ubl", "message_id": mid, "result": s.result, "manual": False,
"record": uc._invoice_text(inv)}
return body
@AR.use_case(kind="command", route="POST /credit-notes/{credit_id}/send", permission=sv.SEND)
def send_credit_einvoice(caller: dict, credit_id: Name) -> dict:
"""An approved credit note to a Peppol customer goes the same way: a UBL CreditNote validated before it leaves, the message id kept."""
def unit():
with uc.repository() as repo, uc._decide():
cn, _v = sv.visible_credit(repo, caller, credit_id)
if cn.status != "approved" or cn.source_kind == "odoo":
raise RuleBroken(f"credit note {credit_id} is {cn.status}: only an approved credit note of this brick is sent")
profile = sv.profile_for(repo, cn.bp_id, uc._today(repo))
if profile is None or profile.einvoice_channel != "peppol":
raise RuleBroken("this customer is not billed over Peppol: a credit note goes by the invoice's own channel")
prior = repo.einvoices.message_of(credit_id)
if prior is not None:
return {"credit_id": credit_id, "message_id": prior["message_id"], "already": True}
bill = json.loads(profile.bill_to_json())
mid, provider, _xml = _hand_to_access_point(repo, caller, doc=cn, kind="credit_note", number=cn.credit_no, bill_to=bill,
issue_date=cn.credit_date, due_date=None, key=f"einv:{credit_id}:1")
repo.einvoices.add_message(mid, credit_id, "credit_note", provider, work.now())
return {"credit_id": credit_id, "message_id": mid, "format": "ubl", "channel": "peppol"}
return _to_409(unit)
# ═══ AR-44: the access point reports ════════════════════════════════════════════════════════════
_RANK = {"delivered": 0, "AB": 1, "IP": 2, "UQ": 3, "CA": 4, "RE": 5, "AP": 5, "PD": 6}
def current_status(statuses: list) -> Optional[str]:
"""The document's standing is the furthest status reported, whatever order they arrived in (AP / RE are final; PD follows AP)."""
return max((x["status"] for x in statuses), key=lambda st: _RANK.get(st, -1), default=None)
def _status_view(repo, message: dict, rejected_reason: Optional[str] = None) -> dict:
rows = repo.einvoices.statuses(message["message_id"])
statuses = [{"status": r["status"], "at": r["status_at"], "reason": r["reason"] or ""} for r in rows]
rejection = rejected_reason or (next((x["reason"] for x in reversed(statuses) if x["status"] == "RE"), None)
if current_status(statuses) == "RE" else None)
return {"invoice_id": message["document_id"], "document_kind": message["document_kind"], "message_id": message["message_id"],
"statuses": statuses, "rejected": rejection}
@AR.use_case(kind="command", route="POST /einvoices/{message_id}/status", permission=sv.EINVOICE, idempotent=True)
def record_einvoice_status(caller: dict, message_id: Name, status: Literal["delivered", "AB", "IP", "UQ", "CA", "RE", "AP", "PD"],
reason: Optional[Note] = None, at: Optional[str] = None) -> dict:
"""The access point (a provider composed beside this brick) reports delivery, then the buyer's Invoice Response: AB acknowledged, IP
in process, UQ under query, CA conditionally accepted, RE rejected, AP accepted, PD paid. The send row records each status with its
time; a rejection shows on the invoice for the biller with its reason. The invoice itself is not changed (a correction is a credit note
and a new invoice). A repeat of (message, status) adds nothing. The provider never writes ar_ tables: it calls this route."""
def unit():
with uc.repository() as repo, uc._decide():
message = repo.einvoices.message(message_id)
if message is None:
raise Refused(404, f"no e-invoice message {message_id}")
if repo.einvoices.status(message_id, status) is not None:
return _status_view(repo, message)
when = _when(at)
if _naive_utc(when) > _naive_utc(work.now()) + timedelta(minutes=5):
raise Refused(422, "at is in the future: a status is reported when it happened")
had = {r["status"] for r in repo.einvoices.statuses(message_id)}
if status in ("AP", "RE") and had & {"AP", "RE"}:
raise Refused(409, f"the buyer's response is already final ({', '.join(sorted(had & {'AP', 'RE'}))}): {status} cannot follow it")
if status == "PD" and "AP" not in had:
raise Refused(409, "PD (paid) follows AP (accepted): the invoice was not accepted")
repo.einvoices.add_status(message_id, status, when, (reason or "").strip()) # plain INSERT keyed (message, status)
if message["document_kind"] == "invoice":
inv, current = repo.invoices.get(message["document_id"])
send = next((s for s in inv.sends if s.message_id == message_id), None)
if send is not None:
seen = json.loads(send.statuses_json or "[]")
seen.append({"status": status, "at": when.isoformat(), "reason": (reason or "").strip()})
send.statuses_json = json.dumps(seen)
repo.invoices.save(inv, current)
return _status_view(repo, message)
return _to_409(unit)
def _naive_utc(d: datetime) -> datetime:
return d.astimezone(timezone.utc).replace(tzinfo=None) if d.tzinfo else d
def _when(at: Optional[str]) -> datetime:
if not at:
return work.now()
try:
d = datetime.fromisoformat(at.replace("Z", "+00:00"))
except ValueError:
raise Refused(422, "at is a date and time such as 2026-11-20T09:30:00Z") from None
return d.astimezone(timezone.utc).replace(tzinfo=None) if d.tzinfo else d
# ═══ AR-46: deferred revenue ════════════════════════════════════════════════════════════════════
def take_back_deferred(repo, cn, today) -> None:
"""A credit note against an invoice with deferred revenue takes back what is not yet released FIRST (the claim rows are plain
INSERTs keyed (invoice, line, "cn:<credit>")); credit_journal then debits deferred revenue for it instead of revenue."""
inv = repo.invoices.get(cn.invoice_id)[0]
if inv.lifecycle != "issued" or inv.source_kind == "odoo" or not sv.defers(repo, inv):
return
totals = repo.releases.totals(inv.invoice_id)
budget, guarded = cn.subtotal, False
for ln in inv.lines:
if budget <= 0:
break
if not deferral.is_deferred(ln, inv.issue_date):
continue
done = sum(totals.get(ln.line_no, [ZERO, ZERO]), ZERO)
open_ = max(ln.net_amount - done, ZERO) # released + taken back never exceeds the line's net
take = min(open_, budget)
if take > 0:
if not guarded:
repo.releases.guard(inv.invoice_id, work.now()) # the same per-invoice slot a release and a void claim
guarded = True
repo.releases.add(inv.invoice_id, ln.line_no, f"cn:{cn.credit_id}", "credit", take, today, work.now())
budget -= take
def _release_body(inv, row: dict, doc: str, line) -> dict:
amount = str(row["amount"])
memo = f"{inv.invoice_no} {row['period_id']} service revenue"
body = {"journal_id": str(seams.journal_id(doc, "approve")), "source_module": "ar", "source_document": doc, "event": "approve",
"journal_date": row["journal_date"].isoformat(), "currency": inv.currency, "memo": memo,
"lines": [{"purpose": "deferred_revenue", "debit": amount, "memo": memo},
{"purpose": "revenue", "credit": amount, "memo": memo, **({"cost_centre": line.cost_centre} if line.cost_centre else {})}]}
if inv.fx_rate:
body["fx_rate"] = str(inv.fx_rate)
if inv.company_code:
body["company_code"] = inv.company_code
return body
@AR.use_case(kind="command", route="POST /revenue/release", permission=sv.RELEASE, idempotent=True)
def release_deferred_revenue(caller: dict, period_id: Name) -> dict:
"""Monthly, when the ledger period closes (or the composition's clock at month end): every service-period line invoiced ahead moves
the days of the month from deferred revenue to revenue - 12,000.00 x 31/365 = 1,019.18 for October on an annual service - keyed
[invoice, line, period] so a second run adds nothing; the last month takes the remainder so the line releases exactly its net.
Months not yet released before this one are caught up. Journals post by purpose through the ledger, dated the month's end."""
first, last = _period(period_id)
with uc.repository() as repo, uc._decide():
today = uc._today(repo)
if last > today:
raise GatesUnmet("release revenue", [unmet("period_ended", f"{period_id} has not ended (today is {today})",
"release it on or after its last day")])
ids = repo.releases.candidates(last)
released, already, posted, pending = [], 0, 0, []
for iid in ids:
def unit(iid=iid):
made = []
with uc.repository() as repo, uc._decide():
inv = repo.invoices.get(iid)[0]
if inv.lifecycle != "issued" or not sv.defers(repo, inv):
return made, 0
skipped, guarded = 0, False
for ln in inv.lines:
if not deferral.is_deferred(ln, inv.issue_date):
continue
for pid in deferral.months(ln, inv.issue_date):
if pid > period_id:
break
if repo.releases.get(iid, ln.line_no, pid) is not None:
skipped += 1
continue
rows = [r for r in repo.releases.rows(iid) if r["line_no"] == ln.line_no]
pf, pl = deferral.bounds(pid)
amount = deferral.release_amount(
net=ln.net_amount, released=sum((r["amount"] for r in rows if r["kind"] == "release"), ZERO),
reversed_=sum((r["amount"] for r in rows if r["kind"] == "credit"), ZERO), service_from=ln.service_from,
service_to=ln.service_to, last_day=pl)
if amount > 0:
if not guarded:
repo.releases.guard(iid, work.now()) # the per-invoice slot a void and a credit take-back also claim
guarded = True
repo.releases.add(iid, ln.line_no, pid, "release", amount, pl, work.now()) # plain INSERT: the claim
made.append((ln.line_no, pid, amount))
return made, skipped
made, skipped = _to_409(unit)
already += skipped
released += [{"invoice_id": iid, "line_no": n, "period_id": pid, "amount": a} for n, pid, a in made]
with uc.repository() as repo: # after the commit: the ledger (also what an earlier run left unposted)
inv = repo.invoices.get(iid)[0]
if sv.ledger_mode(repo, last) != "post":
continue
for row in [r for r in repo.releases.rows(iid) if r["kind"] == "release" and r["period_id"] <= period_id]:
doc = f"rel:{iid}:{row['line_no']}:{row['period_id']}"
line = next(x for x in inv.lines if x.line_no == row["line_no"])
out = uc._post_one(repo, doc, "release", _release_body(inv, row, doc, line), [], row["journal_date"], who(caller))
if out.get("posted") and not out.get("already"):
posted += 1
elif not out.get("posted"):
pending.append({"document_id": doc, "reason": out.get("reason", "")})
return {"period_id": period_id, "released": released, "already_released": already, "posted": posted, "unposted": pending}
def _period(period_id: str) -> tuple:
try:
return deferral.bounds(period_id)
except (ValueError, TypeError):
raise Refused(422, f"{period_id!r} is a period such as 2026-10") from None
# ═══ AR-45: the doubtful-debt provision ═════════════════════════════════════════════════════════
def _run_view(run: ProvisionRun, version: int) -> dict:
return {"run_id": run.run_id, "company_code": run.company_code, "period_id": run.period_id, "lifecycle": run.lifecycle,
"as_at": run.as_at, "lines": run.lines, "total": run.total, "previous_total": run.previous_total, "movement": run.movement,
"proposed_by": run.proposed_by, "approved_by": run.approved_by, "journal_id": run.journal_id,
"rejected_reason": run.rejected_reason, "version": version}
@AR.use_case(kind="query", route="GET /provision-rates", permission=sv.READ)
def list_provision_rates(caller: dict) -> dict:
"""The doubtful-debt percentage per aging bucket (and "specific" for an invoice in dispute), each a dated row."""
with uc.repository() as repo:
return {"rates": repo.provision_rates.rows()}
@AR.use_case(kind="command", route="POST /provision-rates", permission=sv.SETTINGS_WRITE)
def set_provision_rate(caller: dict, bucket: Name, rate_pct: Pct, valid_from: Optional[IsoDate] = None) -> dict:
"""Set a bucket's provision rate as a new dated row, from tomorrow by default (the first row of a bucket may start today). Buckets
are payments' aging buckets (0-30, 31-60, 61-90, 90+); the bucket "specific" is the rate for invoices in dispute."""
with uc.repository() as repo, uc._decide():
today = uc._today(repo)
first = not repo.provision_rates.has_any(bucket)
vf = uc._date(valid_from, "valid_from") or (today if first else today + _ONE_DAY)
check_dated("a provision rate", vf, today, first=first)
try:
repo.provision_rates.add(bucket, vf, rate_pct, who(caller))
except IntegrityError:
raise Refused(409, f"bucket {bucket} already has a rate from {vf}: rows are never edited, date the new one later") from None
return {"bucket": bucket, "valid_from": vf, "rate_pct": rate_pct}
from datetime import timedelta # noqa: E402
_ONE_DAY = timedelta(days=1)
def _provision_in_order(repo, company: str, period_id: str) -> None:
"""The allowance is a running balance: a month earlier than (or equal to) one already approved would post its movement again."""
later = repo.provisions.approved_from(company, period_id)
if later:
raise GatesUnmet("provision in order", [unmet("period_in_order", f"the run of {later} is already approved: {period_id} is not after it",
"provisions are approved in period order; propose the month after the latest approved one")])
@AR.use_case(kind="command", route="POST /provisions", permission=sv.PROVISION_WRITE, idempotent=True)
def propose_doubtful_provision(caller: dict, period_id: Name) -> dict:
"""The month has ended: the biller proposes its provision. payments' aging as at the last day gives each bucket's invoices; the
provision is each bucket's outstanding EXCLUDING VAT (outstanding x net / gross of the frozen invoice) at the bucket's rate, and each
invoice in dispute at the specific rate, on its own line. The run is keyed "prov:<company_code>:<period_id>"; the movement is the
total less the allowance the last approved run left. Proposing again before approval refreshes it (the aging moves)."""
try:
end = provision.period_end(period_id)
except RuleBroken as exc:
raise Refused(422, str(exc)) from None
def unit():
with uc.repository() as repo, uc._decide():
today = uc._today(repo)
if end >= today:
raise GatesUnmet("propose the provision", [unmet("period_ended", f"{period_id} has not ended (today is {today})",
"propose it once the month is over")])
s = repo.settings.in_force(today)
rid = provision.run_id(s.default_company_code or "", period_id)
found = repo.provisions.get(rid)
if found is not None and found[0].lifecycle == "approved":
return _run_view(*found)
_provision_in_order(repo, s.default_company_code or "", period_id)
aging, why = seams.aging(end, s.default_company_code or "")
if aging is None:
raise Refused(502, why)
invoices = {}
loaded = repo.invoices.get_many([i["invoice_id"] for b in aging.get("buckets", []) for i in b.get("invoices", [])])
for b in aging.get("buckets", []):
for i in b.get("invoices", []):
got = loaded.get(i["invoice_id"])
if got is not None and got[0].total:
inv = got[0]
invoices[i["invoice_id"]] = (inv.subtotal, inv.total, inv.fx_rate if inv.currency != s.functional_currency else None)
lines, total = provision.compute(aging, invoices, lambda bucket: repo.provision_rates.in_force(bucket, today))
previous = repo.provisions.last_approved_before(s.default_company_code or "", period_id)
run, version = found if found is not None else (ProvisionRun(rid, company_code=s.default_company_code or "", period_id=period_id,
as_at=end), 0)
run.propose(lines=lines, total=total, previous_total=previous.total if previous else ZERO, by=who(caller), by_sub=sv.sub_of(caller))
return _run_view(run, repo.provisions.save(run, version))
return retry_on_conflict(unit)
@AR.use_case(kind="query", route="GET /provisions", permission=sv.READ)
def list_provisions(caller: dict) -> dict:
"""The provision runs, newest period first."""
with uc.repository() as repo:
return {"runs": [_run_view(*repo.provisions.get(i)) for i in repo.provisions.ids()]}
@AR.use_case(kind="query", route="GET /provisions/{run_id}", permission=sv.READ)
def get_provision(caller: dict, run_id: Name) -> dict:
"""One provision run: each bucket and each disputed invoice with its outstanding net of VAT, rate and provision."""
with uc.repository() as repo:
found = repo.provisions.get(run_id)
if found is None:
raise Refused(404, f"no provision run {run_id}")
return _run_view(*found)
def _post_provision(repo, run: ProvisionRun, actor: str) -> dict:
if not run.movement:
return {"posted": False, "reason": "no movement since last month: nothing to post"}
s = repo.settings.in_force(run.as_at)
if sv.ledger_mode(repo, run.as_at) != "post":
return {"posted": False, "reason": f"the ledger mode on {run.as_at} is not post"}
amount = str(abs(run.movement))
up = run.movement > 0
memo = f"Doubtful-debt provision {run.period_id}"
body = {"journal_id": str(seams.journal_id(run.run_id, "approve")), "source_module": "ar", "source_document": run.run_id,
"event": "approve", "journal_date": run.as_at.isoformat(), "currency": s.functional_currency, "memo": memo,
"lines": [{"purpose": "doubtful_debt_expense", "memo": memo, "debit" if up else "credit": amount},
{"purpose": "doubtful_debt_allowance", "memo": memo, "credit" if up else "debit": amount}]}
if run.company_code:
body["company_code"] = run.company_code
out = uc._post_one(repo, run.run_id, "provision", body, [], run.as_at, actor)
return {"posted": bool(out.get("posted")), **({"reason": out["reason"]} if out.get("reason") else {})}
@AR.use_case(kind="command", route="POST /provisions/{run_id}/approve", permission=sv.PROVISION_APPROVE, idempotent=True)
def approve_doubtful_provision(caller: dict, run_id: Name) -> dict:
"""The approver approves a proposed run (another person than the proposer, matched by email AND sub, where second_approver is on;
the ledger period open). The MOVEMENT since the last approved run is posted dated the period end - debit the doubtful-debt expense,
credit the allowance, a release reverses the sides - by purpose through the ledger, journal id derived from the run, so it posts once.
A ledger that refuses leaves the run approved with `posted` false and a repeat posts it."""
def unit():
with uc.repository() as repo, uc._decide():
found = repo.provisions.get(run_id)
if found is None:
raise Refused(404, f"no provision run {run_id}")
run, current = found
if run.lifecycle == "approved":
return _run_view(run, current)
s = repo.settings.in_force(uc._today(repo))
if seams.period_open(run.as_at) is False:
raise GatesUnmet("approve the provision", [unmet("period_open", f"the ledger period of {run.as_at} is closed",
"ask the accountant to reopen it")])
_provision_in_order(repo, run.company_code, run.period_id)
previous = repo.provisions.last_approved_before(run.company_code, run.period_id)
run.approve(by=who(caller), identities=sv.ids(caller), second_approver=bool(s.second_approver), at=work.now(),
previous_total=previous.total if previous else ZERO)
if run.movement:
run.journal_id = seams.journal_id(run.run_id, "approve")
return _run_view(run, repo.provisions.save(run, current))
view = retry_on_conflict(unit)
with uc.repository() as repo:
run = repo.provisions.get(run_id)[0]
view["posting"] = _post_provision(repo, run, who(caller))
return view
@AR.use_case(kind="command", route="POST /provisions/{run_id}/reject", permission=sv.PROVISION_APPROVE, idempotent=True)
def reject_doubtful_provision(caller: dict, run_id: Name, reason: Reason) -> dict:
"""The approver rejects a proposed run with a reason; the biller proposes it again."""
with uc.repository() as repo, uc._decide():
found = repo.provisions.get(run_id)
if found is None:
raise Refused(404, f"no provision run {run_id}")
run, current = found
if run.lifecycle == "rejected":
return _run_view(run, current)
run.reject(reason=reason)
return _run_view(run, repo.provisions.save(run, current))