"""The tables, and one repository per aggregate that loads and saves it whole.
Every table is prefixed, has a natural primary key (never SERIAL) and, for a part, a key that starts with its
root's — the repository finds stale rows under a root by that prefix. A price per unit is a RATE, NUMERIC(19,4)
(shared.yaml); a line total is NUMERIC(18,2) and is not this module's. schema.sql is written from the same column
model; tests/test_schema.py keeps the two equal.
A price list's dated ROWS and its RULES are not loaded with the list: a list holds thousands of rows and the hot
path (find_price) reads one product's. They are written by keyed upsert, only ever as new rows (a price is never
edited), and a DRAFT row that was never live may be discarded.
"""
import json
from datetime import date, datetime, timezone
from decimal import Decimal
from fractions import Fraction
from typing import Iterable, Optional
from erp_platform.repository import Aggregate, Part, claim_tables, upsert
from sqlalchemy import (
TIMESTAMP,
BigInteger,
Boolean,
Column,
Date,
Integer,
MetaData,
Numeric,
String,
Table,
and_,
delete,
func,
insert,
or_,
select,
)
from brick_product.domain.price_list import PriceList, RowIn, RuleIn
from brick_product.domain.pricing import ListHead, PriceRow, PriceRule
from brick_product.domain.product import Product
from brick_product.domain.values import RowState, Settings
md = MetaData()
RATE = Numeric(19, 4)
prod_product = Table("prod_product", md,
Column("product_id", String, primary_key=True),
Column("name", String),
Column("description", String),
Column("kind", String),
Column("base_unit_code", String),
Column("sales_unit_code", String),
Column("purchase_unit_code", String),
Column("issue_unit_code", String),
Column("category_code", String),
Column("tax_category", String),
Column("billing_kind", String),
Column("billing_plan", String),
Column("invoicing_policy", String),
Column("cost_amount", RATE),
Column("product_template_id", String),
Column("attribute_values", String),
Column("status", String),
Column("version", Integer),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP),
Column("product_type_code", String), # v4 slice B (PRD-35): the type it was added with; kind and stocked_default copied from it
Column("stocked_default", Boolean),
Column("commodity_code", String), # 8-digit CN code (Intrastat, customs), read by tax
Column("country_of_origin", String), # ISO 3166 alpha-2, read by tax and ar
Column("sellable", Boolean), # 00006: copied from the type at creation (PRD-35)
Column("valuation_class", String), # 00006: copied from the type
Column("required_roles", String)) # 00006: copied from the type: JSON list of unit roles activation needs
PROD_PRODUCT_COLUMNS = ("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", "cost_amount", "product_template_id", "attribute_values",
"status", "product_type_code", "stocked_default", "commodity_code", "country_of_origin",
"sellable", "valuation_class", "required_roles")
# What several products have in common (Cola, in three volumes); each variant is its OWN product (a SKU). A price, a document
# line and a stock row name the product, never the template. `attributes` is JSON: [{"name": "volume", "values":
# [{"value": "0.33 L", "code": "033"}, ...]}]. A product with no variants has no template row: it is its own template.
prod_template = Table("prod_template", md,
Column("product_template_id", String, primary_key=True),
Column("name", String),
Column("description", String),
Column("kind", String),
Column("category_code", String),
Column("tax_category", String),
Column("billing_kind", String),
Column("billing_plan", String),
Column("invoicing_policy", String),
Column("base_unit_code", String),
Column("attributes", String),
Column("created_by", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP),
Column("product_type_code", String),
Column("commodity_code", String),
Column("country_of_origin", String))
PROD_TEMPLATE_COLUMNS = ("product_template_id", "name", "description", "kind", "category_code", "tax_category", "billing_kind",
"billing_plan", "invoicing_policy", "base_unit_code", "attributes", "created_by", "product_type_code",
"commodity_code", "country_of_origin")
# What happened to a product, by whom: the record behind "reactivating it is allowed and recorded" (PRD-03).
prod_product_log = Table("prod_product_log", md,
Column("product_id", String, primary_key=True),
Column("seq", Integer, primary_key=True),
Column("event", String),
Column("actor", String),
Column("detail", String),
Column("event_at", TIMESTAMP),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
PROD_PRODUCT_LOG_COLUMNS = ("product_id", "seq", "event", "actor", "detail", "event_at")
PRODUCT = Aggregate(Part(prod_product, ("product_id",)), Part(prod_product_log, ("product_id", "seq")),
version="version")
prod_price_list = Table("prod_price_list", md,
Column("price_list_id", String, primary_key=True),
Column("name", String),
Column("currency", String),
Column("based_on", String),
Column("prices_include_tax", Boolean),
Column("rounding", String),
Column("default_from", Date),
Column("opened_by", String),
Column("status", String),
Column("retired_on", Date),
Column("publish_notes", String),
Column("version", Integer),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
PROD_PRICE_LIST_COLUMNS = ("price_list_id", "name", "currency", "based_on", "prices_include_tax", "rounding",
"default_from", "opened_by", "status", "retired_on", "publish_notes")
PRICE_LIST = Aggregate(Part(prod_price_list, ("price_list_id",)), version="version")
# Which list a customer is priced from, per currency, FROM A DATE: append-only like a price row (a later assignment
# is a new row), so the price on any past date is reproducible. The spec's natural key is [bp_id, currency]; a
# valid_from joins it so an assignment never overwrites the one before.
prod_customer_price_list = Table("prod_customer_price_list", md,
Column("bp_id", BigInteger, primary_key=True),
Column("currency", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("price_list_id", String),
Column("created_by", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# The Odoo id map (shared.yaml conventions.odoo_migration.id_map): what an Odoo product.product / uom.uom became here, written
# by this brick's migration and read by later bricks through GET /migration/map/{model}/{odoo_id}?db=.
prod_migration_map = Table("prod_migration_map", md,
Column("odoo_db", String, primary_key=True),
Column("odoo_model", String, primary_key=True),
Column("odoo_id", String, primary_key=True),
Column("natural_key", String),
Column("created_at", TIMESTAMP))
PROD_CUSTOMER_PRICE_LIST_COLUMNS = ("bp_id", "currency", "valid_from", "price_list_id", "created_by")
prod_category = Table("prod_category", md,
Column("category_code", String, primary_key=True),
Column("name", String),
Column("parent_code", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
PROD_CATEGORY_COLUMNS = ("category_code", "name", "parent_code")
# Which category a product — or a category's PARENT — sits in from a date, so a price rule on a category resolves as the
# tree stood on the date asked. Append-only: a move is a new row from tomorrow, never an edit. The subject is
# 'product:<id>' or 'category:<code>' (a product code and a category code may be the same word); a row from FOREVER is
# the first placement. category_code '' means "no category / no parent" from then.
FOREVER = date(1900, 1, 1)
prod_category_membership = Table("prod_category_membership", md,
Column("product_or_category", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("category_code", String),
Column("created_by", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
prod_unit = Table("prod_unit", md,
Column("unit_code", String, primary_key=True),
Column("name", String),
Column("decimals", Integer),
Column("kind", String), # v4: count | mass | volume | length | area | time | other: a measure's unit must be of its kind
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
PROD_UNIT_COLUMNS = ("unit_code", "name", "decimals", "kind")
# A conversion true for EVERY product (1 KGM = 1000 GRM): "1 of from = numerator / denominator of to". Exact integers.
prod_unit_conversion = Table("prod_unit_conversion", md,
Column("from_unit_code", String, primary_key=True),
Column("to_unit_code", String, primary_key=True),
Column("numerator", BigInteger),
Column("denominator", BigInteger),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
CONVERSIONS_SEED = (("KGM", "GRM", 1000, 1), ("LTR", "MLT", 1000, 1), ("HUR", "MIN", 60, 1), ("MTR", "CMT", 100, 1),
("MTR", "MMT", 1000, 1), ("CMT", "MMT", 10, 1))
# A unit a product can be counted in, and its exact conversion to the product's BASE unit: 1 of it = numerator /
# denominator of the base. Owner is a product id (or, for variants, their template's id): a product's own row overrides
# its template's. `used_at` is set once a document or stock has used it: from then its conversion never changes (a changed
# pack size is a new unit code).
prod_product_unit = Table("prod_product_unit", md,
Column("owner", String, primary_key=True),
Column("unit_code", String, primary_key=True),
Column("numerator", BigInteger),
Column("denominator", BigInteger),
Column("packaging_level", String),
Column("quantity_decimals", Integer),
Column("gtin", String),
Column("einvoice_unit_code", String),
Column("status", String),
Column("used_at", TIMESTAMP),
Column("created_by", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP),
Column("relative_unit_code", String), # v4 slice B (PRD-34): numerator / denominator are stated in THIS unit, not the base
Column("used_ref", String)) # what first used the row ("sales order SO-2026-0012 for COLA-075"), named when a change is refused
PROD_PRODUCT_UNIT_COLUMNS = ("owner", "unit_code", "numerator", "denominator", "packaging_level", "quantity_decimals",
"gtin", "einvoice_unit_code", "status", "used_at", "created_by")
# A GTIN is unique across every product and unit: it is CLAIMED here by a plain INSERT, so two callers racing for one barcode
# cannot both win (the same pattern as the platform's claims). The claim names who holds it.
prod_gtin = Table("prod_gtin", md,
Column("gtin", String, primary_key=True),
Column("owner", String),
Column("unit_code", String),
Column("created_by", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP),
Column("kind", String), # v4: main | additional (NULL reads as main: a v1 row)
Column("status", String), # v4: active | retired (NULL reads as active); a retired GTIN keeps its claim for ever
Column("retired_on", Date))
# v5 (audit MUST-1): ONE row per unit naming its main GTIN. The first GTIN of a unit INSERTs it (a second racer fails 23505, a 409);
# make-main and retire UPDATE the same key, so two changes to one unit's main also meet on one key.
prod_gtin_main = Table("prod_gtin_main", md,
Column("owner", String, primary_key=True),
Column("unit_code", String, primary_key=True),
Column("gtin", String),
Column("updated_by", String),
Column("updated_at", TIMESTAMP))
# v4 (PRD-28): a unit's weights, volume and dimensions FROM A DATE, per ONE of the unit. Append-only: a correction is a new row
# from today; the row for today is replaced only while no document has copied it (`copied`, set by mark_unit_used).
prod_unit_measure = Table("prod_unit_measure", md,
Column("owner", String, primary_key=True),
Column("unit_code", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("gross_weight", Numeric(19, 6)),
Column("net_weight", Numeric(19, 6)),
Column("weight_unit_code", String),
Column("volume", Numeric(19, 6)),
Column("volume_unit_code", String),
Column("length", Numeric(19, 6)),
Column("width", Numeric(19, 6)),
Column("height", Numeric(19, 6)),
Column("dimension_unit_code", String),
Column("set_by", String),
Column("set_at", TIMESTAMP),
Column("copied", Boolean),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
MEASURE_FIELDS = ("gross_weight", "net_weight", "weight_unit_code", "volume", "volume_unit_code", "length", "width", "height",
"dimension_unit_code")
# v4 (PRD-29, PRD-30): a unit ROLE from a date, with the rule (minimum, multiple) that goes with it. One row per (owner, role, date)
# is the whole state of that role from then on; a role change carries the rule forward. The rule is stated in `rule_unit_code`.
prod_unit_role = Table("prod_unit_role", md,
Column("owner", String, primary_key=True),
Column("role", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("unit_code", String),
Column("min_quantity", Numeric(19, 6)),
Column("rounding_multiple", Numeric(19, 6)),
Column("rule_unit_code", String),
Column("set_by", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# v4 slice B ----------------------------------------------------------------------------------------------------------------
# What kind of product this is for every module (PRD-35). Configuration, seeded; a type is retired, never deleted.
prod_product_type = Table("prod_product_type", md,
Column("product_type_code", String, primary_key=True),
Column("name", String),
Column("kind", String),
Column("stocked_default", Boolean),
Column("sellable", Boolean),
Column("purchasable", Boolean),
Column("required_roles", String), # JSON list of unit roles that must be set to activate
Column("code_format", String), # 'typed' or a pattern such as TRD-{seq:5}
Column("valuation_class", String),
Column("production_fields", Boolean), # RESERVED until M7: false
Column("status", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
PROD_PRODUCT_TYPE_COLUMNS = ("product_type_code", "name", "kind", "stocked_default", "sellable", "purchasable", "required_roles",
"code_format", "valuation_class", "production_fields", "status")
# A type's code series: one claim row per number, a plain INSERT (never reused; claimed in the add's own transaction, so a refused
# add rolls its number back and leaves no gap).
prod_code_series = Table("prod_code_series", md,
Column("product_type_code", String, primary_key=True),
Column("number", Integer, primary_key=True),
Column("product_id", String),
Column("claimed_at", TIMESTAMP),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# A document or stock used this product (any unit, the base unit included): its type is fixed from then on.
prod_product_use = Table("prod_product_use", md,
Column("product_id", String, primary_key=True),
Column("used_at", TIMESTAMP),
Column("used_ref", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
prod_characteristic = Table("prod_characteristic", md,
Column("characteristic_code", String, primary_key=True),
Column("name", String),
Column("data_type", String),
Column("unit_code", String),
Column("allowed_values", String), # JSON list (list type only)
Column("multi_value", Boolean),
Column("decimals", Integer),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
PROD_CHARACTERISTIC_COLUMNS = ("characteristic_code", "name", "data_type", "unit_code", "allowed_values", "multi_value", "decimals")
prod_class = Table("prod_class", md,
Column("class_code", String, primary_key=True),
Column("name", String),
Column("characteristics", String), # JSON [{characteristic_code, required}]
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# A product's (or template's) class and its characteristic values FROM A DATE. A row with characteristic_code '' is the class
# membership itself. `value` is a JSON list of canonical strings (numbers in the characteristic's unit, exact). Append-only; a
# plain-INSERT claim, so two writers of one (owner, class, characteristic, date) cannot both win.
prod_classification = Table("prod_classification", md,
Column("owner", String, primary_key=True),
Column("class_code", String, primary_key=True),
Column("characteristic_code", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("value", String),
Column("set_by", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# v4 slice B (PRD-37): how one product is procured, planned and allowed at one site (SAP plant data), FROM A DATE after today.
# Quantities are in the product's BASE unit. The M7 columns are reserved: NULL, refused if set.
prod_site = Table("prod_site", md,
Column("product_id", String, primary_key=True),
Column("site_id", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("site_status", String),
Column("procurement_type", String),
Column("supplying_site_id", String),
Column("planned_delivery_days", Integer),
Column("gr_processing_days", Integer),
Column("planning_method", String),
Column("reorder_point", Numeric(19, 4)),
Column("safety_stock", Numeric(19, 4)),
Column("max_stock", Numeric(19, 4)),
Column("lot_size_method", String),
Column("fixed_lot_size", Numeric(19, 4)),
Column("rounding_profile_code", String),
Column("mrp_controller", String),
Column("planning_time_fence_days", Integer),
Column("lot_size_procedure", String),
Column("production_version", String),
Column("set_by", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
SITE_FIELDS = ("site_status", "procurement_type", "supplying_site_id", "planned_delivery_days", "gr_processing_days", "planning_method",
"reorder_point", "safety_stock", "max_stock", "lot_size_method", "fixed_lot_size", "rounding_profile_code")
# v4 slice B (PRD-38): how a SYSTEM-PROPOSED quantity is rounded to pallets or layers when close to one. Steps largest first.
prod_rounding_profile = Table("prod_rounding_profile", md,
Column("rounding_profile_code", String, primary_key=True),
Column("unit_code", String),
Column("steps", String), # JSON [{multiple, round_up_pct}] as exact decimal strings, largest first
Column("status", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
claim_tables("prod_gtin", "prod_gtin_main", "prod_unit_measure", "prod_unit_role", "prod_code_series", "prod_classification", "prod_site")
# One product's price on one list from a date. Never updated in place; `status` says whether it is live yet.
prod_price = Table("prod_price", md,
Column("price_list_id", String, primary_key=True),
Column("product_id", String, primary_key=True),
Column("unit_code", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("unit_price", RATE),
Column("valid_to", Date),
Column("status", String),
Column("created_by", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP),
Column("pricing_quantity", Integer)) # v4: the price is per this many of unit_code (1..100000; NULL reads as 1)
PROD_PRICE_COLUMNS = ("price_list_id", "product_id", "unit_code", "valid_from", "unit_price", "valid_to", "status", "created_by",
"pricing_quantity")
prod_price_rule = Table("prod_price_rule", md,
Column("price_list_id", String, primary_key=True),
Column("seq", Integer, primary_key=True),
Column("applies_to", String),
Column("target", String),
Column("min_quantity", Numeric(18, 4)),
Column("valid_from", Date),
Column("valid_to", Date),
Column("method", String),
Column("discount_pct", Numeric(9, 4)),
Column("fixed_price", RATE),
Column("surcharge_amount", RATE),
Column("base_list", String),
Column("unit_code", String),
Column("status", String),
Column("created_by", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP),
Column("pricing_quantity", Integer)) # v4: the quantity a fixed price or surcharge is per (NULL reads as 1)
PROD_PRICE_RULE_COLUMNS = ("price_list_id", "seq", "applies_to", "target", "min_quantity", "valid_from", "valid_to",
"method", "discount_pct", "fixed_price", "surcharge_amount", "base_list", "unit_code",
"status", "created_by", "pricing_quantity")
prod_settings = Table("prod_settings", md,
Column("setting_key", String, primary_key=True),
Column("value", String),
Column("changes", String),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
# An AI step's proposal waiting for a person (suggest mode), or the ManualTask when no provider is wired. The key
# is the capability's idempotency key '<step>|<subject_ref>|<version>': asking again for the same version of the
# same record finds the same task.
prod_task = Table("prod_task", md,
Column("task_id", String, primary_key=True),
Column("step", String),
Column("subject_ref", String),
Column("status", String),
Column("capability", String),
Column("input_json", String),
Column("plan", String),
Column("reason", String),
Column("destructive", Boolean),
Column("summary", String),
Column("output_json", String),
Column("created_by", String),
Column("resolved_by", String),
Column("note", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
PROD_TASK_COLUMNS = ("task_id", "step", "subject_ref", "status", "capability", "input_json", "plan", "reason",
"destructive", "summary", "output_json", "created_by", "resolved_by", "note")
# module.yaml config_rows: ONE ROW PER SETTING, typed (key, kind, what it changes, a mode's choices).
# The value is stored as JSON text; NULL = the reference behaviour (domain.values.Settings holds the defaults).
SETTINGS_SEED = (
("product_code_format", "str", "how a product code is made: typed by the catalogue manager, or a pattern such as PRD-{seq:5}", ()),
("publish_needs_second_person", "bool", "whether publishing a price list or a price change waits for a second person's approval (PRD-07)", ()),
("business_timezone", "str", "the IANA time zone in which 'today' is counted: nothing may start today or earlier in this zone", ()),
("unusual_price_pct", "int", "how far (percent) a price may sit from its base list before publishing warns (the fixed check beside ai_price_check)", ()),
("odoo_type.product", "str", "the product type an Odoo storable product (detailed_type 'product', or is_storable) is migrated to (PRD-33, PRD-35); the migration stops on a type with no row", ()),
("odoo_type.consu", "str", "the product type an Odoo consumable (type 'consu', not storable) is migrated to", ()),
("odoo_type.goods", "str", "the product type an Odoo 19 'goods' product that is not storable is migrated to", ()),
("odoo_type.service", "str", "the product type an Odoo service is migrated to", ()),
("odoo_type.combo", "str", "the product type an Odoo 17+ combo is migrated to (no seed: a combo has no type until you choose one)", ()),
("ai.ai_check_duplicate.mode", "mode", "a clean product list: duplicates are the reason two invoices for the same thing never reconcile: off or suggest", ("off", "suggest")),
("ai.ai_suggest_details.mode", "mode", "adding a product takes one line: the category, the unit and the tax category are proposed from the name: off or suggest", ("off", "suggest")),
("ai.ai_price_check.mode", "mode", "a mistyped price (12.00 for 120.00) is caught before a customer is quoted it: off or suggest", ("off", "suggest")),
)
SETTING_KINDS = {k: (kind, choices) for k, kind, _, choices in SETTINGS_SEED}
# The seeded product types (SAP material types): code, name, kind, stocked_default, sellable, purchasable, required roles,
# code format, valuation class.
TYPES_SEED = (
("TRADING", "Trading goods", "goods", True, True, True, ("sales", "purchase"), "TRD-{seq:5}", "trading_goods"),
("FINISHED", "Finished goods", "goods", True, True, False, (), "typed", "finished_goods"),
("RAW", "Raw materials", "goods", True, False, True, (), "typed", "raw_materials"),
("SERVICE", "Services", "service", False, True, True, (), "typed", ""),
("NONSTOCK", "Non-stock goods bought for use", "goods", False, False, True, (), "typed", ""),
("PACKAGING", "Packaging", "goods", True, False, True, (), "typed", "packaging"),
# migration 00006: what an untyped product behaved as (sellable, nothing required, stocking unknown). Nothing new is added with it.
("LEGACY", "Legacy goods (added before product types)", "goods", None, True, True, (), "typed", ""),
)
# UN/ECE Recommendation 20 codes (what a UBL / Peppol invoice line carries): code, name, decimals a quantity may have.
UNITS_SEED = (("C62", "one (each)", 0), ("HUR", "hour", 2), ("DAY", "day", 2), ("WEE", "week", 0), ("MON", "month", 0),
("ANN", "year", 0), ("MIN", "minute", 0), ("KGM", "kilogram", 3), ("GRM", "gram", 0),
("LTR", "litre", 3), ("MLT", "millilitre", 0), ("MTR", "metre", 2), ("MTK", "square metre", 2), ("XBX", "box", 0),
("SET", "set", 0), ("MTQ", "cubic metre", 3), ("CMT", "centimetre", 1), ("MMT", "millimetre", 0),
("LBR", "pound", 3), ("FTQ", "cubic foot", 3), ("INH", "inch", 1))
# What each seeded unit measures (v4): a measure's unit must be of its kind (mass, volume, length).
UNIT_KINDS = {"C62": "count", "XBX": "count", "SET": "count", "HUR": "time", "DAY": "time", "WEE": "time", "MON": "time",
"ANN": "time", "MIN": "time", "KGM": "mass", "GRM": "mass", "LBR": "mass", "LTR": "volume", "MLT": "volume",
"MTQ": "volume", "FTQ": "volume", "MTR": "length", "CMT": "length", "MMT": "length", "INH": "length",
"MTK": "area"}
def _encode(kind: str, choices: tuple, value) -> Optional[str]:
if value is None:
return None
if kind == "bool" and not isinstance(value, bool):
raise ValueError("true or false")
if kind == "int" and (isinstance(value, bool) or not isinstance(value, int)):
raise ValueError("a whole number")
if kind == "date":
value = date.fromisoformat(str(value)).isoformat()
if kind == "mode" and value not in choices:
raise ValueError("one of " + ", ".join(choices))
if kind in ("str", "mode") and not isinstance(value, str):
raise ValueError("text")
return json.dumps(value)
def _decode(raw: Optional[str]):
return None if raw is None else json.loads(raw)
def _pick(obj, names: tuple) -> dict:
return {n: getattr(obj, n) for n in names}
def _make(cls, row: dict, names: tuple):
"""The aggregate from a stored row; a NULL column keeps the field's default."""
return cls(**{n: row[n] for n in names if row[n] is not None})
def _d(v) -> Optional[date]:
return v.date() if isinstance(v, datetime) else v
def _dec(v, default=None) -> Optional[Decimal]:
return default if v is None else Decimal(str(v))
class ProductRepository:
"""Loads and saves a product whole (with its history). Takes an open connection: the use case owns the transaction."""
def __init__(self, conn):
self.conn = conn
self._snapshots: dict = {} # product_id -> what its load read: save() hands it back so only changed rows are written
def get(self, product_id) -> Optional[tuple]:
"""(Product, the version it was loaded at), or None. The repository keeps the load's snapshot for this transaction. Read
once per repository (a batch prices many lines of one product): a second get builds a fresh product from the same rows,
until a save of it."""
loaded = self.__dict__.setdefault("_loaded", {})
if product_id not in loaded:
loaded[product_id] = PRODUCT.load(self.conn, {"product_id": product_id})
rows = loaded[product_id]
if rows is None:
return None
self._snapshots[product_id] = rows.snapshot
agg = _make(Product, rows["prod_product"][0], PROD_PRODUCT_COLUMNS)
agg.cost_amount = _dec(agg.cost_amount, Decimal(0))
agg.history = [{k: r[k] for k in PROD_PRODUCT_LOG_COLUMNS}
for r in sorted(rows["prod_product_log"], key=lambda r: r["seq"])]
return agg, PRODUCT.version_of(rows)
def save(self, agg: Product, expected_version: int) -> int:
"""Writes version expected_version + 1, or raises Conflict (a 409 through both doors)."""
snapshot = self._snapshots.pop(agg.product_id, None) # spent: after this save the rows are no longer what it read
self.__dict__.get("_info", {}).pop(agg.product_id, None)
self.__dict__.get("_loaded", {}).pop(agg.product_id, None)
return PRODUCT.save(self.conn, {
"prod_product": [_pick(agg, PROD_PRODUCT_COLUMNS)],
"prod_product_log": [{k: h[k] for k in PROD_PRODUCT_LOG_COLUMNS} for h in agg.history]},
expected_version=expected_version, snapshot=snapshot)
def add_many(self, aggs: list) -> None:
"""NEW products in bulk (a migration chunk), each at version 1 as `save(agg, 0)` writes it: one statement per table, not per
product. A product that appeared meanwhile fails on its version claim (a plain insert): Conflict, the chunk rolls back."""
if not aggs:
return
for a in aggs:
self.__dict__.get("_info", {}).pop(a.product_id, None)
now = datetime.now(timezone.utc)
roots = [PRODUCT._stamp(prod_product, {**_pick(a, PROD_PRODUCT_COLUMNS), "version": 1}, now) for a in aggs]
logs = [PRODUCT._stamp(prod_product_log, {k: h[k] for k in PROD_PRODUCT_LOG_COLUMNS}, now) for a in aggs for h in a.history]
self.conn.execute(insert(PRODUCT.versions), [{"product_id": a.product_id, "version": 1, "saved_at": now} for a in aggs])
upsert(self.conn, prod_product, roots, ("product_id",), existing=set())
upsert(self.conn, prod_product_log, logs, ("product_id", "seq"), existing=set())
def prefetch(self, product_ids: Iterable[str]) -> None:
"""Load these products whole in one go (one query per table per 100 codes) for the gets that follow (PRD-41's batch)."""
loaded = self.__dict__.setdefault("_loaded", {})
ids = [i for i in dict.fromkeys(product_ids) if i not in loaded]
if ids:
for (pid,), rows in PRODUCT.load_many(self.conn, [{"product_id": i} for i in ids]).items():
loaded[pid] = rows
def get_many(self, product_ids: Iterable[str]) -> dict:
"""product_id → Product for the ones that exist (no history): one read per 200 codes, for an import."""
ids = sorted(set(product_ids))
out: dict = {}
for i in range(0, len(ids), 200):
stmt = select(prod_product).where(prod_product.c.product_id.in_(ids[i:i + 200]))
for r in self.conn.execute(stmt):
agg = _make(Product, r._mapping, PROD_PRODUCT_COLUMNS)
agg.cost_amount = _dec(agg.cost_amount, Decimal(0))
out[agg.product_id] = (agg, int(r._mapping["version"] or 0))
return out
def info(self, product_ids: Iterable[str]) -> dict:
"""product_id -> what a price or a conversion needs to know of the product: its units, category and template. Read once
per id per request (a save of the product forgets it)."""
memo = self.__dict__.setdefault("_info", {})
asked = sorted(set(product_ids))
ids = [i for i in asked if i not in memo]
out: dict = {i: memo[i] for i in asked if i in memo and memo[i] is not None}
for i in range(0, len(ids), 200):
c = prod_product.c
stmt = select(c.product_id, c.base_unit_code, c.sales_unit_code, c.category_code, c.status,
c.product_template_id).where(
c.product_id.in_(ids[i:i + 200]))
for r in self.conn.execute(stmt):
out[r[0]] = {"base": r[1] or "", "sales": r[2] or r[1] or "", "category": r[3] or "", "status": r[4],
"template": r[5] or ""}
memo.update({i: out.get(i) for i in ids})
return {i: dict(out[i]) for i in asked if i in out}
def categories_of(self, product_ids: Iterable[str]) -> dict:
ids = sorted(set(product_ids))
out: dict = {}
for i in range(0, len(ids), 200):
stmt = select(prod_product.c.product_id, prod_product.c.category_code).where(
prod_product.c.product_id.in_(ids[i:i + 200]))
out.update({r[0]: r[1] or "" for r in self.conn.execute(stmt)})
return out
def code_in_use(self, code: str) -> Optional[str]:
"""The stored code that `code` collides with, ignoring case (HOST-M and host-m are one SKU), or None."""
r = self.conn.execute(select(prod_product.c.product_id)
.where(func.upper(prod_product.c.product_id) == code.upper())).first()
return r[0] if r else None
def next_code(self, fmt_prefix: str, fmt_suffix: str, width: int) -> str:
"""The next code of a pattern like PRD-{seq:5}: one more than the highest number in use. Two callers
racing for it collide on the key and one is told to retry — never two products with one code."""
like = f"{fmt_prefix}%{fmt_suffix}"
best = 0
for (pid,) in self.conn.execute(select(prod_product.c.product_id).where(prod_product.c.product_id.like(like))):
if not (pid.startswith(fmt_prefix) and pid.endswith(fmt_suffix)):
continue # '_' in a LIKE pattern is a wildcard: check the real prefix
mid = pid[len(fmt_prefix):len(pid) - len(fmt_suffix)] if fmt_suffix else pid[len(fmt_prefix):]
if mid.isdigit():
best = max(best, int(mid))
return f"{fmt_prefix}{best + 1:0{width}d}{fmt_suffix}"
def search(self, *, text: str = "", category: Optional[Iterable[str]] = None, statuses: Optional[Iterable[str]] = None,
limit: int = 50, after: str = "", in_category_on: Optional[date] = None, owners: Optional[Iterable[str]] = None) -> list:
"""Product summaries ordered by code, a page after `after`: for the product search (a hot path). `owners`: only the
products that ARE one of these codes or are of a template that is."""
cols = [prod_product.c[c] for c in ("product_id", "name", "base_unit_code", "sales_unit_code", "category_code",
"tax_category", "status")]
stmt = select(*cols).order_by(prod_product.c.product_id).limit(limit)
if after:
stmt = stmt.where(prod_product.c.product_id > after)
if statuses is not None:
stmt = stmt.where(prod_product.c.status.in_(list(statuses)))
if owners is not None:
ids = sorted(set(owners))
if not ids:
return []
stmt = stmt.where(or_(prod_product.c.product_id.in_(ids), prod_product.c.product_template_id.in_(ids)))
if category is not None:
cats = list(category)
if in_category_on is None:
stmt = stmt.where(prod_product.c.category_code.in_(cats))
else:
# the category it had ON THAT DATE: its latest placement starting on or before it; a product with no
# placement yet (a draft) has its stored category
m, x = prod_category_membership, prod_category_membership.alias("x")
latest = (select(func.max(x.c.valid_from)).where(and_(
x.c.product_or_category == m.c.product_or_category, x.c.valid_from <= in_category_on)).scalar_subquery())
placed = select(func.substr(m.c.product_or_category, 9)).where(m.c.product_or_category.like("product:%"))
stmt = stmt.where(or_(
prod_product.c.product_id.in_(placed.where(and_(m.c.valid_from == latest, m.c.category_code.in_(cats)))),
and_(~prod_product.c.product_id.in_(placed), prod_product.c.category_code.in_(cats))))
if text:
like = f"%{text.lower()}%"
stmt = stmt.where(or_(func.lower(prod_product.c.name).like(like),
func.lower(prod_product.c.description).like(like),
func.lower(prod_product.c.product_id).like(like)))
return [dict(r._mapping) for r in self.conn.execute(stmt)]
def texts(self, exclude: str = "", limit: int = 20000) -> list:
"""(product_id, name, description) of every product, for the duplicate check."""
stmt = select(prod_product.c.product_id, prod_product.c.name, prod_product.c.description).limit(limit)
return [(r[0], r[1] or "", r[2] or "") for r in self.conn.execute(stmt) if r[0] != exclude]
def ids(self) -> list:
return sorted(row[0] for row in self.conn.execute(select(prod_product.c.product_id)))
class PriceListRepository:
"""Loads and saves a price list whole. Takes an open connection: the use case owns the transaction."""
def __init__(self, conn):
self.conn = conn
self._snapshots: dict = {} # price_list_id -> what its load read (see ProductRepository)
def get(self, price_list_id) -> Optional[tuple]:
"""(PriceList, the version it was loaded at), or None."""
rows = PRICE_LIST.load(self.conn, {"price_list_id": price_list_id})
if rows is None:
return None
self._snapshots[price_list_id] = rows.snapshot
agg = _make(PriceList, rows["prod_price_list"][0], PROD_PRICE_LIST_COLUMNS)
return agg, PRICE_LIST.version_of(rows)
def save(self, agg: PriceList, expected_version: int) -> int:
"""Writes version expected_version + 1, or raises Conflict (a 409 through both doors)."""
self.__dict__.pop("_memo_default_on", None)
snapshot = self._snapshots.pop(agg.price_list_id, None)
return PRICE_LIST.save(self.conn, {"prod_price_list": [_pick(agg, PROD_PRICE_LIST_COLUMNS)]},
expected_version=expected_version, snapshot=snapshot)
def all(self) -> list:
"""(PriceList, version) of every list, by code."""
ids = [pid for (pid,) in self.conn.execute(select(prod_price_list.c.price_list_id).order_by(prod_price_list.c.price_list_id))]
out = []
for (pid,), rows in PRICE_LIST.load_many(self.conn, [{"price_list_id": i} for i in ids]).items():
if rows is not None:
self._snapshots[pid] = rows.snapshot
out.append((_make(PriceList, rows["prod_price_list"][0], PROD_PRICE_LIST_COLUMNS), PRICE_LIST.version_of(rows)))
return out
def default_on(self, currency: str, on: date) -> Optional[str]:
"""The currency's default list on a date: the published (or since retired) list with the LATEST default_from on
or before it. One answer for any date; two lists never share a default_from (publish refuses it). Read once per
(currency, date) per repository: a batch asks it for every line (PRD-41); a save of any list forgets it."""
memo = self.__dict__.setdefault("_memo_default_on", {})
if (currency, on) not in memo:
memo[(currency, on)] = self._default_on(currency, on)
return memo[(currency, on)]
def _default_on(self, currency: str, on: date) -> Optional[str]:
r = self.conn.execute(select(prod_price_list.c.price_list_id).where(and_(
prod_price_list.c.currency == currency, prod_price_list.c.default_from.is_not(None),
prod_price_list.c.default_from <= on, prod_price_list.c.status.in_(["published", "retired"])))
.order_by(prod_price_list.c.default_from.desc()).limit(1)).first()
return r[0] if r else None
def has_had_default(self, currency: str) -> bool:
"""Has the currency ever had a default list (published, or since retired)? The very first may start today."""
return self.conn.execute(select(func.count()).select_from(prod_price_list).where(and_(
prod_price_list.c.currency == currency, prod_price_list.c.default_from.is_not(None),
prod_price_list.c.status.in_(["published", "retired"])))).scalar() > 0
def default_clash(self, currency: str, default_from: date, exclude: str) -> str:
"""Another list of the currency that is already the default from this very date, or ''."""
r = self.conn.execute(select(prod_price_list.c.price_list_id).where(and_(
prod_price_list.c.currency == currency, prod_price_list.c.default_from == default_from,
prod_price_list.c.price_list_id != exclude, prod_price_list.c.status.in_(["published", "retired"])))).first()
return r[0] if r else ""
def next_default_from(self, currency: str, after: date) -> Optional[date]:
"""When the next default of the currency starts after `after` (None: the one running then has no successor)."""
r = self.conn.execute(select(func.min(prod_price_list.c.default_from)).where(and_(
prod_price_list.c.currency == currency, prod_price_list.c.default_from > after,
prod_price_list.c.status.in_(["published", "retired"])))).scalar()
return _d(r) if r else None
def ids(self) -> list:
return sorted(row[0] for row in self.conn.execute(select(prod_price_list.c.price_list_id)))
class AssignmentRepository:
"""Which list a customer is priced from, per currency, from a date (prod_customer_price_list). Append-only: an
assignment is never edited, a later one is a new row."""
def __init__(self, conn):
self.conn = conn
def effective(self, bp_id: int, currency: str, on: date) -> str:
"""The list the customer is assigned to on `on`: the latest assignment starting on or before it, or ''. Read once per
(customer, currency, date) per repository (PRD-41's batch asks it for every line); an add forgets it."""
memo = self.__dict__.setdefault("_memo_effective", {})
if (bp_id, currency, on) not in memo:
memo[(bp_id, currency, on)] = self._effective(bp_id, currency, on)
return memo[(bp_id, currency, on)]
def _effective(self, bp_id: int, currency: str, on: date) -> str:
r = self.conn.execute(select(prod_customer_price_list.c.price_list_id).where(and_(
prod_customer_price_list.c.bp_id == bp_id, prod_customer_price_list.c.currency == currency,
prod_customer_price_list.c.valid_from <= on)).order_by(prod_customer_price_list.c.valid_from.desc()).limit(1)).first()
return (r[0] or "") if r else ""
def of(self, bp_id: int, currency: str) -> list:
stmt = select(prod_customer_price_list).where(and_(prod_customer_price_list.c.bp_id == bp_id,
prod_customer_price_list.c.currency == currency)
).order_by(prod_customer_price_list.c.valid_from)
return [{"bp_id": r._mapping["bp_id"], "currency": r._mapping["currency"], "valid_from": _d(r._mapping["valid_from"]),
"price_list_id": r._mapping["price_list_id"]} for r in self.conn.execute(stmt)]
def add(self, bp_id: int, currency: str, valid_from: date, price_list_id: str, by: str) -> None:
self.__dict__.pop("_memo_effective", None)
upsert(self.conn, prod_customer_price_list, [{"bp_id": bp_id, "currency": currency, "valid_from": valid_from,
"price_list_id": price_list_id, "created_by": by}],
("bp_id", "currency", "valid_from"))
def of_list(self, price_list_id: str) -> int:
return self.conn.execute(select(func.count(func.distinct(prod_customer_price_list.c.bp_id))).where(
prod_customer_price_list.c.price_list_id == price_list_id)).scalar() or 0
def ids(self) -> list:
stmt = select(prod_customer_price_list.c.bp_id, prod_customer_price_list.c.currency).distinct()
return sorted(list(row) for row in self.conn.execute(stmt))
class CategoryRepository:
"""Configuration rows (prod_category): a family of products a rule or a report covers. Not an aggregate: no
lifecycle, no version claim — a keyed upsert, edited under the category permission."""
def __init__(self, conn):
self.conn = conn
def all(self) -> list:
stmt = select(prod_category).order_by(prod_category.c.category_code)
return [{k: r._mapping[k] for k in PROD_CATEGORY_COLUMNS} for r in self.conn.execute(stmt)]
def get(self, code: str) -> Optional[dict]:
return next((c for c in self.all() if c["category_code"] == code), None)
def put(self, row: dict) -> None:
upsert(self.conn, prod_category, [{k: v for k, v in row.items() if k in PROD_CATEGORY_COLUMNS}],
("category_code",))
class MembershipRepository:
"""Where products and categories sit, and when they got there (prod_category_membership): append-only, read as of a
date. Every answer to "which category, which parent" is `as of`, so a move that starts tomorrow changes nothing for
today or earlier."""
def __init__(self, conn):
self.conn = conn
self._by: Optional[dict] = None
self._codes: Optional[list] = None
def _load(self) -> dict:
if self._by is None:
out: dict = {}
for r in self.conn.execute(select(prod_category_membership).order_by(
prod_category_membership.c.product_or_category, prod_category_membership.c.valid_from)):
out.setdefault(r._mapping["product_or_category"], []).append(
(_d(r._mapping["valid_from"]), r._mapping["category_code"] or ""))
self._by = out
return self._by
def at(self, subject: str, on: date) -> str:
"""The category code (or parent) `subject` has on `on`: the latest row starting on or before it; '' if none."""
found = ""
for start, code in self._load().get(subject, []):
if start <= on:
found = code
return found
def rows_of(self, subject: str) -> list:
return list(self._load().get(subject, []))
def product_category(self, product_id: str, on: date) -> str:
return self.at(f"product:{product_id}", on)
def parent(self, code: str, on: date) -> str:
return self.at(f"category:{code}", on)
def path(self, code: str, on: date) -> tuple:
"""`code` and every category above it ON THAT DATE, nearest first."""
out, seen = [], set()
while code and code not in seen and len(out) < 12:
out.append(code)
seen.add(code)
code = self.parent(code, on)
return tuple(out)
def under(self, code: str, on: date) -> list:
"""`code` and every category below it on that date."""
if self._codes is None:
self._codes = [r[0] for r in self.conn.execute(select(prod_category.c.category_code))]
return [c for c in self._codes if code in self.path(c, on)]
def add(self, subject: str, valid_from: date, category_code: str, by: str) -> None:
upsert(self.conn, prod_category_membership, [{"product_or_category": subject, "valid_from": valid_from,
"category_code": category_code, "created_by": by}],
("product_or_category", "valid_from"))
self._by = None
class UnitRepository:
"""Configuration rows (prod_unit): how a quantity is counted. Seeded with UN/ECE Recommendation 20 codes."""
def __init__(self, conn):
self.conn = conn
def all(self) -> list:
"""Every unit, read once per request (put() forgets it)."""
if self.__dict__.get("_all") is None:
stmt = select(prod_unit).order_by(prod_unit.c.unit_code)
self.__dict__["_all"] = [{k: r._mapping[k] for k in PROD_UNIT_COLUMNS} for r in self.conn.execute(stmt)]
return [dict(u) for u in self.__dict__["_all"]]
def codes(self) -> set:
return {r[0] for r in self.conn.execute(select(prod_unit.c.unit_code))}
def put(self, row: dict) -> None:
self.__dict__["_all"] = None
upsert(self.conn, prod_unit, [{k: v for k, v in row.items() if k in PROD_UNIT_COLUMNS}], ("unit_code",))
def seed(self) -> None:
have = self.codes()
for code, name, decimals in UNITS_SEED:
if code not in have:
self.put({"unit_code": code, "name": name, "decimals": decimals, "kind": UNIT_KINDS.get(code, "other")})
for u in self.all(): # a unit that predates v4 learns its kind
if not u.get("kind") and u["unit_code"] in UNIT_KINDS:
self.put({**u, "kind": UNIT_KINDS[u["unit_code"]]})
stored = self.fixed()
for frm, to, n, d in CONVERSIONS_SEED:
if (frm, to) not in stored:
upsert(self.conn, prod_unit_conversion, [{"from_unit_code": frm, "to_unit_code": to, "numerator": n,
"denominator": d}], ("from_unit_code", "to_unit_code"))
self.__dict__["_fixed"] = None
def fixed(self) -> dict:
"""(from, to) -> Fraction: the conversions true for every product (read once per request: only seed() writes them)."""
if self.__dict__.get("_fixed") is not None:
return self.__dict__["_fixed"]
self.__dict__["_fixed"] = out = {(r[0], r[1]): Fraction(r[2], r[3]) for r in self.conn.execute(select(
prod_unit_conversion.c.from_unit_code, prod_unit_conversion.c.to_unit_code,
prod_unit_conversion.c.numerator, prod_unit_conversion.c.denominator))}
return out
class ProductUnitRepository:
"""The units a product (or its template) can be counted in (prod_product_unit): exact conversions, packaging levels
and whether a document or stock has used one. Read as the product sees them: its own row overrides its template's."""
def __init__(self, conn):
self.conn = conn
def prefetch(self, owners: Iterable[str]) -> None:
"""Read the owners' unit rows once: of() answers them from memory afterwards. For a READ path only."""
ids = sorted({o for o in owners if o})
got = self.of_many(ids)
self.__dict__.setdefault("_pre", {}).update({o: dict(sorted(got.get(o, {}).items())) for o in ids})
def of(self, owner: str) -> dict:
"""The owner's own unit rows: read once per request (a write here forgets the owner)."""
pre = self.__dict__.setdefault("_pre", {})
if owner not in pre:
stmt = select(prod_product_unit).where(prod_product_unit.c.owner == owner).order_by(prod_product_unit.c.unit_code)
pre[owner] = {r._mapping["unit_code"]: self._view(r._mapping) for r in self.conn.execute(stmt)}
return {u: dict(r) for u, r in pre[owner].items()}
def _forget(self, *owners: str) -> None:
for o in owners:
self.__dict__.get("_pre", {}).pop(o, None)
def effective(self, product_id: str, template_id: str = "", base: str = "", keep: bool = False) -> dict:
"""unit_code -> row for the product: the template's rows, then its own on top. A row stated RELATIVE to another unit
(PK24 = 24 C62 on a template whose bottle differs per variant, PRD-34) is composed here with the product's own row for that
unit, exactly, so every caller sees base-unit numerator / denominator. One whose unit this product does not have is left out
(`keep` returns it flagged `unresolved`)."""
out = self.of(template_id) if template_id else {}
out.update(self.of(product_id))
pending = {u: r for u, r in out.items() if r["relative_unit_code"]}
if not pending:
return out
base = base or (self.conn.execute(select(prod_product.c.base_unit_code).where(prod_product.c.product_id == product_id)).scalar() or "")
done = {u: r for u, r in out.items() if u not in pending}
moved = True
while pending and moved:
moved = False
for u, r in list(pending.items()):
rel = r["relative_unit_code"]
inner = Fraction(1) if rel == base else (Fraction(done[rel]["numerator"], done[rel]["denominator"]) if rel in done else None)
if inner is None:
continue
f = Fraction(r["numerator"], r["denominator"]) * inner
done[u] = {**r, "numerator": f.numerator, "denominator": f.denominator, "composed_of": rel}
del pending[u]
moved = True
if keep:
done.update({u: {**r, "unresolved": True} for u, r in pending.items()})
return {u: done[u] for u in out if u in done}
def put(self, owner: str, row: dict, by: str) -> None:
self._forget(owner)
upsert(self.conn, prod_product_unit, [{
"owner": owner, "unit_code": row["unit_code"], "numerator": row["numerator"],
"denominator": row["denominator"], "packaging_level": row.get("packaging_level") or "none",
"quantity_decimals": row["quantity_decimals"], "gtin": row.get("gtin") or None,
"einvoice_unit_code": row.get("einvoice_unit_code") or None,
"relative_unit_code": row.get("relative_unit_code") or None,
"status": row.get("status") or "active", "created_by": by}],
("owner", "unit_code"))
def of_many(self, owners: Iterable[str]) -> dict:
"""owner -> {unit_code: row} of the owners' OWN rows (no template composition): one read per 200 owners."""
ids, out = sorted(set(owners)), {}
for i in range(0, len(ids), 200):
stmt = select(prod_product_unit).where(prod_product_unit.c.owner.in_(ids[i:i + 200]))
for r in self.conn.execute(stmt):
out.setdefault(r._mapping["owner"], {})[r._mapping["unit_code"]] = self._view(r._mapping)
return out
def put_many(self, rows: list, by: str, existing: set) -> None:
"""Unit rows (each {owner, unit_code, numerator, denominator, packaging_level, quantity_decimals, gtin, einvoice_unit_code,
relative_unit_code}) in one statement; `existing` = the (owner, unit_code) keys already stored (their created_by is kept)."""
self._forget(*{r["owner"] for r in rows})
upsert(self.conn, prod_product_unit, [{
"owner": r["owner"], "unit_code": r["unit_code"], "numerator": r["numerator"], "denominator": r["denominator"],
"packaging_level": r.get("packaging_level") or "none", "quantity_decimals": r["quantity_decimals"],
"gtin": r.get("gtin") or None, "einvoice_unit_code": r.get("einvoice_unit_code") or None,
"relative_unit_code": r.get("relative_unit_code") or None, "status": r.get("status") or "active", "created_by": by}
for r in rows], ("owner", "unit_code"), existing=existing)
def set_gtin(self, owner: str, unit_code: str, gtin: str) -> None:
self._forget(owner)
upsert(self.conn, prod_product_unit, [{"owner": owner, "unit_code": unit_code, "gtin": gtin or None}], ("owner", "unit_code"))
def knows(self, unit_code: str) -> bool:
"""Does any product (or template) count in this unit code?"""
return self.conn.execute(select(func.count()).select_from(prod_product_unit).where(
prod_product_unit.c.unit_code == unit_code)).scalar() > 0
def mark_used(self, owner: str, unit_code: str, at, ref: str = "") -> None:
row = self.of(owner).get(unit_code)
if row is not None and row["used_at"] is None:
self._forget(owner)
upsert(self.conn, prod_product_unit, [{"owner": owner, "unit_code": unit_code, "used_at": at, "used_ref": ref or None}],
("owner", "unit_code"))
@staticmethod
def _view(m) -> dict:
return {"owner": m["owner"], "unit_code": m["unit_code"], "numerator": int(m["numerator"]),
"denominator": int(m["denominator"]), "packaging_level": m["packaging_level"] or "none",
"quantity_decimals": m["quantity_decimals"], "gtin": m["gtin"] or "", "einvoice_unit_code": m["einvoice_unit_code"] or "", "status": m["status"] or "active", "used_at": m["used_at"],
"relative_unit_code": m["relative_unit_code"] or "", "used_ref": m["used_ref"] or ""}
def _stamp(v) -> datetime:
"""A sort key for a timestamp that may be naive (stored) or aware (a driver), or missing: all compare with each other."""
if v is None:
return datetime.min
return v.astimezone(timezone.utc).replace(tzinfo=None) if v.tzinfo else v
class GtinRepository:
"""GTIN claims (prod_gtin): one barcode, one product-and-unit, for ever across the whole catalogue (retired ones included).
A unit has any number of GTINs and, while it has any active one, exactly one MAIN."""
def __init__(self, conn):
self.conn = conn
@staticmethod
def _view(m) -> dict:
return {"gtin": m["gtin"], "owner": m["owner"], "unit_code": m["unit_code"], "kind": m["kind"] or "main",
"status": m["status"] or "active", "retired_on": _d(m["retired_on"]), "created_at": m["created_at"]}
def holder(self, gtin: str) -> Optional[dict]:
r = self.conn.execute(select(prod_gtin).where(prod_gtin.c.gtin == gtin)).first()
return None if r is None else self._view(r._mapping)
def of_unit(self, owner: str, unit_code: str) -> list:
"""Every GTIN of a unit, retired ones last-written; oldest first."""
stmt = select(prod_gtin).where(and_(prod_gtin.c.owner == owner, prod_gtin.c.unit_code == unit_code))
return sorted((self._view(r._mapping) for r in self.conn.execute(stmt)),
key=lambda g: (_stamp(g["created_at"]), g["gtin"]))
def of_owner(self, owner: str) -> list:
stmt = select(prod_gtin).where(prod_gtin.c.owner == owner)
return sorted((self._view(r._mapping) for r in self.conn.execute(stmt)),
key=lambda g: (g["unit_code"], _stamp(g["created_at"]), g["gtin"]))
def claim(self, gtin: str, owner: str, unit_code: str, by: str, kind: str = "main") -> None:
"""A PLAIN insert: the key is the barcode, so a second claimant fails 23505 (a 409) and never both win."""
self.conn.execute(insert(prod_gtin).values(gtin=gtin, owner=owner, unit_code=unit_code, created_by=by, kind=kind,
status="active", created_at=datetime.now(timezone.utc).replace(tzinfo=None)))
def main_of(self, owner: str, unit_code: str) -> Optional[str]:
"""The gtin the unit's main claim names ('' once its last active GTIN retired), None if the unit has no claim row yet."""
r = self.conn.execute(select(prod_gtin_main.c.gtin).where(and_(
prod_gtin_main.c.owner == owner, prod_gtin_main.c.unit_code == unit_code))).first()
return None if r is None else (r[0] or "")
def claim_main(self, owner: str, unit_code: str, gtin: str, by: str, seen: object = ...) -> None:
"""The unit's one main claim: a plain INSERT when the row did not exist when the caller read the unit (`seen` is what it read;
the loser of a race fails 23505), an update of the same key otherwise."""
now = datetime.now(timezone.utc).replace(tzinfo=None)
held = self.main_of(owner, unit_code) if seen is ... else seen
if held is None:
self.conn.execute(insert(prod_gtin_main).values(owner=owner, unit_code=unit_code, gtin=gtin, updated_by=by, updated_at=now))
elif held != gtin:
upsert(self.conn, prod_gtin_main, [{"owner": owner, "unit_code": unit_code, "gtin": gtin, "updated_by": by, "updated_at": now}],
("owner", "unit_code"))
def holders(self, gtins: Iterable[str]) -> dict:
"""gtin -> claim for the ones held: one read per 200."""
ids, out = sorted(set(gtins)), {}
for i in range(0, len(ids), 200):
for r in self.conn.execute(select(prod_gtin).where(prod_gtin.c.gtin.in_(ids[i:i + 200]))):
out[r._mapping["gtin"]] = self._view(r._mapping)
return out
def of_owners(self, owners: Iterable[str]) -> dict:
"""owner -> its GTINs (all units, retired too), oldest first: one read per 200 owners."""
ids, out = sorted(set(owners)), {}
for i in range(0, len(ids), 200):
for r in self.conn.execute(select(prod_gtin).where(prod_gtin.c.owner.in_(ids[i:i + 200]))):
out.setdefault(r._mapping["owner"], []).append(self._view(r._mapping))
for rows in out.values():
rows.sort(key=lambda g: (g["unit_code"], _stamp(g["created_at"]), g["gtin"]))
return out
def mains_of(self, owners: Iterable[str]) -> dict:
"""(owner, unit_code) -> the gtin its main claim names ('' when none left): one read per 200 owners."""
ids, out = sorted(set(owners)), {}
for i in range(0, len(ids), 200):
for r in self.conn.execute(select(prod_gtin_main).where(prod_gtin_main.c.owner.in_(ids[i:i + 200]))):
out[(r._mapping["owner"], r._mapping["unit_code"])] = r._mapping["gtin"] or ""
return out
def claim_many(self, rows: list, by: str) -> None:
"""Claims in one plain INSERT: rows of {gtin, owner, unit_code, kind}."""
if rows:
now = datetime.now(timezone.utc).replace(tzinfo=None)
self.conn.execute(insert(prod_gtin), [{**r, "created_by": by, "status": "active", "created_at": now} for r in rows])
def claim_main_many(self, fresh: list, moved: list, by: str) -> None:
"""Main claims: `fresh` (units with no claim row yet) as one plain INSERT, `moved` (an existing row) as one upsert."""
now = datetime.now(timezone.utc).replace(tzinfo=None)
if fresh:
self.conn.execute(insert(prod_gtin_main), [{**r, "updated_by": by, "updated_at": now} for r in fresh])
if moved:
upsert(self.conn, prod_gtin_main, [{**r, "updated_by": by, "updated_at": now} for r in moved], ("owner", "unit_code"))
def set_kind(self, gtin: str, kind: str) -> None:
upsert(self.conn, prod_gtin, [{"gtin": gtin, "kind": kind}], ("gtin",))
def retire(self, gtin: str, on: date) -> None:
upsert(self.conn, prod_gtin, [{"gtin": gtin, "status": "retired", "kind": "additional", "retired_on": on}], ("gtin",))
def release(self, gtin: str) -> None:
self.conn.execute(delete(prod_gtin).where(prod_gtin.c.gtin == gtin))
def unit_of(self, gtin: str) -> Optional[tuple]:
"""(owner, unit row or None) behind a barcode, or None. A GTIN on the base unit has no unit row."""
held = self.holder(gtin)
if held is None:
return None
r = self.conn.execute(select(prod_product_unit).where(and_(
prod_product_unit.c.owner == held["owner"], prod_product_unit.c.unit_code == held["unit_code"]))).first()
return held["owner"], (None if r is None else ProductUnitRepository._view(r._mapping))
def _n(v) -> Optional[Decimal]:
return None if v is None else Decimal(str(v))
class MeasureRepository:
"""A unit's dated weights, volume and dimensions (prod_unit_measure). Read as the row in force on a date; a SKU's own
row wins over its template's."""
def __init__(self, conn):
self.conn = conn
@staticmethod
def _view(m) -> dict:
out = {"owner": m["owner"], "unit_code": m["unit_code"], "valid_from": _d(m["valid_from"]), "copied": bool(m["copied"]),
"set_by": m["set_by"] or ""}
out.update({k: (m[k] if k.endswith("_code") else _n(m[k])) for k in MEASURE_FIELDS})
return out
def get(self, owner: str, unit_code: str, valid_from: date) -> Optional[dict]:
r = self.conn.execute(select(prod_unit_measure).where(and_(
prod_unit_measure.c.owner == owner, prod_unit_measure.c.unit_code == unit_code,
prod_unit_measure.c.valid_from == valid_from))).first()
return None if r is None else self._view(r._mapping)
def get_many(self, owners: Iterable[str], valid_from: date) -> dict:
"""(owner, unit_code) -> the row dated `valid_from`, for the owners: one read per 200 owners."""
ids, out = sorted(set(owners)), {}
for i in range(0, len(ids), 200):
for r in self.conn.execute(select(prod_unit_measure).where(and_(
prod_unit_measure.c.owner.in_(ids[i:i + 200]), prod_unit_measure.c.valid_from == valid_from))):
out[(r._mapping["owner"], r._mapping["unit_code"])] = self._view(r._mapping)
return out
def add_many(self, rows: list, by: str, at) -> None:
"""NEW rows (each {owner, unit_code, valid_from, <MEASURE_FIELDS>}) in one plain INSERT."""
self.__dict__["_rows"] = {}
if rows:
self.conn.execute(insert(prod_unit_measure), [{
"owner": r["owner"], "unit_code": r["unit_code"], "valid_from": r["valid_from"], "set_by": by, "set_at": at,
"copied": False, "created_at": at, **{k: r.get(k) for k in MEASURE_FIELDS}} for r in rows])
def rows(self, owner: str, unit_code: str) -> list:
"""The unit's dated rows, read once per request (a write here forgets them)."""
memo = self.__dict__.setdefault("_rows", {})
if (owner, unit_code) not in memo:
stmt = select(prod_unit_measure).where(and_(prod_unit_measure.c.owner == owner, prod_unit_measure.c.unit_code == unit_code))
memo[(owner, unit_code)] = sorted((self._view(r._mapping) for r in self.conn.execute(stmt)), key=lambda x: x["valid_from"])
return [dict(x) for x in memo[(owner, unit_code)]]
def in_force(self, product_id: str, template_id: str, unit_code: str, on: date) -> Optional[dict]:
"""The row in force on `on` for the SKU: its own latest row on or before the date, else its template's."""
for owner in (product_id, template_id):
if not owner:
continue
live = [x for x in self.rows(owner, unit_code) if x["valid_from"] <= on]
if live:
return live[-1]
return None
def add(self, owner: str, unit_code: str, valid_from: date, values: dict, by: str, at) -> None:
"""A NEW row: a plain insert, so two writers racing for one (owner, unit, date) cannot both win (a 409)."""
self.__dict__["_rows"] = {}
self.conn.execute(insert(prod_unit_measure).values(
owner=owner, unit_code=unit_code, valid_from=valid_from, set_by=by, set_at=at, copied=False,
created_at=at, **{k: values.get(k) for k in MEASURE_FIELDS}))
def replace(self, owner: str, unit_code: str, valid_from: date, values: dict, by: str, at) -> None:
"""The row for today again, while no document has copied it: same key, new values (never a delete and an insert)."""
self.__dict__["_rows"] = {}
upsert(self.conn, prod_unit_measure, [{"owner": owner, "unit_code": unit_code, "valid_from": valid_from, "set_by": by,
"set_at": at, **{k: values.get(k) for k in MEASURE_FIELDS}}],
("owner", "unit_code", "valid_from"))
def mark_copied(self, owner: str, unit_code: str, valid_from: date) -> None:
self.__dict__["_rows"] = {}
upsert(self.conn, prod_unit_measure, [{"owner": owner, "unit_code": unit_code, "valid_from": valid_from, "copied": True}],
("owner", "unit_code", "valid_from"))
class RoleRepository:
"""A unit role from a date, and its rounding rule (prod_unit_role): one row per (owner, role, date) is the whole state of
that role from then on. A SKU's own rows win over its template's."""
def __init__(self, conn):
self.conn = conn
@staticmethod
def _view(m) -> dict:
return {"owner": m["owner"], "role": m["role"], "valid_from": _d(m["valid_from"]), "unit_code": m["unit_code"] or "",
"min_quantity": _n(m["min_quantity"]), "rounding_multiple": _n(m["rounding_multiple"]),
"rule_unit_code": m["rule_unit_code"] or "", "set_by": m["set_by"] or ""}
def prefetch(self, owners: Iterable[str]) -> None:
"""Read the owners' role rows once (one query per 200): rows() answers them from memory afterwards. For a READ path only
(a write after it would not be seen)."""
ids = sorted({o for o in owners if o})
pre = self.__dict__.setdefault("_pre", {})
for i in range(0, len(ids), 200):
got = {o: [] for o in ids[i:i + 200]}
for r in self.conn.execute(select(prod_unit_role).where(prod_unit_role.c.owner.in_(ids[i:i + 200]))):
got[r._mapping["owner"]].append(self._view(r._mapping))
pre.update({o: sorted(v, key=lambda x: (x["role"], x["valid_from"])) for o, v in got.items()})
def rows(self, owner: str, role: Optional[str] = None) -> list:
pre = self.__dict__.setdefault("_pre", {})
if owner not in pre: # every role of the owner in one read, once per request (a write forgets it)
stmt = select(prod_unit_role).where(prod_unit_role.c.owner == owner)
pre[owner] = sorted((self._view(r._mapping) for r in self.conn.execute(stmt)), key=lambda x: (x["role"], x["valid_from"]))
return [dict(x) for x in pre[owner] if not role or x["role"] == role]
def get(self, owner: str, role: str, valid_from: date) -> Optional[dict]:
return next((x for x in self.rows(owner, role) if x["valid_from"] == valid_from), None)
def in_force(self, product_id: str, template_id: str, role: str, on: date) -> Optional[dict]:
for owner in (product_id, template_id):
if not owner:
continue
live = [x for x in self.rows(owner, role) if x["valid_from"] <= on]
if live:
return live[-1]
return None
def ever_ruled(self, product_id: str, template_id: str, role: str) -> bool:
"""Has this product (or its template) ever had a rule for the role, on any date?"""
return any(x["min_quantity"] is not None or x["rounding_multiple"] is not None
for owner in (product_id, template_id) if owner for x in self.rows(owner, role))
def add(self, owner: str, row: dict, by: str, at) -> None:
"""A NEW role row: a plain insert (a 409 for the second of two writers of one date)."""
self.__dict__.get("_pre", {}).pop(owner, None)
self.conn.execute(insert(prod_unit_role).values(
owner=owner, role=row["role"], valid_from=row["valid_from"], unit_code=row["unit_code"],
min_quantity=row.get("min_quantity"), rounding_multiple=row.get("rounding_multiple"),
rule_unit_code=row.get("rule_unit_code") or None, set_by=by, created_at=at))
def replace(self, owner: str, row: dict, by: str) -> None:
self.__dict__.get("_pre", {}).pop(owner, None)
upsert(self.conn, prod_unit_role, [{
"owner": owner, "role": row["role"], "valid_from": row["valid_from"], "unit_code": row["unit_code"],
"min_quantity": row.get("min_quantity"), "rounding_multiple": row.get("rounding_multiple"),
"rule_unit_code": row.get("rule_unit_code") or None, "set_by": by}], ("owner", "role", "valid_from"))
class TemplateRepository:
"""Product templates (prod_template): what a family of variants shares. Read only through its products."""
def __init__(self, conn):
self.conn = conn
def get(self, template_id: str) -> Optional[dict]:
r = self.conn.execute(select(prod_template).where(prod_template.c.product_template_id == template_id)).first()
if r is None:
return None
out = {k: r._mapping[k] for k in PROD_TEMPLATE_COLUMNS}
out["attributes"] = json.loads(out["attributes"] or "[]")
return out
def put(self, row: dict) -> None:
data = {k: row.get(k) for k in PROD_TEMPLATE_COLUMNS}
data["attributes"] = json.dumps(row.get("attributes") or [], sort_keys=True)
upsert(self.conn, prod_template, [data], ("product_template_id",))
def products(self, template_id: str) -> list:
"""(product_id, attribute_values dict) of every product of the template."""
stmt = select(prod_product.c.product_id, prod_product.c.attribute_values).where(
prod_product.c.product_template_id == template_id).order_by(prod_product.c.product_id)
return [(r[0], json.loads(r[1] or "{}")) for r in self.conn.execute(stmt)]
def _json_list(raw) -> list:
return json.loads(raw) if raw else []
class TypeRepository:
"""Product types (prod_product_type) and their code series (prod_code_series); what a document or stock has used (prod_product_use)."""
def __init__(self, conn):
self.conn = conn
@staticmethod
def _view(m) -> dict:
out = {k: m[k] for k in PROD_PRODUCT_TYPE_COLUMNS}
out["required_roles"] = _json_list(out["required_roles"])
out["code_format"] = out["code_format"] or "typed"
out["status"] = out["status"] or "active"
return out
def get(self, code: str) -> Optional[dict]:
r = self.conn.execute(select(prod_product_type).where(prod_product_type.c.product_type_code == code)).first()
return None if r is None else self._view(r._mapping)
def all(self) -> list:
return [self._view(r._mapping) for r in self.conn.execute(select(prod_product_type).order_by(prod_product_type.c.product_type_code))]
def put(self, row: dict) -> None:
data = {k: row.get(k) for k in PROD_PRODUCT_TYPE_COLUMNS}
data["required_roles"] = json.dumps(list(row.get("required_roles") or []))
upsert(self.conn, prod_product_type, [data], ("product_type_code",))
def seed(self) -> None:
have = {t["product_type_code"] for t in self.all()}
for code, name, kind, stocked, sellable, purchasable, roles, fmt, valuation in TYPES_SEED:
if code not in have:
self.put({"product_type_code": code, "name": name, "kind": kind, "stocked_default": stocked, "sellable": sellable,
"purchasable": purchasable, "required_roles": roles, "code_format": fmt, "valuation_class": valuation,
"production_fields": False, "status": "active"})
def next_number(self, type_code: str, pre: str, post: str) -> int:
"""One more than the highest number claimed in the series AND the highest in any existing code of the pattern (a migrated
series continues after its highest migrated code)."""
claimed = self.conn.execute(select(func.max(prod_code_series.c.number)).where(
prod_code_series.c.product_type_code == type_code)).scalar() or 0
best = claimed
for (pid,) in self.conn.execute(select(prod_product.c.product_id).where(prod_product.c.product_id.like(f"{pre}%{post}"))):
if pid.startswith(pre) and pid.endswith(post):
mid = pid[len(pre):len(pid) - len(post)] if post else pid[len(pre):]
if mid.isdigit():
best = max(best, int(mid))
return best + 1
def claim_number(self, type_code: str, number: int, product_id: str, at) -> None:
"""A plain INSERT: two adds racing for one number cannot both win (the loser is a 409 and runs again on the next)."""
self.conn.execute(insert(prod_code_series).values(product_type_code=type_code, number=number, product_id=product_id, claimed_at=at, created_at=at))
def used(self, product_id: str) -> Optional[str]:
r = self.conn.execute(select(prod_product_use.c.used_ref).where(prod_product_use.c.product_id == product_id)).first()
return None if r is None else (r[0] or "a document or stock")
def mark_used(self, product_id: str, at, ref: str) -> None:
if self.used(product_id) is None:
upsert(self.conn, prod_product_use, [{"product_id": product_id, "used_at": at, "used_ref": ref}], ("product_id",))
class CharacteristicRepository:
"""Characteristics and classes (configuration), and the dated class / value rows (prod_classification)."""
def __init__(self, conn):
self.conn = conn
def characteristic(self, code: str) -> Optional[dict]:
r = self.conn.execute(select(prod_characteristic).where(prod_characteristic.c.characteristic_code == code)).first()
return None if r is None else self._char(r._mapping)
@staticmethod
def _char(m) -> dict:
out = {k: m[k] for k in PROD_CHARACTERISTIC_COLUMNS}
out["allowed_values"] = _json_list(out["allowed_values"])
out["multi_value"] = bool(out["multi_value"])
out["decimals"] = 4 if out["decimals"] is None else out["decimals"]
return out
def characteristics(self) -> list:
return [self._char(r._mapping) for r in self.conn.execute(select(prod_characteristic).order_by(prod_characteristic.c.characteristic_code))]
def put_characteristic(self, row: dict) -> None:
data = {k: row.get(k) for k in PROD_CHARACTERISTIC_COLUMNS}
data["allowed_values"] = json.dumps(list(row.get("allowed_values") or []))
upsert(self.conn, prod_characteristic, [data], ("characteristic_code",))
def klass(self, code: str) -> Optional[dict]:
r = self.conn.execute(select(prod_class).where(prod_class.c.class_code == code)).first()
return None if r is None else {"class_code": r[0], "name": r[1], "characteristics": _json_list(r[2])}
def classes(self) -> list:
return [{"class_code": r[0], "name": r[1], "characteristics": _json_list(r[2])}
for r in self.conn.execute(select(prod_class.c.class_code, prod_class.c.name, prod_class.c.characteristics).order_by(prod_class.c.class_code))]
def put_class(self, row: dict) -> None:
upsert(self.conn, prod_class, [{"class_code": row["class_code"], "name": row.get("name"),
"characteristics": json.dumps(row.get("characteristics") or [])}], ("class_code",))
@staticmethod
def _row(m) -> dict:
return {"owner": m["owner"], "class_code": m["class_code"], "characteristic_code": m["characteristic_code"],
"valid_from": _d(m["valid_from"]), "value": _json_list(m["value"]), "set_by": m["set_by"] or ""}
def rows(self, owners: Iterable[str]) -> list:
owners = [o for o in set(owners) if o]
if not owners:
return []
stmt = select(prod_classification).where(prod_classification.c.owner.in_(owners))
return sorted((self._row(r._mapping) for r in self.conn.execute(stmt)),
key=lambda x: (x["owner"], x["class_code"], x["characteristic_code"], x["valid_from"]))
def owners_with(self, characteristic_codes: Iterable[str], class_code: Optional[str] = None) -> set:
"""The owners (products or templates) that have a row for EVERY one of the characteristics (and, with `class_code`, in that
class): the superset a characteristic search narrows to in SQL. With no characteristic: the owners in the class."""
codes = sorted(set(characteristic_codes)) or [""]
stmt = select(prod_classification.c.owner, prod_classification.c.characteristic_code).where(
prod_classification.c.characteristic_code.in_(codes)).distinct()
if class_code:
stmt = stmt.where(prod_classification.c.class_code == class_code)
have: dict = {}
for o, c in self.conn.execute(stmt):
have.setdefault(o, set()).add(c)
return {o for o, got in have.items() if got >= set(codes)}
def every_row(self) -> list:
return sorted((self._row(r._mapping) for r in self.conn.execute(select(prod_classification))),
key=lambda x: (x["owner"], x["class_code"], x["characteristic_code"], x["valid_from"]))
def get(self, owner: str, class_code: str, characteristic_code: str, valid_from: date) -> Optional[dict]:
r = self.conn.execute(select(prod_classification).where(and_(
prod_classification.c.owner == owner, prod_classification.c.class_code == class_code,
prod_classification.c.characteristic_code == characteristic_code, prod_classification.c.valid_from == valid_from))).first()
return None if r is None else self._row(r._mapping)
def add(self, owner: str, class_code: str, characteristic_code: str, valid_from: date, value: list, by: str, at) -> None:
"""A NEW row: a plain insert (a 409 for the second of two writers of one date)."""
self.conn.execute(insert(prod_classification).values(owner=owner, class_code=class_code, characteristic_code=characteristic_code,
valid_from=valid_from, value=json.dumps(value), set_by=by, created_at=at))
def replace(self, owner: str, class_code: str, characteristic_code: str, valid_from: date, value: list, by: str) -> None:
upsert(self.conn, prod_classification, [{"owner": owner, "class_code": class_code, "characteristic_code": characteristic_code,
"valid_from": valid_from, "value": json.dumps(value), "set_by": by}],
("owner", "class_code", "characteristic_code", "valid_from"))
class SiteRepository:
"""Site data (prod_site): dated rows per product and site, and the rounding profiles they name (prod_rounding_profile)."""
def __init__(self, conn):
self.conn = conn
@staticmethod
def _view(m) -> dict:
out = {"product_id": m["product_id"], "site_id": m["site_id"], "valid_from": _d(m["valid_from"]), "set_by": m["set_by"] or ""}
for k in SITE_FIELDS:
out[k] = _n(m[k]) if k in ("reorder_point", "safety_stock", "max_stock", "fixed_lot_size") else m[k]
return out
def rows(self, product_id: str, site_id: str) -> list:
stmt = select(prod_site).where(and_(prod_site.c.product_id == product_id, prod_site.c.site_id == site_id))
return sorted((self._view(r._mapping) for r in self.conn.execute(stmt)), key=lambda x: x["valid_from"])
def in_force(self, product_id: str, site_id: str, on: date) -> Optional[dict]:
live = [x for x in self.rows(product_id, site_id) if x["valid_from"] <= on]
return live[-1] if live else None
def at_site(self, site_id: str) -> list:
"""Every row of the site, any product, any date: the planner reads the ones in force on a date."""
stmt = select(prod_site).where(prod_site.c.site_id == site_id)
return sorted((self._view(r._mapping) for r in self.conn.execute(stmt)), key=lambda x: (x["product_id"], x["valid_from"]))
def using(self, profile_code: str) -> list:
stmt = select(prod_site).where(prod_site.c.rounding_profile_code == profile_code)
return [self._view(r._mapping) for r in self.conn.execute(stmt)]
def add(self, row: dict, by: str, at) -> None:
"""A NEW row: a plain insert (a 409 for the second of two writers of one product, site and date)."""
self.conn.execute(insert(prod_site).values(**{k: row.get(k) for k in ("product_id", "site_id", "valid_from", *SITE_FIELDS)},
set_by=by, created_at=at))
def replace(self, row: dict, by: str) -> None:
upsert(self.conn, prod_site, [{**{k: row.get(k) for k in ("product_id", "site_id", "valid_from", *SITE_FIELDS)}, "set_by": by}],
("product_id", "site_id", "valid_from"))
def profile(self, code: str) -> Optional[dict]:
r = self.conn.execute(select(prod_rounding_profile).where(prod_rounding_profile.c.rounding_profile_code == code)).first()
return None if r is None else {"rounding_profile_code": r[0], "unit_code": r[1], "steps": _json_list(r[2]), "status": r[3] or "active"}
def profiles(self) -> list:
return [self.profile(r[0]) for r in self.conn.execute(select(prod_rounding_profile.c.rounding_profile_code).order_by(
prod_rounding_profile.c.rounding_profile_code))]
def put_profile(self, row: dict) -> None:
upsert(self.conn, prod_rounding_profile, [{"rounding_profile_code": row["rounding_profile_code"], "unit_code": row["unit_code"],
"steps": json.dumps(row["steps"]), "status": row.get("status") or "active"}],
("rounding_profile_code",))
class PriceRepository:
"""The dated rows and the rules of every list, and the PriceBook price resolution reads.
As a PriceBook (`head`, `rows`, `rules`) it answers with LIVE rows only: a draft row has priced nothing yet.
One resolution is a few small reads — the list, the product's rows, the list's rules — cached for the life of
this repository (one request)."""
def __init__(self, conn):
self.conn = conn
self._heads: dict = {}
self._rows: dict = {}
self._rules: dict = {}
# ── the PriceBook ─────────────────────────────────────────────────────────────────────────────────────
def head(self, price_list_id: str) -> Optional[ListHead]:
if price_list_id not in self._heads:
r = self.conn.execute(select(prod_price_list).where(prod_price_list.c.price_list_id == price_list_id)).first()
self._heads[price_list_id] = None if r is None else ListHead(
r._mapping["price_list_id"], r._mapping["currency"] or "", r._mapping["based_on"] or "",
r._mapping["rounding"] or "0.01", r._mapping["status"] or "", bool(r._mapping["prices_include_tax"]),
_d(r._mapping["retired_on"]))
return self._heads[price_list_id]
def rows(self, price_list_id: str, product_id: str) -> list:
key = (price_list_id, product_id)
if key not in self._rows:
self._rows[key] = self.rows_of(price_list_id, product_id=product_id, status=RowState.LIVE.value)
return self._rows[key]
def rules(self, price_list_id: str) -> list:
if price_list_id not in self._rules:
self._rules[price_list_id] = self.rules_of(price_list_id, status=RowState.LIVE.value)
return self._rules[price_list_id]
# ── reading rows and rules, any status ────────────────────────────────────────────────────────────────
def rows_of(self, price_list_id: str, *, product_id: Optional[str] = None, status: Optional[str] = None,
limit: Optional[int] = None) -> list:
stmt = (select(prod_price).where(prod_price.c.price_list_id == price_list_id)
.order_by(prod_price.c.product_id, prod_price.c.unit_code, prod_price.c.valid_from))
if product_id is not None:
stmt = stmt.where(prod_price.c.product_id == product_id)
if status is not None:
stmt = stmt.where(prod_price.c.status == status)
if limit:
stmt = stmt.limit(limit)
return [self._row(r._mapping) for r in self.conn.execute(stmt)]
def rules_of(self, price_list_id: str, *, status: Optional[str] = None) -> list:
stmt = select(prod_price_rule).where(prod_price_rule.c.price_list_id == price_list_id).order_by(prod_price_rule.c.seq)
if status is not None:
stmt = stmt.where(prod_price_rule.c.status == status)
return [self._rule(r._mapping) for r in self.conn.execute(stmt)]
def stored_rows(self, price_list_id: str, keys: Iterable[tuple]) -> dict:
"""(product_id, unit_code, valid_from) → (PriceRow, status, author) for the keys that exist, any status."""
keys = sorted(set(keys), key=repr)
out: dict = {}
for i in range(0, len(keys), 200): # bounded statements for a big import
chunk = keys[i:i + 200]
cond = or_(*(and_(prod_price.c.product_id == p, prod_price.c.unit_code == u, prod_price.c.valid_from == v)
for p, u, v in chunk))
for r in self.conn.execute(select(prod_price).where(and_(prod_price.c.price_list_id == price_list_id, cond))):
row = self._row(r._mapping)
out[(row.product_id, row.unit_code, row.valid_from)] = (row, r._mapping["status"], r._mapping["created_by"] or "")
return out
def stored_rules(self, price_list_id: str) -> dict:
"""seq → (PriceRule, status)."""
return {r.seq: (r, s) for r, s in self._rules_with_status(price_list_id)}
def _rules_with_status(self, price_list_id: str) -> list:
stmt = select(prod_price_rule).where(prod_price_rule.c.price_list_id == price_list_id)
return [(self._rule(r._mapping), r._mapping["status"]) for r in self.conn.execute(stmt)]
def pending(self, price_list_id: str) -> tuple:
"""(draft rows, draft rules, the people who wrote them): what publishing would make live."""
stmt = select(prod_price).where(and_(prod_price.c.price_list_id == price_list_id,
prod_price.c.status == RowState.DRAFT.value))
rows = [(self._row(r._mapping), r._mapping["created_by"] or "") for r in self.conn.execute(stmt)]
rules = [(r, who) for r, who in self._rules_pending(price_list_id)]
return [r for r, _ in rows], [r for r, _ in rules], {w for _, w in rows if w} | {w for _, w in rules if w}
def _rules_pending(self, price_list_id: str) -> list:
stmt = select(prod_price_rule).where(and_(prod_price_rule.c.price_list_id == price_list_id,
prod_price_rule.c.status == RowState.DRAFT.value))
return [(self._rule(r._mapping), r._mapping["created_by"] or "") for r in self.conn.execute(stmt)]
def currency_has_published_price(self, currency: str) -> bool:
"""Has anything been priced in this currency yet: a live row or rule on a published (or since retired) list?"""
lists = select(prod_price_list.c.price_list_id).where(and_(
prod_price_list.c.currency == currency, prod_price_list.c.status.in_(["published", "retired"])))
for table in (prod_price, prod_price_rule):
n = self.conn.execute(select(func.count()).select_from(table).where(and_(
table.c.status == RowState.LIVE.value, table.c.price_list_id.in_(lists)))).scalar() or 0
if n:
return True
return False
def count(self, price_list_id: str) -> dict:
def n(table, status):
return self.conn.execute(select(func.count()).select_from(table).where(and_(
table.c.price_list_id == price_list_id, table.c.status == status))).scalar() or 0
return {"rows_live": n(prod_price, "live"), "rows_pending": n(prod_price, "draft"),
"rules_live": n(prod_price_rule, "live"), "rules_pending": n(prod_price_rule, "draft")}
def products_priced(self, price_list_id: str) -> list:
return sorted({r[0] for r in self.conn.execute(select(prod_price.c.product_id).where(
prod_price.c.price_list_id == price_list_id))})
# ── writing: only ever new rows ───────────────────────────────────────────────────────────────────────
def add_rows(self, price_list_id: str, rows: Iterable[RowIn], by: str) -> None:
upsert(self.conn, prod_price, [{
"price_list_id": price_list_id, "product_id": r.product_id, "unit_code": r.unit_code,
"valid_from": r.valid_from, "unit_price": r.unit_price, "valid_to": r.valid_to,
"pricing_quantity": r.pricing_quantity, "status": RowState.DRAFT.value, "created_by": by} for r in rows],
("price_list_id", "product_id", "unit_code", "valid_from"))
def put_rules(self, price_list_id: str, rules: Iterable[RuleIn], by: str) -> None:
upsert(self.conn, prod_price_rule, [{
"price_list_id": price_list_id, "seq": r.seq, "applies_to": r.applies_to, "target": r.target or "",
"min_quantity": r.min_quantity, "valid_from": r.valid_from, "valid_to": r.valid_to, "method": r.method,
"discount_pct": r.discount_pct, "fixed_price": r.fixed_price, "surcharge_amount": r.surcharge_amount,
"base_list": r.base_list or "", "unit_code": r.unit_code or "", "pricing_quantity": r.pricing_quantity,
"status": RowState.DRAFT.value, "created_by": by} for r in rules],
("price_list_id", "seq"))
def go_live(self, price_list_id: str) -> int:
"""Publish: every draft row and rule of the list becomes live, with its price untouched. Returns how many."""
rows, rules, _ = self.pending(price_list_id)
stored = self.stored_rows(price_list_id, [(r.product_id, r.unit_code, r.valid_from) for r in rows])
upsert(self.conn, prod_price, [{
"price_list_id": r.price_list_id, "product_id": r.product_id, "unit_code": r.unit_code,
"valid_from": r.valid_from, "unit_price": r.unit_price, "valid_to": r.valid_to,
"pricing_quantity": r.pricing_quantity, "status": RowState.LIVE.value, "created_by": stored[(r.product_id, r.unit_code, r.valid_from)][2]} for r in rows],
("price_list_id", "product_id", "unit_code", "valid_from"))
creators = {r.seq: who for r, who in self._rules_pending(price_list_id)}
upsert(self.conn, prod_price_rule, [{
"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, "fixed_price": r.fixed_price, "surcharge_amount": r.surcharge_amount,
"base_list": r.base_list, "unit_code": r.unit_code, "pricing_quantity": r.pricing_quantity, "status": RowState.LIVE.value, "created_by": creators.get(r.seq, "")}
for r in rules], ("price_list_id", "seq"))
self._rows.clear()
self._rules.clear()
return len(rows) + len(rules)
def discard_draft(self, price_list_id: str, keys: Iterable[tuple]) -> list:
"""Remove rows that were never live. Returns the keys it removed; a live row is never touched."""
keys = list(keys)
found = self.stored_rows(price_list_id, keys)
removed = []
for key in keys:
hit = found.get(key)
if hit is not None and hit[1] == RowState.DRAFT.value:
self.conn.execute(delete(prod_price).where(and_(
prod_price.c.price_list_id == price_list_id, prod_price.c.product_id == key[0],
prod_price.c.unit_code == key[1], prod_price.c.valid_from == key[2],
prod_price.c.status == RowState.DRAFT.value)))
removed.append(key)
return removed
def discard_draft_rules(self, price_list_id: str, seqs: Iterable[int]) -> list:
"""Remove rules that were never live. Returns the seqs it removed; a live rule is never touched."""
seqs = list(seqs)
have = self.stored_rules(price_list_id)
removed = []
for seq in seqs:
if seq in have and have[seq][1] == RowState.DRAFT.value:
self.conn.execute(delete(prod_price_rule).where(and_(
prod_price_rule.c.price_list_id == price_list_id, prod_price_rule.c.seq == seq,
prod_price_rule.c.status == RowState.DRAFT.value)))
removed.append(seq)
return removed
def dependents(self, price_list_id: str) -> list:
"""The lists (not yet retired) that start from this one: based on it, or holding a rule whose base it is."""
live = {r[0] for r in self.conn.execute(select(prod_price_list.c.price_list_id).where(
prod_price_list.c.status != "retired"))}
based = {r[0] for r in self.conn.execute(select(prod_price_list.c.price_list_id).where(
prod_price_list.c.based_on == price_list_id))}
ruled = {r[0] for r in self.conn.execute(select(prod_price_rule.c.price_list_id).where(
prod_price_rule.c.base_list == price_list_id))}
return sorted(((based | ruled) & live) - {price_list_id})
@staticmethod
def _row(m) -> PriceRow:
return PriceRow(m["price_list_id"], m["product_id"], _d(m["valid_from"]), _dec(m["unit_price"], Decimal(0)),
_d(m["valid_to"]), m["unit_code"] or "", int(m["pricing_quantity"] or 1))
@staticmethod
def _rule(m) -> PriceRule:
return PriceRule(m["price_list_id"], m["seq"], m["applies_to"] or "all", m["target"] or "",
_dec(m["min_quantity"], Decimal(0)), _d(m["valid_from"]), _d(m["valid_to"]),
m["method"] or "discount_pct", _dec(m["discount_pct"], Decimal(0)),
_dec(m["fixed_price"]), _dec(m["surcharge_amount"], Decimal(0)), m["base_list"] or "",
m["unit_code"] or "", int(m["pricing_quantity"] or 1))
class TaskRepository:
"""The tasks an AI step leaves for a person (prod_task): the ManualTask when no provider is wired, or the
provider's proposal. A step asks once per version of its record; a task a person closed is not reopened."""
def __init__(self, conn):
self.conn = conn
def get(self, task_id: str) -> Optional[dict]:
r = self.conn.execute(select(prod_task).where(prod_task.c.task_id == task_id)).first()
return None if r is None else self._view(r._mapping)
def open_for(self, step_prefix_subject: str) -> list:
"""Open tasks about one record, e.g. subject_ref 'prod_product:HOST-M'."""
stmt = select(prod_task).where(and_(prod_task.c.subject_ref == step_prefix_subject, prod_task.c.status == "open"))
return [self._view(r._mapping) for r in self.conn.execute(stmt.order_by(prod_task.c.task_id))]
def all(self, status: Optional[str] = None, limit: int = 200) -> list:
stmt = select(prod_task).order_by(prod_task.c.task_id).limit(limit)
if status:
stmt = stmt.where(prod_task.c.status == status)
return [self._view(r._mapping) for r in self.conn.execute(stmt)]
def open(self, row: dict) -> dict:
"""Record the task unless one with this key exists (then that one — closed or not — is the answer)."""
found = self.get(row["task_id"])
if found is not None:
return found
upsert(self.conn, prod_task, [{**{k: row.get(k) for k in PROD_TASK_COLUMNS}, "status": "open"}], ("task_id",))
return {**row, "status": "open", "task_id": row["task_id"]}
def close(self, task_id: str, status: str, by: str, note: str) -> dict:
found = self.get(task_id)
upsert(self.conn, prod_task, [{**{k: found.get(k) for k in PROD_TASK_COLUMNS}, "status": status,
"resolved_by": by, "note": note}], ("task_id",))
return {**found, "status": status, "resolved_by": by, "note": note}
@staticmethod
def _view(m) -> dict:
out = {k: m[k] for k in PROD_TASK_COLUMNS}
out["input"] = json.loads(out.pop("input_json") or "{}")
out["output"] = json.loads(out.pop("output_json") or "null")
return out
class SettingsRepository:
"""The setting rows (prod_settings): one per setting, typed — int, bool, date, str or an AI mode."""
def __init__(self, conn):
self.conn = conn
def all(self) -> list:
stmt = select(prod_settings).order_by(prod_settings.c.setting_key)
return [{"setting_key": r._mapping["setting_key"], "value": _decode(r._mapping["value"]),
"kind": SETTING_KINDS.get(r._mapping["setting_key"], ("str", ()))[0],
"changes": r._mapping["changes"]} for r in self.conn.execute(stmt)]
def get(self, setting_key: str) -> Optional[dict]:
return next((s for s in self.all() if s["setting_key"] == setting_key), None)
def current(self) -> Settings:
return Settings.from_values({s["setting_key"]: s["value"] for s in self.all()})
def set(self, setting_key: str, value, changes: str) -> None:
"""Raises ValueError when `value` is not of the setting's kind (None puts the reference back)."""
kind, choices = SETTING_KINDS.get(setting_key, ("str", ()))
upsert(self.conn, prod_settings, [{"setting_key": setting_key, "value": _encode(kind, choices, value)