"""The one writer (spec §4): nothing else inserts a decision. A batch is the unit of approval and undo; `record` does its seven steps inside the batch's transaction (validate, lock, read the effective row with the human-first check, the rule's value, insert, materialise through the applier when the row is effective, queue the touched variants); the batch's exit rekeys and folds once, closes the batch, commits once. Sources of truth: this module, `appliers.py`, `effective.py`, `natural_keys.py`, `tests/test_writer.py`, the spec §4 and §5, the brief's signature block. What it cost before: three writers (`overrides.decide`, the merge session, `overrides.move`) each keeping their own idea of what a decision was, and a current-state table with no history and no undo. """ from __future__ import annotations import re import uuid from collections.abc import Iterator from decimal import Decimal from contextlib import contextmanager from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy import select, text from sqlalchemy.orm import Session from app.config import settings from app.models import Account, Brand, Decision, DecisionBatch, Listing, ProductLine, ProductVariant, Suggestion, AttributeAlias from app.models.catalog import IDENTITY_RULES_VERSION from app.models.places import Place from app.services.decisions import natural_keys from app.services.decisions.effective import effective, table_present REFUSAL_CODES = ("DECISION_HELD", "ENTITY_ALIASED", "PRODUCT_MERGED", "PAIR_SIDE_MERGED", "NOTE_REQUIRED", "VALUE_INVALID", "FIELD_UNKNOWN", "DECISION_SUPERSEDED", "ENTITY_MERGED", "UNMERGE_CHAIN", "COLLECTION_RUNNING", "SLUG_TAKEN", "LINE_MISMATCH", "VARIANT_SPLIT", "LISTING_MISSING", "REVERSAL_TARGET_MISSING", "ENTITY_NOT_FOUND", "LINE_NOT_FOUND", "LINE_HOUSES_DIFFER", "LISTING_NOT_FOUND", "PRODUCT_NOT_FOUND", "ENTITY_UNKNOWN", "PAIR_SIDE_GONE", "SPELLING_NOT_ON_ROW", "SPLIT_EMPTIES_SOURCE") class Refused(Exception): """A decision the writer will not record, with the code and the sentence the page shows.""" def __init__(self, code: str, summary: str) -> None: super().__init__(summary) self.code, self.summary = code, summary #: The fields per entity type (spec §1.7); `attribute:` is accepted for any registry kind. FIELDS: dict[str, tuple[str, ...]] = { "product_variant": ("name", "product_line", "merged_into"), "brand": ("name", "alias_of", "hidden", "indexed", "split"), "product_line": ("name", "alias_of", "hidden", "indexed"), "attribute_wording": ("meaning", "display", "kind"), "listing": ("pinned_to", "ignored"), "suggestion": ("decision",), "place": ("name", "hidden", "indexed"), } #: Fields whose value is a reference to another row: the value travels as a natural key, the #: local id sits in `value_ref_id`, and the key path reads the id. REFERENCE_FIELDS = {"product_line": "product_line", "alias_of": None, "merged_into": "product_variant", "pinned_to": "product_variant"} IDENTITY_FIELDS = ("name", "product_line") # plus every attribute: (spec §3.1) MODELS = {"brand": Brand, "product_line": ProductLine, "product_variant": ProductVariant, "listing": Listing, "attribute_wording": AttributeAlias, "suggestion": Suggestion, "place": Place} def field_ok(entity_type: str, field_name: str) -> bool: if field_name in FIELDS.get(entity_type, ()): return True if entity_type == "product_variant" and field_name.startswith("attribute:"): from app.services import attributes try: attributes.kind_of(field_name[len("attribute:"):]) return True except KeyError: return False return False def is_identity_field(field_name: str) -> bool: return field_name in IDENTITY_FIELDS or field_name.startswith("attribute:") @dataclass class Batch: db: Session row: DecisionBatch kind: str mode: str by_id: int | None by_username: str | None pass_id: int | None = None maps: Any = None touched: set[int] = field(default_factory=set) touched_keys: set[str] = field(default_factory=set) decisions: list[Decision] = field(default_factory=list) held: list[dict] = field(default_factory=list) refused: list[dict] = field(default_factory=list) skipped: list[dict] = field(default_factory=list) notes: dict[int, dict] = field(default_factory=dict) # what an applier reports about a row, never written on the row tail: bool = True commit: bool = True folded: dict = field(default_factory=dict) touched_levels: set[str] = field(default_factory=set) @property def uid(self) -> uuid.UUID: return self.row.uid def remember(self, row: Decision) -> None: self.decisions.append(row) def touch(self, variant_id: int | None) -> None: if variant_id is not None: self.touched.add(variant_id) def _username(db: Session, by) -> tuple[int | None, str | None]: if by is None: return None, None if isinstance(by, Account): return by.id, by.username if isinstance(by, str): row = db.scalar(select(Account).where(Account.username == by)) return (row.id, row.username) if row else (None, by) row = db.get(Account, int(by)) return (row.id, row.username) if row else (int(by), None) @contextmanager def batch(db: Session, kind: str, mode: str, by, scope: dict | None = None, pass_id: int | None = None, note: str | None = None, *, commit: bool = True, tail: bool = True, acting_as_id: int | None = None, reverses_batch_id: int | None = None, uid: uuid.UUID | None = None, replayed_from: str | None = None) -> Iterator[Batch]: """Open a batch: the row inserted at open (its id known), one `Maps` for the batch; at exit, once: rekey the touched variants, fold what then agrees (every merge stamped with this batch), close the batch, one commit. An exception rolls the whole batch back and records nothing.""" from app.services import keying by_id, by_username = _username(db, by) row = DecisionBatch(origin_host=settings.app_env, kind=kind, mode=mode, by_account_id=by_id, by_username=by_username, acting_as_id=acting_as_id, pass_id=pass_id, scope=scope, note=note, reverses_batch_id=reverses_batch_id, replayed_from=replayed_from, **({"uid": uid} if uid is not None else {})) db.add(row) db.flush() keying.invalidate() b = Batch(db=db, row=row, kind=kind, mode=mode, by_id=by_id, by_username=by_username, pass_id=pass_id, maps=keying.load_maps(db), tail=tail, commit=commit) try: yield b except Exception: db.rollback() keying.invalidate() raise finish(b) def finish(b: Batch, close: bool = True) -> None: """The batch's tail: rekey and fold once, close (unless the batch was closed before, a replay retry reusing it), commit once.""" from app.services import keying, merges db = b.db summary: dict[str, Any] = {"decisions": len(b.decisions), "held": len(b.held), "refused": len(b.refused)} if b.skipped: summary["skipped"] = b.skipped if b.tail and b.touched: keying.invalidate() maps = keying.load_maps(db) variants = [v for v in (db.get(ProductVariant, i) for i in sorted(b.touched)) if v is not None] touched_keys = merges.rekey_product_variants(db, variants, maps) | b.touched_keys b.folded = merges._fold(db, touched_keys, b.by_id, batch=b) summary["variants_rekeyed"] = len(variants) summary["groups_merged"] = b.folded.get("groups", 0) summary["rows_merged"] = b.folded.get("merged_rows", 0) if close and b.row.closed_at is None: b.row.closed_at = datetime.now(UTC) b.row.summary = summary db.flush() if b.commit: db.commit() keying.invalidate() # --------------------------------------------------------------------------- record def _resolve_entity(b: Batch, entity_type: str, entity, allow_merged: bool = False, allow_aliased: bool = False) -> Any: db = b.db model = MODELS[entity_type] if isinstance(entity, str): detail = None row = natural_keys.resolve(db, entity_type, entity, detail, mint=entity_type in ("attribute_wording", "suggestion")) if isinstance(row, natural_keys.Parked): raise Refused(row.code, f"{entity!r}: {row.detail or row.code}") entity = row elif isinstance(entity, int): row = db.get(model, entity) if row is None: raise Refused("ENTITY_NOT_FOUND", f"No such {entity_type} {entity}.") entity = row if entity_type == "product_variant" and entity.merged_into_id is not None and not allow_merged: raise Refused("PRODUCT_MERGED", "This product variant was merged away; decide on its survivor.") if entity_type in ("brand", "product_line") and entity.alias_of_id is not None and not allow_aliased: raise Refused("ENTITY_ALIASED", f"This {entity_type.replace('_', ' ')} is an alias; decide on the one it points at.") return entity def _lock(db: Session, entity_type: str, entity_id: int) -> None: if db.get_bind().dialect.name == "postgresql": db.execute(text("SET LOCAL lock_timeout = '5s'")) db.execute(text("SELECT pg_advisory_xact_lock(hashtext(:k))"), {"k": f"{entity_type}:{entity_id}"}) def _reference(b: Batch, entity_type: str, field_name: str, value, ref_detail: dict | None = None) -> tuple[Any, int | None, Any]: """A reference value as (natural key string, local id, the row); accepts a row, an id or a key. A key resolves with `ref_detail` (the target's own natural-key detail, carried in the row's `detail.ref`), so a target born after the fork still resolves by slug, barcode or listings.""" db = b.db target_type = REFERENCE_FIELDS.get(field_name) if target_type is None and field_name == "alias_of": target_type = entity_type model = MODELS[target_type] if isinstance(value, model): row = value elif isinstance(value, int): row = db.get(model, value) if row is None: raise Refused("VALUE_INVALID", f"No such {target_type} {value}.") elif isinstance(value, str): row = natural_keys.resolve(db, target_type, value, ref_detail) if isinstance(row, natural_keys.Parked): raise Refused(row.code, f"{value!r}: {row.detail or row.code}") else: raise Refused("VALUE_INVALID", f"{field_name} names a {target_type}.") key, _ = natural_keys.build(row) return key, row.id, row def _validate(b: Batch, entity_type: str, entity, field_name: str, value, effect: str, ref_detail: dict | None = None, reversal: bool = False): """Step (a): the field, the value's shape, a reference resolved to a live local row.""" from app.services import attributes if not field_ok(entity_type, field_name): raise Refused("FIELD_UNKNOWN", f"A {entity_type.replace('_', ' ')} has no decidable field {field_name!r}.") if effect == "release": return None, None, None ref_id = ref_row = None if field_name in REFERENCE_FIELDS or field_name == "alias_of": value, ref_id, ref_row = _reference(b, entity_type, field_name, value, ref_detail) if field_name == "product_line" and not reversal: # An UNDO restores what was in force and is not a person's new choice, so the brand # check does not apply to it: undoing a brand split reverses its consequences first, # which puts a variant's decided product line back while the variant itself is still # on the new brand row for one more statement of the same transaction (K9). _line_of_the_brand(b, entity, ref_row) if field_name == "alias_of" and ref_row.id == entity.id: raise Refused("VALUE_INVALID", "A row cannot be an alias of itself.") if field_name == "merged_into" and ref_row.id == entity.id: raise Refused("VALUE_INVALID", "A variant cannot merge into itself.") elif field_name == "name": if not isinstance(value, str) or not value.strip(): raise Refused("VALUE_INVALID", "A name is a non-empty string.") value = value.strip()[:400 if entity_type == "product_variant" else 200] elif field_name in ("hidden", "indexed", "ignored"): if not isinstance(value, bool): raise Refused("VALUE_INVALID", f"{field_name} is true or false.") elif field_name.startswith("attribute:"): try: value = attributes.validate(attributes.kind_of(field_name[len("attribute:"):]), value) except ValueError as exc: raise Refused("VALUE_INVALID", str(exc)) from exc elif field_name == "decision": if not isinstance(value, dict) or value.get("decision") not in ("same", "separate"): raise Refused("VALUE_INVALID", "A pair decision is {decision: same | separate, survivor, name, note}.") value = {"decision": value["decision"], "survivor": value.get("survivor"), "name": value.get("name"), "note": value.get("note")} elif field_name == "split": # The value carries the whole claim, and no reference: the brand it names does not exist # before the decision is written, and on a replay it is minted on the target host by the # applier's prepare step (`services/brands.py` says why the claim is the portable shape). if not isinstance(value, dict) or not isinstance(value.get("spellings"), list) or not value["spellings"]: raise Refused("VALUE_INVALID", "A split is {brand_slug, name, spellings: [...]}.") slug = str(value.get("brand_slug") or "").strip() name = str(value.get("name") or "").strip() if not slug or not re.fullmatch(r"[a-z0-9][a-z0-9-]{0,159}", slug) or not name: raise Refused("VALUE_INVALID", "A split names a slug-shaped address and a non-empty brand name.") spellings = sorted({str(x).strip() for x in value["spellings"] if str(x).strip()}) if not spellings: raise Refused("VALUE_INVALID", "A split names at least one listed spelling to move.") value = {"brand_slug": slug, "name": name[:160], "spellings": spellings} elif field_name in ("meaning", "display", "kind"): if not isinstance(value, str) or not value.strip(): raise Refused("VALUE_INVALID", f"{field_name} is a non-empty string.") value = value.strip().lower()[:80] if field_name != "display" else value.strip()[:80] return value, ref_id, ref_row def _line_of_the_brand(b: Batch, variant: ProductVariant, line: ProductLine) -> None: from app.services.product_lines import resolve_alias if variant.brand_id is None: return brand = resolve_alias(b.maps.brands, variant.brand_id) if b.maps else None if brand is not None and line.brand_id != brand.id and line.brand_id != variant.brand_id: raise Refused("LINE_HOUSES_DIFFER", "That product line belongs to another brand; confirm the brand alias first.") def _rule_value(b: Batch, entity_type: str, entity, field_name: str, prev: Decision | None) -> Any: """Step (d), spec §1.7: `name` is keep-first (the previous row's rule value, else the column); `product_line` the rule's line for the variant; an attribute the registry's reading; others NULL.""" from app.services import attributes, keying if field_name == "name": if prev is not None and prev.rule_value not in (None, ""): return prev.rule_value return entity.name if entity_type == "listing" and field_name == "pinned_to": return prev.rule_value if prev is not None and prev.rule_value is not None else entity.variant_id if entity_type == "attribute_wording": from app.services import product_lines if prev is not None and prev.rule_value is not None: return prev.rule_value rule = product_lines.canonical_attribute(entity.raw) return rule if field_name == "meaning" else (product_lines.display_attribute(rule) if field_name == "display" and rule else None) if entity_type != "product_variant": return None if field_name == "product_line": if prev is not None and prev.rule_value is not None: return prev.rule_value brand = b.maps.brand_of(entity.brand_id, entity.brand, entity.vertical) if b.maps else None line = keying.product_line_for(b.db, b.maps, brand, keying.keyed_name(entity, b.maps), entity.brand, entity.vertical, create=False, category=entity.category) if brand is not None else None return line.id if line is not None else entity.product_line_id if field_name.startswith("attribute:"): kind = field_name[len("attribute:"):] if prev is not None and prev.rule_value is not None: return prev.rule_value if kind in ("concentration", "color", "flavor"): return b.maps.attribute(keying.keyed_name(entity, b.maps), entity.vertical, category=entity.category) or None if kind == "quantity": q = attributes.quantity_of(entity) return None if q is None or q.state == "none" else {"value": q.value, "unit": q.unit, "form": q.form, "pack_count": q.pack_count, "set_contents": q.set_contents} current = attributes.get(entity, kind) return current if current not in ("", None) else None return None def _aware(value: datetime) -> datetime: """SQLite hands a naive datetime back; the ledger's clock is UTC.""" return value if value.tzinfo is not None else value.replace(tzinfo=UTC) def jsonable(value: Any) -> Any: """A column value as JSON: a Decimal is an int or a float, a datetime its ISO form.""" if isinstance(value, Decimal): f = float(value) return int(f) if f.is_integer() else f if isinstance(value, datetime): return value.isoformat() if isinstance(value, dict): return {k: jsonable(v) for k, v in value.items()} if isinstance(value, (list, tuple)): return [jsonable(v) for v in value] return value def record(b: Batch, entity_type: str, entity, field_name: str, value, *, origin: str = "person", proposal=None, reason: str | None = None, decided_at: datetime | None = None, effect: str = "set", cause: Decision | None = None, detail: dict | None = None, decided_by=None, mode: str | None = None, reverses: Decision | None = None, uid: uuid.UUID | None = None, replayed_from: str | None = None, restores: Decision | None = None) -> Decision: """One decision, the seven steps of spec §4. Raises `Refused(code)`. `reverses` makes it an undo row (`reverses_id`, `restores_id`, spec §7); `uid` and `replayed_from` are replay's.""" from app.services.decisions import appliers db = b.db entity = _resolve_entity(b, entity_type, entity, allow_merged=(field_name == "merged_into" and effect == "release"), allow_aliased=(field_name == "alias_of")) if decided_at is None: decided_at = datetime.now(UTC) # A replayed timestamp arrives as text and comes back naive from a database that stores no # zone (the tests' SQLite); the ledger's clock is UTC, so read it as UTC rather than # comparing a naive value to an aware one and raising TypeError inside the batch. decided_at = _aware(decided_at) if decided_at > datetime.now(UTC) + timedelta(minutes=5): raise Refused("VALUE_INVALID", "A decision cannot be dated in the future.") value, ref_id, ref_row = _validate(b, entity_type, entity, field_name, value, effect, (detail or {}).get("ref"), reversal=reverses is not None) if ref_row is not None: detail = {**(detail or {}), "ref": natural_keys.build(ref_row)[1]} # the target's own detail rides with the row # (b) the lock, (c) the effective row and the human-first check. _lock(db, entity_type, entity.id) prev = effective(db, entity_type, [entity.id], [field_name]).get((entity.id, field_name)) mode = mode or b.mode if mode == "bulk" and reverses is None and prev is not None and prev.effect == "set" and prev.value != value \ and (prev.origin == "person" or prev.mode == "individual"): # an undo is a person's own act raise Refused("DECISION_HELD", f"{field_name} was decided by {prev.decided_by_username or 'a person'} on " f"{prev.decided_at:%Y-%m-%d}; a bulk decision does not overrule it (decision {prev.uid}).") # (d) what the rules had, and what was in force. rule_value = jsonable(_rule_value(b, entity_type, entity, field_name, prev)) prior_value = jsonable(prev.value if prev is not None else rule_value) value = jsonable(value) # (e) the row: its detail complete at insert (the ledger is append-only), the applier's # prepare step first, then whether it will be effective (a replayed older row is not). applier = appliers.for_field(entity_type, field_name) if effect == "set" and applier.prepare is not None: extra = applier.prepare(b, entity, ref_row if ref_row is not None else value) or {} detail = {**(detail or {}), **extra} is_effective = prev is None or _aware(decided_at) > _aware(prev.decided_at) or _aware(decided_at) == _aware(prev.decided_at) if not is_effective: detail = {**(detail or {}), "conflict": {"target_uid": str(prev.uid), "kept": "target"}} natural_key, key_detail = natural_keys.build(entity) if uid is None: if cause is not None: uid = uuid.uuid5(cause.uid, f"{entity_type}|{natural_key}|{field_name}") if db.scalar(select(Decision.id).where(Decision.uid == uid)) is not None: uid = uuid.uuid5(uid, str(len(b.decisions))) else: uid = uuid.uuid4() by_id, by_username = (b.by_id, b.by_username) if decided_by is None else _username(db, decided_by) row = Decision( uid=uid, decided_at=decided_at, origin_host=settings.app_env, entity_type=entity_type, entity_id=entity.id, natural_key=natural_key, natural_key_detail=key_detail, field=field_name, effect=effect, value=value if effect == "set" else None, value_ref_id=ref_id, rule_value=rule_value, prior_value=prior_value, supersedes_id=prev.id if prev is not None else None, caused_by_id=cause.id if cause is not None else None, rules_version=IDENTITY_RULES_VERSION, origin=origin, proposal_id=getattr(proposal, "id", proposal), pass_id=b.pass_id if b.pass_id is not None else getattr(proposal, "pass_id", None), batch_id=b.row.id, mode=mode, decided_by=by_id, decided_by_username=by_username, reason=reason, detail=detail, replayed_from=replayed_from, reverses_id=reverses.id if reverses is not None else None, restores_id=restores.id if restores is not None else (reverses.supersedes_id if reverses is not None and effect == "set" else None), ) db.add(row) db.flush() b.remember(row) # (f) materialise only when this row is effective (always, for a native write); nothing on the # row changes after its insert. if is_effective: if effect == "set": applier.materialise(b, row, entity, ref_row if ref_row is not None else value) applier.consequences(b, row, entity, ref_row if ref_row is not None else value) else: applier.release(b, row, entity, reverses if reverses is not None else prev) # (g) the touched variants. if entity_type == "product_variant": b.touch(entity.id) return row def refuse_if_collecting(db: Session, force: bool = False) -> None: """The gate every batch route shares (spec §6.4, test 22).""" from app.models import CollectionRun from sqlalchemy import func if force or not table_present(db, "collection_runs"): return running = db.scalar(select(func.count(CollectionRun.id)).where(CollectionRun.status == "running")) or 0 if running: raise Refused("COLLECTION_RUNNING", f"{running} collection run(s) are running; wait, or force.") def ledger_present(db: Session) -> bool: return table_present(db)