dodil/erp-brick-sales-order
DODIL ERP brick: sales order
src/brick_sales_order/infrastructure/tables.py
"""The tables, and one repository per aggregate that loads and saves it whole.
Every table is prefixed `so_`, has a natural primary key (never SERIAL) and, for a part, a key that starts with its
root's. Money is NUMERIC(18,2); a price, a quantity or a percentage is NUMERIC(19,4). schema.sql is generated from
the same column model; tests/test_schema.py keeps the two equal.
"""
import dataclasses
import json
from datetime import date, datetime, timezone
from typing import Optional
from erp_platform.repository import Aggregate, Part, claim_tables, upsert
from sqlalchemy import TIMESTAMP, BigInteger, Boolean, Column, Date, Integer, MetaData, Numeric, String, Table, select
from sqlalchemy.exc import IntegrityError
from brick_sales_order.domain import pricing as pr
from brick_sales_order.domain import rules as rl
from brick_sales_order.domain.number_series import NumberSeries
from brick_sales_order.domain.rebate import RebateAccrual, RebateAgreement, RebateSettlement
from brick_sales_order.domain.sales_order import (
BillingMilestone,
Delivery,
HeaderCondition,
InvoiceRequest,
LineCondition,
OrderBlock,
OrderLine,
ReturnInspection,
Revision,
SalesOrder,
ScheduleLine,
Send,
Suggestion,
)
from brick_sales_order.domain.values import EPOCH, SETTING_DEFAULTS, SETTING_KINDS, SETTINGS
md = MetaData()
MONEY, RATE, WEIGHT = Numeric(18, 2), Numeric(19, 4), Numeric(19, 6)
STAMPS = (Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
def _t(name, *cols):
return Table(name, md, *cols, *(Column(c.name, c.type) for c in STAMPS))
so_order = Table("so_order", md,
Column("sales_order_id", String, primary_key=True),
Column("order_no", String), Column("bp_id", BigInteger), Column("owner", String), Column("currency", String),
Column("price_list_id", String), Column("lifecycle", String), Column("revision", Integer),
Column("customer_po_ref", String), Column("ordered_on", Date), Column("confirmed_at", TIMESTAMP),
Column("delivery_terms", String), Column("payment_terms_id", String), Column("deposit_pct", RATE),
Column("subtotal", MONEY), Column("tax_total", MONEY), Column("total", MONEY),
Column("source_kind", String), Column("source_id", String),
Column("version", Integer), Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP),
# appended columns (additive): the business date of confirmation, the credit override, the cancel reason
Column("confirmed_on", Date), Column("credit_override_by", String), Column("credit_override_reason", String),
Column("cancel_reason", String),
# v6 slice A2 (additive): the order type, the ship-to and its country, the price group, the reference, what is shown on the order
Column("order_type_code", String), Column("ship_to_bp_id", BigInteger), Column("ship_to_country", String), Column("price_group", String),
Column("ref_sales_order_id", String), Column("ref_line_no", Integer), Column("ref_invoice_id", String), Column("warnings_json", String),
Column("freight_unknown", String),
# v6 slice B (additive): the partners, the promised delivery, the delivery rules copied from the customer's policy, the return reason
Column("bill_to_bp_id", BigInteger), Column("payer_bp_id", BigInteger), Column("requested_delivery_on", Date), Column("priority", String),
Column("partial_delivery", String), Column("max_partials", Integer), Column("under_tolerance_pct", RATE), Column("over_tolerance_pct", RATE),
Column("combine_allowed", Boolean), Column("min_shelf_life_days", Integer), Column("return_reason", String),
# v6 slice C (additive): the deposit deduction mode, the requester's sign-in subject, who approved a memo request or a free-of-charge order
Column("deposit_deduction_mode", String), Column("owner_sub", String), Column("approved_by_email", String), Column("approved_by_sub", String),
Column("approved_at", TIMESTAMP),
# v8 (additive): the ship-to address object (the country is ship_to_country)
Column("ship_to_address_lines", String), Column("ship_to_address_city", String), Column("ship_to_address_postcode", String),
Column("ship_to_address_region", String))
SO_ORDER_COLUMNS = ("sales_order_id", "order_no", "bp_id", "owner", "currency", "price_list_id", "lifecycle", "revision",
"customer_po_ref", "ordered_on", "confirmed_at", "delivery_terms", "payment_terms_id", "deposit_pct",
"subtotal", "tax_total", "total", "source_kind", "source_id", "confirmed_on", "credit_override_by",
"credit_override_reason", "cancel_reason", "order_type_code", "ship_to_bp_id", "ship_to_country", "price_group",
"ref_sales_order_id", "ref_line_no", "ref_invoice_id", "warnings_json", "freight_unknown",
"bill_to_bp_id", "payer_bp_id", "requested_delivery_on", "priority", "partial_delivery", "max_partials", "under_tolerance_pct",
"over_tolerance_pct", "combine_allowed", "min_shelf_life_days", "return_reason",
"deposit_deduction_mode", "owner_sub", "approved_by_email", "approved_by_sub", "approved_at",
"ship_to_address_lines", "ship_to_address_city", "ship_to_address_postcode", "ship_to_address_region")
so_line = _t("so_line",
Column("sales_order_id", String, primary_key=True), Column("line_no", Integer, primary_key=True),
Column("product_id", String), Column("description", String), Column("unit_code", String), Column("quantity", RATE),
Column("unit_numerator", BigInteger), Column("unit_denominator", BigInteger),
Column("list_price", RATE), Column("unit_price", RATE), Column("discount_pct", RATE), Column("priced_on", Date),
Column("priced_from", String), Column("tax_category", String), Column("tax_code", String),
Column("net_amount", MONEY), Column("tax_amount", MONEY), Column("invoicing_policy", String),
Column("billing_kind", String), Column("billing_plan", String), Column("quote_ref", String),
Column("odoo_line_id", String), Column("pricing_quantity", BigInteger), Column("pricing_unit_code", String),
Column("pricing_numerator", BigInteger), Column("pricing_denominator", BigInteger), Column("stocked", Boolean),
Column("reserved_base", RATE), Column("short_base", RATE),
Column("category_code", String), Column("product_type_code", String), Column("usage", String), Column("item_category_code", String),
Column("gross_weight", WEIGHT), Column("weight_unit_code", String), Column("measure_valid_from", Date), Column("procedural", Boolean),
# v6 slice C (additive): what a return, memo or consignment line refers to, instalments, the statistical value, purchasing's request
Column("ref_sales_order_id", String), Column("ref_line_no", Integer), Column("ref_delivery_ref", String), Column("ref_invoice_id", String),
Column("ref_invoice_line_no", Integer), Column("instalments", Integer), Column("statistical_value", MONEY), Column("purchase_request_id", String))
SO_LINE_COLUMNS = tuple(f.name for f in dataclasses.fields(OrderLine))
so_delivery = _t("so_delivery",
Column("sales_order_id", String, primary_key=True), Column("line_no", Integer, primary_key=True),
Column("delivery_ref", String, primary_key=True), Column("quantity", RATE), Column("kind", String),
Column("corrects_ref", String), Column("delivered_on", Date), Column("recorded_by", String), Column("source", String),
Column("quantity_base", RATE))
SO_DELIVERY_COLUMNS = tuple(f.name for f in dataclasses.fields(Delivery))
so_invoice_request = _t("so_invoice_request",
Column("sales_order_id", String, primary_key=True), Column("request_no", Integer, primary_key=True),
Column("kind", String), Column("lines_json", String), Column("invoice_id", String),
Column("requested_at", TIMESTAMP), Column("requested_by", String),
Column("amount", MONEY), Column("deducted", MONEY), Column("status", String),
Column("deductions_json", String), Column("source_ref", String))
SO_INVOICE_REQUEST_COLUMNS = ("sales_order_id", "request_no", "kind", "lines_json", "invoice_id", "requested_at",
"requested_by", "amount", "deducted", "status", "deductions_json", "source_ref")
so_revision = _t("so_revision",
Column("sales_order_id", String, primary_key=True), Column("revision", Integer, primary_key=True),
Column("reason", String), Column("changed_lines_json", String), Column("approved_by", String),
Column("approved_at", TIMESTAMP), Column("status", String), Column("proposed_by", String), Column("proposed_at", TIMESTAMP),
Column("proposed_sub", String))
SO_REVISION_COLUMNS = tuple(f.name for f in dataclasses.fields(Revision))
so_send = _t("so_send",
Column("sales_order_id", String, primary_key=True), Column("sent_at", TIMESTAMP, primary_key=True),
Column("channel", String), Column("address", String), Column("revision", Integer), Column("sent_by", String),
Column("schedule_version", Integer))
SO_SEND_COLUMNS = tuple(f.name for f in dataclasses.fields(Send))
so_suggestion = _t("so_suggestion",
Column("sales_order_id", String, primary_key=True), Column("step", String, primary_key=True),
Column("suggestion_no", Integer, primary_key=True), Column("model", String), Column("inputs_ref", String),
Column("output_json", String), Column("decision", String), Column("decided_by", String),
Column("decided_at", TIMESTAMP), Column("decisions_json", String))
SO_SUGGESTION_COLUMNS = ("sales_order_id", "step", "suggestion_no", "model", "inputs_ref", "output_json", "decision",
"decided_by", "decided_at", "decisions_json")
so_line_condition = _t("so_line_condition",
Column("sales_order_id", String, primary_key=True), Column("line_no", Integer, primary_key=True), Column("step", Integer, primary_key=True),
Column("revision", Integer, primary_key=True), Column("condition_type_code", String), Column("rate", RATE), Column("amount", MONEY),
Column("basis", MONEY), Column("source", String), Column("statistical", Boolean), Column("header_share_of", String))
SO_LINE_CONDITION_COLUMNS = tuple(f.name for f in dataclasses.fields(LineCondition))
so_header_condition = _t("so_header_condition",
Column("sales_order_id", String, primary_key=True), Column("condition_type_code", String, primary_key=True),
Column("revision", Integer, primary_key=True), Column("step", Integer), Column("rate", RATE), Column("amount", MONEY), Column("source", String))
SO_HEADER_CONDITION_COLUMNS = tuple(f.name for f in dataclasses.fields(HeaderCondition))
so_order_block = _t("so_order_block",
Column("sales_order_id", String, primary_key=True), Column("block_no", Integer, primary_key=True), Column("kind", String),
Column("reason_code", String), Column("event", String), Column("note", String), Column("by_email", String), Column("by_sub", String),
Column("at", TIMESTAMP))
SO_ORDER_BLOCK_COLUMNS = tuple(f.name for f in dataclasses.fields(OrderBlock))
so_schedule_line = _t("so_schedule_line",
Column("sales_order_id", String, primary_key=True), Column("line_no", Integer, primary_key=True),
Column("schedule_version", Integer, primary_key=True), Column("seq", Integer, primary_key=True),
Column("requested_on", Date), Column("requested_quantity", WEIGHT), Column("confirmed_on", Date), Column("material_available_on", Date),
Column("confirmed_quantity", WEIGHT), Column("confirmed_base", WEIGHT), Column("reason", String), Column("at", TIMESTAMP),
Column("trigger_key", String))
SO_SCHEDULE_LINE_COLUMNS = tuple(f.name for f in dataclasses.fields(ScheduleLine))
so_billing_milestone = _t("so_billing_milestone",
Column("sales_order_id", String, primary_key=True), Column("line_no", Integer, primary_key=True), Column("milestone_no", Integer, primary_key=True),
Column("pct", RATE), Column("event", String), Column("name", String), Column("due_on", Date), Column("released_on", Date),
Column("released_by", String), Column("amount", MONEY), Column("billed_request_no", Integer))
SO_BILLING_MILESTONE_COLUMNS = tuple(f.name for f in dataclasses.fields(BillingMilestone))
so_return_inspection = _t("so_return_inspection",
Column("sales_order_id", String, primary_key=True), Column("line_no", Integer, primary_key=True), Column("inspection_ref", String, primary_key=True),
Column("received_quantity", RATE), Column("accepted_quantity", RATE), Column("rejected_quantity", RATE), Column("decision", String),
Column("reason", String), Column("inspected_on", Date), Column("recorded_at", TIMESTAMP))
SO_RETURN_INSPECTION_COLUMNS = tuple(f.name for f in dataclasses.fields(ReturnInspection))
# something a person should do because the system changed a promise; one per (order, kind, business day), a plain INSERT claim
so_task = _t("so_task",
Column("sales_order_id", String, primary_key=True), Column("kind", String, primary_key=True), Column("raised_on", Date, primary_key=True),
Column("owner", String), Column("text", String), Column("closed_by", String), Column("closed_at", TIMESTAMP))
SALES_ORDER = Aggregate(Part(so_order, ("sales_order_id",)),
Part(so_line, ("sales_order_id", "line_no")),
Part(so_delivery, ("sales_order_id", "line_no", "delivery_ref")),
Part(so_invoice_request, ("sales_order_id", "request_no")),
Part(so_revision, ("sales_order_id", "revision")),
Part(so_send, ("sales_order_id", "sent_at")),
Part(so_suggestion, ("sales_order_id", "step", "suggestion_no")),
Part(so_line_condition, ("sales_order_id", "line_no", "step", "revision")),
Part(so_header_condition, ("sales_order_id", "condition_type_code", "revision")),
Part(so_order_block, ("sales_order_id", "block_no")),
Part(so_schedule_line, ("sales_order_id", "line_no", "schedule_version", "seq")),
Part(so_billing_milestone, ("sales_order_id", "line_no", "milestone_no")),
Part(so_return_inspection, ("sales_order_id", "line_no", "inspection_ref")),
version="version")
so_number_series = Table("so_number_series", md,
Column("series_id", String, primary_key=True), Column("high_water_mark", Integer),
Column("version", Integer), Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
SO_NUMBER_SERIES_COLUMNS = ("series_id", "high_water_mark")
NUMBER_SERIES = Aggregate(Part(so_number_series, ("series_id",)), version="version")
so_order_no_claim = _t("so_order_no_claim",
Column("order_no", String, primary_key=True), Column("sales_order_id", String), Column("claimed_at", TIMESTAMP))
so_order_type = _t("so_order_type",
Column("order_type_code", String, primary_key=True), Column("valid_from", Date, primary_key=True), Column("name", String),
Column("series_id", String), Column("direction", String), Column("billing", String), Column("delivery_relevant", Boolean),
Column("reference", String), Column("default_block", String), Column("pricing_procedure_code", String), Column("workflow", String),
Column("priority", String), Column("credit_check", Boolean), Column("special_stock", String), Column("approval_limit", String),
Column("status", String))
so_item_category = _t("so_item_category",
Column("item_category_code", String, primary_key=True), Column("fulfilment", String), Column("billing_relevant", Boolean),
Column("pricing", String), Column("schedule_lines", Boolean), Column("returns", Boolean))
so_item_category_rule = _t("so_item_category_rule",
Column("order_type_code", String, primary_key=True), Column("product_type_code", String, primary_key=True),
Column("usage", String, primary_key=True), Column("valid_from", Date, primary_key=True), Column("item_category_code", String),
Column("allowed_alternatives", String))
so_condition_type = _t("so_condition_type",
Column("condition_type_code", String, primary_key=True), Column("calculation", String), Column("level", String), Column("header_spread", String),
Column("sign", String), Column("access_sequence_code", String), Column("manual_allowed", Boolean), Column("manual_limit_pct", RATE),
Column("statistical", Boolean), Column("scale_basis", String), Column("account_purpose", String))
so_access_sequence = _t("so_access_sequence",
Column("access_sequence_code", String, primary_key=True), Column("step", Integer, primary_key=True), Column("key_fields", String),
Column("exclusive", Boolean))
so_pricing_procedure = _t("so_pricing_procedure",
Column("pricing_procedure_code", String, primary_key=True), Column("step", Integer, primary_key=True), Column("condition_type_code", String),
Column("from_step", Integer), Column("to_step", Integer), Column("subtotal_name", String), Column("required", Boolean), Column("statistical", Boolean))
so_condition_record = _t("so_condition_record",
Column("condition_type_code", String, primary_key=True), Column("key_fields", String, primary_key=True),
Column("key_value", String, primary_key=True), Column("valid_from", Date, primary_key=True), Column("valid_to", Date),
Column("bp_id", BigInteger), Column("price_group", String), Column("product_id", String), Column("category_code", String),
Column("ship_to_country", String), Column("rate", RATE), Column("minimum", MONEY), Column("scale_json", String), Column("currency", String),
Column("set_by", String))
so_rebate_agreement = _t("so_rebate_agreement",
Column("rebate_agreement_id", String, primary_key=True), Column("bp_id", BigInteger), Column("price_group", String), Column("basis", String),
Column("basis_values_json", String), Column("valid_from", Date), Column("valid_to", Date), Column("scale_json", String),
Column("retroactive", Boolean), Column("accrual_rate_pct", RATE), Column("currency", String), Column("lifecycle", String),
Column("set_by_email", String), Column("set_by_sub", String), Column("version", Integer))
SO_REBATE_COLUMNS = tuple(f.name for f in dataclasses.fields(RebateAgreement) if f.name not in ("accruals", "settlements"))
so_rebate_accrual = _t("so_rebate_accrual",
Column("rebate_agreement_id", String, primary_key=True), Column("document_id", String, primary_key=True),
Column("line_no", Integer, primary_key=True), Column("kind", String), Column("issue_date", Date), Column("basis_amount", MONEY),
Column("accrual_amount", MONEY), Column("tax_category", String), Column("journal_id", String), Column("posting_status", String))
SO_REBATE_ACCRUAL_COLUMNS = tuple(f.name for f in dataclasses.fields(RebateAccrual))
so_rebate_settlement = _t("so_rebate_settlement",
Column("rebate_agreement_id", String, primary_key=True), Column("settlement_no", Integer, primary_key=True), Column("kind", String),
Column("basis_amount", MONEY), Column("rate_pct", RATE), Column("earned_amount", MONEY), Column("settled_before", MONEY), Column("amount", MONEY),
Column("true_up_amount", MONEY), Column("requested_by_email", String), Column("requested_by_sub", String), Column("approved_by_email", String),
Column("approved_by_sub", String), Column("credit_id", String), Column("status", String), Column("reason", String), Column("lines_json", String),
Column("requested_on", Date), Column("approved_on", Date))
SO_REBATE_SETTLEMENT_COLUMNS = tuple(f.name for f in dataclasses.fields(RebateSettlement))
REBATE_AGREEMENT = Aggregate(Part(so_rebate_agreement, ("rebate_agreement_id",)),
Part(so_rebate_accrual, ("rebate_agreement_id", "document_id", "line_no")),
Part(so_rebate_settlement, ("rebate_agreement_id", "settlement_no")), version="version")
# ONE claim key per agreement event: accrue, settle, end, cancel and a scale change all claim (agreement, seq) with a plain INSERT, so two
# different operations on the agreement's one bounded total can never both win the same step
so_rebate_claim = _t("so_rebate_claim",
Column("rebate_agreement_id", String, primary_key=True), Column("seq", Integer, primary_key=True), Column("operation", String),
Column("claimed_at", TIMESTAMP))
# what the ledger was asked, per source document and event: pending with its reason until the ledger takes it
so_posting = _t("so_posting",
Column("document_id", String, primary_key=True), Column("event", String, primary_key=True), Column("journal_id", String),
Column("status", String), Column("reason", String), Column("attempts", Integer), Column("rebate_agreement_id", String),
Column("payload_json", String))
# ONE claim key per bounded thing, whichever operation takes from it: a credit (a memo request's, a return's, an unused deposit's) against an
# invoice line's net, a return against an original order line's delivered quantity. A plain INSERT at (claim_key, seq): two operations that read
# the same seq collide at the insert or the commit, and the loser is refused with what is left.
so_credit_claim = _t("so_credit_claim",
Column("claim_key", String, primary_key=True), Column("seq", Integer, primary_key=True), Column("kind", String), Column("sales_order_id", String),
Column("request_no", Integer), Column("amount", MONEY), Column("quantity", RATE), Column("claimed_at", TIMESTAMP))
# a deposit is deducted per deposit: one claim row per (deposit, seq), plain INSERT; the deposit's remaining is its amount less these
so_deposit_deduction = _t("so_deposit_deduction",
Column("deposit_invoice_id", String, primary_key=True), Column("seq", Integer, primary_key=True), Column("kind", String), Column("sales_order_id", String),
Column("request_no", Integer), Column("amount", MONEY), Column("tax_category", String), Column("claimed_at", TIMESTAMP))
so_return_reason = _t("so_return_reason",
Column("return_reason_code", String, primary_key=True), Column("name", String), Column("inspect", Boolean),
Column("restocking_fee_pct", RATE), Column("credit_on", String))
# every table a plain-INSERT claim writes: a duplicate at the insert or the commit is a 409, never a 500 and never a silent overwrite
claim_tables("so_order_no_claim", "so_condition_record", "so_order_type", "so_item_category_rule", "so_rebate_claim",
"so_customer_partner", "so_incompletion_rule", "so_block_reason", "so_task", "so_credit_claim", "so_deposit_deduction", "so_return_reason")
so_customer_policy = _t("so_customer_policy",
Column("bp_id", BigInteger, primary_key=True), Column("valid_from", Date, primary_key=True),
Column("po_required", Boolean), Column("deposit_pct", RATE), Column("delivery_terms", String), Column("set_by", String),
Column("price_group", String),
Column("partial_delivery", String), Column("max_partials", Integer), Column("under_tolerance_pct", RATE), Column("over_tolerance_pct", RATE),
Column("combine_allowed", Boolean), Column("min_shelf_life_days", Integer), Column("deposit_deduction_mode", String))
SO_CUSTOMER_POLICY_COLUMNS = ("bp_id", "valid_from", "po_required", "deposit_pct", "delivery_terms", "set_by", "price_group", "partial_delivery",
"max_partials", "under_tolerance_pct", "over_tolerance_pct", "combine_allowed", "min_shelf_life_days", "deposit_deduction_mode")
so_customer_partner = _t("so_customer_partner",
Column("sold_to_bp_id", BigInteger, primary_key=True), Column("function", String, primary_key=True),
Column("partner_bp_id", BigInteger, primary_key=True), Column("valid_from", Date, primary_key=True), Column("is_default", Boolean),
Column("transit_days_json", String), Column("status", String), Column("set_by", String))
so_incompletion_rule = _t("so_incompletion_rule",
Column("order_type_code", String, primary_key=True), Column("field", String, primary_key=True), Column("valid_from", Date, primary_key=True),
Column("blocks", String), Column("condition", String), Column("reason_text", String))
so_block_reason = _t("so_block_reason",
Column("block_reason_code", String, primary_key=True), Column("kind", String), Column("name", String))
so_settings = _t("so_settings",
Column("setting_key", String, primary_key=True), Column("valid_from", Date, primary_key=True),
Column("value", String), Column("set_by", String), Column("set_at", TIMESTAMP), Column("changes", String))
# the Odoo id map (shared.yaml conventions.odoo_migration.id_map): (odoo db, model, id) -> our natural key
so_migration_map = _t("so_migration_map",
Column("odoo_db", String, primary_key=True), Column("odoo_model", String, primary_key=True),
Column("odoo_id", BigInteger, primary_key=True), Column("sales_order_id", String), Column("line_no", Integer),
Column("delivery_ref", String), Column("migrated_at", TIMESTAMP),
# what a NOT-CARRIED line (a section, a note, a down payment) was: its reason in delivery_ref, its amounts here
Column("amount", MONEY), Column("tax_amount", MONEY))
def _default_of(cls, name):
for f in dataclasses.fields(cls):
if f.name == name:
return None if f.default is dataclasses.MISSING else f.default
return None
def _make(cls, row: dict, names: tuple):
return cls(**{n: (row[n] if row.get(n) is not None else _default_of(cls, n)) for n in names})
def _pick(obj, names: tuple) -> dict:
return {n: getattr(obj, n) for n in names}
def _sorted(rows: list, key: tuple) -> list:
return sorted(rows, key=lambda r: tuple(str(r[k]).zfill(12) if isinstance(r[k], int) else str(r[k]) for k in key))
def _build(rows: dict) -> SalesOrder:
agg = _make(SalesOrder, rows["so_order"][0], SO_ORDER_COLUMNS)
agg.lines = [_make(OrderLine, x, SO_LINE_COLUMNS) for x in _sorted(rows["so_line"], ("line_no",))]
agg.deliveries = [_make(Delivery, x, SO_DELIVERY_COLUMNS) for x in _sorted(rows["so_delivery"], ("line_no", "delivery_ref"))]
agg.invoice_requests = [_make(InvoiceRequest, x, SO_INVOICE_REQUEST_COLUMNS) for x in _sorted(rows["so_invoice_request"], ("request_no",))]
agg.revisions = [_make(Revision, x, SO_REVISION_COLUMNS) for x in _sorted(rows["so_revision"], ("revision",))]
agg.sends = [_make(Send, x, SO_SEND_COLUMNS) for x in _sorted(rows["so_send"], ("sent_at",))]
agg.suggestions = [_make(Suggestion, x, SO_SUGGESTION_COLUMNS) for x in _sorted(rows["so_suggestion"], ("step", "suggestion_no"))]
agg.conditions = [_make(LineCondition, x, SO_LINE_CONDITION_COLUMNS) for x in _sorted(rows["so_line_condition"], ("line_no", "step", "revision"))]
agg.header_conditions = [_make(HeaderCondition, x, SO_HEADER_CONDITION_COLUMNS) for x in _sorted(rows["so_header_condition"], ("condition_type_code", "revision"))]
agg.blocks = [_make(OrderBlock, x, SO_ORDER_BLOCK_COLUMNS) for x in _sorted(rows.get("so_order_block", []), ("block_no",))]
agg.schedule = [_make(ScheduleLine, x, SO_SCHEDULE_LINE_COLUMNS) for x in _sorted(rows.get("so_schedule_line", []), ("line_no", "schedule_version", "seq"))]
agg.milestones = [_make(BillingMilestone, x, SO_BILLING_MILESTONE_COLUMNS) for x in _sorted(rows.get("so_billing_milestone", []), ("line_no", "milestone_no"))]
agg.inspections = [_make(ReturnInspection, x, SO_RETURN_INSPECTION_COLUMNS) for x in _sorted(rows.get("so_return_inspection", []), ("line_no", "inspection_ref"))]
return agg
class SalesOrderRepository:
"""Loads and saves a sales order whole. Takes an open connection: the use case owns the transaction."""
def __init__(self, conn):
self.conn = conn
self._snapshots: dict = {} # sales_order_id -> what its load read: save() hands it back so only changed rows are written
def get(self, sales_order_id) -> Optional[tuple]:
"""(SalesOrder, the version it was loaded at), or None. The repository keeps the load's snapshot for this transaction. Read once per
request: a second get builds a fresh order from the same rows (until a save of it). An order not stored yet is saved without
re-reading its 13 tables' keys (nothing_stored: its version claim still refuses a racer)."""
loaded = self.__dict__.setdefault("_loaded", {})
if sales_order_id not in loaded:
loaded[sales_order_id] = SALES_ORDER.load(self.conn, {"sales_order_id": sales_order_id})
rows = loaded[sales_order_id]
if rows is None:
self._snapshots.setdefault(sales_order_id, SALES_ORDER.nothing_stored({"sales_order_id": sales_order_id}))
return None
self._snapshots[sales_order_id] = rows.snapshot
return _build(rows), SALES_ORDER.version_of(rows)
def get_many(self, ids) -> dict:
"""{sales_order_id: (SalesOrder, version)} for a set of orders: one query per table per 100 orders, in one round trip (a loop of get() is
one round trip per order)."""
ids = list(dict.fromkeys(ids))
out = {}
for (oid,), rows in SALES_ORDER.load_many(self.conn, [{"sales_order_id": i} for i in ids]).items():
if rows is None:
continue
self._snapshots[oid] = rows.snapshot
out[oid] = (_build(rows), SALES_ORDER.version_of(rows))
return out
def snapshot_of(self, sales_order_id):
"""What this repository read of an order (None if it did not), to carry into another transaction's repository with adopt()."""
return self._snapshots.get(sales_order_id)
def adopt(self, sales_order_id, snapshot) -> None:
"""Take a snapshot read by another repository: the save still checks the version, so a stale one is a Conflict, never a lost write."""
if snapshot is not None:
self._snapshots[sales_order_id] = snapshot
def confirmed_scheduled(self, products=None) -> list:
"""[(sales_order_id, priority, ordered_on)] of the confirmed orders that have schedule lines (the only ones a re-promise can touch), optionally
only those with a line of one of `products`."""
q = select(so_order.c.sales_order_id, so_order.c.priority, so_order.c.ordered_on).where(
so_order.c.lifecycle == "confirmed", so_order.c.sales_order_id.in_(select(so_schedule_line.c.sales_order_id)))
if products is not None:
q = q.where(so_order.c.sales_order_id.in_(select(so_line.c.sales_order_id).where(so_line.c.product_id.in_(list(products) or [""]))))
return [tuple(r) for r in self.conn.execute(q)]
def confirmed_with_product(self, product_id: str) -> list:
"""The ids of the CONFIRMED orders that have a line of this product (SO-50: demand)."""
q = select(so_order.c.sales_order_id).where(
so_order.c.lifecycle == "confirmed", so_order.c.sales_order_id.in_(select(so_line.c.sales_order_id).where(so_line.c.product_id == product_id)))
return sorted(r[0] for r in self.conn.execute(q))
def products_of(self, sales_order_id: str) -> set:
return {r[0] for r in self.conn.execute(select(so_line.c.product_id).where(so_line.c.sales_order_id == sales_order_id))}
def save(self, agg: SalesOrder, expected_version: int) -> int:
"""Writes version expected_version + 1, or raises Conflict (a 409 through both doors)."""
snapshot = self._snapshots.pop(agg.sales_order_id, None) # spent: after this save the rows are no longer what it read
self.__dict__.get("_loaded", {}).pop(agg.sales_order_id, None)
return SALES_ORDER.save(self.conn, {
"so_order": [_pick(agg, SO_ORDER_COLUMNS)],
"so_line": [_pick(x, SO_LINE_COLUMNS) for x in agg.lines],
"so_delivery": [_pick(x, SO_DELIVERY_COLUMNS) for x in agg.deliveries],
"so_invoice_request": [_pick(x, SO_INVOICE_REQUEST_COLUMNS) for x in agg.invoice_requests],
"so_revision": [_pick(x, SO_REVISION_COLUMNS) for x in agg.revisions],
"so_send": [_pick(x, SO_SEND_COLUMNS) for x in agg.sends],
"so_suggestion": [_pick(x, SO_SUGGESTION_COLUMNS) for x in agg.suggestions],
"so_line_condition": [_pick(x, SO_LINE_CONDITION_COLUMNS) for x in agg.conditions],
"so_header_condition": [_pick(x, SO_HEADER_CONDITION_COLUMNS) for x in agg.header_conditions],
"so_order_block": [_pick(x, SO_ORDER_BLOCK_COLUMNS) for x in agg.blocks],
"so_schedule_line": [_pick(x, SO_SCHEDULE_LINE_COLUMNS) for x in agg.schedule],
"so_billing_milestone": [_pick(x, SO_BILLING_MILESTONE_COLUMNS) for x in agg.milestones],
"so_return_inspection": [_pick(x, SO_RETURN_INSPECTION_COLUMNS) for x in agg.inspections]},
expected_version=expected_version, snapshot=snapshot)
def page(self, *, owner: Optional[str] = None, also_confirmed: bool = False, lifecycle: Optional[str] = None,
source_kind: Optional[str] = None, bp_id: Optional[int] = None, ordered_from: Optional[date] = None, ordered_to: Optional[date] = None,
limit: int = 50, offset: int = 0) -> tuple:
"""(orders, total): a page of orders with their lines, deliveries and invoice requests, in FOUR queries however
long the page is (the list derives its statuses from them). `also_confirmed`: a fulfiller sees confirmed orders
that are not theirs."""
where = []
if owner is not None:
own = so_order.c.owner == owner
where.append(own | (so_order.c.lifecycle == "confirmed") if also_confirmed else own)
if lifecycle:
where.append(so_order.c.lifecycle == lifecycle)
if source_kind:
where.append(so_order.c.source_kind == source_kind)
if bp_id is not None:
where.append(so_order.c.bp_id == int(bp_id))
if ordered_from:
where.append(so_order.c.ordered_on >= ordered_from)
if ordered_to:
where.append(so_order.c.ordered_on <= ordered_to)
base = select(so_order).where(*where)
total = len(self.conn.execute(select(so_order.c.sales_order_id).where(*where)).all())
heads = [dict(r._mapping) for r in self.conn.execute(
base.order_by(so_order.c.ordered_on.desc(), so_order.c.sales_order_id).limit(limit).offset(offset))]
ids = [h["sales_order_id"] for h in heads]
parts = {}
for t in (so_line, so_delivery, so_invoice_request, so_billing_milestone):
parts[t.name] = [dict(r._mapping) for r in self.conn.execute(select(t).where(t.c.sales_order_id.in_(ids)))] if ids else []
out = []
for h in heads:
rows = {"so_order": [h], **{n: [r for r in rs if r["sales_order_id"] == h["sales_order_id"]] for n, rs in parts.items()},
"so_revision": [], "so_send": [], "so_suggestion": [], "so_line_condition": [], "so_header_condition": []}
out.append(_build(rows))
return out, total
def all_from(self, source_kind: str) -> list:
"""Every order of a source (the migration's reconciliation reads them all), a page at a time."""
out, offset = [], 0
while True:
page, total = self.sales_orders_page(source_kind, offset)
out += page
offset += len(page)
if not page or offset >= total:
return out
def sales_orders_page(self, source_kind: str, offset: int) -> tuple:
return self.page(source_kind=source_kind, limit=500, offset=offset)
def exists(self, sales_order_id: str) -> bool:
found = self.conn.execute(select(so_order.c.sales_order_id).where(so_order.c.sales_order_id == sales_order_id)).first() is not None
if not found: # a new order is saved without re-reading its 13 tables' keys (its version claim refuses a racer)
self._snapshots.setdefault(sales_order_id, SALES_ORDER.nothing_stored({"sales_order_id": sales_order_id}))
return found
class NumberSeriesRepository:
def __init__(self, conn):
self.conn = conn
def get(self, series_id) -> Optional[tuple]:
rows = NUMBER_SERIES.load(self.conn, {"series_id": series_id})
snaps = self.__dict__.setdefault("_snapshots", {})
if rows is None: # a series not stored yet: saved without re-reading its key
snaps[series_id] = NUMBER_SERIES.nothing_stored({"series_id": series_id})
return None
snaps[series_id] = rows.snapshot # save() writes the diff, without re-reading the key
return _make(NumberSeries, rows["so_number_series"][0], SO_NUMBER_SERIES_COLUMNS), NUMBER_SERIES.version_of(rows)
def marks(self) -> list:
"""[(series_id, high_water_mark)] of every series."""
return [(r[0], int(r[1] or 0)) for r in self.conn.execute(select(so_number_series.c.series_id, so_number_series.c.high_water_mark))]
def get_or_new(self, series_id) -> tuple:
return self.get(series_id) or (NumberSeries(series_id), 0)
def save(self, agg: NumberSeries, expected_version: int) -> int:
snapshot = self.__dict__.get("_snapshots", {}).pop(agg.series_id, None)
return NUMBER_SERIES.save(self.conn, {"so_number_series": [_pick(agg, SO_NUMBER_SERIES_COLUMNS)]},
expected_version=expected_version, snapshot=snapshot)
class OrderNoClaimRepository:
"""An order number is claimed once, by one order (a guarded claim: DataK3 has no UNIQUE beyond the PK)."""
def __init__(self, conn):
self.conn = conn
def claim(self, order_no: str, sales_order_id: str) -> bool:
"""True when the number is now (or already was) this order's; False when another order holds it."""
row = self.conn.execute(select(so_order_no_claim.c.sales_order_id).where(so_order_no_claim.c.order_no == order_no)).first()
if row is not None:
return row[0] == sales_order_id
# a PLAIN insert, never an upsert: two orders racing for one number, the second is refused (a duplicate key at the insert
# or at the commit; both are a 409 through claim_tables), never silently given the same number
self.conn.execute(so_order_no_claim.insert().values(order_no=order_no, sales_order_id=sales_order_id,
claimed_at=datetime.now(timezone.utc).replace(tzinfo=None)))
return True
def _memo(fn):
"""Read once per repository (= per transaction): the configuration and the settings are small tables asked for many times in one request, and on
the live wire every ask is a round trip. A write through the same repository forgets what was read."""
def wrapper(self, *args):
key = (fn.__name__, *args)
if key not in self._memo:
self._memo[key] = fn(self, *args)
return self._memo[key]
wrapper.__name__ = fn.__name__
wrapper.__doc__ = fn.__doc__
return wrapper
class CustomerOrderPolicyRepository:
"""so_customer_policy: dated, append-only. The row in force on a date is the latest valid_from on or before it."""
def __init__(self, conn):
self.conn = conn
self._memo = {}
@_memo
def all(self, bp_id: Optional[int] = None) -> list:
stmt = select(so_customer_policy).order_by(so_customer_policy.c.bp_id, so_customer_policy.c.valid_from)
if bp_id is not None:
stmt = stmt.where(so_customer_policy.c.bp_id == int(bp_id))
return [{k: r._mapping[k] for k in SO_CUSTOMER_POLICY_COLUMNS} for r in self.conn.execute(stmt)]
def in_force(self, bp_id: int, day: date) -> Optional[dict]:
rows = [r for r in self.all(bp_id) if r["valid_from"] <= day]
return rows[-1] if rows else None
def put(self, row: dict) -> None:
self._memo.clear()
upsert(self.conn, so_customer_policy, [{k: v for k, v in row.items() if k in SO_CUSTOMER_POLICY_COLUMNS}], ("bp_id", "valid_from"))
def _encode(kind: str, choices: tuple, value) -> str:
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 == "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)
class SettingsRepository:
"""so_settings: one dated row per change [setting_key, valid_from]. Seeded defaults start 1970-01-01; every later
row starts AFTER the day it is set (the use case checks), and a correction is a new row, never an edit."""
def __init__(self, conn):
self.conn = conn
self._memo = {}
@_memo
def _all(self) -> list:
stmt = select(so_settings).order_by(so_settings.c.setting_key, so_settings.c.valid_from)
return [{**{k: r._mapping[k] for k in ("setting_key", "valid_from", "set_by", "set_at", "changes")},
"value": json.loads(r._mapping["value"])} for r in self.conn.execute(stmt)]
def rows(self, key: Optional[str] = None) -> list:
return [r for r in self._all() if not key or r["setting_key"] == key]
def value_on(self, key: str, day: date):
got = [r for r in self.rows(key) if r["valid_from"] <= day]
return got[-1]["value"] if got else SETTING_DEFAULTS.get(key) # an unseeded bucket behaves as the defaults say
def put(self, key: str, valid_from: date, value, set_by: str, set_at: datetime, changes: str = "") -> None:
"""Raises ValueError when `value` is not of the setting's kind."""
kind, choices = SETTING_KINDS[key]
self._memo.clear()
upsert(self.conn, so_settings, [{"setting_key": key, "valid_from": valid_from, "value": _encode(kind, choices, value),
"set_by": set_by, "set_at": set_at, "changes": changes}], ("setting_key", "valid_from"))
def seed(self) -> None:
self._memo.clear()
have = {r[0] for r in self.conn.execute(select(so_settings.c.setting_key))}
for key, _kind, default, changes, _c in SETTINGS:
if key not in have:
self.put(key, date.fromisoformat(EPOCH), default, "default", datetime(1970, 1, 1), changes)
class MigrationMapRepository:
def __init__(self, conn):
self.conn = conn
def get(self, db: str, model: str, odoo_id: int) -> Optional[dict]:
r = self.conn.execute(select(so_migration_map).where(so_migration_map.c.odoo_db == db, so_migration_map.c.odoo_model == model,
so_migration_map.c.odoo_id == int(odoo_id))).first()
return dict(r._mapping) if r else None
def put(self, db: str, model: str, odoo_id: int, sales_order_id: str, line_no: Optional[int], delivery_ref: str, at: datetime,
amount=None, tax_amount=None) -> None:
upsert(self.conn, so_migration_map, [{"odoo_db": db, "odoo_model": model, "odoo_id": int(odoo_id), "sales_order_id": sales_order_id,
"line_no": line_no, "delivery_ref": delivery_ref, "migrated_at": at,
"amount": amount, "tax_amount": tax_amount}],
("odoo_db", "odoo_model", "odoo_id"))
def all(self) -> list:
return [dict(r._mapping) for r in self.conn.execute(select(so_migration_map))]
def _rows(conn, table, cls) -> list:
names = tuple(f.name for f in dataclasses.fields(cls))
return [_make(cls, dict(r._mapping), names) for r in conn.execute(select(table))]
class Duplicate(Exception):
"""A dated configuration row that already exists: the use case turns it into a 409."""
def _put(conn, table, obj, key: tuple) -> None:
"""A PLAIN insert after a look: a row that is there is a 409 (a correction is a new row from a later date, never an edit); two writers
racing for it, the second is refused at the insert or the commit (the table is a claim table)."""
names = tuple(c.name for c in table.columns if c.name not in ("created_at", "updated_at"))
row = {n: getattr(obj, n) for n in names}
now = datetime.now(timezone.utc).replace(tzinfo=None)
if conn.execute(select(table.c[key[0]]).where(*[table.c[k] == row[k] for k in key])).first() is not None:
raise Duplicate(f"{table.name} already has a row {', '.join(str(row[k]) for k in key)}")
conn.execute(table.insert().values(**row, created_at=now, updated_at=now))
class ConfigRepository:
"""The dated configuration of orders and prices: order types, item categories and their rules, condition types, access sequences, the
pricing procedure and the condition records. An empty table behaves as its seed says (the same rows `seed` writes)."""
def __init__(self, conn):
self.conn = conn
self._memo = {}
@_memo
def order_types(self) -> list:
return _rows(self.conn, so_order_type, pr.OrderType) or pr.default_types()
@_memo
def categories(self) -> list:
return _rows(self.conn, so_item_category, pr.ItemCategory) or pr.default_categories()
@_memo
def rules(self) -> list:
return _rows(self.conn, so_item_category_rule, pr.ItemCategoryRule) or pr.default_rules()
@_memo
def condition_types(self) -> dict:
return {t.condition_type_code: t for t in (_rows(self.conn, so_condition_type, pr.ConditionType) or pr.default_condition_types())}
@_memo
def access(self) -> list:
return _rows(self.conn, so_access_sequence, pr.AccessStep) or pr.default_access()
@_memo
def procedure(self, code: str = "STANDARD") -> list:
return [s for s in (_rows(self.conn, so_pricing_procedure, pr.ProcStep) or pr.default_procedure()) if s.pricing_procedure_code == code]
@_memo
def records(self) -> list:
return _rows(self.conn, so_condition_record, pr.Record)
def order_type(self, code: str, on: date) -> Optional[pr.OrderType]:
return pr.in_force(self.order_types(), on, ("order_type_code", code))
def add_order_type(self, row: pr.OrderType) -> None:
self._memo.clear()
_put(self.conn, so_order_type, row, ("order_type_code", "valid_from"))
def add_rule(self, row: pr.ItemCategoryRule) -> None:
self._memo.clear()
_put(self.conn, so_item_category_rule, row, ("order_type_code", "product_type_code", "usage", "valid_from"))
def add_record(self, row: pr.Record) -> None:
self._memo.clear()
_put(self.conn, so_condition_record, row, ("condition_type_code", "key_fields", "key_value", "valid_from"))
@_memo
def partners(self) -> list:
return _rows(self.conn, so_customer_partner, rl.CustomerPartner)
def add_partner(self, row: rl.CustomerPartner) -> None:
self._memo.clear()
_put(self.conn, so_customer_partner, row, ("sold_to_bp_id", "function", "partner_bp_id", "valid_from"))
@_memo
def incompletion_rules(self) -> list:
return _rows(self.conn, so_incompletion_rule, rl.IncompletionRule) or rl.default_incompletion()
def add_incompletion_rule(self, row: rl.IncompletionRule) -> None:
self._memo.clear()
_put(self.conn, so_incompletion_rule, row, ("order_type_code", "field", "valid_from"))
@_memo
def block_reasons(self) -> list:
return _rows(self.conn, so_block_reason, rl.BlockReason) or rl.default_block_reasons()
def add_block_reason(self, row: rl.BlockReason) -> None:
self._memo.clear()
_put(self.conn, so_block_reason, row, ("block_reason_code",))
@_memo
def return_reasons(self) -> list:
return _rows(self.conn, so_return_reason, rl.ReturnReason) or rl.default_return_reasons()
def add_return_reason(self, row: rl.ReturnReason) -> None:
self._memo.clear()
_put(self.conn, so_return_reason, row, ("return_reason_code",))
def seed(self) -> None:
self._memo.clear()
for table, rows, key in ((so_return_reason, rl.default_return_reasons(), ("return_reason_code",)),
(so_incompletion_rule, rl.default_incompletion(), ("order_type_code", "field", "valid_from")),
(so_block_reason, rl.default_block_reasons(), ("block_reason_code",)),
(so_order_type, pr.default_types(), ("order_type_code", "valid_from")),
(so_item_category, pr.default_categories(), ("item_category_code",)),
(so_item_category_rule, pr.default_rules(), ("order_type_code", "product_type_code", "usage", "valid_from")),
(so_condition_type, pr.default_condition_types(), ("condition_type_code",)),
(so_access_sequence, pr.default_access(), ("access_sequence_code", "step")),
(so_pricing_procedure, pr.default_procedure(), ("pricing_procedure_code", "step"))):
if self.conn.execute(select(table.c[key[0]])).first() is None:
for r in rows:
_put(self.conn, table, r, key)
class RebateRepository:
def __init__(self, conn):
self.conn = conn
def get(self, rebate_agreement_id: str) -> Optional[tuple]:
rows = REBATE_AGREEMENT.load(self.conn, {"rebate_agreement_id": rebate_agreement_id})
if rows is None:
return None
agg = _make(RebateAgreement, rows["so_rebate_agreement"][0], SO_REBATE_COLUMNS)
agg.accruals = [_make(RebateAccrual, x, SO_REBATE_ACCRUAL_COLUMNS) for x in _sorted(rows["so_rebate_accrual"], ("document_id", "line_no"))]
agg.settlements = [_make(RebateSettlement, x, SO_REBATE_SETTLEMENT_COLUMNS) for x in _sorted(rows["so_rebate_settlement"], ("settlement_no",))]
return agg, REBATE_AGREEMENT.version_of(rows)
def save(self, agg: RebateAgreement, expected_version: int) -> int:
return REBATE_AGREEMENT.save(self.conn, {"so_rebate_agreement": [_pick(agg, SO_REBATE_COLUMNS)],
"so_rebate_accrual": [_pick(x, SO_REBATE_ACCRUAL_COLUMNS) for x in agg.accruals],
"so_rebate_settlement": [_pick(x, SO_REBATE_SETTLEMENT_COLUMNS) for x in agg.settlements]},
expected_version=expected_version)
def claim(self, rebate_agreement_id: str, operation: str) -> int:
"""Claim the agreement's next step with a plain INSERT after a look; a racer that looked at the same step loses at the insert or the
commit (the table is a claim table): a 409, never two operations on one total."""
from erp_platform.repository import Conflict
from sqlalchemy import func
t = so_rebate_claim
seq = (self.conn.execute(select(func.max(t.c.seq)).where(t.c.rebate_agreement_id == rebate_agreement_id)).scalar() or 0) + 1
now = datetime.now(timezone.utc).replace(tzinfo=None)
try:
self.conn.execute(t.insert().values(rebate_agreement_id=rebate_agreement_id, seq=seq, operation=operation, claimed_at=now,
created_at=now, updated_at=now))
except IntegrityError:
raise Conflict() from None
return seq
def all(self) -> list:
return [_make(RebateAgreement, dict(r._mapping), SO_REBATE_COLUMNS) for r in self.conn.execute(select(so_rebate_agreement))]
class ClaimRepository:
"""The claim rows of slice C: credits and returns at one key, deposits per deposit. Every claim is a PLAIN INSERT at the next seq of its key
in the caller's transaction (no savepoint, no delete-then-insert): a racer that read the same seq loses at the insert or the commit (the tables
are claim tables) and is a 409, never two operations on one bounded total."""
def __init__(self, conn):
self.conn = conn
def _next(self, table, key_col, key) -> int:
from sqlalchemy import func
return (self.conn.execute(select(func.max(table.c.seq)).where(table.c[key_col] == key)).scalar() or 0) + 1
def _insert(self, table, **row) -> int:
from erp_platform.repository import Conflict
now = datetime.now(timezone.utc).replace(tzinfo=None)
try:
self.conn.execute(table.insert().values(**row, claimed_at=now, created_at=now, updated_at=now))
except IntegrityError:
raise Conflict() from None
return row["seq"]
def credited(self, claim_key: str):
"""The net already taken from an invoice line by credits (a release is a negative row)."""
from decimal import Decimal
rows = self.conn.execute(select(so_credit_claim.c.amount).where(so_credit_claim.c.claim_key == claim_key, so_credit_claim.c.kind == "credit")).all()
return sum((Decimal(str(r[0] or 0)) for r in rows), Decimal(0))
def returned(self, claim_key: str):
"""The quantity of an original order line already claimed by returns (a cancelled return is a negative row)."""
from decimal import Decimal
rows = self.conn.execute(select(so_credit_claim.c.quantity).where(so_credit_claim.c.claim_key == claim_key, so_credit_claim.c.kind == "return")).all()
return sum((Decimal(str(r[0] or 0)) for r in rows), Decimal(0))
def claim(self, claim_key: str, kind: str, sales_order_id: str, request_no: int = 0, amount=0, quantity=0) -> int:
return self._insert(so_credit_claim, claim_key=claim_key, seq=self._next(so_credit_claim, "claim_key", claim_key), kind=kind,
sales_order_id=sales_order_id, request_no=request_no, amount=amount, quantity=quantity)
def deducted(self, deposit_invoice_id: str):
from decimal import Decimal
rows = self.conn.execute(select(so_deposit_deduction.c.amount).where(so_deposit_deduction.c.deposit_invoice_id == deposit_invoice_id)).all()
return sum((Decimal(str(r[0] or 0)) for r in rows), Decimal(0))
def claim_deposit(self, deposit_invoice_id: str, kind: str, sales_order_id: str, request_no: int, amount, tax_category: str, after: int = 0) -> int:
"""`after`: rows this same transaction already inserted at this deposit (a staged write is not read back), so the next seq follows them."""
return self._insert(so_deposit_deduction, deposit_invoice_id=deposit_invoice_id,
seq=self._next(so_deposit_deduction, "deposit_invoice_id", deposit_invoice_id) + after,
kind=kind, sales_order_id=sales_order_id, request_no=request_no, amount=amount, tax_category=tax_category)
def is_deposit_invoice(self, invoice_id: str) -> bool:
"""Is this (ar's) invoice id the invoice of a deposit request of any order? Its credits and deductions then share ONE claim key."""
return self.conn.execute(select(so_invoice_request.c.sales_order_id).where(so_invoice_request.c.invoice_id == invoice_id,
so_invoice_request.c.kind == "deposit").limit(1)).first() is not None
class PostingRepository:
"""so_posting: one row per (source document, event). A journal the ledger refused stays pending with its reason and is retried."""
def __init__(self, conn):
self.conn = conn
def add(self, row: dict) -> None:
upsert(self.conn, so_posting, [row], ("document_id", "event"))
def get(self, document_id: str, event: str) -> Optional[dict]:
r = self.conn.execute(select(so_posting).where(so_posting.c.document_id == document_id, so_posting.c.event == event)).first()
return dict(r._mapping) if r else None
def pending(self, rebate_agreement_id: str) -> list:
return [dict(r._mapping) for r in self.conn.execute(select(so_posting).where(
so_posting.c.rebate_agreement_id == rebate_agreement_id, so_posting.c.status == "pending").order_by(so_posting.c.document_id, so_posting.c.event))]
def settle(self, document_id: str, event: str, status: str, reason: str, attempts: int, journal_id: str) -> None:
upsert(self.conn, so_posting, [{"document_id": document_id, "event": event, "status": status, "reason": reason, "attempts": attempts,
"journal_id": journal_id}], ("document_id", "event"))
def pending_sends(self, sales_order_id: str) -> list:
"""The calls to neighbours an order's committed action still owes (billing schedules, purchase requests, draft cancellations): sent after the commit."""
rows = self.conn.execute(select(so_posting).where(so_posting.c.document_id == sales_order_id, so_posting.c.status == "pending",
so_posting.c.event.like("send:%")).order_by(so_posting.c.event))
return [dict(r._mapping) for r in rows]
def pending_repromises(self) -> list:
"""The cancelled orders whose freed stock has not been re-promised yet."""
rows = self.conn.execute(select(so_posting.c.document_id).where(so_posting.c.status == "pending", so_posting.c.event == "repromise:cancelled"))
return sorted({r[0] for r in rows})
def orders_with_pending_sends(self) -> list:
"""Every order that owes a neighbour a call after its committed action (rows send:%), for the daily run."""
rows = self.conn.execute(select(so_posting.c.document_id).where(so_posting.c.status == "pending", so_posting.c.event.like("send:%")))
return sorted({r[0] for r in rows})
def pending_reservations(self) -> list:
"""The orders whose re-promised schedule was committed but not yet re-sent to inventory (SO-36): sent again by the next run."""
rows = self.conn.execute(select(so_posting).where(so_posting.c.status == "pending", so_posting.c.event.like("reserve:%"))
.order_by(so_posting.c.document_id, so_posting.c.event))
return [dict(r._mapping) for r in rows]
class TaskRepository:
"""so_task: something a person should do because the system moved a promise. One per (order, kind, business day): a plain INSERT claim."""
def __init__(self, conn):
self.conn = conn
def add(self, sales_order_id: str, kind: str, raised_on: date, owner: str, text: str) -> None:
from erp_platform.repository import Conflict
now = datetime.now(timezone.utc).replace(tzinfo=None)
try:
self.conn.execute(so_task.insert().values(sales_order_id=sales_order_id, kind=kind, raised_on=raised_on, owner=owner, text=text,
created_at=now, updated_at=now))
except IntegrityError:
raise Conflict() from None
def open_for(self, owner: Optional[str] = None, sales_order_id: Optional[str] = None, kind: Optional[str] = None) -> list:
q = select(so_task).where(so_task.c.closed_at.is_(None))
if owner is not None:
q = q.where(so_task.c.owner == owner)
if sales_order_id is not None:
q = q.where(so_task.c.sales_order_id == sales_order_id)
if kind is not None:
q = q.where(so_task.c.kind == kind)
return [dict(r._mapping) for r in self.conn.execute(q.order_by(so_task.c.raised_on, so_task.c.sales_order_id, so_task.c.kind))]
class Repository:
"""Every repository on one connection: what `use_cases.repository()` yields."""
def __init__(self, conn):
self.conn = conn
self.sales_orders = SalesOrderRepository(conn)
self.number_series = NumberSeriesRepository(conn)
self.order_no_claims = OrderNoClaimRepository(conn)
self.policies = CustomerOrderPolicyRepository(conn)
self.settings = SettingsRepository(conn)
self.migration_map = MigrationMapRepository(conn)
self.config = ConfigRepository(conn)
self.rebates = RebateRepository(conn)
self.postings = PostingRepository(conn)
self.tasks = TaskRepository(conn)
self.claims = ClaimRepository(conn)
def seed(self) -> None:
"""The rows a new bucket starts with: the settings, the order types, the item categories and their rules, the condition types,
their access sequences and the standard pricing procedure."""
self.settings.seed()
self.config.seed()