From fcbf1393f5a1f2abe6502ac324a60a0a0867369b Mon Sep 17 00:00:00 2001 From: Ben Stull Date: Thu, 11 Jun 2026 15:51:03 -0700 Subject: [PATCH] =?UTF-8?q?feat(products):=20import=5Fvalidate=20=E2=86=92?= =?UTF-8?q?=20draft,=20preview=20records,=20discard=20+=20TEL-1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5 --- backend/app/domains/products/__init__.py | 2 + backend/app/domains/products/service.py | 88 +++++++++++++++++ backend/app/platform/telemetry.py | 13 +++ backend/tests/test_products_service.py | 117 +++++++++++++++++++++++ 4 files changed, 220 insertions(+) create mode 100644 backend/app/domains/products/service.py create mode 100644 backend/app/platform/telemetry.py create mode 100644 backend/tests/test_products_service.py diff --git a/backend/app/domains/products/__init__.py b/backend/app/domains/products/__init__.py index 8115618..154ef2c 100644 --- a/backend/app/domains/products/__init__.py +++ b/backend/app/domains/products/__init__.py @@ -16,9 +16,11 @@ from .errors import ( RunNotFound, ) from .models import MAX_DATA_ROWS, MAX_FILE_BYTES +from .service import discard_draft, get_draft, get_draft_records, import_validate __all__ = [ "ProductsError", "FileRejected", "DraftNotFound", "DraftExpired", "PreviewStale", "NothingToApply", "RunNotFound", "MAX_DATA_ROWS", "MAX_FILE_BYTES", + "import_validate", "get_draft", "get_draft_records", "discard_draft", ] diff --git a/backend/app/domains/products/service.py b/backend/app/domains/products/service.py new file mode 100644 index 0000000..32bd2ac --- /dev/null +++ b/backend/app/domains/products/service.py @@ -0,0 +1,88 @@ +"""products service — the import/export use-case orchestration (SD-0002 §6.5). + +Coordinates codec → validate → diff → repo; owns transaction boundaries (repo +never commits). Preview is read-only against catalog tables (INV-11): validation +writes exactly one row — the import_draft. TEL events per §9.1. +""" +from __future__ import annotations + +import time +from datetime import datetime, timezone + +import psycopg + +from app.platform import telemetry + +from . import codec, diff, repo, validate +from .errors import DraftExpired, DraftNotFound + + +def import_validate(conn: psycopg.Connection, storefront_id: int, account_id: int, + file_name: str, data: bytes) -> dict: + """Upload → validate → diff → persist draft (PUC-2/3; INV-11). Raises FileRejected.""" + started = time.monotonic() + # Commit the sweep before parsing: a FileRejected mid-parse must not roll + # back expired-draft cleanup along with it. + repo.sweep_expired_drafts(conn) + conn.commit() + parsed = codec.parse_csv(data) + products = validate.build_products(parsed) + catalog = repo.load_catalog(conn, storefront_id) + diff_result = diff.compute_diff(catalog, products) + draft = repo.insert_draft( + conn, storefront_id, account_id, file_name, parsed.dialect, data, + diff_result.summary, diff_result.records, diff_result.fingerprint, + parsed.unknown_columns, + ) + conn.commit() + telemetry.emit( + "import_draft_created", + storefront_id=storefront_id, + dialect=parsed.dialect, + row_count=len(parsed.rows), + adds=diff_result.summary["adds"], + updates=diff_result.summary["updates"], + unchanged=diff_result.summary["unchanged"], + errors=diff_result.summary["errors"], + unknown_columns_count=len(parsed.unknown_columns), + duration_ms=int((time.monotonic() - started) * 1000), + ) + return draft + + +def _live_draft_row(conn: psycopg.Connection, storefront_id: int, draft_id: int) -> dict: + """The draft row if it exists and hasn't expired; expiry deletes lazily (§6.3).""" + row = repo.get_draft_row(conn, storefront_id, draft_id) + if row is None: + raise DraftNotFound() + if row["expires_at"] < datetime.now(timezone.utc): + repo.delete_draft(conn, storefront_id, draft_id) + conn.commit() + raise DraftExpired() + return row + + +def get_draft(conn: psycopg.Connection, storefront_id: int, draft_id: int) -> dict: + """The §6.4 draft payload — never file_bytes or the full records list.""" + row = _live_draft_row(conn, storefront_id, draft_id) + return { + "id": row["id"], + "file_name": row["file_name"], + "dialect": row["dialect"], + "summary": row["summary"], + "unknown_columns": row["unknown_columns"], + "expires_at": row["expires_at"].isoformat(), + } + + +def get_draft_records(conn: psycopg.Connection, storefront_id: int, draft_id: int, + kind: str | None = None, limit: int = 100, offset: int = 0) -> list[dict]: + """The draft's preview records, paged, optionally filtered by kind (PUC-3).""" + _live_draft_row(conn, storefront_id, draft_id) + return repo.draft_records(conn, storefront_id, draft_id, kind, limit, offset) + + +def discard_draft(conn: psycopg.Connection, storefront_id: int, draft_id: int) -> None: + """Delete the draft, no trace kept; idempotent — an absent draft is fine (PUC-3a).""" + repo.delete_draft(conn, storefront_id, draft_id) + conn.commit() diff --git a/backend/app/platform/telemetry.py b/backend/app/platform/telemetry.py new file mode 100644 index 0000000..8abd7e6 --- /dev/null +++ b/backend/app/platform/telemetry.py @@ -0,0 +1,13 @@ +"""Structured log-event telemetry (SD-0002 §9.1). One JSON object per event on the +`ecomm.telemetry` logger — counts and durations only; never file names, URLs, +catalog content, or secret bytes (§6.3-handbook).""" +from __future__ import annotations + +import json +import logging + +_logger = logging.getLogger("ecomm.telemetry") + + +def emit(event: str, **fields: object) -> None: + _logger.info(json.dumps({"event": event, **fields}, sort_keys=True, default=str)) diff --git a/backend/tests/test_products_service.py b/backend/tests/test_products_service.py new file mode 100644 index 0000000..0c22b43 --- /dev/null +++ b/backend/tests/test_products_service.py @@ -0,0 +1,117 @@ +"""products service — drafts: validate/preview/discard (PUC-2/3/3a/5a; INV-11).""" +import json +import logging + +import psycopg +import pytest + +from app.domains import products +from app.platform import db + +GOOD_CSV = b"Handle,Title,Vendor,Variant Price\nmoon-mug,Moon Mug,Acme,18.00\nstar-tee,Star Tee,Acme,24.00\n" + + +@pytest.fixture() +def migrated_conn(fresh_db_url): + with psycopg.connect(fresh_db_url) as conn: + db.migrate(conn) + yield conn + + +@pytest.fixture() +def merchant(migrated_conn): + acct = migrated_conn.execute( + "INSERT INTO account (email) VALUES ('m@example.com') RETURNING id").fetchone()[0] + sf = migrated_conn.execute( + "INSERT INTO storefront (name) VALUES ('Shop') RETURNING id").fetchone()[0] + migrated_conn.execute( + "INSERT INTO storefront_membership (account_id, storefront_id) VALUES (%s,%s)", (acct, sf)) + migrated_conn.commit() + return {"account_id": acct, "storefront_id": sf} + + +def test_import_validate_creates_draft_with_summary(migrated_conn, merchant): + draft = products.import_validate( + migrated_conn, merchant["storefront_id"], merchant["account_id"], "cat.csv", GOOD_CSV) + assert draft["dialect"] == "canonical" + assert draft["summary"] == {"adds": 2, "updates": 0, "unchanged": 0, "errors": 0} + assert draft["expires_at"] + + +def test_validate_writes_nothing_to_catalog_inv11(migrated_conn, merchant): + products.import_validate( + migrated_conn, merchant["storefront_id"], merchant["account_id"], "cat.csv", GOOD_CSV) + assert migrated_conn.execute("SELECT count(*) FROM product").fetchone()[0] == 0 + assert migrated_conn.execute("SELECT count(*) FROM variant").fetchone()[0] == 0 + + +def test_file_rejection_leaves_no_draft(migrated_conn, merchant): + with pytest.raises(products.FileRejected): + products.import_validate( + migrated_conn, merchant["storefront_id"], merchant["account_id"], "bad.csv", + b"Vendor,Price\nAcme,1\n") + assert migrated_conn.execute("SELECT count(*) FROM import_draft").fetchone()[0] == 0 + + +def test_records_paging_and_kind_filter(migrated_conn, merchant): + draft = products.import_validate( + migrated_conn, merchant["storefront_id"], merchant["account_id"], "cat.csv", GOOD_CSV) + recs = products.get_draft_records(migrated_conn, merchant["storefront_id"], draft["id"]) + assert [r["handle"] for r in recs] == ["moon-mug", "star-tee"] + adds = products.get_draft_records( + migrated_conn, merchant["storefront_id"], draft["id"], kind="add", limit=1) + assert len(adds) == 1 and adds[0]["kind"] == "add" + + +def test_discard_deletes_no_trace_puc3a(migrated_conn, merchant): + draft = products.import_validate( + migrated_conn, merchant["storefront_id"], merchant["account_id"], "cat.csv", GOOD_CSV) + products.discard_draft(migrated_conn, merchant["storefront_id"], draft["id"]) + assert migrated_conn.execute("SELECT count(*) FROM import_draft").fetchone()[0] == 0 + products.discard_draft(migrated_conn, merchant["storefront_id"], draft["id"]) # idempotent + + +def test_draft_scoped_to_storefront_inv14(migrated_conn, merchant): + draft = products.import_validate( + migrated_conn, merchant["storefront_id"], merchant["account_id"], "cat.csv", GOOD_CSV) + other_sf = migrated_conn.execute( + "INSERT INTO storefront (name) VALUES ('Other') RETURNING id").fetchone()[0] + migrated_conn.commit() + with pytest.raises(products.DraftNotFound): + products.get_draft(migrated_conn, other_sf, draft["id"]) + + +def test_expired_draft_raises_and_lazily_deletes(migrated_conn, merchant): + draft = products.import_validate( + migrated_conn, merchant["storefront_id"], merchant["account_id"], "cat.csv", GOOD_CSV) + migrated_conn.execute( + "UPDATE import_draft SET expires_at = now() - interval '1 minute' WHERE id = %s", + (draft["id"],)) + migrated_conn.commit() + with pytest.raises(products.DraftExpired): + products.get_draft(migrated_conn, merchant["storefront_id"], draft["id"]) + assert migrated_conn.execute("SELECT count(*) FROM import_draft").fetchone()[0] == 0 + + +@pytest.fixture() +def telemetry_propagation(): + """create_app() sets propagate=False on the parent "ecomm" logger + (main._ensure_app_logging), which hides ecomm.telemetry records from caplog's + root-logger handler whenever an API test ran first. Restore propagation here.""" + lg = logging.getLogger("ecomm") + prior = lg.propagate + lg.propagate = True + yield + lg.propagate = prior + + +def test_tel1_emitted(migrated_conn, merchant, caplog, telemetry_propagation): + with caplog.at_level(logging.INFO, logger="ecomm.telemetry"): + products.import_validate( + migrated_conn, merchant["storefront_id"], merchant["account_id"], "cat.csv", GOOD_CSV) + events = [json.loads(r.message) for r in caplog.records if r.name == "ecomm.telemetry"] + assert any( + e["event"] == "import_draft_created" and e["adds"] == 2 and e["row_count"] == 2 + and "duration_ms" in e and e["unknown_columns_count"] == 0 + for e in events + )