"""The use cases — what a person (or their agent) can do with products and prices.
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 (`_sees_*`, 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).
Row visibility (module.yaml): everyone sees active products and published prices; draft products and lists, and the
standard cost, are for the catalogue and pricing roles. Cost is projected out of every answer to a caller without
product:product:write.
The price of a product is NEVER stored: `find_price` derives it on read from dated rows and rules
(domain/pricing.py), and names the list, the rule and the row it used. That is the seam quote calls (PRD-20).
"""
import json
import re
from contextlib import contextmanager
from datetime import date, datetime, timedelta, timezone
from decimal import Decimal, InvalidOperation
from fractions import Fraction
from typing import Annotated, Literal, Optional, Union
from zoneinfo import ZoneInfo
from erp_platform import auth, work
from erp_platform.bricks import Brick, Refused
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_product.application.inputs import (
AttributeIn,
BatchLineIn,
CatalogueIn,
CharValueIn,
ClassCharIn,
Code,
Currency,
DiscardIn,
Measure,
OdooProductIn,
OdooUomIn,
Pct,
PriceRowIn,
PriceRuleIn,
Quantity,
Rate,
Reason,
RuleSeq,
StepIn,
UnitIn,
day,
)
from brick_product.domain import measures, pricing, rounding, unece, units, variants
from brick_product.domain import site as site_rules
from brick_product.domain.errors import RuleBroken
from brick_product.domain.price_list import PriceList, RowIn, RuleIn, plan_changes, valid_currency
from brick_product.domain.pricing import Subject, rate
from brick_product.domain.product import CODE, Product, likely_duplicates
from brick_product.domain.values import (
CODE_PATTERN,
CONTROL_KEYS,
PriceListState,
ProductState,
RuleScope,
check_setting,
)
from brick_product.infrastructure import ai_steps, seams
from brick_product.infrastructure.tables import FOREVER
# Every decision is a PERMISSION (module.yaml permissions.catalog); no role's name is ever compared. Which people
# hold which is the composition's role catalogue — roles.yaml beside this package is the recommended default.
READ, WRITE, ARCHIVE = "product:product:read", "product:product:write", "product:product:archive"
CATEGORY_WRITE = "product:category:write"
CLASS_WRITE, SITE_WRITE = "product:class:write", "product:site:write" # v3: classify products; set a product's data for a site
LIST_READ, LIST_WRITE, PUBLISH = "product:price_list:read", "product:price_list:write", "product:price_list:publish"
CUSTOMER_WRITE, SETTINGS_WRITE = "product:customer_price_list:write", "product:settings:write"
CONTROLS = "product:settings:controls" # the controls on pricing authors (second person, unusual-price threshold)
UNIT_USED = "product:unit:mark_used" # machine: a selling brick records that a document used a unit (freezes it)
SUGGEST, CHECK = "product:product:suggest", "product:price_list:check" # machine: held by a service client only
MIGRATION_READ = "product:migration:read" # machine: purchasing, sales-order, inventory and ar resolve Odoo ids through the map
IMPORT = "product:catalogue:import" # machine: the one-off catalogue import — adds DRAFTS, never publishes, never sees cost
_EINVOICE_FIX = "give it (UN/ECE Rec 20 or 21, for example XPK for a package, XBX for a box) before a document uses it"
# The spec this build implements: erp-method modules/product v7 — v7 (PRD-41, POST /prices/batch) signed 2026-10-25; was a DRAFT for
# sign-off (spec branch spec/product-batch-price, 97f5444).
SPEC_VERSION = 7
PRODUCT = Brick("product", version="0.2.2", ids_as_strings=True, machine_permissions=(SUGGEST, CHECK, IMPORT, UNIT_USED, MIGRATION_READ), instructions=(
"Products & price lists: what the business sells, with its unit, how it is billed and taxed, and the prices each "
"customer pays. Ask find_price for the price of a product (quantity, customer, currency, date): it is derived "
"from dated rows and rules, names the list, rule and row it used, and never shows cost. A price row is never "
"edited: a different price starts on a later date, so a document priced last year re-prices identically. "
"Orders and invoices copy a product's description, unit, tax category and invoicing policy onto the line. "
"Archiving a product never changes a document that already names it."))
# The process view (BRICKS.md): the signed workflows as BPMN, GET /process/bpmn.
PRODUCT.declare_process("process.yaml")
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 (a GTIN, a code-series number) can surface at COMMIT (the loser's
COMMIT raises 23505), after the use case has returned: 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."""
from brick_product.infrastructure.tables import Repository
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
@contextmanager
def _repository_default():
from erp_platform import db # imported late: it needs BUCKET
with _on_engine(db.engine) as repo:
yield repo
repository = _repository_default # tests and a composition swap in their own
@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 _who(caller: dict) -> str:
"""The caller as a stored author is: their email, lower-case, else their sub."""
return (caller.get("email") or caller.get("sub") or "").strip().lower()
def _today(repo) -> date:
"""Business today: the date NOW (work.now(): a demo may move the clock) in the setting business_timezone — "after
today" means after this date, everywhere a rule says it."""
return work.now().astimezone(ZoneInfo(repo.settings.current().business_timezone)).date()
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 (these writes are keyed and repeat safely). 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 _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")
# ── who sees what ─────────────────────────────────────────────────────────────────────────────
def _sees_drafts(caller: dict) -> bool:
"""Draft and archived products, draft and retired lists: the catalogue and pricing roles (and the system's AI
steps). Sales sees what is offered."""
return any(auth.has_permission(caller, p) for p in (WRITE, ARCHIVE, LIST_WRITE, PUBLISH, SUGGEST, CHECK, CONTROLS))
def _sees_cost(caller: dict) -> bool:
return auth.has_permission(caller, WRITE)
def _manages_prices(caller: dict) -> bool:
return any(auth.has_permission(caller, p) for p in (LIST_WRITE, PUBLISH, CUSTOMER_WRITE, CHECK, CONTROLS))
def _product(repo, caller: dict, product_id: str) -> tuple:
found = repo.products.get(product_id)
if found is None or (not found[0].offered() and not _sees_drafts(caller)):
raise Refused(404, f"no product {product_id}")
return found
def _list(repo, caller: dict, price_list_id: str) -> tuple:
found = repo.price_lists.get(price_list_id)
if found is None or (found[0].status != PriceListState.PUBLISHED and not _manages_prices(caller)):
raise Refused(404, f"no price list {price_list_id}")
return found
# ── how things read ───────────────────────────────────────────────────────────────────────────
def _product_view(repo, agg: Product, version: int, caller: dict) -> dict:
out = {f: getattr(agg, f) for f in ("product_id", "name", "description", "kind", "base_unit_code", "sales_unit_code",
"purchase_unit_code", "issue_unit_code", "category_code",
"tax_category", "billing_kind", "billing_plan", "invoicing_policy", "status")}
if _sees_cost(caller):
out["cost_amount"] = rate(agg.cost_amount)
if _sees_drafts(caller):
out["history"] = [{**h, "event_at": h["event_at"].isoformat() if h["event_at"] else None}
for h in agg.history]
out["product_template_id"] = agg.product_template_id or agg.product_id # no variants: it is its own template
out["attribute_values"] = json.loads(agg.attribute_values or "{}")
rows = repo.memberships.rows_of(f"product:{agg.product_id}")
if rows: # the category TODAY; a move that starts later is shown beside it
today = _today(repo)
out["category_code"] = repo.memberships.product_category(agg.product_id, today)
later = next(((d, c) for d, c in rows if d > today), None)
if later:
out["category_next"] = {"category_code": later[1], "valid_from": later[0]}
out["product_type_code"] = agg.product_type_code or None
out["version"] = version
return out
def _list_view(repo, agg: PriceList, version: int, caller: dict, contents: Optional[dict] = None) -> dict:
"""A list as the caller may see it. `contents` is passed by a write that has just changed the rows: a read inside
the open transaction would not see its own writes (DataK3 stages them until commit), so it works the counts out."""
out = {f: getattr(agg, f) for f in ("price_list_id", "name", "currency", "based_on", "prices_include_tax",
"rounding", "default_from", "status", "retired_on")}
if _manages_prices(caller):
out["opened_by"] = agg.opened_by
out["contents"] = contents if contents is not None else repo.prices.count(agg.price_list_id)
out["customers"] = repo.assignments.of_list(agg.price_list_id)
out["publish_warnings"] = json.loads(agg.publish_notes or "[]")
out["version"] = version
return out
def _row_view(r: pricing.PriceRow) -> dict:
return {"price_list_id": r.price_list_id, "product_id": r.product_id, "unit_code": r.unit_code,
"valid_from": r.valid_from, "unit_price": rate(r.unit_price), "valid_to": r.valid_to,
"pricing_quantity": r.pricing_quantity}
def _rule_view(r: pricing.PriceRule) -> dict:
return {"price_list_id": r.price_list_id, "seq": r.seq, "applies_to": r.applies_to, "target": r.target,
"min_quantity": r.min_quantity, "valid_from": r.valid_from, "valid_to": r.valid_to, "method": r.method,
"discount_pct": r.discount_pct, "unit_code": r.unit_code, "pricing_quantity": r.pricing_quantity, "fixed_price": None if r.fixed_price is None else rate(r.fixed_price),
"surcharge_amount": rate(r.surcharge_amount or 0), "base_list": r.base_list}
def _trail(r: pricing.Resolved) -> dict:
"""One step of how a price was made: the list, the rule or row, and what it started from."""
return {"price_list_id": r.price_list_id, "via": r.via, "unit_price": r.unit_price,
"rule": _rule_view(r.rule) if r.rule else None, "price_row": _row_view(r.row) if r.row else None,
"note": r.note, "base": _trail(r.base) if r.base else None}
def _deepest_row(r: pricing.Resolved) -> Optional[pricing.PriceRow]:
while r is not None:
if r.row is not None:
return r.row
r = r.base
return None
# ── references the spec's rows must resolve ───────────────────────────────────────────────────
def _check_refs(repo, *, unit_code=None, category_code=None) -> None:
if unit_code and unit_code not in repo.units.codes():
raise Refused(422, f"unit '{unit_code}' is not in the unit list (UN/ECE Recommendation 20 codes: C62, HUR, DAY, "
"MON, KGM …) — add it in the settings first")
if category_code and repo.categories.get(category_code) is None:
raise Refused(422, f"category '{category_code}' does not exist — add it in the settings first")
def _unit_factors(repo, product_id: str, info: dict) -> dict:
"""unit_code -> how many base units one of it is, for the units the product (or its template) names itself."""
return {u: Fraction(r["numerator"], r["denominator"])
for u, r in repo.product_units.effective(product_id, info.get("template", "")).items()}
def _factor(repo, product_id: str, info: dict, unit_code: str) -> Optional[Fraction]:
"""How many base units one `unit_code` is for this product — exact — or None: it cannot be counted in it."""
return units.factor_of(unit_code, info["base"], _unit_factors(repo, product_id, info), repo.units.fixed())
def _subject(repo, product_id: str, on: date, fallback: str = "", unit_code: str = "") -> Subject:
"""The product, its categories ON THAT DATE (a move starts tomorrow, so today and earlier keep their old answers),
its template and the unit asked, with the exact factors a price is converted by. `fallback`: its stored category, for
a product not yet placed (a draft)."""
cat = repo.memberships.product_category(product_id, on) or (fallback if not repo.memberships.rows_of(f"product:{product_id}") else "")
info = repo.products.info([product_id]).get(product_id) or {"base": "", "sales": "", "template": ""}
unit = unit_code or info["sales"] or info["base"]
own, fixed = _unit_factors(repo, product_id, info), repo.units.fixed()
factor = units.factor_of(unit, info["base"], own, fixed) or Fraction(1)
every = {info["base"]: Fraction(1), **own}
for (frm, to), f in fixed.items(): # the conversions true for everyone, either way round
if to == info["base"]:
every.setdefault(frm, f)
elif frm == info["base"]:
every.setdefault(to, 1 / f)
return Subject(product_id, repo.memberships.path(cat, on), info["template"] or product_id, unit, info["base"], factor, every)
def _move(repo, subject: str, category_code: str, caller: dict) -> None:
"""Place a product (or re-parent a category) from TOMORROW: a new row, never an edit, so no price asked for today or
earlier changes. The same placement again is nothing; a different one for the same day is refused."""
start = _today(repo) + timedelta(days=1)
latest = repo.memberships.rows_of(subject)
if latest and latest[-1][1] == category_code:
return
if any(d == start for d, _ in latest):
raise Refused(409, f"{subject.split(':', 1)[1]} is already moving to another category from {start}: a move is "
"never edited — make the next one after it takes effect")
repo.memberships.add(subject, start, category_code, _who(caller))
# ═══ products ════════════════════════════════════════════════════════════════════════════════
@PRODUCT.use_case(kind="query", route="GET /products", permission=READ)
def list_products(caller: dict, q: Optional[Name] = None, category_code: Optional[Code] = None,
status: Optional[Code] = None, limit: Limit = 50, after: Optional[Code] = None,
char: Optional[list[text(200, min_length=1)]] = None, class_code: Optional[Code] = None,
on: Optional[IsoDate] = None) -> dict:
"""Search the product list, by code, name or description (`q`), by category (and everything under it) and by
status, ordered by code. `after` is the last code of the previous page. Sales sees active products only; the
catalogue and pricing roles also see drafts and archived ones. Cost is never in a list. `char` (repeat it) and `class_code` search
by what a product is described by (PRD-36): `char=MATERIAL:=A2`, `char=LENGTH:>=20 MMT`, `char=THREAD:in M3,M4` — a number is
compared after an exact conversion to the characteristic's unit, by the values in force on `on` (default today); each product
then carries its `class_code` and `values`."""
with repository() as repo:
if char or class_code:
return _search_by_characteristics(repo, caller, q, category_code, status, limit, after, char or [], class_code, _date(on, "on"))
statuses = None if _sees_drafts(caller) else [ProductState.ACTIVE.value]
if status:
statuses = [status] if statuses is None or status in statuses else []
cats = repo.memberships.under(category_code, _today(repo)) if category_code else None
rows = repo.products.search(text=q or "", category=cats, statuses=statuses, limit=limit,
after=after or "", in_category_on=_today(repo)) if statuses != [] else []
return {"products": rows, "next": rows[-1]["product_id"] if len(rows) == limit else None}
def _search_by_characteristics(repo, caller, q, category_code, status, limit, after, char, class_code, on) -> dict:
asked_on = on or _today(repo)
statuses = None if _sees_drafts(caller) else [ProductState.ACTIVE.value]
if status:
statuses = [status] if statuses is None or status in statuses else []
if statuses == []:
return {"products": [], "next": None}
filters = []
for raw in char:
code, op, wanted = _parse_char(raw)
c = repo.classification.characteristic(code)
if c is None:
raise Refused(422, f"no characteristic '{code}'")
if c["data_type"] == "number":
_number_in(repo, wanted.split(",")[0], None, c) # a malformed number is a 422/409 now, not an empty answer
filters.append((c, op, wanted))
cats = repo.memberships.under(category_code, asked_on) if category_code else None
# SQL narrows to the products that (or whose template) have a row for every asked characteristic; the exact test (dates, units,
# operators) then runs on a window of 200 candidates at a time, until the page is full. Never every product in memory.
owners = repo.classification.owners_with([c["characteristic_code"] for c, _op, _w in filters], class_code)
found, cursor = [], after or ""
while owners and len(found) < limit:
cands = repo.products.search(text=q or "", category=cats, statuses=statuses, limit=200, after=cursor, in_category_on=asked_on,
owners=owners)
if not cands:
break
cursor = cands[-1]["product_id"]
infos = repo.products.info([c["product_id"] for c in cands])
by_owner: dict = {}
for r in repo.classification.rows([c["product_id"] for c in cands] + [i["template"] for i in infos.values()]):
by_owner.setdefault(r["owner"], []).append(r)
for cand in cands:
pid, tmpl = cand["product_id"], infos.get(cand["product_id"], {}).get("template", "")
classes, values = _class_state(by_owner.get(pid, []) + by_owner.get(tmpl, []), pid, tmpl, asked_on)
if not classes or (class_code and class_code not in classes):
continue
flat = _flat({c: v for c, v in values.items() if not class_code or c == class_code}, classes)
if all(c["characteristic_code"] in flat and _char_matches(repo, c, flat[c["characteristic_code"]], op, w) for c, op, w in filters):
found.append({"product_id": pid, "name": cand["name"], "class_code": class_code or sorted(classes)[0],
"values": {ch: _show(repo, ch, v) for ch, v in flat.items()}})
if len(found) == limit:
break
if len(cands) < 200:
break
return {"products": found, "next": found[-1]["product_id"] if len(found) == limit else None}
@PRODUCT.use_case(kind="query", route="GET /products/{product_id}", permission=READ)
def get_product(caller: dict, product_id: Code, on: Optional[IsoDate] = None) -> dict:
"""One product, whole, with its version. This is what an order or invoice line COPIES when it names the product
(description, the unit and its conversion, tax_category, billing_kind, invoicing_policy) — it never asks again. The standard cost
is shown only to those who keep the catalogue; a draft or archived product is a 404 to sales. `on` (default today) is the date
the dated parts are read for (v4): each role (`roles`) and, per unit, its measures and GTINs, as they were in force then —
`units` lists the base unit first, each with its exact conversion, decimals, GTINs (main or additional) and measures (null
while never set, never 0). A document copies what it reads and never asks again."""
with repository() as repo:
agg, version = _product(repo, caller, product_id)
asked_on = _date(on, "on") or _today(repo)
out = _product_view(repo, agg, version, caller)
roles, defaulted = _roles_on(repo, agg, asked_on)
out.update({"lifecycle": agg.status, "on": asked_on, "roles": roles, "roles_defaulted": defaulted,
"units": _unit_answers(repo, agg, asked_on), "product_type_code": agg.product_type_code or None,
"stocked_default": agg.stocked_default, "valuation_class": agg.valuation_class or (_type_of(repo, agg) or {}).get("valuation_class") or None,
"commodity_code": agg.commodity_code or None, "country_of_origin": agg.country_of_origin or None})
classes, values = _class_state(repo.classification.rows([product_id, _template_of(agg)]), product_id, _template_of(agg), asked_on)
out["classes"] = sorted(classes)
out["characteristics"] = {ch: _show(repo, ch, v) for ch, v in _flat(values, classes).items()}
if _sees_cost(caller):
out["open_tasks"] = [t["task_id"] for t in repo.tasks.open_for(f"prod_product:{product_id}")]
return out
def _duplicate_check(repo, caller: dict, agg: Product, version: int, settings) -> dict:
"""PRD-12: the fixed baseline (words in common) plus the ai.suggest step; both only SUGGEST. A person sees the
likely duplicates and may carry on; nothing is merged."""
mode = settings.mode("ai_check_duplicate")
if mode == "off":
return {"likely_duplicates": [], "ai": {"step": "ai_check_duplicate", "mode": "off", "kind": "off"}}
found = likely_duplicates(agg.name, agg.description, repo.products.texts(exclude=agg.product_id))
context = {"name": agg.name, "description": agg.description or "", "base_unit_code": agg.base_unit_code or "",
**{f"candidate_{i + 1}": f"{c['product_id']} | {c['name']} | similarity {c['score']}"
for i, c in enumerate(found)}}
answer = ai_steps.ask("ai_check_duplicate", mode, f"prod_product:{agg.product_id}", version, ["duplicates"], context)
task_id = _keep_task(repo, answer, f"prod_product:{agg.product_id}", version, context, _who(caller),
baseline=found, only_if=bool(found))
return {"likely_duplicates": found, "ai": answer.as_dict(), "task_id": task_id}
def _keep_task(repo, answer, subject_ref: str, version: int, context: dict, by: str, *, baseline=None,
only_if: bool = True) -> Optional[str]:
"""A proposal waits for a person as a task: the ManualTask when no provider answered, or the provider's draft."""
if answer.kind in ("off", "pending") or (answer.kind == "manual" and not only_if):
return None
key = f"{answer.step}|{subject_ref}|{version}"
task = answer.task
repo.tasks.open({
"task_id": key, "step": answer.step, "subject_ref": subject_ref, "capability": "ai.suggest",
"input_json": json.dumps(context, sort_keys=True), "plan": task.plan if task else "",
"reason": task.reason if task else "", "destructive": False,
"summary": task.summary if task else f"{answer.step}: {answer.rationale}"[:300],
"output_json": json.dumps({"baseline": baseline or [], "suggestion": answer.suggestion,
"confidence": answer.confidence}, default=str), "created_by": by})
return key
@PRODUCT.use_case(kind="command", route="POST /products", permission=WRITE)
def add_product(caller: dict, name: Name, product_id: Optional[Code] = None, description: Optional[Note] = None,
kind: Optional[Code] = None, base_unit_code: Optional[Code] = None, category_code: Optional[Code] = None,
tax_category: Optional[Code] = None, billing_kind: Optional[Code] = None,
billing_plan: Optional[Code] = None, invoicing_policy: Optional[Code] = None,
cost_amount: Optional[Rate] = None, product_type_code: Optional[Code] = None,
commodity_code: Optional[text(8)] = None, country_of_origin: Optional[text(2)] = None) -> dict:
"""Add a product once, with its code (a SKU), unit and how it is billed and taxed (PRD-01). It starts as a draft
and is offered to sales when it is complete (activate_product). The code is fixed from now on; a second product
with the same code is refused. If the settings give a code pattern such as PRD-{seq:5}, leave `product_id` out
to get the next one. `product_type_code` is REQUIRED (PRD-35: TRADING, FINISHED, RAW, SERVICE, NONSTOCK, PACKAGING or your own;
a missing or unknown one is a 422 listing the types). The type's kind, stocked default, sellable, valuation class and required unit
roles are COPIED onto the product now (a later edit of the type reaches no existing product); a type with a code format such as TRD-{seq:5}
numbers the product from its own series when `product_id` is left out. The answer also lists likely duplicates (PRD-12) —
nothing is merged."""
return retry_on_conflict(lambda: _add_product(caller, name, product_id, description, kind, base_unit_code, category_code, tax_category,
billing_kind, billing_plan, invoicing_policy, cost_amount, product_type_code,
commodity_code, country_of_origin))
def _add_product(caller, name, product_id, description, kind, base_unit_code, category_code, tax_category, billing_kind, billing_plan,
invoicing_policy, cost_amount, product_type_code, commodity_code, country_of_origin) -> dict:
with repository() as repo, _decide():
settings = repo.settings.current()
fmt = settings.product_code_format
ptype = _need_type(repo, product_type_code)
series = None
if ptype["code_format"] != "typed":
fmt = ptype["code_format"]
if kind and kind != ptype["kind"]:
raise Refused(422, f"{ptype['product_type_code']} products are {ptype['kind']}, not {kind}: leave kind out, or pick another type")
if fmt == "typed":
if not product_id:
raise Refused(422, "give the product code (the settings ask for codes to be typed)")
code = product_id
elif product_id:
code = product_id
if not _fits(fmt, code):
raise Refused(422, f"the code {code} does not follow the pattern {fmt}")
elif fmt == ptype["code_format"]:
m = CODE_PATTERN.match(fmt)
series = repo.types.next_number(ptype["product_type_code"], m["pre"], m["post"])
code = f"{m['pre']}{series:0{int(m['width'])}d}{m['post']}"
else:
m = CODE_PATTERN.match(fmt)
code = repo.products.next_code(m["pre"], m["post"], int(m["width"]))
if repo.templates.get(code.strip()):
raise Refused(409, f"{code.strip()} is a product template: a product code is never a template's")
clash = repo.products.code_in_use(code.strip())
if clash:
raise Refused(409, f"{clash} already exists" if clash == code.strip() else
f"{clash} already exists (codes are the same ignoring case: {code})")
_check_refs(repo, unit_code=base_unit_code, category_code=category_code)
agg = Product.add_product(code, by=_who(caller), name=name, description=description or "", kind=kind,
base_unit_code=base_unit_code, category_code=category_code, tax_category=tax_category,
billing_kind=billing_kind, billing_plan=billing_plan,
invoicing_policy=invoicing_policy, cost_amount=cost_amount,
type_row=ptype, commodity_code=commodity_code, country_of_origin=country_of_origin)
if series is not None: # claimed in THIS transaction: a refusal above (or below) rolls the number back, no gap
repo.types.claim_number(ptype["product_type_code"], series, agg.product_id, datetime.now(timezone.utc).replace(tzinfo=None))
version = repo.products.save(agg, 0)
return {**_product_view(repo, agg, version, caller), **_duplicate_check(repo, caller, agg, version, settings)}
def _fits(fmt: str, code: str) -> bool:
import re
m = CODE_PATTERN.match(fmt)
return bool(m and re.fullmatch(re.escape(m["pre"]) + r"\d{%d,}" % int(m["width"]) + re.escape(m["post"]), code))
@PRODUCT.use_case(kind="command", route="POST /products/{product_id}/ai-check-duplicate", permission=SUGGEST,
if_match="version")
def ai_check_duplicate(caller: dict, product_id: Code, version: Optional[int] = None) -> dict:
"""The system checks a DRAFT product for a likely duplicate (suggest mode; PRD-12). It lists up to three
products that look the same, and leaves a task for the catalogue manager; nothing is merged and the product
does not change. Switched off in the settings, it refuses."""
with repository() as repo, _decide():
agg, current = _product(repo, caller, product_id)
_same_version(f"product {product_id}", current, version)
settings = repo.settings.current()
agg.ai_check_duplicate(settings.mode("ai_check_duplicate"), by=_who(caller))
return {**_product_view(repo, agg, current, caller), **_duplicate_check(repo, caller, agg, current, settings)}
@PRODUCT.use_case(kind="command", route="POST /products/{product_id}/ai-suggest-details", permission=SUGGEST,
if_match="version")
def ai_suggest_details(caller: dict, product_id: Code, version: Optional[int] = None) -> dict:
"""The system proposes a DRAFT product's empty category, unit and tax category from its name (suggest mode).
The proposal waits as a task for the catalogue manager, who accepts it by completing the product (complete_product)
or changes it; nothing is filled in on its own."""
with repository() as repo, _decide():
agg, current = _product(repo, caller, product_id)
_same_version(f"product {product_id}", current, version)
settings = repo.settings.current()
agg.ai_suggest_details(settings.mode("ai_suggest_details"), by=_who(caller))
empty = [f for f in ("category_code", "base_unit_code", "tax_category") if not getattr(agg, f)]
context = {"name": agg.name, "description": agg.description or "",
"categories": ", ".join(c["category_code"] for c in repo.categories.all()),
"units": ", ".join(sorted(repo.units.codes()))}
answer = ai_steps.ask("ai_suggest_details", settings.mode("ai_suggest_details"),
f"prod_product:{product_id}", current, empty, context)
task_id = _keep_task(repo, answer, f"prod_product:{product_id}", current, context, _who(caller))
return {**_product_view(repo, agg, current, caller), "fields_to_suggest": empty, "ai": answer.as_dict(),
"task_id": task_id}
@PRODUCT.use_case(kind="command", route="POST /products/{product_id}/complete-product", permission=WRITE,
if_match="version")
def complete_product(caller: dict, product_id: Code, version: Optional[int] = None, kind: Optional[Code] = None,
base_unit_code: Optional[Code] = None, category_code: Optional[Code] = None,
tax_category: Optional[Code] = None, billing_kind: Optional[Code] = None,
billing_plan: Optional[Code] = None, invoicing_policy: Optional[Code] = None,
cost_amount: Optional[Rate] = None, commodity_code: Optional[text(8)] = None,
country_of_origin: Optional[text(2)] = None) -> dict:
"""Complete a DRAFT product: its unit (UN/ECE Rec 20 code), category, tax category (a category such as
'standard', never a tax code or rate), how it is billed (billing_kind one_off | recurring | metered, billing_plan
single | periodic | arrears) and its invoicing policy (ordered | delivered). Fields left out keep their value."""
with repository() as repo, _decide():
agg, current = _product(repo, caller, product_id)
_same_version(f"product {product_id}", current, version)
_check_refs(repo, unit_code=base_unit_code, category_code=category_code)
if base_unit_code and base_unit_code != agg.base_unit_code and repo.product_units.of(product_id):
raise Refused(409, f"{product_id} has units expressed in {agg.base_unit_code}: its base unit is fixed")
agg.complete_product(by=_who(caller), kind=kind, base_unit_code=base_unit_code, category_code=category_code,
tax_category=tax_category, billing_kind=billing_kind, billing_plan=billing_plan,
invoicing_policy=invoicing_policy, cost_amount=cost_amount, commodity_code=commodity_code,
country_of_origin=country_of_origin)
return _product_view(repo, agg, repo.products.save(agg, current), caller)
@PRODUCT.use_case(kind="command", route="POST /products/{product_id}/activate-product", permission=WRITE,
idempotent=True, if_match="version")
def activate_product(caller: dict, product_id: Code, version: Optional[int] = None) -> dict:
"""Offer a draft product to sales (PRD-02). Refused, naming what is missing, unless it has a unit, a tax
category, a billing type and an invoicing policy. A repeat changes nothing."""
with repository() as repo, _decide():
agg, current = _product(repo, caller, product_id)
_same_version(f"product {product_id}", current, version)
if agg.status == ProductState.DRAFT:
_activation_gates(repo, agg)
if agg.activate_product(by=_who(caller)) is None:
return _product_view(repo, agg, current, caller)
if agg.category_code and not repo.memberships.rows_of(f"product:{product_id}"):
repo.memberships.add(f"product:{product_id}", FOREVER, agg.category_code, _who(caller)) # its first placement
return _product_view(repo, agg, repo.products.save(agg, current), caller)
@PRODUCT.use_case(kind="command", route="POST /products/{product_id}/rename-product", permission=WRITE,
if_match="version")
def rename_product(caller: dict, product_id: Code, version: Optional[int] = None, name: Optional[Name] = None,
description: Optional[Note] = None, category_code: Optional[Code] = None,
cost_amount: Optional[Rate] = None) -> dict:
"""Change an ACTIVE product's name, description, category (PRD-04) or standard cost. A new category starts TOMORROW:
price rules on it apply to the product from then, and a price asked for today or earlier still uses the old one.
Documents already raised keep their own copy. Its code, unit, tax category and billing are fixed once it is on sale.
A new name or description is checked for duplicates again."""
with repository() as repo, _decide():
agg, current = _product(repo, caller, product_id)
_same_version(f"product {product_id}", current, version)
_check_refs(repo, category_code=category_code)
before = (agg.name, agg.description)
agg.rename_product(by=_who(caller), name=name, description=description, category_code=category_code,
cost_amount=cost_amount)
if category_code:
_move(repo, f"product:{product_id}", category_code, caller)
saved = repo.products.save(agg, current)
out = _product_view(repo, agg, saved, caller)
if (agg.name, agg.description) != before:
out |= _duplicate_check(repo, caller, agg, saved, repo.settings.current())
return out
@PRODUCT.use_case(kind="command", route="POST /products/{product_id}/archive-product", permission=ARCHIVE,
idempotent=True, if_match="version")
def archive_product(caller: dict, product_id: Code, version: Optional[int] = None) -> dict:
"""Take an active product off sale (PRD-03): it is no longer offered for new lines and find_price refuses it.
Invoices, orders and quotes that already name it keep their own copy, unchanged — nothing is deleted. It can
be reactivated, with a reason. A repeat changes nothing."""
with repository() as repo, _decide():
agg, current = _product(repo, caller, product_id)
_same_version(f"product {product_id}", current, version)
if agg.archive_product(by=_who(caller)) is None:
return _product_view(repo, agg, current, caller)
return _product_view(repo, agg, repo.products.save(agg, current), caller)
@PRODUCT.use_case(kind="command", route="POST /products/{product_id}/reactivate-product", permission=WRITE,
idempotent=True, if_match="version")
def reactivate_product(caller: dict, product_id: Code, reason: Reason, version: Optional[int] = None) -> dict:
"""Put an archived product back on sale. A reason is required and is recorded, with who and when, in the
product's history. A repeat changes nothing."""
with repository() as repo, _decide():
agg, current = _product(repo, caller, product_id)
_same_version(f"product {product_id}", current, version)
if agg.reactivate_product(reason=reason, by=_who(caller)) is None:
return _product_view(repo, agg, current, caller)
return _product_view(repo, agg, repo.products.save(agg, current), caller)
# ── units of measure (PRD-23): exact conversions to the base unit ─────────────────────────────
def _unit_ratio(repo, product_id: str, info: dict, u: UnitIn, known: dict) -> Optional[Fraction]:
"""The exact ratio a UnitIn states (base units per one of it), or None when it states none (a conversion true for
every product). `known`: unit -> Fraction already settled for this product, so a pack can be made of bottles."""
if u.numerator is not None or u.denominator is not None:
if u.numerator is None or u.denominator is None:
raise Refused(422, f"{u.unit_code}: give both the numerator and the denominator")
stated = units.ratio(u.numerator, u.denominator, u.unit_code)
if u.relative_unit_code:
inner = known.get(u.relative_unit_code) or units.factor_of(u.relative_unit_code, info["base"], known, repo.units.fixed())
if inner is None:
raise Refused(422, f"{u.unit_code} is stated in {u.relative_unit_code}, which has no conversion to the base unit yet")
return stated * inner
return stated
if u.contains_unit:
if u.contains_quantity is None or u.contains_quantity <= 0:
raise Refused(422, f"{u.unit_code}: say how many {u.contains_unit} it contains")
inner = known.get(u.contains_unit) or units.factor_of(u.contains_unit, info["base"], known, repo.units.fixed())
if inner is None:
raise Refused(422, f"{u.unit_code} is made of {u.contains_unit}, which has no conversion to the base unit yet")
return units.exact(u.contains_quantity) * inner
return None
def _store_units(repo, product_id: str, owner: str, info: dict, unit_rows, by: str, template_owner: bool = False) -> dict:
"""Write the unit rows asked for, in order (a later one may be made of an earlier one). Returns unit -> Fraction for
every unit the product now has. A row already used by a document or stock is never changed: a changed pack size is a
new unit code."""
have = repo.product_units.of(owner) if template_owner else repo.product_units.effective(product_id, info.get("template", ""), info["base"])
known = {u: Fraction(r["numerator"], r["denominator"]) for u, r in have.items() if not r["relative_unit_code"]}
fixed = repo.units.fixed()
for u in unit_rows:
if not re.fullmatch(r"[A-Z0-9][A-Z0-9._-]{0,15}", u.unit_code):
raise Refused(422, f"a unit code is capital letters, digits and . _ - (up to 16): a UN/ECE Recommendation 20 code "
f"such as C62, or your own pack code such as PK24 — not '{u.unit_code}'")
if u.unit_code == info["base"]:
raise Refused(409, f"{u.unit_code} is the base unit of {product_id}: it is 1 by definition")
relative = ""
inner_code = u.contains_unit or u.relative_unit_code
if template_owner and inner_code and inner_code != info["base"] and units.factor_of(inner_code, info["base"], known, fixed) is None:
# PRD-34: a template states a pack in a unit only its variants have (the bottle differs per SKU): kept RELATIVE, composed
# exactly with each variant's own row for that unit when the variant reads it.
relative = inner_code
if u.contains_unit:
if u.contains_quantity is None or u.contains_quantity <= 0:
raise Refused(422, f"{u.unit_code}: say how many {u.contains_unit} it contains")
f = units.exact(u.contains_quantity)
elif u.numerator is not None and u.denominator is not None:
f = units.ratio(u.numerator, u.denominator, u.unit_code)
else:
raise Refused(422, f"{u.unit_code}: give how many {inner_code} it is")
if not 0 < f.numerator <= units.BIG or f.denominator > units.BIG:
raise Refused(422, f"{u.unit_code}: the conversion is too large")
else:
f = _unit_ratio(repo, product_id, info, u, known)
if u.einvoice_unit_code and not unece.is_einvoice_code(u.einvoice_unit_code):
raise Refused(422, f"{u.unit_code}: '{u.einvoice_unit_code}' is not a UN/ECE Recommendation 20 or 21 code "
"(XPK package, XBX box, XPX pallet, KGM kilogram …)")
if f is None:
if units.factor_of(u.unit_code, info["base"], {}, fixed) is None:
raise Refused(422, f"{u.unit_code}: give its exact conversion to {info['base']} (a numerator and a "
"denominator, or how many of another unit it contains)")
continue # true for every product: no product row, ever
if units.factor_of(u.unit_code, info["base"], {}, fixed) is not None:
raise Refused(409, f"{u.unit_code} converts to {info['base']} the same way for every product: it has no row of its own")
try:
if not relative:
units.check_pack_code(u.unit_code, info["base"], f)
except RuleBroken as e:
raise Refused(409, str(e)) from None
old = have.get(u.unit_code)
gtin = variants.gtin14(u.gtin) if u.gtin else (old or {}).get("gtin", "")
if u.gtin and owner != product_id:
raise Refused(422, f"{u.unit_code}: a GTIN names one product's packaging level — give it on the product, not on its template")
einvoice = (u.einvoice_unit_code or (old or {}).get("einvoice_unit_code")
or (u.unit_code if unece.is_einvoice_code(u.unit_code) else ""))
row = {"unit_code": u.unit_code, "numerator": f.numerator, "denominator": f.denominator, "gtin": gtin,
"einvoice_unit_code": einvoice, "relative_unit_code": relative,
"packaging_level": u.packaging_level or (old or {}).get("packaging_level") or "none",
"quantity_decimals": u.quantity_decimals if u.quantity_decimals is not None
else (old["quantity_decimals"] if old else _unit_decimals(repo, u.unit_code))}
if old is not None:
same = (Fraction(old["numerator"], old["denominator"]) == f and (old.get("relative_unit_code") or "") == relative
and old["packaging_level"] == row["packaging_level"]
and old["quantity_decimals"] == row["quantity_decimals"]
and (old.get("einvoice_unit_code") or einvoice) == einvoice)
if old["used_at"] is not None and not same and template_owner:
raise Refused(409, f"{u.unit_code} on template {old['owner']} is used by {old.get('used_ref') or 'a document or stock'}: its "
"conversion, packaging level and decimals are fixed — add the changed pack as a NEW unit code "
"(for example PK12 beside PK24)")
if old["used_at"] is not None and not same:
where = "" if old["owner"] == owner else f" (through its template {old['owner']})"
raise Refused(409, f"{u.unit_code} of {product_id} is already used by a document or stock{where}, so its conversion, "
"packaging level, decimals and e-invoice code are fixed: add the changed pack as a NEW unit "
"code (for example PK12 beside PK24)")
if same and (old["owner"] != owner or (old.get("gtin") or "") == (gtin or "")) \
and (old.get("einvoice_unit_code") or "") == (einvoice or ""):
if not relative:
known[u.unit_code] = f
continue # the same row again (or the template's own, unchanged): nothing
_claim_gtin(repo, owner, u.unit_code, old, gtin, by)
repo.product_units.put(owner, row, by)
if not relative:
known[u.unit_code] = f
return known
def _claim_gtin(repo, owner: str, unit_code: str, old: Optional[dict], gtin: str, by: str) -> None:
"""A GTIN given with the unit (set_units / add_unit) is that unit's MAIN barcode: claimed by a plain insert, unique across every
product and unit (a barcode someone else holds is a 409). The old main, if any, stays as an additional barcode."""
if not gtin or gtin == (old or {}).get("gtin", "") or (old or {}).get("gtin", "").zfill(14) == gtin:
return
held = repo.gtins.holder(gtin)
if held is not None and (held["owner"], held["unit_code"]) != (owner, unit_code):
raise Refused(409, f"{gtin} is {held['owner']} {held['unit_code']}: a barcode names one product and unit for ever")
if old and old.get("gtin") and old["used_at"] is not None:
raise Refused(409, f"{unit_code}'s GTIN {old['gtin']} is already on documents or stock: a changed main barcode is a new unit code "
"(add_gtin gives a used unit another one)")
if held is None:
try:
repo.gtins.claim(gtin, owner, unit_code, by, "additional")
except IntegrityError:
raise Refused(409, f"GTIN {gtin} was claimed by someone else a moment ago: a barcode names one product and unit") from None
if held is None or held["status"] == "active":
_promote(repo, owner, unit_code, gtin, by)
def _unit_decimals(repo, unit_code: str) -> int:
return next((u["decimals"] or 0 for u in repo.units.all() if u["unit_code"] == unit_code), 0)
def _role_units_convert(repo, agg: Product, known: dict, info: dict) -> None:
for role in ("sales_unit_code", "purchase_unit_code", "issue_unit_code"):
unit = getattr(agg, role)
if unit and units.factor_of(unit, agg.base_unit_code, known, repo.units.fixed()) is None:
raise RuleBroken(f"{unit} does not convert to {agg.base_unit_code}, the base unit of {agg.product_id}: give "
"its exact conversion (a numerator and a denominator) first")
@PRODUCT.use_case(kind="command", route="POST /products/{product_id}/set-units", permission=WRITE, if_match="version")
def set_units(caller: dict, product_id: Code, version: Optional[int] = None, base_unit_code: Optional[Code] = None,
sales_unit_code: Optional[Code] = None, purchase_unit_code: Optional[Code] = None,
issue_unit_code: Optional[Code] = None, production_unit_code: Optional[Code] = None,
warehouse_unit_code: Optional[Code] = None, delivery_unit_code: Optional[Code] = None,
units_of: Optional[list[UnitIn]] = None) -> dict:
"""Set a DRAFT product's units (PRD-23): the BASE unit it is stocked in, the SALES unit a price is asked in, the
PURCHASE and ISSUE units, and the other units it can be counted in (`units_of`: a pack, a pallet, with its exact
conversion to the base unit — a numerator and a denominator, or "24 of the bottle"). Every unit must convert exactly
to the base unit; a conversion true for every product (kg-g, l-ml, h-min) needs no row. 2 packs bought = 48 bottles =
36 litres in stock, with no decimal factor anywhere."""
with repository() as repo, _decide():
agg, current = _product(repo, caller, product_id)
_same_version(f"product {product_id}", current, version)
_check_refs(repo, unit_code=base_unit_code) # the base unit is a real UN/ECE unit; the others may be packs
if base_unit_code and agg.base_unit_code and base_unit_code != agg.base_unit_code and repo.product_units.of(product_id):
raise Refused(409, f"{product_id} has units expressed in {agg.base_unit_code}: remove them before changing its "
"base unit")
agg.set_units(by=_who(caller), base_unit_code=base_unit_code, sales_unit_code=sales_unit_code,
purchase_unit_code=purchase_unit_code, issue_unit_code=issue_unit_code)
info = {"base": agg.base_unit_code, "template": agg.product_template_id}
if units_of and not agg.base_unit_code:
raise Refused(422, "set the base unit first: every other unit is expressed in it")
known = _store_units(repo, product_id, product_id, info, units_of or [], _who(caller))
_role_units_convert(repo, agg, known, info)
today = _today(repo)
for role, unit in (("sales", sales_unit_code), ("purchase", purchase_unit_code), ("issue", issue_unit_code),
("production", production_unit_code), ("warehouse", warehouse_unit_code), ("delivery", delivery_unit_code)):
if unit:
if units.factor_of(unit, agg.base_unit_code, known, repo.units.fixed()) is None:
raise RuleBroken(f"{unit} does not convert to {agg.base_unit_code}, the base unit of {product_id}: give "
"its exact conversion (a numerator and a denominator) first")
_put_role(repo, agg, product_id, role, unit, today, _who(caller))
return _product_view(repo, agg, repo.products.save(agg, current), caller) | {"units": _units_view(repo, agg)}
def _units_view(repo, agg: Product) -> list:
info = {"base": agg.base_unit_code, "template": agg.product_template_id}
rows = repo.product_units.effective(agg.product_id, info["template"])
out = [{"unit_code": agg.base_unit_code, "numerator": 1, "denominator": 1, "packaging_level": "none", "base": True}]
for u, r in rows.items():
out.append({"unit_code": u, "numerator": r["numerator"], "denominator": r["denominator"],
"packaging_level": r["packaging_level"], "quantity_decimals": r["quantity_decimals"],
"used": r["used_at"] is not None})
return out
@PRODUCT.use_case(kind="command", route="POST /products/{product_id}/add-unit", permission=WRITE, if_match="version")
def add_unit(caller: dict, product_id: Code, unit: UnitIn, version: Optional[int] = None,
use_for: Optional[list[Literal["purchase", "sales", "issue"]]] = None) -> dict:
"""Add a unit or a pack size to an ACTIVE product (PRD-23), with its exact conversion to the base unit; `use_for` makes
it the purchase, sales or issue unit from now. The same unit again with the same conversion is nothing. A conversion that
a document or stock has already used is NEVER changed (409): a changed pack size is a new unit code."""
with repository() as repo, _decide():
agg, current = _product(repo, caller, product_id)
_same_version(f"product {product_id}", current, version)
info = {"base": agg.base_unit_code, "template": agg.product_template_id}
known = _store_units(repo, product_id, product_id, info, [unit], _who(caller))
if units.factor_of(unit.unit_code, agg.base_unit_code, known, repo.units.fixed()) is None:
raise Refused(422, f"{unit.unit_code} has no conversion to {agg.base_unit_code}")
if not use_for and unit.unit_code in {agg.sales_unit_code, agg.purchase_unit_code, agg.issue_unit_code}:
use_for = []
agg.add_unit(unit.unit_code, by=_who(caller), roles=use_for or [])
return _product_view(repo, agg, repo.products.save(agg, current), caller) | {"units": _units_view(repo, agg)}
@PRODUCT.use_case(kind="command", route="POST /products/{product_id}/units/{unit_code}/used", permission=UNIT_USED,
idempotent=True)
def mark_unit_used(caller: dict, product_id: Code, unit_code: Code, document_ref: Optional[Name] = None,
measures_from: Optional[IsoDate] = None) -> dict:
"""A selling brick (quote, order, invoice, stock — a service client holding the machine permission product:unit:mark_used)
calls this once a document used a unit of a product. A pack must have its e-invoice unit code (Rec 20/21) first. From then
that unit's conversion, packaging level, decimals and e-invoice code are fixed for good — a changed pack size is a new unit code. A repeat changes nothing. A unit with
no row of its own (the base unit, or a conversion true for every product) is always fixed and nothing is recorded."""
with repository() as repo:
if not _product_visible(repo, caller, product_id):
raise Refused(404, f"no product {product_id}")
info = repo.products.info([product_id])[product_id]
rows = repo.product_units.effective(product_id, info.get("template", ""))
if unit_code != info["base"] and unit_code not in rows and \
units.factor_of(unit_code, info["base"], {}, repo.units.fixed()) is None:
raise Refused(404, f"{product_id} is not counted in {unit_code}")
row = rows.get(unit_code)
ref = f"{document_ref or 'a document'} for {product_id}"
repo.types.mark_used(product_id, datetime.now(timezone.utc).replace(tzinfo=None), ref) # any unit, the base unit too: its type is fixed
if row is not None and not row.get("einvoice_unit_code"):
raise Refused(409, f"{unit_code} of {product_id} has no e-invoice unit code yet", gate="einvoice_unit_code", fix=_EINVOICE_FIX) # the body purchasing's and inventory's signed specs read
if row is not None and row["used_at"] is None:
repo.product_units.mark_used(row["owner"], unit_code, datetime.now(timezone.utc).replace(tzinfo=None), ref)
if measures_from: # the document copied the unit's measures row dated `measures_from` (v3): it is now copied
when = _date(measures_from, "measures_from")
for owner in (product_id, info.get("template", "")):
if owner and (m := repo.measures.get(owner, unit_code, when)) is not None:
if not m["copied"]:
repo.measures.mark_copied(owner, unit_code, when)
break
return {"product_id": product_id, "unit_code": unit_code, "used": True, "fixed_by": "its own row" if row else
"the base unit or a conversion true for every product"}
def _product_visible(repo, caller: dict, product_id: str) -> bool:
found = repo.products.get(product_id)
return found is not None and (found[0].offered() or _sees_drafts(caller))
@PRODUCT.use_case(kind="query", route="GET /products/{product_id}/convert", permission=READ)
def convert_quantity(caller: dict, product_id: Code, quantity: Quantity, from_unit: Code, to_unit: Code) -> dict:
"""Convert a quantity between two units of ONE product, exactly: 2 packs of COLA-075 = 48 bottles = 36 LTR. The answer is
the exact fraction and the decimal rounded half up to the target unit's quantity_decimals — the only rounding."""
with repository() as repo:
return _convert_quantity(repo, caller, product_id, quantity, from_unit, to_unit)
def _convert_quantity(repo, caller: dict, product_id, quantity, from_unit, to_unit) -> dict:
if not _product_visible(repo, caller, product_id):
raise Refused(404, f"no product {product_id}")
info = repo.products.info([product_id])[product_id]
a, b = (_factor(repo, product_id, info, u) for u in (from_unit, to_unit))
for f, u in ((a, from_unit), (b, to_unit)):
if f is None:
raise Refused(422, f"{product_id} is not counted in {u}")
exact_q = units.convert(quantity, a, b)
return {"product_id": product_id, "from_unit": from_unit, "to_unit": to_unit, "quantity": quantity,
"exact": f"{exact_q.numerator}/{exact_q.denominator}", "converted": units.round_half_up(
exact_q, _decimals(repo, product_id, info, to_unit)),
"in_base_unit": units.round_half_up(units.exact(quantity) * a, _decimals(repo, product_id, info, info["base"])),
"base_unit_code": info["base"]}
# ── roles, measures, GTINs and rounding (v4: PRD-27 .. PRD-30, PRD-32) ───────────────────────────
LEGACY_ROLE = {"sales": "sales_unit_code", "purchase": "purchase_unit_code", "issue": "issue_unit_code"}
def _num(v) -> Optional[str]:
"""A stored decimal as the doors show it: plain digits without trailing zeros (19.200000 -> 19.2); None stays None."""
return None if v is None else format(Decimal(v).normalize(), "f")
def _template_of(agg: Product) -> str:
return agg.product_template_id if agg.product_template_id and agg.product_template_id != agg.product_id else ""
def _roles_on(repo, agg: Product, on: date) -> tuple:
"""({role: unit_code}, [roles that fell back to the base unit]) as in force on `on`: the SKU's own dated row, else its
template's, else (sales, purchase, issue) the column set before roles were dated; unset ones default to the base unit."""
tmpl, out, defaulted = _template_of(agg), {}, []
for role in rounding.ROLES:
row = repo.roles.in_force(agg.product_id, tmpl, role, on)
unit = (row["unit_code"] if row else "") or (getattr(agg, LEGACY_ROLE[role]) if role in LEGACY_ROLE else "")
if not unit and role != "sales":
unit = agg.base_unit_code
defaulted.append(role)
out[role] = unit
return out, defaulted
def _measure_view(row: Optional[dict]) -> Optional[dict]:
if row is None:
return None
out = {k: (row[k] or None) if k.endswith("_code") else _num(row[k]) for k in
("gross_weight", "net_weight", "weight_unit_code", "volume", "volume_unit_code", "length", "width", "height",
"dimension_unit_code")}
return out | {"valid_from": row["valid_from"]}
def _gtin_view(g: dict) -> dict:
return {"gtin": g["gtin"], "kind": g["kind"], "status": g["status"]}
def _unit_answers(repo, agg: Product, on: date) -> list:
"""Every unit of the product as GET /products/{id} shows it, the base unit first."""
info = {"base": agg.base_unit_code, "template": _template_of(agg)}
rows = repo.product_units.effective(agg.product_id, info["template"])
gtins = repo.gtins.of_owner(agg.product_id)
out = []
for code in [agg.base_unit_code, *rows]:
if not code:
continue
r = rows.get(code)
f = Fraction(1) if r is None else Fraction(r["numerator"], r["denominator"])
out.append({"unit_code": code, "numerator": f.numerator, "denominator": f.denominator,
"quantity_decimals": _decimals(repo, agg.product_id, info, code),
"gtins": [{"gtin": g["gtin"], "kind": g["kind"]} for g in gtins if g["unit_code"] == code and g["status"] == "active"],
"measures": _measure_view(repo.measures.in_force(agg.product_id, info["template"], code, on))})
return out
def _owner(repo, agg: Product, scope: str) -> str:
"""Who a per-unit row belongs to: the SKU, or (scope template) its template, which every variant without its own row inherits."""
if scope != "template":
return agg.product_id
tmpl = _template_of(agg)
if not tmpl:
raise Refused(422, f"{agg.product_id} has no template: set the row on the product itself")
return tmpl
def _need_unit(repo, agg: Product, unit_code: str) -> None:
info = {"base": agg.base_unit_code, "template": _template_of(agg)}
if _factor(repo, agg.product_id, info, unit_code) is None:
raise Refused(409, f"{unit_code} is not a unit of {agg.product_id}: its units are {agg.base_unit_code} (base)"
f"{''.join(', ' + u for u in sorted(repo.product_units.effective(agg.product_id, info['template'])))}")
def _put_role(repo, agg: Product, owner: str, role: str, unit_code: str, valid_from: date, by: str, rule: Optional[dict] = None) -> None:
"""Write the role's row for `valid_from`: a new row (a plain insert: two writers of one date collide, a 409), or the same date
again, which is replaced (only dates a document cannot have used reach here). The rule in force is carried forward."""
tmpl = _template_of(agg)
before = repo.roles.in_force(agg.product_id, tmpl, role, valid_from)
carried = rule if rule is not None else ({k: before[k] for k in ("min_quantity", "rounding_multiple", "rule_unit_code")} if before else {})
row = {"role": role, "valid_from": valid_from, "unit_code": unit_code, **carried}
same_day = repo.roles.get(owner, role, valid_from)
if same_day is not None:
repo.roles.replace(owner, row, by)
return
if before is not None and all(before.get(k) == row.get(k) for k in ("unit_code", "min_quantity", "rounding_multiple")) \
and before["valid_from"] <= valid_from and (rule is None):
return # the same state again: nothing
try:
repo.roles.add(owner, row, by, datetime.now(timezone.utc).replace(tzinfo=None))
except IntegrityError:
raise Refused(409, f"the {role} role of {owner} from {valid_from} was set by someone else a moment ago") from None
@PRODUCT.use_case(kind="command", route="POST /products/{product_id}/set-unit-role", permission=WRITE)
def set_unit_role(caller: dict, product_id: Code, role: Literal["sales", "purchase", "issue", "production", "warehouse", "delivery"],
unit_code: Code, valid_from: Optional[IsoDate] = None, scope: Literal["product", "template"] = "product") -> dict:
"""Change which of the product's own units a role defaults to (PRD-29): a DATED row, so the change is the default for new
lines from that date and never touches a document or a price for today or earlier. On an active product it starts after today
(the first rows of a draft may start today). `scope` template sets it for every variant without its own row. The role's
rounding rule (PRD-30) carries forward."""
with repository() as repo, _decide():
agg, current = _product(repo, caller, product_id)
today = _today(repo)
start = _date(valid_from, "valid_from") or (today if agg.status == ProductState.DRAFT else today + timedelta(days=1))
if start < today or (start == today and agg.status != ProductState.DRAFT):
raise Refused(409, f"a role changes from a date after today ({today}); only a draft product's first rows may start today")
_need_unit(repo, agg, unit_code)
owner = _owner(repo, agg, scope)
if scope == "template" and start <= today:
sibs = repo.products.info([x[0] for x in repo.templates.products(owner)])
live = sorted(k for k, v in sibs.items() if v["status"] != ProductState.DRAFT)
if live:
raise Refused(409, f"a template role changes from a date after today ({today}) unless every variant is a draft: "
f"{', '.join(live)} {'is' if len(live) == 1 else 'are'} not")
_put_role(repo, agg, owner, role, unit_code, start, _who(caller))
if start <= today and role in LEGACY_ROLE and scope == "product":
agg.set_units(by=_who(caller), **{LEGACY_ROLE[role]: unit_code})
repo.products.save(agg, current)
roles, defaulted = _roles_on(repo, agg, start)
return {"product_id": product_id, "role": role, "unit_code": unit_code, "valid_from": start, "roles_from_then": roles,
"roles_defaulted": defaulted}
@PRODUCT.use_case(kind="command", route="POST /products/{product_id}/set-rounding", permission=WRITE)
def set_rounding(caller: dict, product_id: Code, role: Literal["sales", "purchase", "issue", "production", "warehouse", "delivery"],
rounding_multiple: Optional[Measure] = None, min_quantity: Optional[Measure] = None,
unit_code: Optional[Code] = None, valid_from: Optional[IsoDate] = None,
scope: Literal["product", "template"] = "product") -> dict:
"""Set a role's minimum and multiple ("at least 80, in multiples of 80 PK24": a full pallet; "multiples of 6 C62") from a date
AFTER today (default tomorrow). The rule keeps the unit it was written in (`unit_code`, default the role's unit). A dated row on
the template (every variant without its own inherits it) or the SKU. Product only answers the rule
(GET /products/{id}/rounding): purchasing and sales-order decide what an unmet rule does."""
with repository() as repo, _decide():
agg, _ = _product(repo, caller, product_id)
today = _today(repo)
start = _date(valid_from, "valid_from") or today + timedelta(days=1)
if start <= today:
raise Refused(409, f"a rounding rule starts after today ({today}): a document may already have been checked against the old one")
rounding.check_rule(min_quantity, rounding_multiple)
roles, _ = _roles_on(repo, agg, start)
role_unit = roles[role] or agg.base_unit_code
rule_unit = unit_code or role_unit
_need_unit(repo, agg, rule_unit)
owner = _owner(repo, agg, scope)
_put_role(repo, agg, owner, role, role_unit, start, _who(caller),
rule={"min_quantity": min_quantity, "rounding_multiple": rounding_multiple, "rule_unit_code": rule_unit})
return {"product_id": product_id, "role": role, "valid_from": start, "owner": owner,
"rule": {"unit_code": rule_unit, "min_quantity": _num(min_quantity), "rounding_multiple": _num(rounding_multiple),
"valid_from": start}}
@PRODUCT.use_case(kind="query", route="GET /products/{product_id}/rounding", permission=READ)
def get_rounding(caller: dict, product_id: Code, role: Literal["purchase", "sales", "issue", "production", "warehouse", "delivery"],
quantity: Quantity, unit_code: Code, on: Optional[IsoDate] = None) -> dict:
"""Does a quantity meet the product's rounding rule for a role on a date (PRD-32)? Derived, repeatable, nothing stored.
The rule is the dated row in force on `on` (default today): the SKU's own, else its template's. The answer is in the rule's own unit:
`quantity_in_rule_unit`, whether it is `met`, the nearest allowed quantities below and above, and a `reason`. Product
enforces nothing: purchasing and sales-order turn an unmet answer into their own gate."""
with repository() as repo:
return _get_rounding(repo, caller, product_id, role, quantity, unit_code, on)
def _get_rounding(repo, caller: dict, product_id, role, quantity, unit_code, on) -> dict:
agg, _ = _product(repo, caller, product_id)
asked_on = _date(on, "on") or _today(repo)
info = {"base": agg.base_unit_code, "template": _template_of(agg)}
asked = _factor(repo, product_id, info, unit_code)
if asked is None:
raise Refused(422, f"{product_id} is not counted in {unit_code}")
tmpl = _template_of(agg)
row = repo.roles.in_force(product_id, tmpl, role, asked_on)
if row is None or (row["min_quantity"] is None and row["rounding_multiple"] is None):
why = f"no rule on {asked_on}" if repo.roles.ever_ruled(product_id, tmpl, role) else f"no rule for role {role}"
return {"product_id": product_id, "role": role, "on": asked_on, "rule": None, "unit_code": unit_code,
"quantity_in_rule_unit": _num(quantity), "met": True, "next_down": None, "next_up": None, "reason": why}
rule_unit = row["rule_unit_code"] or row["unit_code"]
rule_f = _factor(repo, product_id, info, rule_unit)
if rule_f is None:
raise Refused(409, f"the rule on {product_id} is in {rule_unit}, which {product_id} no longer counts in")
q = units.exact(quantity) * asked / rule_f
verdict = rounding.judge(product_id, role, rule_unit, q, row["min_quantity"], row["rounding_multiple"])
out = lambda v: None if v is None else rounding.fmt(v) # noqa: E731
exact = "" if q.denominator == 1 else f"{q.numerator}/{q.denominator}"
return {"product_id": product_id, "role": role, "on": asked_on,
"rule": {"unit_code": rule_unit, "min_quantity": _num(row["min_quantity"]),
"rounding_multiple": _num(row["rounding_multiple"]), "valid_from": row["valid_from"]},
"rule_row": {"owner": row["owner"], "role": role, "valid_from": row["valid_from"]},
"unit_code": unit_code, "quantity_in_rule_unit": out(q), "exact": exact or None, "met": verdict.met,
"next_down": out(verdict.next_down), "next_up": out(verdict.next_up), "reason": verdict.reason}
@PRODUCT.use_case(kind="command", route="POST /products/{product_id}/set-measures", permission=WRITE)
def set_measures(caller: dict, product_id: Code, unit_code: Code, gross_weight: Optional[Measure] = None,
net_weight: Optional[Measure] = None, weight_unit_code: Optional[Code] = None, volume: Optional[Measure] = None,
volume_unit_code: Optional[Code] = None, length: Optional[Measure] = None, width: Optional[Measure] = None,
height: Optional[Measure] = None, dimension_unit_code: Optional[Code] = None, valid_from: Optional[IsoDate] = None,
scope: Literal["product", "template"] = "product") -> dict:
"""Set a unit's weights, volume and dimensions (PRD-28), each per ONE of the unit and in a unit of the right kind (mass, volume,
length); the net weight never above the gross. A DATED row from `valid_from` (default today, never earlier): a correction is a
new row, never an edit, and a document keeps the copy it took. The row for today is replaced only while no document has copied
it. Measures left out keep their value from the row in force; one never set is unknown (null), never 0."""
with repository() as repo, _decide():
agg, _ = _product(repo, caller, product_id)
today = _today(repo)
start = _date(valid_from, "valid_from") or today
if start < today:
raise Refused(409, f"a measure starts today ({today}) or later, not {start}: a document may already have copied the old one")
_need_unit(repo, agg, unit_code)
owner, tmpl = _owner(repo, agg, scope), _template_of(agg)
given = {"gross_weight": gross_weight, "net_weight": net_weight, "weight_unit_code": weight_unit_code, "volume": volume,
"volume_unit_code": volume_unit_code, "length": length, "width": width, "height": height,
"dimension_unit_code": dimension_unit_code}
if all(v is None for v in given.values()):
raise Refused(422, "give at least one measure")
before = repo.measures.in_force(agg.product_id, tmpl, unit_code, start)
kinds = {u["unit_code"]: u.get("kind") for u in repo.units.all()}
measures.check({k: v for k, v in given.items() if k.endswith("_code")}, kinds) # a unit of the wrong kind, first
merged = {k: (before or {}).get(k) for k in given}
for fields, unit_field, _kind in measures.GROUPS:
if given[unit_field] and (before or {}).get(unit_field) and given[unit_field] != before[unit_field] \
and any(before.get(f) is not None and given[f] is None for f in fields):
raise Refused(422, f"{unit_field} changes from {before[unit_field]} to {given[unit_field]}: restate every "
f"{', '.join(fields)} in the new unit")
merged.update({k: v for k, v in given.items() if v is not None})
measures.check(merged, kinds)
if before is not None and before["valid_from"] < start and all(before[k] == merged[k] for k in given):
return {"product_id": product_id, "unit_code": unit_code, "valid_from": before["valid_from"],
"measures": _measure_view(before), "changed": False}
at = datetime.now(timezone.utc).replace(tzinfo=None)
same_day = repo.measures.get(owner, unit_code, start)
if same_day is not None:
if same_day["copied"]:
raise Refused(409, "a document copied today's row: correct from tomorrow" if start == today else
f"a document copied the row of {start}: correct from the next day")
repo.measures.replace(owner, unit_code, start, merged, _who(caller), at)
else:
try:
repo.measures.add(owner, unit_code, start, merged, _who(caller), at)
except IntegrityError:
raise Refused(409, f"the measures of {unit_code} from {start} were set by someone else a moment ago") from None
return {"product_id": product_id, "unit_code": unit_code, "valid_from": start,
"measures": _measure_view({**merged, "valid_from": start}), "changed": True}
def _gtin_answer(repo, owner: str, unit_code: str) -> dict:
return {"product_id": owner, "unit_code": unit_code, "gtins": [_gtin_view(g) for g in repo.gtins.of_unit(owner, unit_code)]}
def _sync_main(repo, owner: str, unit_code: str) -> None:
"""prod_product_unit.gtin (v1) is kept equal to the unit's main active GTIN."""
row = repo.product_units.of(owner).get(unit_code)
if row is not None:
main = next((g["gtin"] for g in repo.gtins.of_unit(owner, unit_code) if g["kind"] == "main" and g["status"] == "active"), "")
if (row["gtin"] or "") != main:
repo.product_units.set_gtin(owner, unit_code, main)
def _promote(repo, owner: str, unit_code: str, gtin: str, by: str = "") -> None:
"""Make `gtin` the unit's one main: the old main becomes additional."""
for g in repo.gtins.of_unit(owner, unit_code):
if g["gtin"] != gtin and g["kind"] == "main" and g["status"] == "active":
repo.gtins.set_kind(g["gtin"], "additional")
repo.gtins.set_kind(gtin, "main")
try:
repo.gtins.claim_main(owner, unit_code, gtin, by)
except IntegrityError:
raise Refused(409, f"another change to the main GTIN of {owner} {unit_code} landed a moment ago: ask again") from None
_sync_main(repo, owner, unit_code)
@PRODUCT.use_case(kind="command", route="POST /products/{product_id}/add-gtin", permission=WRITE)
def add_gtin(caller: dict, product_id: Code, unit_code: Code, gtin: text(14), main: bool = False,
scope: Literal["product", "template"] = "product") -> dict:
"""Give a unit of the product another barcode (PRD-27): the supplier's case beside your own. It needs a right GS1 check digit
(the answer to a wrong one names the digit expected) and must be on no other product or unit, retired ones included ("... is
COLA-075 PK24"). The first GTIN of a unit is its main one; `main` makes this one main and the old main an additional one: a unit
has exactly one main while it has any. The same GTIN on the same unit again is nothing. `scope` template is refused while the
template has more than one product ("COLA has 3 products: a GTIN names one, set it on the SKU")."""
for attempt in (1, 2): # losing the unit's main claim to a racing first GTIN: start again, now as an additional one
try:
return _add_gtin(caller, product_id, unit_code, g_in := gtin, main, scope)
except _MainRaced:
if attempt == 2:
raise Refused(409, f"the main GTIN of {product_id} {unit_code} changed while {g_in} was added: ask again") from None
class _MainRaced(Exception):
"""A racing change took the unit's main claim first (rolled back, nothing written)."""
def _add_gtin(caller: dict, product_id: str, unit_code: str, gtin: str, main: bool, scope: str) -> dict:
with repository() as repo, _decide():
agg, _ = _product(repo, caller, product_id)
if scope == "template":
tmpl = _template_of(agg)
if not tmpl:
raise Refused(422, f"{product_id} has no template: give the GTIN on the product itself")
count = len(repo.templates.products(tmpl))
if count > 1:
raise Refused(409, f"{tmpl} has {count} products: a GTIN names one, set it on the SKU")
g = variants.gtin14(gtin)
_need_unit(repo, agg, unit_code)
held = repo.gtins.holder(g)
if held is not None:
if (held["owner"], held["unit_code"]) != (product_id, unit_code):
raise Refused(409, f"{g} is {held['owner']} {held['unit_code']}"
f"{' (retired)' if held['status'] == 'retired' else ''}: a barcode names one product and unit for ever")
if held["status"] == "retired":
raise Refused(409, f"{g} was retired from {held['owner']} {held['unit_code']} and is never reissued")
if main and held["kind"] != "main":
_promote(repo, product_id, unit_code, g, _who(caller))
return _gtin_answer(repo, product_id, unit_code)
seen_main = repo.gtins.main_of(product_id, unit_code) # read BEFORE the unit's GTINs: the claim row is the race's one key
mine = [x for x in repo.gtins.of_unit(product_id, unit_code) if x["status"] == "active"]
try:
repo.gtins.claim(g, product_id, unit_code, _who(caller), "main" if not mine else "additional")
except IntegrityError:
raise Refused(409, f"{g} was claimed by someone else a moment ago: a barcode names one product and unit") from None
if not mine:
try:
repo.gtins.claim_main(product_id, unit_code, g, _who(caller), seen_main)
except IntegrityError:
raise _MainRaced from None
if main and mine:
_promote(repo, product_id, unit_code, g, _who(caller))
_sync_main(repo, product_id, unit_code)
return _gtin_answer(repo, product_id, unit_code)
@PRODUCT.use_case(kind="command", route="POST /products/{product_id}/retire-gtin", permission=WRITE)
def retire_gtin(caller: dict, product_id: Code, gtin: text(14)) -> dict:
"""Retire a barcode (PRD-27): it no longer answers a scan for a NEW document, still names its product and unit on every document
that scanned it, and is never given to another product. Retiring a unit's main GTIN makes its oldest active additional one the main."""
with repository() as repo, _decide():
_product(repo, caller, product_id)
g = variants.gtin14(gtin)
held = repo.gtins.holder(g)
if held is None or held["owner"] != product_id:
raise Refused(404, f"{product_id} has no GTIN {g}")
if held["status"] == "retired":
return _gtin_answer(repo, product_id, held["unit_code"])
repo.gtins.retire(g, _today(repo))
rest = [x for x in repo.gtins.of_unit(product_id, held["unit_code"]) if x["status"] == "active" and x["gtin"] != g]
if held["kind"] == "main" and rest:
_promote(repo, product_id, held["unit_code"], rest[0]["gtin"], _who(caller))
elif held["kind"] == "main":
try:
repo.gtins.claim_main(product_id, held["unit_code"], "", _who(caller))
except IntegrityError:
raise Refused(409, f"another change to the main GTIN of {product_id} {held['unit_code']} landed a moment ago: ask again") from None
_sync_main(repo, product_id, held["unit_code"])
return _gtin_answer(repo, product_id, held["unit_code"])
@PRODUCT.use_case(kind="command", route="POST /products/{product_id}/make-gtin-main", permission=WRITE)
def make_gtin_main(caller: dict, product_id: Code, gtin: text(14)) -> dict:
"""Make one of a unit's active GTINs its main one (PRD-27); the old main becomes an additional one."""
with repository() as repo, _decide():
_product(repo, caller, product_id)
g = variants.gtin14(gtin)
held = repo.gtins.holder(g)
if held is None or held["owner"] != product_id:
raise Refused(404, f"{product_id} has no GTIN {g}")
if held["status"] == "retired":
raise Refused(409, f"{g} is retired: it cannot be the main GTIN")
_promote(repo, product_id, held["unit_code"], g, _who(caller))
return _gtin_answer(repo, product_id, held["unit_code"])
# ── templates and variants (PRD-24), barcodes (PRD-25) ────────────────────────────────────────
@PRODUCT.use_case(kind="command", route="POST /templates/{product_template_id}/generate-variants", permission=WRITE,
idempotent=True)
def generate_variants(caller: dict, product_template_id: Code, name: Name, attributes: list[AttributeIn],
description: Optional[Note] = None, kind: Optional[Code] = None,
base_unit_code: Optional[Code] = None, category_code: Optional[Code] = None,
tax_category: Optional[Code] = None, billing_kind: Optional[Code] = None,
billing_plan: Optional[Code] = None, invoicing_policy: Optional[Code] = None,
units_of: Optional[list[UnitIn]] = None, product_type_code: Optional[Code] = None,
commodity_code: Optional[text(8)] = None, country_of_origin: Optional[text(2)] = None,
class_code: Optional[Code] = None) -> dict:
"""Describe a product that comes in sizes or colours ONCE, as a template with attributes (volume: 0.33 L, 0.75 L, 1.5 L),
and get one DRAFT product — its own SKU — per combination (COLA-033, COLA-075, COLA-150: the template's code, then each
value's `code`, or its letters and digits). Each variant carries the template's category, tax category, billing, base unit
and the units the template gives (`units_of`: one conversion for all of them, e.g. a pack of 12 litres; a variant sets its
own with set_units, which wins). Running it again adds only the combinations that are missing and touches nothing that
exists; new values may be added to an attribute, a new attribute may not. Prices, documents and stock name the
variant (the product), never the template. The type is REQUIRED (a template that already has one keeps it when none is given); the template's type, commodity code and country of origin are COPIED onto each variant
when it is made and are the variant's own from then on (a later template change reaches no existing variant). An attribute that
names a `characteristic_code` gives each variant that characteristic's value (volume 0.75 L) in `class_code`.
"""
def unit():
with repository() as repo, _decide():
_check_refs(repo, unit_code=base_unit_code, category_code=category_code)
wanted = variants.check_attributes([a.model_dump() for a in attributes])
template = repo.templates.get(product_template_id)
if template is None and repo.products.code_in_use(product_template_id):
raise Refused(409, f"{product_template_id} is already a product: a template's code is its own, so a price, a "
"document or a stock row never confuses it with a product")
attrs = variants.merge_attributes(template["attributes"], wanted) if template else wanted
ptype = _need_type(repo, product_type_code or (template or {}).get("product_type_code"))
if kind and kind != ptype["kind"]:
raise Refused(422, f"{ptype['product_type_code']} products are {ptype['kind']}, not {kind}")
wants_chars = [a for a in attrs if a.get("characteristic_code")]
if wants_chars and not (class_code and repo.classification.klass(class_code)):
raise Refused(422, "an attribute that names a characteristic needs class_code: the class those characteristics belong to")
fields = {"kind": kind or ptype["kind"], "base_unit_code": base_unit_code, "category_code": category_code,
"tax_category": tax_category, "billing_kind": billing_kind, "billing_plan": billing_plan,
"invoicing_policy": invoicing_policy, "product_type_code": ptype["product_type_code"], "commodity_code": commodity_code,
"country_of_origin": country_of_origin}
row = template or {"product_template_id": product_template_id, "name": name, "description": description or ""}
row.update({"name": name, "attributes": attrs, "created_by": row.get("created_by") or _who(caller)})
for k, v in fields.items():
if v is not None:
row[k] = v
row.setdefault(k, "")
if description is not None:
row["description"] = description
known = {p: v for p, v in repo.templates.products(product_template_id)}
have = {json.dumps(v, sort_keys=True) for v in known.values()}
made, kept, clash = [], [], []
todo = []
for sku, values in variants.combinations(product_template_id, attrs):
if json.dumps(values, sort_keys=True) in have:
kept.append(sku)
elif repo.products.code_in_use(sku):
mine = repo.products.get(sku) # a racing caller made it a moment ago: same template and values = existing
if mine is not None and mine[0].product_template_id == product_template_id \
and json.loads(mine[0].attribute_values or "{}") == values:
kept.append(sku)
else:
clash.append(sku)
else:
todo.append((sku, values))
if clash:
raise Refused(409, f"{', '.join(clash[:5])} already exist as other products: give those values another code")
repo.templates.put(row)
if units_of:
_store_template_units(repo, product_template_id, row, units_of, _who(caller))
for sku, values in todo:
agg = Product.add_product(sku, by=_who(caller), name=variants.variant_name(name, values),
description=row.get("description") or "", kind=row.get("kind") or None,
base_unit_code=row.get("base_unit_code") or None,
category_code=row.get("category_code") or None,
tax_category=row.get("tax_category") or None,
billing_kind=row.get("billing_kind") or None,
billing_plan=row.get("billing_plan") or None,
invoicing_policy=row.get("invoicing_policy") or None,
product_template_id=product_template_id,
attribute_values=json.dumps(values, sort_keys=True),
type_row=ptype,
commodity_code=row.get("commodity_code") or None,
country_of_origin=row.get("country_of_origin") or None)
repo.products.save(agg, 0)
if wants_chars:
at = datetime.now(timezone.utc).replace(tzinfo=None)
repo.classification.add(sku, class_code, "", _today(repo), [class_code], _who(caller), at)
for a in wants_chars:
char = repo.classification.characteristic(a["characteristic_code"])
if char is None:
raise Refused(422, f"no characteristic '{a['characteristic_code']}': add it in the settings first")
repo.classification.add(sku, class_code, a["characteristic_code"], _today(repo),
_canonical(repo, char, CharValueIn(characteristic_code=char["characteristic_code"], value=values[a["name"]])),
_who(caller), at)
made.append(sku)
return {"product_template_id": product_template_id, "name": name, "attributes": attrs, "created": made,
"existing": kept, "products": sorted(kept + made)}
# two callers generating at once: the loser's save conflicts, runs again, finds the variants the winner made and
# reports them as existing — 200, never a 409
return retry_on_conflict(unit)
def _store_template_units(repo, template_id: str, row: dict, unit_rows, by: str) -> None:
"""Units every variant of a template shares (a pack of 12 litres): rows owned by the template; a variant's own wins."""
base = row.get("base_unit_code") or ""
if not base:
raise Refused(422, "set the template's base unit first: every other unit is expressed in it")
for u in unit_rows:
if u.gtin:
raise Refused(422, f"{u.unit_code}: a GTIN names one product's packaging level — give it on the product")
_store_units(repo, template_id, template_id, {"base": base, "template": ""}, unit_rows, by, template_owner=True)
@PRODUCT.use_case(kind="command", route="POST /templates/{product_template_id}/set-units", permission=WRITE, idempotent=True)
def set_template_units(caller: dict, product_template_id: Code, units_of: list[UnitIn]) -> dict:
"""Set the units every variant of a template shares (PRD-34): a pack stated RELATIVE to a unit only the variants have —
`{"unit_code": "PK24", "contains_unit": "C62", "contains_quantity": "24"}` where the bottle (C62) is per SKU (0.33, 0.75, 1.5 LTR)
— is composed exactly with each variant's own row when it is read (COLA-075's PK24 is 24 x 0.75 = 18 LTR). A variant's own row for
the same unit wins, whole. Once a document or stock has used a template row it is never changed: a changed pack is a NEW unit code."""
with repository() as repo, _decide():
template = repo.templates.get(product_template_id)
if template is None:
raise Refused(404, f"no template {product_template_id}")
_store_template_units(repo, product_template_id, template, units_of, _who(caller))
rows = repo.product_units.of(product_template_id)
return {"product_template_id": product_template_id, "units": [
{"unit_code": u, "numerator": r["numerator"], "denominator": r["denominator"], "relative_unit_code": r["relative_unit_code"] or None,
"packaging_level": r["packaging_level"], "used": r["used_at"] is not None} for u, r in rows.items()]}
@PRODUCT.use_case(kind="query", route="GET /gtin/{gtin}", permission=READ)
def find_by_gtin(caller: dict, gtin: text(14)) -> dict:
"""Scan a barcode (PRD-27): the product, the unit, whether this is the unit's main GTIN or an additional one, whether it is
active or retired, and how much of the base unit that is — scanning the supplier's case of COLA-075 answers PK24, 24 bottles,
18 LTR. The scanner resolves every barcode here (8, 12, 13 or 14 digits, left-padded to 14). A retired GTIN still answers, with
status retired, so an old document can name its unit; the caller refuses it for a new line. 404 for one never issued."""
with repository() as repo:
g = gtin.strip()
held = repo.gtins.holder(g.zfill(14)) if g.isdigit() and len(g) in (8, 12, 13, 14) else None
if held is None or not _product_visible(repo, caller, held["owner"]):
raise Refused(404, f"no product has the GTIN {gtin}")
owner = held["owner"]
product = repo.products.get(owner)[0]
info = repo.products.info([owner])[owner]
f = _factor(repo, owner, info, held["unit_code"]) or Fraction(1)
row = repo.product_units.effective(owner, info.get("template", "")).get(held["unit_code"])
sales = _factor(repo, owner, info, info["sales"]) or Fraction(1)
base_q = units.round_half_up(f, _decimals(repo, owner, info, info["base"]))
return {"gtin": held["gtin"], "product_id": owner, "unit_code": held["unit_code"], "kind": held["kind"],
"status": held["status"], "quantity_base": base_q, "base_unit_code": info["base"], "name": product.name,
"packaging_level": (row or {}).get("packaging_level") or "none", "quantity_in_base_unit": base_q,
"sales_unit_code": info["sales"],
"quantity_in_sales_unit": units.round_half_up(f / sales, _decimals(repo, owner, info, info["sales"])),
"exact": {"base": f"{f.numerator}/{f.denominator}", "sales": f"{(f / sales).numerator}/{(f / sales).denominator}"}}
# ═══ price lists ═════════════════════════════════════════════════════════════════════════════
@PRODUCT.use_case(kind="query", route="GET /price-lists", permission=LIST_READ)
def list_price_lists(caller: dict) -> dict:
"""The price lists the caller may see: sales sees the published ones; the pricing roles also see drafts and
retired lists. A summary each; get one for its contents."""
with repository() as repo:
return {"price_lists": [
{"price_list_id": a.price_list_id, "name": a.name, "currency": a.currency, "status": a.status,
"default_from": a.default_from, "based_on": a.based_on, "version": v}
for a, v in repo.price_lists.all() if a.status == PriceListState.PUBLISHED or _manages_prices(caller)]}
@PRODUCT.use_case(kind="query", route="GET /price-lists/{price_list_id}", permission=LIST_READ)
def get_price_list(caller: dict, price_list_id: Code) -> dict:
"""One price list with its version; the pricing roles also see what is waiting to be published, what publishing
flagged and how many customers are assigned. A draft list is a 404 to sales."""
with repository() as repo:
agg, version = _list(repo, caller, price_list_id)
return _list_view(repo, agg, version, caller)
@PRODUCT.use_case(kind="query", route="GET /price-lists/{price_list_id}/rows", permission=LIST_READ)
def list_price_rows(caller: dict, price_list_id: Code, product_id: Optional[Code] = None,
status: Optional[Literal["live", "draft"]] = None, limit: Limit = 200) -> dict:
"""The dated price rows of a list, by product then date: one product's history (`product_id`), or what is
waiting to be published (`status` draft). Sales sees the live rows of published lists."""
with repository() as repo:
agg, _ = _list(repo, caller, price_list_id)
if not _manages_prices(caller):
status = "live"
rows = repo.prices.rows_of(agg.price_list_id, product_id=product_id, status=status, limit=limit)
return {"price_list_id": agg.price_list_id, "rows": [_row_view(r) for r in rows]}
@PRODUCT.use_case(kind="query", route="GET /price-lists/{price_list_id}/prices", permission=LIST_READ)
def list_prices_on(caller: dict, price_list_id: Code, on: Optional[IsoDate] = None, limit: Annotated[int, Field(ge=1, le=1000)] = 500,
offset: Annotated[int, Field(ge=0, le=10_000_000)] = 0) -> dict:
"""PRD-40: a whole PUBLISHED price list in force on a date, for a register's snapshot: every product with a price on the list is
answered once per (unit_code, from_quantity), with the price find_price would give for that quantity (a price derived by a rule is
its result, never the rule), its pricing quantity and unit and priced_from. `next_change_on` is the first date after `on` on which an
answered row or rule changes, so the register knows when its snapshot goes stale. A draft or retired list is a 404. Ordered by
product_id, unit_code, from_quantity; `more` says another page follows."""
with repository() as repo:
found = repo.price_lists.get(price_list_id)
if found is None or found[0].status != PriceListState.PUBLISHED:
raise Refused(404, f"no published price list {price_list_id}")
head = repo.prices.head(price_list_id)
asked_on = _date(on, "on") or _today(repo)
rows = repo.prices.rows_of(price_list_id, status="live")
rules_ = repo.prices.rules_of(price_list_id, status="live")
keys = sorted({(r.product_id, r.unit_code) for r in rows if r.valid_from <= asked_on and (r.valid_to is None or r.valid_to >= asked_on)})
breaks = sorted({q for r in rules_ if r.min_quantity and r.min_quantity > 1
and (r.valid_from is None or r.valid_from <= asked_on) and (r.valid_to is None or r.valid_to >= asked_on)
for q in [r.min_quantity]})
out, changes = [], []
for pid, unit in keys:
got_product = repo.products.get(pid)
if got_product is None or not got_product[0].offered():
continue
product = got_product[0]
subject = _subject(repo, pid, asked_on, product.category_code, unit)
info = repo.products.info([pid])[pid]
factor = _factor(repo, pid, info, subject.unit_code)
if factor is None:
continue
last = None
for base_qty in [Decimal(0)] + [Decimal(b) for b in breaks]:
qty_unit = Fraction(1) if base_qty == 0 else units.exact(base_qty) / factor
asked = pricing.resolve(repo.prices, [price_list_id], subject, units.exact(qty_unit) * factor, asked_on, head.currency if head else "GBP")
got = asked.resolved
if got is None:
continue
row = _deepest_row(got)
shown = (got.unit_price, got.pricing_quantity, got.pricing_unit or subject.unit_code)
if shown == last:
continue # a break that changes nothing is not a row
last = shown
from_q = units.round_half_up(qty_unit, 6).normalize()
out.append({"product_id": pid, "unit_code": subject.unit_code, "from_quantity": format(from_q, "f"), "unit_price": rate(got.unit_price),
"pricing_quantity": got.pricing_quantity, "pricing_unit_code": got.pricing_unit or subject.unit_code,
"valid_from": row.valid_from if row else None, "valid_to": row.valid_to if row else None,
"priced_from": {"price_list_id": got.price_list_id, "rule_seq": got.rule.seq if got.rule else None,
"row": _row_view(row) if row else None}})
answered = {o["product_id"] for o in out}
for r in rows:
if r.product_id in answered:
changes += [d for d in (r.valid_from, r.valid_to + timedelta(days=1) if r.valid_to else None) if d and d > asked_on]
for r in rules_:
changes += [d for d in (r.valid_from, r.valid_to + timedelta(days=1) if r.valid_to else None) if d and d > asked_on]
page = out[offset:offset + limit]
return {"price_list_id": price_list_id, "currency": head.currency if head else None, "prices_include_tax": bool(found[0].prices_include_tax),
"on": asked_on, "next_change_on": min(changes) if changes else None, "prices": page, "more": offs