dodil/erp-brick-sales-order
DODIL ERP brick: sales order
src/brick_sales_order/application/schedule.py
"""Promised dates: promise a draft's dates (SO-35) and the daily re-promise (SO-36).
SO-36 in order of things, so a retry or a racer never double-writes: for each open order, in priority order (rush first) then order date (first
come, first served), ONE transaction asks inventory's availability (a read), appends the new schedule version, the owner's task and a pending
re-send of the reservation, and saves the order (its version is the claim: a cancel, an amend or another run of the same order that saved first
makes this a 409 and nothing is written); only AFTER that commit is inventory's reservation re-sent, with the same revision and the new
schedule version (keyed there, so a retry changes nothing; one that failed is retried by the next run). The next order's availability then
already counts this one's new reservation.
"""
import json
from datetime import date
from decimal import Decimal
from typing import Literal, Optional
from erp_platform.bricks import Refused
from erp_platform.repository import Conflict
from brick_sales_order.application import promising
from brick_sales_order.application import use_cases as uc
from brick_sales_order.application.inputs import Ref
from brick_sales_order.domain import promise as pm
from brick_sales_order.infrastructure.seams import ports
SO, READ, WRITE, RESCHEDULE = uc.SO, uc.READ, uc.WRITE, uc.RESCHEDULE
@SO.use_case(kind="command", route="POST /sales-orders/{sales_order_id}/promise-dates", permission=WRITE, if_match="version")
def promise_dates(caller: dict, sales_order_id: Ref, version: Optional[int] = None) -> dict:
"""Promise the delivery dates of a DRAFT order from inventory's availability: each stocked line gets schedule lines (a quantity for a date;
a shortfall is a later date or an unconfirmed quantity, never a refusal), shown on the order and printed on the confirmation. Nothing is
reserved (confirming reserves). Without inventory, or without a delivering warehouse, the requested date is kept and the line says nothing
was checked. Confirming promises again."""
def unit():
with uc.repository() as repo, uc._decide():
o, current = uc._load(repo, caller, sales_order_id)
uc._same_version(f"sales order {sales_order_id}", current, version)
if o.lifecycle != "draft":
raise Refused(409, f"the order is {o.lifecycle}: dates are promised on a draft (a confirmed order is re-promised daily)")
promising.promise_order(repo, o, uc._today(repo), at=uc._bnow(), reason="promised", exclude_own=False)
return uc._saved(repo, caller, o, current)
return uc._retrying(version, unit)
@SO.use_case(kind="command", route="POST /sales-orders/reschedule", permission=RESCHEDULE, idempotent=True)
def reschedule_orders(caller: dict, trigger: Literal["daily", "cancelled"] = "daily", cancelled_sales_order_id: Optional[Ref] = None) -> dict:
"""The re-promise (a machine's): at the start of each business day, and when an order is cancelled. Every confirmed order's open schedule
lines (not yet delivered) are re-checked against inventory's availability, rush orders first then by order date; a changed promise is a NEW
schedule version with its reason, the owner gets a task naming what moved later ("tell 7311: 120 C62 now 19 Nov"), and inventory's
reservation is re-sent for the same revision with that schedule version. An EARLIER date is applied only to quantity not yet promised to the
customer (no confirmation sent since) unless reschedule_mode is both; otherwise it is offered in a task. Once per business date and trigger: a second
run the same day writes nothing."""
key = "daily" if trigger == "daily" else f"cancelled:{cancelled_sales_order_id or ''}"
return run(trigger=key)
def run(trigger: str) -> dict:
"""Candidates come from ONE query (confirmed orders that have schedule lines; after a cancellation only those with a line of a product the cancelled
order held), are loaded a chunk at a time with one query per table, and are judged in one transaction per chunk; only an order whose promise
changed is written (its own transaction)."""
with uc.repository() as repo:
today = uc._today(repo)
products = None
freed = repo.postings.pending_repromises()
named = trigger.split(":", 1)[1] if trigger.startswith("cancelled:") else ""
if named or (freed and trigger != "daily"): # the freed stock only helps lines of the same products (one default warehouse)
products = set().union(*(repo.sales_orders.products_of(x) for x in {*freed, *([named] if named else [])}))
cands = repo.sales_orders.confirmed_scheduled(products)
order_ids = [c[0] for c in sorted(cands, key=lambda c: (c[1] != "rush", c[2] or date.min, c[0]))]
retry = sorted({json.loads(p["payload_json"])["sales_order_id"] for p in repo.postings.pending_reservations()})
owed = repo.postings.orders_with_pending_sends()
key = f"{trigger}:{today}"
sent = [oid for oid in retry if _resend(oid)]
for oid in owed: # a call a neighbour missed after the action committed (billing schedule, purchase request, draft cancel)
try:
uc._complete_sends(oid)
except (Refused, Conflict):
continue # stays pending: the next run sends it again
moved, tasks = [], 0
for at in range(0, len(order_ids), 50):
chunk = order_ids[at:at + 50]
with uc.repository() as repo, uc._decide():
loaded = repo.sales_orders.get_many(chunk)
plans = [p for p in (_plan(repo, *loaded[oid], today, key, trigger) for oid in chunk if oid in loaded) if p]
for plan in plans:
try:
res = _write(plan, today)
except Conflict:
continue # someone saved the order first: the next run sees it
if res:
moved.append(plan["o"].sales_order_id)
tasks += res
_resend(plan["o"].sales_order_id)
with uc.repository() as repo:
for oid in freed: # the cancelled orders' freed stock has now been offered to the open lines
repo.postings.add({"document_id": oid, "event": "repromise:cancelled", "journal_id": "", "status": "posted", "reason": "", "attempts": 1,
"rebate_agreement_id": "", "payload_json": json.dumps({"sales_order_id": oid})})
pending = len(repo.postings.pending_reservations())
return {"business_date": today.isoformat(), "trigger": trigger, "re_promised": moved, "tasks": tasks, "resent": sent, "reservation_pending": pending}
def _plan(repo, o, current: int, today: date, key: str, trigger: str) -> Optional[dict]:
"""One loaded order: its re-promise worked out in memory (nothing written). None when nothing changes."""
wh = promising.warehouse(repo, today)
if o.lifecycle != "confirmed" or not o.schedule or not wh or not ports.inventory_installed():
return None
if any(r.trigger_key == key for r in o.schedule):
return None # this trigger already re-promised the order today
mode = repo.settings.value_on("reschedule_mode", today) or "later_only"
told = {s.schedule_version for s in o.sends} # the schedule versions a confirmation carried to the customer
version, later, offers = o.schedule_version() + 1, [], []
reason = f"re-promised {today.isoformat()} ({trigger.split(':')[0]})"
changed = False
for ln in o.lines:
rows = o.in_force(ln.line_no)
delivered = o.delivered(ln.line_no)
if not rows or ln.quantity - delivered <= 0 or o.closed(ln):
continue
old = pm.open_chunks([pm.Chunk(r.confirmed_on, r.requested_quantity, r.confirmed_base, r.material_available_on) for r in rows], delivered)
new, why = promising.propose(repo, o, ln, today, quantity=ln.quantity - delivered, requested_on=rows[0].requested_on, exclude_own=True, wh=wh)
if why:
continue # inventory did not answer: what is promised stands, the next run asks again
merged, early = pm.merge(old, new, keep_told=rows[0].schedule_version in told and mode != "both")
tell = pm.moved_later(old, merged)
for chunks, bucket in ((tell, later), (early, offers)):
for c in chunks:
bucket.append((ln, c))
if pm.same(merged, old):
continue
o.append_schedule(version, ln.line_no, merged, requested_on=rows[0].requested_on, reason=reason, at=uc._bnow(), trigger_key=key)
changed = True
if not changed:
return None
return {"o": o, "current": current, "version": version, "later": later, "offers": offers,
"snapshot": repo.sales_orders.snapshot_of(o.sales_order_id)}
def _write(plan: dict, today: date) -> int:
"""The planned order's tasks, pending reservation and new schedule version in ONE transaction. -> the number of tasks raised."""
o, current, version, later, offers = plan["o"], plan["current"], plan["version"], plan["later"], plan["offers"]
oid = o.sales_order_id
with uc.repository() as repo, uc._decide():
made = 0
for kind, items, phrase in (("repromise", later, lambda ln, c: f"{pm_q(c.quantity)} {ln.unit_code} now {c.on.day} {c.on:%b}" if c.on else f"{pm_q(c.quantity)} {ln.unit_code} not yet confirmed"),
("offer", offers, lambda ln, c: f"{pm_q(c.quantity)} {ln.unit_code} could go on {c.on.day} {c.on:%b}")):
if not items or (kind == "offer" and repo.tasks.open_for(sales_order_id=oid, kind="offer")):
continue
many = len(o.lines) > 1
text = ("tell " if kind == "repromise" else "offer ") + f"{o.bp_id}: " + "; ".join((f"line {ln.line_no}: " if many else "") + phrase(ln, c) for ln, c in items)
repo.tasks.add(oid, kind, today, o.owner, text)
made += 1
repo.postings.add({"document_id": f"{oid}@r{o.revision}", "event": f"reserve:v{version}", "journal_id": "", "status": "pending", "reason": "",
"attempts": 0, "rebate_agreement_id": "", "payload_json": json.dumps({"sales_order_id": oid})})
repo.sales_orders.adopt(oid, plan.get("snapshot")) # the plan's load is the baseline: only the rows the re-promise changed are written (the version still guards it)
repo.sales_orders.save(o, current)
return made
def pm_q(q: Decimal) -> str:
return format(Decimal(q).normalize(), "f")
def _resend(oid: str) -> bool:
"""Re-send the reservation of an order whose schedule was committed (same revision, the version in force); settles its pending rows."""
try:
with uc.repository() as repo, uc._decide():
found = repo.sales_orders.get(oid)
if found is None:
return False
o, current = found
if o.lifecycle == "confirmed":
uc._reserve(o, promising.warehouse(repo, uc._today(repo)))
repo.sales_orders.save(o, current)
for p in repo.postings.pending_reservations():
if json.loads(p["payload_json"])["sales_order_id"] == oid:
repo.postings.settle(p["document_id"], p["event"], "posted", "", int(p.get("attempts") or 0) + 1, "")
return True
except (Refused, Conflict):
return False # stays pending with its row: the next run sends it again
@SO.use_case(kind="query", route="GET /tasks", permission=READ)
def list_tasks(caller: dict, mine: bool = True) -> dict:
"""What a person should do because the system moved a promise ("tell 7311: 120 C62 now 19 Nov"): your own open tasks, or everyone's with
so:sales_order:read_all."""
scope = uc._scope(caller, mine)
with uc.repository() as repo:
return {"tasks": uc._plain(repo.tasks.open_for(owner=scope.lower() if scope else None))}