dodil/erp-brick-ar
DODIL ERP brick: accounts receivable
src/brick_ar/application/use_cases.py
"""The use cases: what a person (or their agent), a source brick and the system's clock can do with invoices.
Each is declared ONCE and served as an API route and as an MCP tool. Two things are decided here and nowhere else: WHO MAY (a
PERMISSION on the declaration, never a role's name) and WHICH ROWS (`services.visible_*`, from the caller: a row the caller may not
see is a 404, never a 403). Each command is one unit of work: load, let the domain decide, save whole at the version it was loaded
(`If-Match` for a screen). The domain holds every rule; this file asks the neighbours (infrastructure/seams.py) and decides who.
Row visibility (module.yaml): finance sees every invoice; sales sees status, balance and hold for a customer, never lines or margin.
The sales-facing `customer_status` is a DIFFERENT route with a narrow shape, not a filter on the full one.
Settlement (open, part paid, paid, written off) is never stored here: every answer that shows an outstanding amount reads
payments' settlement. invoicing never writes a pay_ table; a credit note or write-off reduces a balance through payments' allocate.
"""
import hashlib
import json
import re
import time
import uuid
from contextlib import contextmanager
from datetime import date, timedelta
from decimal import Decimal
from typing import Annotated, Literal, Optional
from erp_platform import work
from erp_platform.bricks import Brick, Refused
from erp_platform.capabilities import CapabilitySeam, ManualTask
from erp_platform.repository import retry_on_conflict
from erp_platform.types import BigId, IsoDate, Limit, Name, Note, text
from pydantic import Field
from sqlalchemy.exc import IntegrityError
from brick_ar.application import services as sv
from brick_ar.application.inputs import (
Code,
CreditLineIn,
Currency,
DeductionIn,
DepositIn,
DoorsIn,
FxRate,
LineIn,
LineNo,
Money,
Pct,
Qty,
Rate,
Reason,
day,
)
from brick_ar.application.services import (
CHECK,
CLAIM,
COLLECT,
CONTROLS,
CREDIT_APPROVE,
CREDIT_WRITE,
CUSTOMER_WRITE,
FINISH,
ISSUE,
LOAD,
MIGRATION_READ,
MIGRATION_SIGN,
PLAN,
POST,
RAISE,
READ,
RECONCILE,
SCHEDULE_READ,
SEND,
SETTINGS_WRITE,
STATUS,
STOP,
VOID,
WRITE,
WRITEOFF_APPROVE,
WRITEOFF_WRITE,
who,
)
from brick_ar.domain import numbering, reconcile
from brick_ar.domain import odoo as odoo_map
from brick_ar.domain.credit_note import CreditNote
from brick_ar.domain.customer import Profile, profile_gate
from brick_ar.domain.dates import check_dated, first_of_next_month
from brick_ar.domain.errors import RuleBroken
from brick_ar.domain.invoice import Invoice, Line, deposit_deduction_lines
from brick_ar.domain.migration_run import MigrationRun
from brick_ar.domain.money import ZERO, money
from brick_ar.domain.readiness import Gate, GatesUnmet, met, unmet
from brick_ar.domain.schedule import Schedule
from brick_ar.domain.values import (
FIRST_OF_MONTH_KEYS,
ID_PREFIX,
SETTINGS,
InvoiceKind,
check_setting,
)
from brick_ar.domain.writeoff import Writeoff
from brick_ar.infrastructure import ai_steps, clock, seams
from brick_ar.infrastructure.tables import Repository
Days = Annotated[int, Field(ge=0, le=3650)]
AR = Brick("ar", version="0.1.2", ids_as_strings=True, machine_permissions=sv.MACHINE, requires=("mail.send",), instructions=(
"Invoicing: send invoices for what you sold (from an order, a repeating schedule, delivered work, or by hand), correct them "
"with credit notes, and keep the books in step. A draft is raised from a source with raise_draft (keyed on the source, so "
"asking twice returns the same invoice). issue_invoice fixes the number, the dates, the contents, the tax and the bill-to "
"details; an issued invoice is never edited, a correction is a credit note. invoice_readiness answers every gate of the "
"next step as {gate, met, reason, fix}. What is outstanding always comes from payments, never from this brick."))
# The process view (BRICKS.md): the workflows as BPMN, GET /process/bpmn.
AR.declare_process("process.yaml")
mail_send = CapabilitySeam("mail.send")
@contextmanager
def _repository_default():
from erp_platform import db # imported late: it needs BUCKET
with _on_engine(db.engine) as repo:
yield repo
def _lost_a_unique_race(exc: IntegrityError) -> bool:
text_ = f"{getattr(exc, 'orig', '')} {exc}"
return "23505" in text_ or "duplicate key" in text_ or "UNIQUE constraint" in text_
@contextmanager
def _on_engine(engine):
"""One transaction. On DataK3 a write conflict on a claim row can surface at COMMIT (the loser's COMMIT raises 23505), after the use case
has returned - so it is caught here, around the commit, and answered as the same retry-safe 409 a conflict seen mid-use-case gets, never a 500."""
try:
with engine.begin() as conn:
yield Repository(conn)
except IntegrityError as exc:
if not _lost_a_unique_race(exc):
raise
raise Refused(409, "Someone else wrote the same record a moment ago - load it again and retry") from None
repository = _repository_default # tests and a composition swap in their own
def _odoo_default():
import os
from brick_ar.infrastructure.odoo import XmlRpcOdoo
missing = [k for k in ("ODOO_URL", "ODOO_DB", "ODOO_USER", "ODOO_KEY") if not os.environ.get(k)]
if missing:
raise Refused(503, "Odoo is not configured for this brick: set " + ", ".join(missing) + " (a read-only API key)")
return XmlRpcOdoo()
odoo_reader = _odoo_default # tests swap in fake Odoo rows
@contextmanager
def _decide():
"""A broken business rule is a 409 with the rule's own words, through both doors."""
try:
yield
except RuleBroken as broken:
raise Refused(409, str(broken)) from None
def _today(repo) -> date:
return clock.today(repo)
def _date(v: Optional[str], what: str) -> Optional[date]:
try:
return day(v)
except ValueError:
raise Refused(422, f"{what} is a date like 2026-11-01") from None
def _retrying(version: Optional[int], unit):
"""Run one unit of work. A caller who sent NO version asked for "this change on whatever is there": a race lost to another
writer is re-run on the latest. One who sent a version is told it changed (409) and reloads."""
return unit() if version is not None else retry_on_conflict(unit)
def _claiming(unit, attempts: int = 8):
"""A document number is claimed with a plain INSERT; the loser of a race for one number gets 23505 and tries the next."""
for attempt in range(1, attempts + 1):
try:
return unit()
except Exception as exc: # noqa: BLE001 — re-raised unless it is a lost number claim
lost = any(t in str(exc) for t in ("ar_number_claims", "ar_invoice_no_claims", "ar_credit_claim", "ar_deposit_claim", "ar_deferral_release", "ar_einvoice_status", "ar_provision_run")) and isinstance(getattr(exc, "orig", exc), Exception) and (
isinstance(exc, IntegrityError) or "23505" in str(exc) or "uplicate" in str(exc))
if not lost or attempt == attempts:
raise
time.sleep(0.02 * attempt)
raise AssertionError("unreachable")
def _same_version(what: str, current: int, asked: Optional[int]) -> None:
if asked is not None and int(asked) != current:
raise Refused(409, f"{what} changed since you opened it (you have version {asked}, it is at {current}) — load it again")
def _invoice(repo, caller, invoice_id) -> tuple:
return sv.visible_invoice(repo, caller, invoice_id)
def _settled(inv_ids: list) -> tuple:
sr = seams.settlement(inv_ids)
return sr.by_invoice, sr.reason
def _discount_statement(inv: Invoice) -> str:
return (f"A settlement discount of {inv.settlement_discount_pct.normalize():f}% applies if you pay by {inv.settlement_discount_until}: "
f"{inv.settlement_discount_amount} {inv.currency}. No credit note will be issued for it: you may only recover the VAT actually paid.")
def _full_view(inv: Invoice, version: int, repo=None) -> dict:
out = sv.invoice_view(inv, version)
if inv.settlement_discount_pct and inv.lifecycle == "issued" and repo is not None:
if repo.settings.in_force(_today(repo)).settlement_discount_vat == "invoice_wording": # printed as the owner's setting reads today
out["settlement_discount_statement"] = _discount_statement(inv)
if inv.lifecycle == "issued":
by, why = _settled([inv.invoice_id])
out["settlement"] = sv.settlement_view(by.get(inv.invoice_id), why)
return out
# ═══ queries: invoices ══════════════════════════════════════════════════════════════════════════
@AR.use_case(kind="query", route="GET /invoices", permission=READ)
def list_invoices(caller: dict, lifecycle: Optional[Literal["draft", "issued", "void", "cancelled"]] = None,
bp_id: Optional[BigId] = None, currency: Optional[Currency] = None,
source_kind: Optional[Literal["sales_order", "schedule", "acceptance", "manual", "contract", "odoo"]] = None,
issued_from: Optional[IsoDate] = None, issued_to: Optional[IsoDate] = None, limit: Limit = 50,
offset: int = 0) -> dict:
"""The invoices the caller may see, newest first, a page at a time: filter by state, customer, currency, source and issue date.
Each row shows the number, the customer, the total and the outstanding amount (from payments, one read per page)."""
with repository() as repo:
if lifecycle in ("draft", "cancelled") and not sv.sees_drafts(caller):
return {"invoices": [], "total": 0, "limit": limit, "offset": offset}
rows, total = repo.invoices.page(lifecycle=lifecycle, bp_id=bp_id, currency=currency.upper() if currency else None,
source_kind=source_kind, issued_from=_date(issued_from, "issued_from"),
issued_to=_date(issued_to, "issued_to"), limit=limit, offset=max(0, offset))
if not sv.sees_drafts(caller):
rows = [r for r in rows if r["lifecycle"] not in ("draft", "cancelled")]
issued = [r["invoice_id"] for r in rows if r["lifecycle"] == "issued"]
by, why = _settled(issued)
return {"invoices": [sv.invoice_row(r, by.get(r["invoice_id"]), why) for r in rows], "total": total,
"limit": limit, "offset": offset}
@AR.use_case(kind="query", route="GET /invoices/{invoice_id}", permission=READ, etag="version")
def get_invoice(caller: dict, invoice_id: Name) -> dict:
"""One invoice, whole, with its version, its frozen tax and (once issued) what is outstanding, read from payments."""
with repository() as repo:
return _full_view(*_invoice(repo, caller, invoice_id), repo)
@AR.use_case(kind="query", route="GET /invoices/{invoice_id}/readiness", permission=READ)
def invoice_readiness(caller: dict, invoice_id: Name) -> dict:
"""Every gate of the invoice's next steps as {gate, met, reason, fix} (fix names the screen or person who clears it), so
no screen re-derives a rule: issue for a draft; void for an issued invoice."""
with repository() as repo:
inv, _v = _invoice(repo, caller, invoice_id)
today = _today(repo)
if inv.is_draft:
gates, _q, _p = sv.issue_gates(repo, inv, today)
return {"invoice_id": inv.invoice_id, "transition": "issue", "ready": all(g.met for g in gates),
"gates": [g.as_dict() for g in gates]}
if inv.lifecycle == "issued":
void = sv.void_gates(repo, inv, today)
others = {}
mode = sv.ledger_mode(repo, inv.issue_date or today)
if mode == "post" and inv.source_kind != "odoo":
posted = repo.postings.get(inv.invoice_id, "issue") is not None
body, needs = sv.invoice_journal(repo, inv, "issue", inv.issue_date or today)
others["post_to_ledger"] = [met("account_map") if body is not None or posted else
unmet("account_map", "no account in force today for: " + ", ".join(needs),
"Settings: account map")]
if mode == "export" and inv.source_kind != "odoo":
others["export"] = [unmet("not_exported", f"already in batch {inv.export_batch_id}", "none needed: it is exported once")
if inv.export_batch_id else met("not_exported")]
transitions = {"void_invoice": void, **others}
return {"invoice_id": inv.invoice_id, "transition": "void_invoice", "ready": all(g.met for g in void),
"gates": [g.as_dict() for g in void],
"transitions": {k: {"ready": all(g.met for g in v), "gates": [g.as_dict() for g in v]} for k, v in transitions.items()}}
return {"invoice_id": inv.invoice_id, "transition": None, "ready": False, "gates": [],
"note": f"the invoice is {inv.lifecycle}: it has no next step"}
@AR.use_case(kind="query", route="GET /sources/{source_kind}/{source_id}/billed", permission=READ)
def billed_per_source(caller: dict, source_kind: Literal["sales_order"], source_id: Name) -> dict:
"""Billed vs sold for an order: per order line, the quantity and net billed on ISSUED invoices less approved credit notes.
sales-order reads this back; it never writes ar_ tables. `source_id` is the sales_order_id."""
with repository() as repo:
per: dict = {}
for _so, line_no, qty, net in repo.invoices.billed(source_id):
q = per.setdefault(line_no, [Decimal(0), ZERO])
q[0] += qty
q[1] += net
for _so, line_no, qty, net in repo.credits.billed(source_id):
q = per.setdefault(line_no, [Decimal(0), ZERO])
q[0] -= qty
q[1] -= net
return {"source_kind": source_kind, "source_id": source_id, "lines": [
{"so_line_no": n, "quantity_billed": q, "net_billed": money(m)} for n, (q, m) in sorted(per.items())]}
@AR.use_case(kind="query", route="GET /billed-lines", permission=READ)
def billed_lines(caller: dict, bp_id: BigId, issued_from: Optional[IsoDate] = None, issued_to: Optional[IsoDate] = None,
include_drafts: bool = False) -> list:
"""The lines of one customer's issued invoices and approved credit notes dated in the window (the rebate basis sales-order reads).
With include_drafts the draft invoice lines are included too (status draft, issue_date null, whatever the window), with the net
they have now, so sales-order's credit headroom counts what is about to be billed. A credit-note line keeps its own positive net:
`kind` says it is a credit."""
start, end = _date(issued_from, "issued_from"), _date(issued_to, "issued_to")
products: dict = {}
def category(pid: str) -> str:
if pid and pid not in products:
products[pid] = (seams.get_product(pid) or {}).get("category_code") or ""
return products.get(pid, "")
with repository() as repo:
out = [{"document_id": d, "kind": "invoice", "status": "issued" if st == "issued" else "draft", "line_no": n,
"issue_date": on if st == "issued" else None, "product_id": pid or "", "category_code": category(pid),
"net_amount": money(net), "tax_category": tc or "", "sales_order_id": so or "", "so_line_no": sl}
for d, st, n, on, pid, net, tc, so, sl in repo.invoices.billed_lines(bp_id, start, end, include_drafts)]
out += [{"document_id": d, "kind": "credit_note", "status": "issued", "line_no": n, "issue_date": on, "product_id": pid or "",
"category_code": category(pid), "net_amount": money(net), "tax_category": tc or "", "sales_order_id": so or "",
"so_line_no": sl} for d, n, on, pid, net, tc, so, sl in repo.credits.billed_lines(bp_id, start, end)]
return out
@AR.use_case(kind="query", route="GET /invoice-totals", permission=READ)
def invoice_totals(caller: dict, invoice_ids: Optional[text(8000, min_length=1)] = None, bp_id: Optional[BigId] = None,
limit: Limit = 1000, offset: int = 0) -> dict:
"""The total, currency, due date and customer of up to 200 invoices (comma-separated ids) - or, without ids, of the ISSUED
invoices (of one customer with bp_id), a page at a time (limit, offset; `more` says another page follows): what payments
reads to derive settlement, open items, the overdue list and its migration check. Void and cancelled invoices are named as
such, so they drop out of every balance."""
with repository() as repo:
out, more = {}, False
if invoice_ids:
ids = [i for i in invoice_ids.split(",") if i][:200]
else:
rows, total = repo.invoices.page(lifecycle="issued", bp_id=bp_id, limit=limit, offset=max(0, offset))
ids, more = [r["invoice_id"] for r in rows], max(0, offset) + len(rows) < total
repo.invoices.prefetch(ids) # every invoice in one read per table, not one load each
for iid in ids:
try:
inv, _v = _invoice(repo, caller, iid)
except Refused:
continue
out[iid] = {"bp_id": inv.bp_id, "lifecycle": inv.lifecycle, "total": inv.total, "currency": inv.currency,
"issue_date": inv.issue_date, "due_date": inv.due_date, "invoice_no": inv.invoice_no,
"fx_rate": inv.fx_rate, "settlement_discount_pct": inv.settlement_discount_pct,
"settlement_discount_until": inv.settlement_discount_until}
return {"invoices": out, "more": more}
@AR.use_case(kind="query", route="GET /overdue", permission=COLLECT)
def list_overdue(caller: dict, bp_id: Optional[BigId] = None, limit: Limit = 100) -> dict:
"""Issued invoices past their due date with something outstanding (read from payments): the collector's list to chase by hand.
The chase and the receipt are recorded in payments; this brick keeps no dunning state."""
with repository() as repo:
today = _today(repo)
rows, _t = repo.invoices.page(lifecycle="issued", bp_id=bp_id, limit=500)
late = [r for r in rows if r["due_date"] and r["due_date"] < today]
by, why = _settled([r["invoice_id"] for r in late])
out = []
for r in late:
s = by.get(r["invoice_id"])
if s is not None and s.outstanding is not None and s.outstanding > 0:
out.append({**sv.invoice_row(r, s, why), "days_overdue": (today - r["due_date"]).days})
out.sort(key=lambda x: -x["days_overdue"])
return {"as_of": today, "invoices": out[:limit], "settlement_known": not why, "reason": why}
@AR.use_case(kind="query", route="GET /customers/{bp_id}/status", permission=STATUS)
def customer_status(caller: dict, bp_id: BigId) -> dict:
"""What a sales rep needs before quoting a customer again: the overdue amount, the oldest age and whether they are on credit
hold. Read-only and narrow: never lines, margin or other customers. (A different route from the full invoice, not a filter.)"""
with repository() as repo:
today = _today(repo)
issued = [r for r in repo.invoices.issued_for(bp_id) if r["due_date"] and r["due_date"] < today]
by, why = _settled([r["invoice_id"] for r in issued])
overdue: dict = {}
oldest = 0
for r in issued:
s = by.get(r["invoice_id"])
if s is not None and s.outstanding and s.outstanding > 0:
overdue[r["currency"]] = overdue.get(r["currency"], ZERO) + s.outstanding # amounts are never summed across currencies
oldest = max(oldest, (today - r["due_date"]).days)
hold = seams.credit_status(bp_id)
single = next(iter(overdue.values())) if len(overdue) == 1 else (ZERO if not overdue else None)
return {"bp_id": bp_id, "overdue_amount": single if not why else None, "overdue_by_currency": overdue if not why else None,
"oldest_days_overdue": oldest if not why else None, "on_credit_hold": hold.hold if hold.known else None,
"hold_reason": hold.reason, "known": not why and hold.known, "note": why or ("" if hold.known else hold.reason)}
# ═══ drafts: from a source, from a schedule, by hand ════════════════════════════════════════════
def _check_customer(repo, caller: dict, bp_id: int) -> None:
c = seams.check_partner(bp_id, caller)
if c.status == "unknown":
raise Refused(422, c.note)
if repo.invoices.migration_in_progress(bp_id):
raise Refused(409, f"customer {bp_id} has Odoo history loaded and not yet signed off: invoicing takes writes for them once "
"the owner signs the reconciliation (Migration)")
def _auto_issue(repo, caller: dict, inv: Invoice, version: int, today: date, source: Optional[str] = None) -> tuple:
"""A draft raised by the system issues itself only where the owner's setting names its source (default: none), the caller may
claim a number and every gate is met; otherwise it waits for the biller's review."""
s = repo.settings.in_force(today)
if (source or inv.source_kind) not in s.auto_issue or not sv.has(caller, CLAIM):
return inv, version
gates, quote, profile = sv.issue_gates(repo, inv, today)
if not all(g.met for g in gates) or quote is None:
return inv, version
return _issue_now(repo, caller, inv, version, today, quote, profile)
def _issue_now(repo, caller: dict, inv: Invoice, version: int, today: date, quote, profile) -> tuple:
s = repo.settings.in_force(today)
due = sv.terms_due(repo, profile.terms_id, today)
for ln in inv.lines: # AR-47: a revenue line with no cost centre takes the customer's default
if not ln.cost_centre and not ln.deducts_invoice_id:
ln.cost_centre = profile.default_cost_centre
number, sid = sv.claim_number(repo, s, "invoice", inv.invoice_id, today, work.now(), company=inv.company_code)
terms = repo.terms.get(profile.terms_id) or {}
inv.issue(today=today, number=number, series_id=sid, due_date=due, bill_to_json=(sv.address_profile(repo, inv, profile, today) or profile).bill_to_json(),
terms_id=profile.terms_id, quote=quote, by=who(caller), discount_pct=terms.get("early_discount_pct") or None,
discount_days=terms.get("early_discount_days") if terms.get("early_discount_pct") else None)
for dep_id, net in sorted(sv.deducted_net(inv).items()): # a claim per deposit deduction: plain INSERT, keyed (deposit, n)
repo.deposits.claim(dep_id, inv.invoice_id, net, work.now())
return inv, repo.invoices.save(inv, version)
def _create_draft(repo, caller: dict, *, invoice_id: str, bp_id: int, currency: str, source_kind: str, source_id: str, lines: list,
kind: str = "regular", customer_po_ref: str = "", tax_point_date: Optional[date] = None,
deposits: Optional[list] = None, auto_issue: bool = True, fx_rate=None, company_code: str = "",
auto_source: Optional[str] = None, partners: Optional[dict] = None) -> dict:
"""The one way a draft is made (raise_draft, the daily pass, raise_for_acceptance, draft_invoice). Keyed on the source, so a
second call returns the first invoice and creates nothing."""
found = repo.invoices.get(invoice_id)
if found is not None:
return _full_view(*found, repo)
_check_customer(repo, caller, bp_id)
today = _today(repo)
sv.check_order_lines(lines)
if deposits:
if kind == InvoiceKind.DEPOSIT.value:
raise RuleBroken("a deposit invoice does not deduct deposits: deduct on a part invoice (kind regular) or a final one")
for item in deposits: # an id (all that remains) or (id, net amount) (a part)
dep_id, amount = (item, None) if isinstance(item, str) else (item[0], item[1])
dep = repo.invoices.get(dep_id)
if dep is None:
raise RuleBroken(f"deposit invoice {dep_id} does not exist")
if dep[0].bp_id != bp_id or dep[0].currency != currency.upper():
raise RuleBroken(f"deposit invoice {dep_id} is for another customer or currency")
if amount is None:
amount = sv.deposit_remaining(repo, dep[0])
lines = lines + deposit_deduction_lines(dep[0], len(lines) + 1, amount)
profile = sv.profile_for(repo, bp_id, today)
partner = (profile.intercompany_company_code if profile else "") or ""
company_code = company_code or repo.settings.in_force(today).default_company_code or ""
if partner and partner == company_code:
raise RuleBroken(f"{partner} is the issuing company itself: an intercompany invoice goes to ANOTHER company")
inv = Invoice.raise_draft(invoice_id, by=who(caller), bp_id=bp_id, currency=currency, source_kind=source_kind,
source_id=source_id, lines=lines, kind=kind, customer_po_ref=customer_po_ref,
tax_point_date=tax_point_date, fx_rate=fx_rate, company_code=company_code, trading_partner=partner, **(partners or {}))
sv.draft_tax(inv, today)
version = repo.invoices.save(inv, 0)
if auto_issue:
inv, version = _auto_issue(repo, caller, inv, version, today, auto_source)
return _full_view(inv, version, repo)
@AR.use_case(kind="command", route="POST /invoices/raise", permission=RAISE, idempotent=True)
def raise_draft(caller: dict, source_kind: Literal["sales_order", "schedule", "acceptance", "contract", "expense", "pos"], source_id: Name,
bp_id: BigId, currency: Currency, lines: list[LineIn], kind: Literal["regular", "deposit", "final"] = "regular",
deposit_invoice_ids: Optional[list[Name]] = None, customer_po_ref: Optional[Name] = None,
tax_point_date: Optional[IsoDate] = None, fx_rate: Optional[FxRate] = None,
deposit_deductions: Optional[list[DepositIn]] = None, company_code: Optional[Name] = None,
deductions: Optional[list[DeductionIn]] = None, sold_to_bp_id: Optional[BigId] = None,
ship_to_bp_id: Optional[BigId] = None, bill_to_bp_id: Optional[BigId] = None) -> dict:
"""A source asks for a draft invoice (a sales order's request "so:<sales_order_id>:<request_no>", a deposit, a contract
adapter's "<adapter>:<source>"). One draft per source: asking again returns the same invoice and creates nothing. Each line
keeps the order line it bills (sales_order_id + so_line_no) and its unit_code; tax is computed by the tax module at the
invoice's tax point; the draft waits for the biller's review unless the owner's auto_issue_sources names this source.
A final invoice lists the issued deposit invoices it deducts (deposit_invoice_ids: all that remains of each) or, with
deposit_deductions, the NET it takes from each (a part): every deduction line is pinned to the deposit's frozen tax code and rate,
and issue refuses a deduction above what remains of the deposit (gate deposit_remaining). v6: `deductions`
([{deposit_invoice_id, net_amount, tax_category}], sales-order's shape) are accepted on a part invoice (kind regular) as well as a
final one, never on a deposit; sold_to_bp_id, ship_to_bp_id and bill_to_bp_id are stored, the invoice is addressed from bill_to's
billing profile and bp_id stays the payer.
v7: `deductions` is the one name for what a final or part invoice takes from deposits (`deposit_deductions` and `deposit_invoice_ids` are
deprecated and accepted until the next major); a line priced per `pricing_quantity` units of `pricing_unit_code` carries `quantity_priced`
and its net is quantity_priced x unit_price / pricing_quantity x (1 - discount), rounded once; a given `net_amount` is kept as sent.
Sources expense ("exp:...", expenses' re-bill) and pos ("pos:...", an on-account till sale) raise like any other; a draft waits for the
biller unless auto_issue_sources names the source."""
deposits = (list(deposit_invoice_ids or []) + [(d.invoice_id, d.amount) for d in (deposit_deductions or [])]
+ [(d.deposit_invoice_id, d.net_amount) for d in (deductions or [])])
prefix = ID_PREFIX.get(source_kind)
if prefix and not source_id.startswith(prefix):
raise Refused(422, f"a {source_kind} invoice is keyed '{prefix}…', not {source_id!r}")
if not prefix and ":" not in source_id:
raise Refused(422, f"a {source_kind} invoice is keyed '<adapter>:<source>', not {source_id!r}")
def unit():
with repository() as repo, _decide():
return _create_draft(repo, caller, invoice_id=source_id, bp_id=bp_id, currency=currency, source_kind=source_kind,
source_id=source_id, lines=sv.build_lines(lines), kind=kind,
customer_po_ref=customer_po_ref or "", tax_point_date=_date(tax_point_date, "tax_point_date"),
deposits=deposits or None, fx_rate=fx_rate, company_code=company_code or "",
partners={"sold_to_bp_id": sold_to_bp_id, "ship_to_bp_id": ship_to_bp_id, "bill_to_bp_id": bill_to_bp_id})
return _freeze_tax(retry_on_conflict(unit))
@AR.use_case(kind="command", route="POST /invoices/raise-for-acceptance", permission=RAISE, idempotent=True)
def raise_for_acceptance(caller: dict, item_id: Name, bp_id: BigId, currency: Currency, lines: list[LineIn],
customer_po_ref: Optional[Name] = None, accepted_on: Optional[IsoDate] = None) -> dict:
"""An accepted one-off delivery item is billed on acceptance: one draft keyed "acc:<item_id>"; calling again returns it.
`accepted_on` (the supply date the delivery system records) becomes the tax point."""
def unit():
with repository() as repo, _decide():
return _create_draft(repo, caller, invoice_id=f"acc:{item_id}", bp_id=bp_id, currency=currency,
source_kind="acceptance", source_id=f"acc:{item_id}", lines=sv.build_lines(lines),
customer_po_ref=customer_po_ref or "", tax_point_date=_date(accepted_on, "accepted_on"))
return retry_on_conflict(unit)
@AR.use_case(kind="command", route="POST /invoices", permission=WRITE)
def draft_invoice(caller: dict, bp_id: BigId, currency: Currency, lines: list[LineIn], customer_po_ref: Optional[Name] = None,
tax_point_date: Optional[IsoDate] = None, fx_rate: Optional[FxRate] = None,
company_code: Optional[Name] = None) -> dict:
"""Raise an invoice by hand, for work with no order. Line totals, the per-rate tax summary and the total are computed by
the system, never typed; the invoice is a draft until the biller issues it. A usage line (usage true) leaves its quantity
empty and issue is refused until it is filled."""
with repository() as repo, _decide():
iid = f"man:{uuid.uuid4().hex}"
return _create_draft(repo, caller, invoice_id=iid, bp_id=bp_id, currency=currency, source_kind="manual", source_id=iid,
lines=sv.build_lines(lines), customer_po_ref=customer_po_ref or "",
tax_point_date=_date(tax_point_date, "tax_point_date"), auto_issue=False, fx_rate=fx_rate,
company_code=company_code or "")
@AR.use_case(kind="command", route="POST /invoices/{invoice_id}/edit", permission=WRITE, if_match="version")
def edit_draft(caller: dict, invoice_id: Name, lines: Optional[list[LineIn]] = None, customer_po_ref: Optional[Name] = None,
tax_point_date: Optional[IsoDate] = None, version: Optional[int] = None) -> dict:
"""Change a DRAFT: replace its lines, its customer PO reference or its tax point. Totals and tax are recomputed. An issued
invoice refuses every edit ("credit it instead")."""
with repository() as repo, _decide():
inv, current = _invoice(repo, caller, invoice_id)
_same_version(f"invoice {invoice_id}", current, version)
new = sv.build_lines(lines) if lines is not None else None
if new is not None:
sv.check_order_lines(new)
inv.edit(by=who(caller), lines=new, customer_po_ref=customer_po_ref,
tax_point_date=_date(tax_point_date, "tax_point_date"))
sv.draft_tax(inv, _today(repo))
return _full_view(inv, repo.invoices.save(inv, current), repo)
@AR.use_case(kind="command", route="POST /invoices/{invoice_id}/fill-usage", permission=WRITE, if_match="version")
def fill_usage(caller: dict, invoice_id: Name, line_no: LineNo, quantity: Qty, version: Optional[int] = None) -> dict:
"""Enter the period's usage on a metered draft's usage line. Metered billing is never estimated: until this is done issue is
refused. A zero month is cancelled (cancel_draft_invoice) and leaves no number."""
with repository() as repo, _decide():
inv, current = _invoice(repo, caller, invoice_id)
_same_version(f"invoice {invoice_id}", current, version)
inv.fill_usage(line_no, quantity, by=who(caller))
sv.draft_tax(inv, _today(repo))
return _full_view(inv, repo.invoices.save(inv, current), repo)
@AR.use_case(kind="command", route="POST /invoices/{invoice_id}/review", permission=WRITE, if_match="version")
def review_draft(caller: dict, invoice_id: Name, version: Optional[int] = None) -> dict:
"""The biller records that they reviewed a draft (who, when). Drafts raised by the system wait for this; none is issued
automatically unless the owner's setting names its source."""
with repository() as repo, _decide():
inv, current = _invoice(repo, caller, invoice_id)
_same_version(f"invoice {invoice_id}", current, version)
inv.review(by=who(caller), at=work.now())
return _full_view(inv, repo.invoices.save(inv, current), repo)
@AR.use_case(kind="command", route="POST /invoices/{invoice_id}/cancel", permission=WRITE, idempotent=True, if_match="version")
def cancel_draft_invoice(caller: dict, invoice_id: Name, version: Optional[int] = None) -> dict:
"""Cancel a draft that should never have existed (a zero-usage period, a duplicate). It leaves no number behind. A repeat
changes nothing."""
with repository() as repo, _decide():
inv, current = _invoice(repo, caller, invoice_id)
_same_version(f"invoice {invoice_id}", current, version)
if inv.lifecycle == "cancelled":
return _full_view(inv, current, repo)
inv.cancel(by=who(caller))
return _full_view(inv, repo.invoices.save(inv, current), repo)
# ═══ AI pre-issue check: suggestions a person accepts or dismisses ═════════════════════════════════
@AR.use_case(kind="command", route="POST /invoices/{invoice_id}/ai-check", permission=CHECK)
def ai_check_draft(caller: dict, invoice_id: Name) -> dict:
"""The pre-issue check (when on in the settings): lists unusual amounts, a missing PO number or a duplicate of a recent invoice
as FINDINGS on the draft, each with its evidence. A finding is a suggestion: it never edits the draft and issue never waits
on one. A person accepts or dismisses each (decide_finding)."""
def unit():
with repository() as repo, _decide():
inv, current = _invoice(repo, caller, invoice_id)
today = _today(repo)
if repo.settings.in_force(today).mode("ai_check_draft") == "off":
return {"invoice_id": invoice_id, "mode": "off", "findings": [], "version": current}
recent = repo.invoices.recent_totals(inv.bp_id, invoice_id)
digest = hashlib.sha256(json.dumps([str(inv.total), inv.customer_po_ref, [str(t) for t, _d, _p in recent]]).encode()).hexdigest()[:16]
have = {(f.finding, f.evidence) for f in inv.findings if f.decision == "open"}
for finding, evidence in ai_steps.baseline_findings(inv.total, inv.customer_po_ref, recent, today):
if (finding, evidence) not in have:
inv.add_finding("ai_check_draft", ai_steps.BASELINE_MODEL, digest, finding, evidence)
return {**_full_view(inv, repo.invoices.save(inv, current), repo), "mode": "suggest"}
return retry_on_conflict(unit)
@AR.use_case(kind="command", route="POST /invoices/{invoice_id}/findings/{step}/{run_no}/decide", permission=WRITE)
def decide_finding(caller: dict, invoice_id: Name, step: Literal["ai_check_draft"], run_no: LineNo,
decision: Literal["accepted", "dismissed"], reason: Optional[Reason] = None) -> dict:
"""Accept or dismiss one AI finding on a draft, with a reason when dismissing. The decision is recorded (step, model, evidence,
who, when). Nothing in the draft changes: if you accept, you edit the line yourself."""
def unit():
with repository() as repo, _decide():
inv, current = _invoice(repo, caller, invoice_id)
inv.decide_finding(step, run_no, decision, reason or "", by=who(caller), at=work.now())
return _full_view(inv, repo.invoices.save(inv, current), repo)
return retry_on_conflict(unit)
# ═══ issue, send, void ══════════════════════════════════════════════════════════════════════════
@AR.use_case(kind="command", route="POST /invoices/{invoice_id}/issue", permission=ISSUE, idempotent=True, if_match="version")
def issue_invoice(caller: dict, invoice_id: Name, version: Optional[int] = None) -> dict:
"""Issue a draft: it is stamped with the next number from the series, an issue date of TODAY (in the business time zone,
never typed), a tax point (the supply date its source recorded, else the issue date) and a due date from the customer's
terms; the tax is frozen; the bill-to details in force today are copied onto it. From then on every edit is refused
("credit it instead"). When a gate fails nothing is issued and each gate is named {gate, met, reason, fix}. A repeat of an
issued invoice changes nothing."""
def unit():
with repository() as repo, _decide():
inv, current = _invoice(repo, caller, invoice_id)
_same_version(f"invoice {invoice_id}", current, version)
if inv.lifecycle == "issued":
return _full_view(inv, current, repo)
today = _today(repo)
gates, quote, profile = sv.issue_gates(repo, inv, today)
if not all(g.met for g in gates):
raise GatesUnmet("issue the invoice", gates)
inv, v = _issue_now(repo, caller, inv, current, today, quote, profile)
body = _full_view(inv, v, repo)
for ln in inv.lines:
seams.mark_unit_used(ln.product_id, ln.unit_code, inv.invoice_id)
return body
return _freeze_tax(_claiming(unit))
def _freeze_tax(body: dict) -> dict:
"""Once an issue has COMMITTED, the tax module freezes the document's tax under the invoice's id (POST
/documents/{id}/freeze, with the number, the dates, our details and the bill-to copied at issue): the tax the customer
received, forever. After the commit, so a refused freeze can never leave tax frozen on a draft; a repeat of the issue (or
of the source's raise) asks again, and freezing twice is a no-op there. The answer says whether it is frozen."""
if body.get("lifecycle") != "issued" or body.get("source_kind") == "odoo":
return body
if body.get("trading_partner"): # AR-41: an intercompany invoice goes to the partner company's purchasing
from brick_ar.application import use_cases_v4
body = use_cases_v4.hand_over_after_issue(body)
if not seams.TAX.installed:
return body
with repository() as repo:
inv, _v = repo.invoices.get(body["invoice_id"])
s = repo.settings.in_force(inv.issue_date)
ok, why = seams.tax_freeze(document_ref=inv.invoice_id, bill_to_json=inv.bill_to_json, issue_date=inv.issue_date,
supplier=sv.supplier(s), invoice_number=inv.invoice_no)
return {**body, "tax_frozen": ok, **({} if ok else {"tax_frozen_reason": why})}
def _invoice_text(inv: Invoice) -> str:
lines = [f"Invoice {inv.invoice_no}", f"Issued {inv.issue_date}, due {inv.due_date}", ""]
lines += [f"{ln.line_no}. {ln.description} {ln.quantity} x {ln.unit_price} = {ln.net_amount}" for ln in inv.lines]
lines += ["", f"Subtotal {inv.subtotal} {inv.currency}", f"Tax {inv.tax_total}", f"Total {inv.total} {inv.currency}"]
return "\n".join(lines)
@AR.use_case(kind="command", route="POST /invoices/{invoice_id}/send", permission=SEND)
def send_invoice(caller: dict, invoice_id: Name, address: Optional[Name] = None) -> dict:
"""A person presses Send: the document is produced from the FROZEN invoice and handed to the mail capability (a person sends
it where no mail provider is connected); the send is recorded with the date and address. Sending again records a second
send, never a second invoice. A migrated invoice is history and is not sent."""
def unit():
with repository() as repo, _decide():
inv, current = _invoice(repo, caller, invoice_id)
if json.loads(inv.bill_to_json or "{}").get("einvoice_channel") == "peppol" and not address:
from brick_ar.application import use_cases_v4b # AR-43: the customer is billed over Peppol (frozen at issue)
if inv.lifecycle != "issued" or inv.source_kind == "odoo":
inv.record_send(at=work.now(), address="", channel="peppol", by=who(caller)) # the domain's own refusal
return use_cases_v4b.send_einvoice(repo, caller, inv, current)
to = address or json.loads(inv.bill_to_json or "{}").get("invoice_email") or ""
if not to:
raise RuleBroken("no address to send to: the invoice carries no invoice email, name one")
n = len(inv.sends) + 1
res = mail_send.call(f"send:{invoice_id}:{n}", {
"from_email": caller.get("email") or "invoicing@localhost", "to": [to], "cc": [],
"subject": f"Invoice {inv.invoice_no}", "body_text": _invoice_text(inv), "body_html": None}, actor=who(caller))
outcome = "no mail provider connected: a person sends it" if isinstance(res, ManualTask) else \
("handed to the mail capability" if res.status == "done" else f"mail capability: {res.status}")
s = inv.record_send(at=work.now(), address=to, channel="email", by=who(caller))
s.result = outcome
body = _full_view(inv, repo.invoices.save(inv, current), repo)
body["send"] = {"address": to, "result": outcome, "manual": isinstance(res, ManualTask)}
return body
return retry_on_conflict(unit)
@AR.use_case(kind="command", route="POST /invoices/{invoice_id}/void", permission=VOID, idempotent=True, if_match="version")
def void_invoice(caller: dict, invoice_id: Name, reason: Reason, version: Optional[int] = None) -> dict:
"""The approver voids an issued invoice, with a reason: it keeps its number, drops out of every balance and (in post mode) its
posting is reversed on today's date. Allowed only as the owner's void_policy says (default never_sent: never sent, no
allocations, tax point not in a filed VAT return); otherwise a full credit note is the only path and the gates say so."""
with repository() as repo, _decide():
inv, current = _invoice(repo, caller, invoice_id)
_same_version(f"invoice {invoice_id}", current, version)
if inv.lifecycle == "void":
return _full_view(inv, current, repo)
if inv.lifecycle != "issued":
raise RuleBroken(f"only an issued invoice can be voided; this one is {inv.lifecycle}")
gates = sv.void_gates(repo, inv, _today(repo))
if not all(g.met for g in gates):
raise GatesUnmet("void the invoice", gates)
inv.void(today=_today(repo), reason=reason, by=who(caller))
repo.releases.guard(inv.invoice_id, work.now()) # beside a release / take-back of this invoice: the second to commit collides
if inv.kind == "deposit": # a slot on the deposit's claim key: a final issuing beside this void collides on it (23505 -> 409)
repo.deposits.claim(inv.invoice_id, f"void:{inv.invoice_id}", ZERO, work.now())
return _full_view(inv, repo.invoices.save(inv, current), repo)
# ═══ credit notes ═══════════════════════════════════════════════════════════════════════════════
def _credit_tax(repo, cn: CreditNote, today: date) -> Optional[object]:
"""Tax for a credit note: at the original invoice's tax point (so, at its rates) for a correction, else today."""
tax_point = today
if cn.invoice_id:
found = repo.invoices.get(cn.invoice_id)
if found:
tax_point = found[0].tax_point_date or found[0].issue_date or today
return seams.tax_compute(cn.lines, bp_id=cn.bp_id, tax_point=tax_point, currency=cn.currency, document_ref=cn.credit_id)
@AR.use_case(kind="command", route="POST /credit-notes", permission=CREDIT_WRITE)
def draft_credit(caller: dict, reason: Reason, invoice_id: Optional[Name] = None, full: bool = False,
lines: Optional[list[CreditLineIn]] = None, bp_id: Optional[BigId] = None,
currency: Optional[Currency] = None) -> dict:
"""Draft a credit note: against an issued invoice (in full with full=true, or the lines you name), or a goodwill credit on the
customer (bp_id, currency, lines; no invoice). Its tax is the tax module's, at the original invoice's rates for a
correction. Approving it (a different permission) numbers it, dates it today and settles the invoice through payments."""
with repository() as repo, _decide():
inv = None
if invoice_id:
inv = sv.visible_invoice(repo, caller, invoice_id)[0]
if inv.lifecycle != "issued":
raise RuleBroken(f"only an issued invoice is corrected with a credit note; {invoice_id} is {inv.lifecycle}")
bp_id, currency = inv.bp_id, inv.currency
if bp_id is None or not currency:
raise RuleBroken("a goodwill credit names the customer (bp_id) and the currency")
if full:
if inv is None:
raise RuleBroken("full needs an invoice to credit in full")
built = [Line(line_no=ln.line_no, description=ln.description, quantity=ln.quantity, unit_price=ln.unit_price,
discount_pct=ln.discount_pct, unit_code=ln.unit_code, product_id=ln.product_id,
tax_category=ln.tax_category, tax_code=ln.tax_code, sales_order_id=ln.sales_order_id,
so_line_no=ln.so_line_no, cost_centre=ln.cost_centre) for ln in inv.lines if not ln.deducts_invoice_id or ln.net_amount]
else:
built = sv.build_lines([ln.model_dump(exclude_none=True) for ln in (lines or [])])
cn = CreditNote.draft(f"cn:{uuid.uuid4().hex}", by=who(caller), bp_id=bp_id, currency=currency, reason=reason,
lines=built, invoice_id=invoice_id, by_sub=sv.sub_of(caller))
outcome = _credit_tax(repo, cn, _today(repo)) if cn.lines else None
if outcome is not None and outcome.quote is not None:
cn.apply_tax(outcome.quote)
return sv.credit_view(cn, repo.credits.save(cn, 0))
@AR.use_case(kind="query", route="GET /credit-notes/{credit_id}", permission=READ, etag="version")
def get_credit_note(caller: dict, credit_id: Name) -> dict:
"""One credit note, whole, with its frozen tax."""
with repository() as repo:
return sv.credit_view(*sv.visible_credit(repo, caller, credit_id))
@AR.use_case(kind="query", route="GET /credit-notes", permission=READ)
def list_credit_notes(caller: dict, invoice_id: Optional[Name] = None, bp_id: Optional[BigId] = None,
status: Optional[Literal["draft", "approved", "cancelled"]] = None, limit: Limit = 50, offset: int = 0) -> dict:
"""The credit notes the caller may see (drafts only to those who author or approve them), a page at a time."""
with repository() as repo:
if status in ("draft", "cancelled") and not sv.sees_drafts(caller):
return {"credit_notes": []}
rows = repo.credits.page(status=status, bp_id=bp_id, invoice_id=invoice_id, limit=limit, offset=max(0, offset))
if not sv.sees_drafts(caller):
rows = [r for r in rows if r["status"] == "approved"]
return {"credit_notes": [{k: r[k] for k in ("credit_id", "credit_no", "bp_id", "invoice_id", "credit_date", "currency",
"status", "total")} for r in rows]}
def _outstanding_of(invoice_id: str) -> tuple:
by, why = _settled([invoice_id])
s = by.get(invoice_id)
return (s.outstanding if s else None), why
@AR.use_case(kind="command", route="POST /credit-notes/{credit_id}/approve", permission=CREDIT_APPROVE, idempotent=True,
if_match="version")
def approve_credit(caller: dict, credit_id: Name, version: Optional[int] = None) -> dict:
"""The approver approves a drafted credit note: it needs lines and tax, must not exceed what remains creditable on its
invoice, and (where the owner asks for a second person) must not be approved by its author. It is numbered, dated TODAY and
its tax frozen; when it names an invoice it is allocated against it through payments (the lesser of the credit and what is
outstanding), so the balance falls and the original stays readable and unchanged. The allocation is asked for once the
approval has COMMITTED - payments reads the approved credit note back from this brick, which an open transaction would hide
- so a payments that refuses or is down leaves it approved with `allocation_pending`, and a repeat asks again (payments keys
the allocation on the credit note and the invoice: it is made once)."""
def unit():
with repository() as repo, _decide():
cn, current = sv.visible_credit(repo, caller, credit_id)
_same_version(f"credit note {credit_id}", current, version)
if cn.status == "approved":
return sv.credit_view(cn, current)
today = _today(repo)
s = repo.settings.in_force(today)
if cn.lines and not cn.tax:
out = _credit_tax(repo, cn, today)
if out.quote is None:
raise RuleBroken(f"the credit note's tax cannot be computed: {out.reason}")
cn.apply_tax(out.quote)
remaining, inv_state = None, "issued"
if cn.invoice_id:
inv = repo.invoices.get(cn.invoice_id)
inv_state = inv[0].lifecycle if inv else "missing"
remaining = sv.creditable_left(repo, inv[0], repo.credits.credited_total(cn.invoice_id)) if inv else None
gates = cn.gates(remaining_creditable=remaining, second_approver=s.second_approver, approver=who(caller),
identities=sv.ids(caller),
invoice_lifecycle=inv_state)
number, sid = sv.claim_number(repo, s, "credit", cn.credit_id, today, work.now(),
company=(inv[0].company_code if cn.invoice_id and inv else ""))
if cn.invoice_id and all(g.met for g in gates):
repo.credits.claim_slot(cn.invoice_id, cn.credit_id, work.now())
cn.approve(today=today, number=number, series_id=sid, by=who(caller), at=work.now(), quote=None, gates=gates)
if cn.invoice_id:
from brick_ar.application import use_cases_v4b
use_cases_v4b.take_back_deferred(repo, cn, today)
v = repo.credits.save(cn, current)
return sv.credit_view(cn, v)
body = _claiming(unit)
body["allocated"] = None
if body.get("status") == "approved" and body.get("invoice_id"): # AR-41: a credit against an intercompany invoice goes to the partner too
with repository() as repo:
found = repo.invoices.get(body["invoice_id"])
if found is not None and found[0].trading_partner:
from brick_ar.application import use_cases_v4
try:
got = use_cases_v4.hand_over_credit_note({"sub": "ar", "email": "ar@system", "permissions": [sv.HAND_OVER, sv.READ]}, credit_id)
body["handed_to_partner"] = got["handed"]
except (RuleBroken, Refused) as exc:
body["handed_to_partner"], body["hand_over_reason"] = False, str(exc)
if body.get("status") == "approved" and body.get("invoice_id") and seams.PAYMENTS.installed:
out, why = _outstanding_of(body["invoice_id"])
total = Decimal(str(body["total"]))
amount = min(total, out) if out is not None else total
if amount > 0:
a = seams.allocate("credit_note", credit_id, body["invoice_id"], amount, work.now().date(),
who(caller))
if a.ok:
body["allocated"] = {"invoice_id": body["invoice_id"], "amount": amount}
else:
body["allocation_pending"] = a.reason + " - the credit note is approved; approve it again to retry the allocation"
return body
@AR.use_case(kind="command", route="POST /credit-notes/{credit_id}/allocate", permission=CREDIT_WRITE, idempotent=True)
def allocate_credit(caller: dict, credit_id: Name, invoice_id: Name, amount: Money) -> dict:
"""Allocate an approved credit note (or what is left of it) against an invoice of the same customer and currency, through
payments: one allocation row; the invoice's outstanding falls. This brick never writes a pay_ table."""
with repository() as repo, _decide():
cn, _v = sv.visible_credit(repo, caller, credit_id)
inv = sv.visible_invoice(repo, caller, invoice_id)[0]
if cn.status != "approved":
raise RuleBroken(f"credit note {credit_id} is {cn.status}: only an approved one is allocated")
if inv.lifecycle != "issued" or inv.bp_id != cn.bp_id or inv.currency != cn.currency:
raise RuleBroken("a credit is allocated to an issued invoice of the same customer and the same currency")
left = seams.credit_unallocated([(credit_id, cn.bp_id)]).get(credit_id)
if left is not None and amount > left:
raise RuleBroken(f"only {left} of this credit note is unallocated")
out, _why = _outstanding_of(invoice_id)
if out is not None and amount > out:
raise RuleBroken(f"invoice {inv.invoice_no} only has {out} outstanding")
a = seams.allocate("credit_note", credit_id, invoice_id, amount, _today(repo), who(caller))
if not a.ok:
raise Refused(502, a.reason)
return {"credit_id": credit_id, "invoice_id": invoice_id, "allocated": amount}
@AR.use_case(kind="command", route="POST /credit-notes/{credit_id}/cancel", permission=CREDIT_WRITE, idempotent=True,
if_match="version")
def cancel_credit(caller: dict, credit_id: Name, version: Optional[int] = None) -> dict:
"""Cancel a draft credit note. An approved one is final."""
with repository() as repo, _decide():
cn, current = sv.visible_credit(repo, caller, credit_id)
_same_version(f"credit note {credit_id}", current, version)
if cn.status == "cancelled":
return sv.credit_view(cn, current)
cn.cancel(by=who(caller))
return sv.credit_view(cn, repo.credits.save(cn, current))
# ═══ write-offs ═════════════════════════════════════════════════════════════════════════════════
def _wo_view(w: Writeoff, version: int) -> dict:
out = {k: getattr(w, k) for k in ("writeoff_id", "invoice_id", "amount", "reason", "status", "proposed_by", "writeoff_date",
"approved_by", "approved_at", "rejected_reason")}
out["version"] = version
return out
@AR.use_case(kind="command", route="POST /writeoffs", permission=WRITEOFF_WRITE)
def propose_writeoff(caller: dict, invoice_id: Name, amount: Money, reason: Reason) -> dict:
"""Propose writing off a remainder that will never be paid, with a reason. An approver decides (approve_writeoff)."""
with repository() as repo, _decide():
inv = sv.visible_invoice(repo, caller, invoice_id)[0]
if inv.lifecycle != "issued":
raise RuleBroken(f"only an issued invoice is written off; {invoice_id} is {inv.lifecycle}")
w = Writeoff.propose(f"wo:{uuid.uuid4().hex}", by=who(caller), invoice_id=invoice_id, amount=amount, reason=reason,
by_sub=sv.sub_of(caller))
return _wo_view(w, repo.writeoffs.save(w, 0))
@AR.use_case(kind="command", route="POST /writeoffs/{writeoff_id}/approve", permission=WRITEOFF_APPROVE, idempotent=True,
if_match="version")
def approve_writeoff(caller: dict, writeoff_id: Name, confirm_above_threshold: bool = False, version: Optional[int] = None) -> dict:
"""The approver approves a proposed write-off: it may not exceed what is outstanding, a write-off above the owner's threshold
needs confirm_above_threshold, and where the owner asks for a second person the proposer cannot approve. The write-off is
allocated against the invoice through payments like a receipt, so the balance clears; the loss posts as bad debt where
the ledger is installed."""
def unit():
with repository() as repo, _decide():
found = repo.writeoffs.get(writeoff_id)
if found is None:
raise Refused(404, f"no write-off {writeoff_id}")
w, current = found
_same_version(f"write-off {writeoff_id}", current, version)
if w.status == "approved":
return _wo_view(w, current)
today = _today(repo)
s = repo.settings.in_force(today)
if not seams.PAYMENTS.installed:
raise RuleBroken("payments is not installed: a write-off clears the balance through payments' allocation")
out, why = _outstanding_of(w.invoice_id)
if out is None:
raise Refused(502, why or "payments could not say what is outstanding")
gates = w.gates(outstanding=out, threshold=s.writeoff_threshold, confirmed=confirm_above_threshold,
second_approver=s.second_approver, approver=who(caller), identities=sv.ids(caller))
w.approve(today=today, by=who(caller), at=work.now(), gates=gates)
v = repo.writeoffs.save(w, current)
a = seams.allocate("writeoff", w.writeoff_id, w.invoice_id, w.amount, today, who(caller))
if not a.ok:
raise Refused(502, a.reason) # rolls the approval back; the repeat is safe
return _wo_view(w, v)
return retry_on_conflict(unit)
@AR.use_case(kind="command", route="POST /writeoffs/{writeoff_id}/reject", permission=WRITEOFF_APPROVE, idempotent=True)
def reject_writeoff(caller: dict, writeoff_id: Name, reason: Reason) -> dict:
"""The approver rejects a proposed write-off, with a reason."""
with repository() as repo, _decide():
found = repo.writeoffs.get(writeoff_id)
if found is None:
raise Refused(404, f"no write-off {writeoff_id}")
w, current = found
if w.status == "rejected":
return _wo_view(w, current)
w.reject(reason=reason, by=who(caller), at=work.now())
return _wo_view(w, repo.writeoffs.save(w, current))
# ═══ schedules: a repeating line, billed once per period ═════════════════════════════════════════
def _schedule_view(repo, s: Schedule, version: int) -> dict:
raised = repo.invoices.raised_periods(s.schedule_id)
out = {f: getattr(s, f) for f in ("schedule_id", "bp_id", "source_kind", "source_id", "line_id", "product_id", "description",
"billing_kind", "billing_plan", "starts_on", "periods", "quantity", "unit_price",
"discount_pct", "unit_code", "tax_category", "currency", "in_advance", "billing_day", "status",
"stopped_at", "stop_reason")}
out["periods_raised"] = sorted(raised)
out["periods_remaining"] = s.periods - len(raised)
out["next_billing_date"] = s.next_due(raised)
out["version"] = version
return out
@AR.use_case(kind="query", route="GET /schedules", permission=SCHEDULE_READ)
def list_schedules(caller: dict, status: Optional[Literal["active", "finished", "stopped"]] = None,
source_id: Optional[Name] = None, limit: Limit = 100) -> dict:
"""Billing schedules per source: periods raised, the next billing date and what remains (all derived, never counted)."""
with repository() as repo:
out = []
ids = repo.schedules.ids(status=status, source_id=source_id, limit=limit)
got = repo.schedules.get_many(ids)
for sid in ids:
s, v = got[sid]
out.append(_schedule_view(repo, s, v))
return {"schedules": out}
@AR.use_case(kind="query", route="GET /schedules/{schedule_id}", permission=SCHEDULE_READ)
def get_schedule(caller: dict, schedule_id: Name) -> dict:
"""One schedule with its periods, which are raised, and the next billing date."""
with repository() as repo:
found = repo.schedules.get(schedule_id)
if found is None:
raise Refused(404, f"no schedule {schedule_id}")
return _schedule_view(repo, *found)
@AR.use_case(kind="command", route="POST /schedules", permission=PLAN, idempotent=True)
def plan_billing(caller: dict, source_kind: Literal["sales_order", "contract"], source_id: Name, line_id: Name, bp_id: BigId,
starts_on: IsoDate, periods: LineNo, unit_price: Rate, currency: Currency,
billing_kind: Literal["recurring", "instalment", "usage"] = "recurring",
billing_plan: Literal["monthly", "quarterly", "annual", "periodic"] = "monthly", quantity: Optional[Qty] = None,
discount_pct: Optional[Pct] = None, unit_code: Optional[Code] = None, product_id: Optional[Code] = None,
description: Optional[Note] = None, tax_category: Optional[Code] = None,
in_advance: Optional[bool] = None) -> dict:
"""A source confirmed a line that repeats: ONE schedule per line, keyed "<source_id>|<line_id>", active, starting on the date the
source line records (an accounting fact; it may be in the past). A period whose billing day has already passed is raised
in the next daily pass, covering its own period, each draft waiting for the biller's review. Calling twice changes nothing."""
sid = f"{source_id}|{line_id}"
def unit():
with repository() as repo, _decide():
found = repo.schedules.get(sid)
if found is not None:
return _schedule_view(repo, *found)
today = _today(repo)
gate = profile_gate(sv.profile_for(repo, bp_id, today), today)
if not gate.met:
raise GatesUnmet("plan billing", [Gate("customer_profile", False, "the customer has no billing profile",
"Customer billing: set bill-to and terms (AR-25)")])
s = repo.settings.in_force(today)
sch = Schedule.plan(sid, by=who(caller), bp_id=bp_id, source_kind=source_kind, source_id=source_id, line_id=line_id,
product_id=product_id or "", description=description or "", billing_kind=billing_kind,
billing_plan=billing_plan, starts_on=_date(starts_on, "starts_on"), 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 "",
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 _schedule_view(repo, sch, repo.schedules.save(sch, 0))
return retry_on_conflict(unit)
@AR.use_case(kind="command", route="POST /schedules/stop", permission=STOP, idempotent=True)
def stop_billing(caller: dict, source_id: Name, reason: Reason, stop_date: Optional[IsoDate] = None) -> dict:
"""The source of a schedule ended early (an order cancelled, a contract terminated): billing stops after the period in
progress. The stop is dated TODAY, never earlier (a stop_date before today is refused, with the fix); issued invoices
are untouched and an earlier end the customer is owed is a credit note against them. Applies to every schedule of the source."""
with repository() as repo, _decide():
today = _today(repo)
if stop_date and _date(stop_date, "stop_date") != today:
raise GatesUnmet("stop billing", [Gate("not_back_dated", False, "a stop cannot be dated before today",
"stop today and credit the periods already issued (AR-04)")])
stopped = []
ids = repo.schedules.ids(source_id=source_id, limit=500)
got = repo.schedules.get_many(ids)
for sid in ids:
s, v = got[sid]
if s.status == "active":
s.stop(today=today, reason=reason, by=who(caller))
repo.schedules.save(s, v)
stopped.append(sid)
return {"source_id": source_id, "stopped": stopped, "stopped_on": today}
def _period_line(s: Schedule, k: int) -> Line:
first, last = s.bounds(k)
qty = None if s.billing_kind == "usage" else (s.quantity or Decimal(1))
if qty is not None:
share = s.proration(k)
qty = (qty * share).quantize(Decimal("0.0001")) if share != 1 else qty
so_line = int(s.line_id) if s.source_kind == "sales_order" and str(s.line_id).isdigit() else None
return Line(line_no=1, description=f"{s.description or s.product_id or 'Subscription'} ({first} to {last})", quantity=qty,
unit_price=s.unit_price, discount_pct=s.discount_pct, unit_code=s.unit_code, product_id=s.product_id,
tax_category=s.tax_category, tax_code=s.tax_code, schedule_id=s.schedule_id, service_from=first, service_to=last,
sales_order_id=s.source_id if so_line is not None else "", so_line_no=so_line, usage=s.billing_kind == "usage")
@AR.use_case(kind="command", route="POST /schedules/raise-due", permission=RAISE, idempotent=True)
def raise_due(caller: dict, limit: Limit = 200) -> dict:
"""The daily pass: for every schedule, raise the draft of each period whose billing day has come and that has none yet, keyed
"rec:<schedule_id>:<period>" so a second pass creates nothing. Periods already past (a schedule starting in the past) are
raised now, issued today, each waiting for the biller's review; a usage period arrives with its quantity empty. A stopped
schedule raises only the period in progress and earlier ones."""
with repository() as repo:
today = _today(repo)
ids = []
for status in ("active", "stopped"):
offset = 0
while True: # every page, not just the first
page = repo.schedules.ids(status=status, limit=limit, offset=offset)
ids += page
if len(page) < limit:
break
offset += limit
raised_now, skipped, failed = [], 0, []
for sid in ids:
with repository() as repo:
s, _v = repo.schedules.get(sid)
due = s.due_periods(today, repo.invoices.raised_periods(sid))
for k in due:
def unit(k=k, s=s):
with repository() as repo, _decide():
iid = f"rec:{s.schedule_id}:{k}"
if repo.invoices.exists(iid):
return False
_create_draft(repo, caller, invoice_id=iid, bp_id=s.bp_id, currency=s.currency, source_kind="schedule",
source_id=iid, lines=[_period_line(s, k)], tax_point_date=None,
auto_source="manual" if s.source_kind == "manual" else None)
return True
try:
if retry_on_conflict(unit):
raised_now.append(f"rec:{sid}:{k}")
else:
skipped += 1
except Refused as exc:
failed.append({"schedule_id": sid, "period": k, "reason": exc.reason})
return {"as_of": today, "raised": raised_now, "already_raised": skipped, "failed": failed}
@AR.use_case(kind="command", route="POST /schedules/finish", permission=FINISH, idempotent=True)
def finish_schedules(caller: dict) -> dict:
"""In the same pass as raise_due: finish every active schedule whose every period is raised (set once); nothing more is raised."""
finished = []
with repository() as repo:
ids = repo.schedules.ids(status="active", limit=1000)
got = repo.schedules.get_many(ids)
for sid in ids:
s, v = got[sid]
if s.finish(raised=repo.invoices.raised_periods(sid), by=who(caller)):
repo.schedules.save(s, v)
finished.append(sid)
return {"finished": finished}
# ═══ customers, terms, settings, account map ════════════════════════════════════════════════════
def _profile_view(p: Profile) -> dict:
return {"bp_id": p.bp_id, "valid_from": p.valid_from, "bill_to_name": p.bill_to_name, "bill_to_address": p.bill_to_address,
"invoice_email": p.invoice_email, "terms_id": p.terms_id, "currency": p.currency, "set_by": p.set_by,
"intercompany_company_code": p.intercompany_company_code, "einvoice_channel": p.einvoice_channel or "email",
"peppol_participant_id": p.peppol_participant_id, "default_cost_centre": p.default_cost_centre}
@AR.use_case(kind="query", route="GET /customers/{bp_id}/billing", permission=READ)
def get_billing_profile(caller: dict, bp_id: BigId) -> dict:
"""How a customer is billed: the profile in force today and any that start later (bill-to, invoice email, terms, currency)."""
with repository() as repo:
today = _today(repo)
rows = repo.customers.rows(bp_id)
now = sv.profile_for(repo, bp_id, today)
return {"bp_id": bp_id, "in_force": _profile_view(now) if now else None,
"later": [_profile_view(p) for p in rows if p.valid_from > today], "complete": bool(now and not now.missing())}
@AR.use_case(kind="command", route="POST /customers/{bp_id}/billing", permission=CUSTOMER_WRITE)
def set_billing_profile(caller: dict, bp_id: BigId, bill_to_name: Name, bill_to_address: Note, invoice_email: Name, terms_id: Code,
currency: Currency, valid_from: Optional[IsoDate] = None,
intercompany_company_code: Optional[Name] = None, einvoice_channel: Optional[Literal["email", "peppol"]] = None,
peppol_participant_id: Optional[Name] = None, default_cost_centre: Optional[Code] = None) -> dict:
"""Set a customer's bill-to name, address, invoice email, terms and currency as a NEW dated row, from tomorrow by default. A
first profile may start today; one dated today (for a customer who has one) or earlier is refused. At issue the profile in
force is copied onto the invoice, so a later change never alters an issued invoice. The party master keeps identity."""
with repository() as repo, _decide():
today = _today(repo)
rows = repo.customers.rows(bp_id)
vf = _date(valid_from, "valid_from") or (today if not rows else today + timedelta(days=1))
check_dated("a billing profile", vf, today, first=not rows)
terms = repo.terms.get(terms_id)
if terms is None or not terms["active"]:
raise RuleBroken(f"there are no active payment terms {terms_id}: set them first (set_terms)")
c = seams.check_partner(bp_id, caller)
if c.status == "unknown":
raise Refused(422, c.note)
if einvoice_channel == "peppol" and not re.match(r"^\d{4}:\S+$", peppol_participant_id or ""):
raise RuleBroken("a customer on Peppol needs their participant id as scheme:id, such as 0088:5060012349998 (a GLN)")
p = Profile(bp_id=bp_id, valid_from=vf, bill_to_name=bill_to_name, bill_to_address=bill_to_address, invoice_email=invoice_email,
terms_id=terms_id, currency=currency.upper(), intercompany_company_code=intercompany_company_code or "",
einvoice_channel=einvoice_channel or "email", peppol_participant_id=peppol_participant_id or "",
default_cost_centre=default_cost_centre or "", set_by=who(caller), set_at=work.now())
try:
repo.customers.add(p)
except IntegrityError:
raise Refused(409, f"the customer already has a billing profile from {vf}: rows are never edited, date the new one later") from None
return _profile_view(p)
@AR.use_case(kind="query", route="GET /terms", permission=READ)
def list_terms(caller: dict) -> dict:
"""The payment terms: days, basis (invoice_date | end_of_month) and any early-payment discount."""
with repository() as repo:
return {"terms": repo.terms.all()}
@AR.use_case(kind="command", route="POST /terms", permission=SETTINGS_WRITE)
def set_terms(caller: dict, terms_id: Code, days: Days = 30, basis: Literal["invoice_date", "end_of_month"] = "invoice_date",
description: Optional[Name] = None, early_discount_pct: Optional[Pct] = None, active: bool = True,
early_discount_days: Optional[Days] = None) -> dict:
"""Add payment terms, or deactivate old ones. Terms an issued invoice used are never changed (the invoice carries its due date
anyway): different terms are a new terms_id, the old one deactivated."""
with repository() as repo, _decide():
old = repo.terms.get(terms_id)
if bool(early_discount_pct) != bool(early_discount_days is not None):
raise RuleBroken("an early-payment discount is a percentage AND the days it holds for (2/10: 2% within 10 days)")
shown = (early_discount_pct or Decimal(0), early_discount_days or 0)
if old and repo.terms.used(terms_id) and (old["days"], old["basis"], old["early_discount_pct"], old["early_discount_days"]) != (days, basis, *shown):
raise RuleBroken(f"terms {terms_id} were used by an issued invoice: add different terms under a new id and deactivate these")
row = {"terms_id": terms_id, "days": days, "basis": basis, "description": description or f"{days} days",
"early_discount_pct": early_discount_pct or Decimal(0), "active": active,
"early_discount_days": early_discount_days or 0}
repo.terms.put(row)
return row
def _setting_rows(repo, caller: dict) -> dict:
today = _today(repo)
out = {}
for key, (_kind, default, control, _choices) in SETTINGS.items():
if control and not (sv.has(caller, CONTROLS) or sv.has(caller, SETTINGS_WRITE)):
continue
rows = [r for r in repo.settings.rows(key)]
out[key] = {"value": getattr(repo.settings.in_force(today), key) if key in SETTINGS else default, "default": default,
"control": control, "rows": [{"valid_from": r["valid_from"], "value": r["value"], "set_by": r["set_by"]} for r in rows]}
return out
@AR.use_case(kind="query", route="GET /settings", permission=READ)
def get_settings(caller: dict) -> dict:
"""The settings in force today, each with its dated rows. The owner's controls are shown to those who hold or administer them."""
with repository() as repo:
return {"today": _today(repo), "settings": _setting_rows(repo, caller)}
def _set(repo, caller: dict, key: str, value, valid_from: Optional[str], *, control: bool) -> dict:
if key not in SETTINGS or (SETTINGS[key][2] != control):
raise Refused(403 if key in SETTINGS else 422,
(f"{key} is a control on the people who invoice and approve: it needs {CONTROLS}" if key in SETTINGS and
SETTINGS[key][2] else f"{key} is not a control: use set_setting") if key in SETTINGS else f"there is no setting {key}")
try:
value = check_setting(key, value)
except ValueError as exc:
raise Refused(422, str(exc)) from None
today = _today(repo)
first = not repo.settings.has_any(key)
vf = _date(valid_from, "valid_from") or (today if first else today + timedelta(days=1))
check_dated(f"a change to {key}", vf, today, first=first)
if key in ("invoice_number_format", "number_restarts_yearly"):
in_force = repo.settings.in_force(vf)
fmt = value if key == "invoice_number_format" else in_force.invoice_number_format
restarts = value if key == "number_restarts_yearly" else in_force.number_restarts_yearly
try:
numbering.check_pair(fmt, restarts)
except ValueError as exc:
raise Refused(409, str(exc)) from None
if key in FIRST_OF_MONTH_KEYS and vf.day != 1:
raise RuleBroken(f"{key} takes effect from the first day of a month, so no month is both posted and exported: "
f"{vf} is not a first (try {first_of_next_month(today)})")
try:
repo.settings.add(key, vf, value, who(caller), work.now())
except IntegrityError:
raise Refused(409, f"{key} already has a row from {vf}: rows are never edited, date the new one later") from None
return {"setting": key, "value": value, "valid_from": vf, "set_by": who(caller)}
@AR.use_case(kind="command", route="POST /settings", permission=SETTINGS_WRITE)
def set_setting(caller: dict, key: Name, value: Optional[bool | int | str], valid_from: Optional[IsoDate] = None) -> dict:
"""Change a day-to-day setting (export format, billing day, advance/arrears, proration, document layout) as a new dated row, from
tomorrow by default. The owner's controls are refused here: they are set with set_control by whoever holds ar:settings:controls."""
with repository() as repo, _decide():
return _set(repo, caller, key, value, valid_from, control=False)
@AR.use_case(kind="command", route="POST /controls", permission=CONTROLS)
def set_control(caller: dict, key: Name, value: Optional[bool | int | str], valid_from: Optional[IsoDate] = None) -> dict:
"""The owner changes a CONTROL on the people who invoice and approve (business_timezone, invoice_number_format,
number_restarts_yearly, ledger_mode, auto_issue_sources, void_policy, writeoff_threshold, second_approver, the AI step modes) as
a new dated row. A control dated today or earlier is refused, the first row of a key excepted; a ledger_mode switch takes
effect from the first day of a later month. Drafts raised today still wait."""
with repository() as repo, _decide():
return _set(repo, caller, key, value, valid_from, control=True)
@AR.use_case(kind="query", route="GET /account-map", permission=READ)
def get_ar_account_map(caller: dict) -> dict:
"""Which ledger account each posting hits (receivable, revenue, tax by code, bad debt, deposit, FX), with the dates they start."""
with repository() as repo:
return {"rows": repo.accounts.rows()}
@AR.use_case(kind="command", route="POST /account-map", permission=SETTINGS_WRITE)
def set_ar_account_map(caller: dict, purpose: Literal["receivable", "revenue", "tax", "bad_debt", "fx_gain", "fx_loss", "deposit"],
account_id: BigId, tax_code: Optional[Code] = None, valid_from: Optional[IsoDate] = None) -> dict:
"""Map a posting purpose (and, for tax, a tax code) to a ledger account as a new dated row, from tomorrow by default. A
posting uses the row in force on the document's own date; a row dated today or earlier is refused (the first of a key excepted)."""
with repository() as repo, _decide():
today = _today(repo)
tc = tax_code or ""
first = not repo.accounts.has_any(purpose, tc)
vf = _date(valid_from, "valid_from") or (today if first else today + timedelta(days=1))
check_dated("an account-map row", vf, today, first=first)
try:
repo.accounts.add(purpose, tc, vf, account_id, who(caller))
except IntegrityError:
raise Refused(409, "the account map already has a row for that purpose and day") from None
return {"purpose": purpose, "tax_code": tc, "valid_from": vf, "account_id": account_id}
# ═══ the ledger and the accountant's export ═════════════════════════════════════════════════════
def _post_one(repo, document_id: str, event: str, body: Optional[dict], needs: list, on: date, actor: str) -> dict:
if repo.postings.get(document_id, event) is not None:
return {"document_id": document_id, "event": event, "posted": True, "already": True}
if body is None:
return {"document_id": document_id, "event": event, "posted": False,
"gate": Gate("account_map", False, "no account in force on " + str(on) + " for: " + ", ".join(needs),
"Settings: account map").as_dict()}
res = seams.post_journal(body, actor)
if not res.ok:
return {"document_id": document_id, "event": event, "posted": False, "reason": res.reason}
repo.postings.record(document_id, event, int(body["journal_id"]), on, actor)
return {"document_id": document_id, "event": event, "posted": True, "journal_id": body["journal_id"]}
def _post_document(repo, document_id: str, actor: str) -> list:
"""Post what a document owes the ledger: an issued invoice its journal; a void one the reversal; an approved credit note and
an approved write-off theirs. Only in post mode, and only for the document's own date. A migrated document is never posted."""
out = []
if repo.invoices.exists(document_id):
inv = repo.invoices.get(document_id)[0]
if inv.source_kind == "odoo" or inv.lifecycle not in ("issued", "void"):
return [{"document_id": document_id, "posted": False, "reason": f"{inv.lifecycle} {inv.source_kind} invoices are never posted"}]
if sv.ledger_mode(repo, inv.issue_date) != "post":
return [{"document_id": document_id, "posted": False, "reason": f"the ledger mode on {inv.issue_date} is not post"}]
body, needs = sv.invoice_journal(repo, inv, "issue", inv.issue_date)
out.append(_post_one(repo, document_id, "issue", body, needs, inv.issue_date, actor))
if inv.lifecycle == "void" and out[-1].get("posted"):
body, needs = sv.invoice_journal(repo, inv, "void", inv.voided_on)
out.append(_post_one(repo, document_id, "void", body, needs, inv.voided_on, actor))
return out
if repo.credits.exists(document_id):
cn = repo.credits.get(document_id)[0]
if cn.status != "approved" or cn.source_kind == "odoo":
return [{"document_id": document_id, "posted": False, "reason": "only an approved credit note of this brick is posted"}]
if sv.ledger_mode(repo, cn.credit_date) != "post":
return [{"document_id": document_id, "posted": False, "reason": f"the ledger mode on {cn.credit_date} is not post"}]
body, needs = sv.credit_journal(repo, cn, cn.credit_date)
return [_post_one(repo, document_id, "approve", body, needs, cn.credit_date, actor)]
found = repo.writeoffs.get(document_id)
if found and found[0].status == "approved":
w = found[0]
inv = repo.invoices.get(w.invoice_id)[0]
if sv.ledger_mode(repo, w.writeoff_date) != "post":
return [{"document_id": document_id, "posted": False, "reason": f"the ledger mode on {w.writeoff_date} is not post"}]
body, needs = sv.writeoff_journal(repo, w.invoice_id, w.writeoff_id, w.amount, inv.currency, w.writeoff_date)
return [_post_one(repo, document_id, "approve", body, needs, w.writeoff_date, actor)]
raise Refused(404, f"no invoice, credit note or write-off {document_id}")
@AR.use_case(kind="command", route="POST /documents/{document_id}/post", permission=POST, idempotent=True)
def post_to_ledger(caller: dict, document_id: Name) -> dict:
"""In post mode, post an issued invoice, a voided one (the reversal, dated today), an approved credit note or write-off to
the ledger as a balanced journal, through the ledger's route. The journal id is derived from the document, so posting twice
changes the ledger once; a closed period is refused and named. A migrated document is never posted."""
with repository() as repo, _decide():
return {"results": _post_document(repo, document_id, who(caller))}
@AR.use_case(kind="command", route="POST /documents/post-pending", permission=POST, idempotent=True)
def post_pending(caller: dict, limit: Limit = 200) -> dict:
"""Post every issued, voided, approved-credit and approved-write-off document that the ledger has not got yet (the clock
worker's pass). Documents the ledger refuses stay for the next pass with the reason."""
with repository() as repo:
states = repo.invoices.ledger_states()
ids = [x[0] for x in states] + repo.credits.approved_ids() + repo.writeoffs.approved_ids()
done = {i: lc for i, lc, day, issue_posted, void_posted in states
if issue_posted and (lc == "issued" or void_posted) and sv.ledger_mode(repo, day) == "post"}
results = []
# The limit counts the documents still to post: taking it before skipping the posted ones meant that past
# `limit` posted invoices nothing new was ever posted. Already-posted ones fill what is left of the limit.
pending = [d for d in ids if d not in done]
for did in pending[:limit] + [d for d in ids if d in done][:max(0, limit - len(pending))]:
if did in done: # the ledger has its journals: the answer _post_document gives, without loading it
results += [{"document_id": did, "event": ev, "posted": True, "already": True}
for ev in (("issue",) if done[did] == "issued" else ("issue", "void"))]
continue
try:
results += _post_document(repo, did, who(caller))
except RuleBroken as exc:
results.append({"document_id": did, "posted": False, "reason": str(exc)})
return {"posted": sum(1 for r in results if r.get("posted") and not r.get("already")), "results": results}
@AR.use_case(kind="command", route="POST /exports", permission=sv.EXPORT, idempotent=True)
def run_export(caller: dict, period: text(7, min_length=7)) -> dict:
"""The monthly export to the accountant (ledger mode export): one file with every issued and voided invoice and approved
credit note of the month, once; a batch record "<period>:<format>"; each document marked with the batch. Running the same
month again exports nothing new. In post mode nothing is exported (export and post together would book revenue twice)."""
def unit():
with repository() as repo, _decide():
try:
start = date.fromisoformat(period + "-01")
except ValueError:
raise Refused(422, "the period is a month such as 2026-10") from None
end = first_of_next_month(start) - timedelta(days=1)
if sv.ledger_mode(repo, start) != "export":
raise GatesUnmet("export", [Gate("ledger_mode", False, f"the ledger mode on {start} is {sv.ledger_mode(repo, start)}, not export",
"Settings: ledger mode (the owner)")])
fmt = repo.settings.in_force(_today(repo)).export_format
batch_id = f"{period}:{fmt}"
invs, creds = repo.invoices.in_period_unexported(start, end), repo.credits.in_period_unexported(start, end)
old = repo.exports.get(batch_id)
if not invs and not creds:
return {"batch_id": batch_id, "exported": 0, "document_count": old["document_count"] if old else 0,
"file_key": old["file_key"] if old else None}
text_ = sv.export_csv(invs, creds, with_head=old is None)
repo.exports.put({"batch_id": batch_id, "period": period, "format": fmt, "created_by": who(caller),
"document_count": (old["document_count"] if old else 0) + len(invs) + len(creds),
"file_key": f"exports/ar/{batch_id}.csv", "content": (old["content"] if old else "") + text_})
loaded_invs = repo.invoices.get_many([h["invoice_id"] for h in invs])
loaded_creds = repo.credits.get_many([h["credit_id"] for h in creds])
for h in invs:
inv, v = loaded_invs[h["invoice_id"]]
inv.mark_exported(batch_id)
repo.invoices.save(inv, v)
for h in creds:
cn, v = loaded_creds[h["credit_id"]]
cn.export_batch_id = batch_id
repo.credits.save(cn, v)
return {"batch_id": batch_id, "exported": len(invs) + len(creds), "file_key": f"exports/ar/{batch_id}.csv"}
return retry_on_conflict(unit)
@AR.use_case(kind="query", route="GET /exports/{batch_id}", permission=sv.EXPORT)
def get_export(caller: dict, batch_id: Name) -> dict:
"""An export batch with its file's content (CSV)."""
with repository() as repo:
b = repo.exports.get(batch_id)
if b is None:
raise Refused(404, f"no export batch {batch_id}")
return b
# ═══ migration from Odoo: history, read-only, reconciled and signed ═════════════════════════════
def _run_view(run: MigrationRun, version: int) -> dict:
out = {k: getattr(run, k) for k in ("run_id", "source", "source_db", "cutover_date", "status", "loaded_at", "checksums_json",
"signed_by", "signed_at")}
out["doors"] = json.loads(run.doors_json or "{}")
out["counts"] = json.loads(run.counts_json or "{}")
out["checks"] = [{"check": c.check_key, "slice": c.slice, "odoo": c.odoo_value, "ours": c.ours_value, "difference": c.difference,
"equal": c.equal, "must": c.must, "explanation": c.explanation, "explained_by": c.explained_by}
for c in run.checks]
out["readiness"] = {"transition": "sign_reconciliation", "gates": [g.as_dict() for g in run.gates()]} if run.checks else None
out["version"] = version
return out
def _n(v):
"""A comparable form: 10.0 and 10.0000 are the same number, a date is its ISO text."""
if isinstance(v, Decimal):
return format(v.normalize(), "f")
return None if v in (None, "") else str(v)
def _doc_signature(doc) -> str:
"""What a reload compares: if Odoo's data and our mapping are unchanged, the second load writes nothing."""
lines = [tuple(_n(x) for x in (ln.line_no, ln.description, ln.quantity, ln.unit_price, ln.net_amount, ln.tax_amount, ln.tax_code,
ln.product_id, ln.unit_code, ln.sales_order_id, ln.so_line_no, ln.odoo_line_id)) for ln in doc.lines]
tax = [tuple(_n(x) for x in (t.tax_code, t.rate_pct, t.net_amount, t.tax_amount)) for t in doc.tax]
head = [_n(getattr(doc, k, None)) for k in ("bp_id", "lifecycle", "status", "invoice_no", "credit_no", "issue_date", "credit_date",
"tax_point_date", "due_date", "terms_id", "currency", "fx_rate", "subtotal", "tax_total",
"total", "total_functional", "customer_po_ref", "invoice_id", "credit_id")]
return hashlib.sha256(json.dumps([head, lines, tax]).encode()).hexdigest()
@AR.use_case(kind="query", route="GET /migration/runs", permission=MIGRATION_READ)
def list_invoice_migration_runs(caller: dict) -> dict:
"""The migration runs and their state (loaded, reconciled, signed, discarded)."""
with repository() as repo:
return {"runs": repo.runs.list()}
@AR.use_case(kind="query", route="GET /migration/runs/{run_id}", permission=MIGRATION_READ)
def get_invoice_migration_run(caller: dict, run_id: Name) -> dict:
"""A run with its doors, counts and checks (Odoo, ours, difference, must, explanation) and the sign button's readiness
{gate, met, reason, fix}."""
with repository() as repo:
found = repo.runs.get(run_id)
if found is None:
raise Refused(404, f"no migration run {run_id}")
return _run_view(*found)
@AR.use_case(kind="command", route="POST /migration/odoo/load", permission=LOAD, idempotent=True)
def load_from_odoo(caller: dict, cutover_date: IsoDate, doors: Optional[DoorsIn] = None) -> dict:
"""Load Odoo's customer invoices and credit notes as read-only HISTORY: one invoice per move keyed "odoo:<db>:<move id>",
issued with Odoo's number, dates, lines, tax and totals exactly as Odoo stored them (never recomputed, never re-posted to the
ledger, never re-sent, never raised again by a schedule); credit notes keyed the same and linked to the invoice they reverse.
Needs the doors decided (defaults are the customer's signed decisions: continue Odoo's numbering, six years of history, Odoo drafts
re-raised by hand) and business-partner and tax to have mapped every partner and tax the moves use. A second load changes nothing;
a load after a fix keeps the same keys and the explanations of unchanged checks."""
d = doors or DoorsIn()
cutover = _date(cutover_date, "cutover_date")
door = {"history_years": 6 if d.history_years is None else d.history_years, "numbering": d.numbering or "continue",
"carry_drafts": bool(d.carry_drafts), "instalments": d.instalments or "as_odoo", "company_id": d.company_id,
"ledger": d.ledger}
reader = odoo_reader()
gates = [met("doors_decided") if door["numbering"] in ("continue", "new_prefix") and door["instalments"] in ("as_odoo", "simplified")
else unmet("doors_decided", "a migration door is undecided or not one of its choices", "Migration: answer it")]
moves = [m for m in reader.moves(door["company_id"], door["carry_drafts"])
if odoo_map.in_scope(m, cutover, door["history_years"], door["carry_drafts"])]
lines = reader.lines([m["id"] for m in moves])
db = reader.db
partner_ids = sorted({odoo_map.m2o(m.get("commercial_partner_id")) or odoo_map.m2o(m.get("partner_id")) for m in moves} - {None})
tax_ids = sorted({t for ls in lines.values() for ln in ls for t in (ln.get("tax_ids") or [])} |
{odoo_map.m2o(ln.get("tax_line_id")) for ls in lines.values() for ln in ls if ln.get("tax_line_id")} - {None})
bp = seams.migration_map("business-partner", db, "res.partner", partner_ids)
taxes = seams.migration_map("tax", db, "account.tax", tax_ids)
miss_p = [p for p in partner_ids if bp is None or str(p) not in bp]
miss_t = [t for t in tax_ids if taxes is None or str(t) not in taxes]
gates.append(met("partners_mapped") if not miss_p else unmet(
"partners_mapped", "Odoo partners used by the moves have no bp_id" + (f": {miss_p[:10]}" if bp is not None else
" (business-partner is not answering)"),
"run the business-partner migration first"))
gates.append(met("taxes_mapped") if not miss_t else unmet(
"taxes_mapped", "Odoo taxes used by the moves have no tax_code" + (f": {miss_t[:10]}" if taxes is not None else
" (tax is not answering)"),
"run the tax migration first"))
if not all(g.met for g in gates):
raise Refused(409, GatesUnmet("load the Odoo history", gates).args[0])
run_id = f"odoo:{db}:{cutover}"
ctx = odoo_map.MapContext(db=db, run_id=run_id, company_currency=reader.company_currency(),
bp={int(k): int(v) for k, v in bp.items()}, tax={int(k): v for k, v in taxes.items()})
prod_ids = sorted({odoo_map.m2o(ln.get("product_id")) for ls in lines.values() for ln in ls if ln.get("product_id")} - {None})
ctx.product = {int(k): v for k, v in (seams.migration_map("product", db, "product.product", prod_ids) or {}).items()}
so_ids = sorted({i for ls in lines.values() for ln in ls for i in (ln.get("sale_line_ids") or [])})
ctx.so_line = {int(k): (str(v).split("|")[0], int(str(v).split("|")[1])) for k, v in
(seams.migration_map("sales-order", db, "sale.order.line", so_ids) or {}).items() if "|" in str(v)}
ctx.partners = {p["id"]: p for p in reader.partners(partner_ids)}
term_ids = sorted({odoo_map.m2o(m.get("invoice_payment_term_id")) for m in moves if m.get("invoice_payment_term_id")} - {None})
terms, term_lines = reader.payment_terms(term_ids)
def unit():
counts = {"invoices": 0, "credit_notes": 0, "unchanged": 0, "updated": 0, "voided": 0, "drafts": 0, "differs": []}
signed_runs: dict = {}
with repository() as repo, _decide():
run_found = repo.runs.get(run_id)
if run_found and run_found[0].status in ("signed", "discarded"):
raise RuleBroken(f"migration run {run_id} is {run_found[0].status}: a later load is a new run with a new cutover date")
now_settings = repo.settings.in_force(_today(repo))
fmt_now, restarts_now = now_settings.invoice_number_format, now_settings.number_restarts_yearly
for t in terms:
repo.terms.put(odoo_map.map_term(t, [ln for ln in term_lines if odoo_map.m2o(ln.get("payment_id")) == t["id"]]))
repo.maps.put(db, "account.payment.term", t["id"], f"odoo:{t['id']}", run_id)
for m in sorted(moves, key=lambda m: m["id"]):
doc = odoo_map.map_move(m, lines.get(m["id"], []), ctx)
is_inv = isinstance(doc, Invoice)
repo_ = repo.invoices if is_inv else repo.credits
key = doc.invoice_id if is_inv else doc.credit_id
existing = repo_.get(key)
carried = doc.invoice_no if is_inv else doc.credit_no
if carried and not repo.numbers.register(carried, key, work.now()): # a number belongs to ONE document
counts["duplicate_numbers"] = counts.get("duplicate_numbers", 0) + 1
if existing is not None:
if _doc_signature(existing[0]) == _doc_signature(doc):
counts["unchanged"] += 1
continue
owner = existing[0].migration_run_id
if owner and owner != run_id:
if owner not in signed_runs:
held = repo.runs.get(owner)
signed_runs[owner] = bool(held and held[0].status == "signed")
if signed_runs[owner]: # signed history is final: Odoo's change is a difference to read
counts["differs"].append(key)
continue
repo_.save(doc, existing[1])
counts["updated"] += 1
else:
repo_.save(doc, 0)
counts["invoices" if is_inv else "credit_notes"] += 1
counts["drafts"] += (doc.lifecycle if is_inv else doc.status) == "draft"
counts["voided"] += (doc.lifecycle if is_inv else doc.status) in ("void", "cancelled")
repo.maps.put(db, "account.move", m["id"], key, run_id)
no = doc.invoice_no if is_inv else doc.credit_no
parsed = numbering.parse_number(no)
if parsed and door["numbering"] == "continue":
repo.numbers.raise_mark(*parsed)
resolved = numbering.match(fmt_now if is_inv else numbering.credit_format(fmt_now), no, restarts_now)
if resolved: # the series the owner's format ACTUALLY counts in
repo.numbers.raise_mark(*resolved)
for pid, bpid in ctx.bp.items():
repo.maps.put(db, "res.partner", pid, str(bpid), run_id)
p = ctx.partners.get(pid, {})
if not repo.customers.rows(bpid):
term = odoo_map.m2o(p.get("property_payment_term_id"))
repo.customers.add(Profile(bp_id=bpid, valid_from=cutover, bill_to_name=p.get("name") or "",
bill_to_address=", ".join(str(p[k]) for k in ("street", "street2", "zip", "city") if p.get(k)),
invoice_email=p.get("email") or "", terms_id=f"odoo:{term}" if term else "",
currency=ctx.company_currency, set_by=f"odoo:{db}", set_at=work.now()))
for tid, code in ctx.tax.items():
repo.maps.put(db, "account.tax", tid, code, run_id)
counts["multi_tax_lines"] = ctx.multi_tax_lines
run = run_found[0] if run_found else Migr