From 627e257d4ccedeacab53271848f64ff31df20a5e Mon Sep 17 00:00:00 2001 From: Ben Stull Date: Thu, 11 Jun 2026 21:57:48 -0700 Subject: [PATCH] =?UTF-8?q?feat(products):=20canonical=20serializer=20skel?= =?UTF-8?q?eton=20=E2=80=94=20header=20+=20single=20product=20row=20(SD-00?= =?UTF-8?q?02=20=C2=A76.5.5)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/domains/products/serialize.py | 97 +++++++++++++++++++++++ backend/tests/test_products_serialize.py | 53 +++++++++++++ 2 files changed, 150 insertions(+) create mode 100644 backend/app/domains/products/serialize.py create mode 100644 backend/tests/test_products_serialize.py diff --git a/backend/app/domains/products/serialize.py b/backend/app/domains/products/serialize.py new file mode 100644 index 0000000..47b2eaf --- /dev/null +++ b/backend/app/domains/products/serialize.py @@ -0,0 +1,97 @@ +"""Canonical serializer — CatalogProduct snapshot → canonical CSV (SD-0002 §6.5.5). + +The export half of "one codec, two directions": this writes exactly the columns +codec.py/validate.py parse, in a row grammar (§6.5.1) the validator regroups +identically — so re-importing an unmodified export diffs to nothing (INV-12). +DB-free, like diff.py: it consumes the same CatalogProduct snapshot the diff +engine reads (repo.load_catalog), and the service streams the result. +""" +from __future__ import annotations + +import csv +import io +from collections.abc import Iterable, Iterator +from decimal import Decimal + +from .diff import CatalogProduct +from .models import ( + IMAGE_COLUMNS, + OPTION_VALUE_COLUMNS, + PRODUCT_COLUMNS, + VARIANT_COLUMNS, +) + +# Canonical column order: Handle, then product, option-value, variant, image +# columns. A superset of everything the parser knows (models.KNOWN_COLUMNS minus +# the #15-reserved Component columns, which export never emits). +HEADER: list[str] = [ + "Handle", + *PRODUCT_COLUMNS, + *OPTION_VALUE_COLUMNS, + *VARIANT_COLUMNS, + *IMAGE_COLUMNS, +] + +# field name -> column name, inverting the model registry (the parser reads +# column->field; the serializer writes field->column). +_PRODUCT_FIELD_TO_COL = {field: col for col, field in PRODUCT_COLUMNS.items()} +_VARIANT_FIELD_TO_COL = {field: col for col, field in VARIANT_COLUMNS.items()} + + +def _cell(value: object) -> str: + """Serialize one value to its canonical cell text (the parse inverse).""" + if value is None: + return "" + if isinstance(value, bool): + return "TRUE" if value else "FALSE" + if isinstance(value, Decimal): + return str(value) + if isinstance(value, (list, tuple)): + return ", ".join(str(v) for v in value) + return str(value) + + +def catalog_to_csv(products: Iterable[CatalogProduct]) -> Iterator[str]: + """Stream canonical CSV text, header first, one product block at a time.""" + buf = io.StringIO() + writer = csv.writer(buf) + writer.writerow(HEADER) + yield _drain(buf) + for product in products: + for row in _product_rows(product): + writer.writerow([row.get(col, "") for col in HEADER]) + yield _drain(buf) + + +def _drain(buf: io.StringIO) -> str: + text = buf.getvalue() + buf.seek(0) + buf.truncate(0) + return text + + +def _product_rows(product: CatalogProduct) -> list[dict[str, str]]: + """The product's CSV rows (§6.5.1 grammar). Task 1: first variant only.""" + base = {"Handle": product.handle, "Title": product.title} + for field, col in _PRODUCT_FIELD_TO_COL.items(): + if field in product.fields: + base[col] = _cell(product.fields[field]) + for slot, name in enumerate(product.option_names, start=1): + if name: + base[f"Option{slot} Name"] = name + variant = product.variants[0] + row = dict(base) + _write_variant(row, product, variant) + return [row] + + +def _write_variant(row: dict[str, str], product: CatalogProduct, variant) -> None: + """Fill a row's option-value + variant columns for one variant.""" + for slot, value in enumerate(variant.options, start=1): + # Only emit an option value when the product actually has that option + # (a no-option product's single variant carries all-NULL options). + if product.option_names[slot - 1] and value is not None: + row[f"Option{slot} Value"] = value + for field, col in _VARIANT_FIELD_TO_COL.items(): + if field in variant.fields: + row[col] = _cell(variant.fields[field]) diff --git a/backend/tests/test_products_serialize.py b/backend/tests/test_products_serialize.py new file mode 100644 index 0000000..0c5fb47 --- /dev/null +++ b/backend/tests/test_products_serialize.py @@ -0,0 +1,53 @@ +"""Canonical serializer + INV-12 round-trip lock (SD-0002 §6.5.5, §6.8).""" +import csv +import io + +from app.domains.products import serialize +from app.domains.products.diff import CatalogImage, CatalogProduct, CatalogVariant + + +def _product(**kw) -> CatalogProduct: + """A minimal no-option catalog product (one all-NULL variant).""" + base = dict( + id=1, handle="moon-mug", title="Moon Mug", + option_names=(None, None, None), + fields={ + "title": "Moon Mug", "description_html": None, "vendor": "Acme", + "product_type": "standalone", "google_product_category": None, + "tags": [], "status": "active", "published": True, + }, + variants=[CatalogVariant( + id=1, options=(None, None, None), position=1, + fields={"sku": "WG-MUG", "barcode": None, "price": None, "cost": None, + "weight": None, "weight_unit": None, "volume": None, + "volume_unit": None, "tax_id_1": None, "tax_id_2": None, + "inventory_tracker": None, "inventory_qty": None, + "variant_image": None})], + images=[], + ) + base.update(kw) + return CatalogProduct(**base) + + +def _rows(products) -> list[dict]: + text = "".join(serialize.catalog_to_csv(products)) + return list(csv.DictReader(io.StringIO(text))) + + +def test_header_is_full_canonical_set(): + text = "".join(serialize.catalog_to_csv([_product()])) + header = next(csv.reader(io.StringIO(text))) + # Handle + Title first; every known column present exactly once. + assert header[0] == "Handle" + assert "Title" in header and "Variant SKU" in header and "Image Src" in header + assert len(header) == len(set(header)) + + +def test_single_no_option_product_one_row(): + rows = _rows([_product()]) + assert len(rows) == 1 + assert rows[0]["Handle"] == "moon-mug" + assert rows[0]["Title"] == "Moon Mug" + assert rows[0]["Variant SKU"] == "WG-MUG" + # A no-option product emits no option values. + assert rows[0]["Option1 Value"] == ""