From 1c2a6af9869ead73acd19881cbd9cc4a91768bd3 Mon Sep 17 00:00:00 2001 From: Ben Stull Date: Thu, 11 Jun 2026 22:03:09 -0700 Subject: [PATCH] =?UTF-8?q?feat(products):=20streamed=20export=20service?= =?UTF-8?q?=20+=20TEL-3=20+=20EmptyCatalog=20(PUC-9,=20=C2=A79.1)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/domains/products/__init__.py | 6 ++-- backend/app/domains/products/errors.py | 4 +++ backend/app/domains/products/service.py | 40 +++++++++++++++++++-- backend/tests/test_products_export.py | 46 +++++++++++++++++++++++- 4 files changed, 91 insertions(+), 5 deletions(-) diff --git a/backend/app/domains/products/__init__.py b/backend/app/domains/products/__init__.py index c2608ff..45bc81e 100644 --- a/backend/app/domains/products/__init__.py +++ b/backend/app/domains/products/__init__.py @@ -11,6 +11,7 @@ from pathlib import Path from .errors import ( DraftExpired, DraftNotFound, + EmptyCatalog, FileRejected, NothingToApply, PreviewStale, @@ -21,6 +22,7 @@ from .models import MAX_DATA_ROWS, MAX_FILE_BYTES from .service import ( confirm_draft, discard_draft, + export_catalog, get_draft, get_draft_records, get_run, @@ -34,8 +36,8 @@ SAMPLE_CSV_PATH = Path(__file__).parent / "sample.csv" __all__ = [ "ProductsError", "FileRejected", "DraftNotFound", "DraftExpired", - "PreviewStale", "NothingToApply", "RunNotFound", + "PreviewStale", "NothingToApply", "RunNotFound", "EmptyCatalog", "MAX_DATA_ROWS", "MAX_FILE_BYTES", "SAMPLE_CSV_PATH", "import_validate", "get_draft", "get_draft_records", "discard_draft", - "confirm_draft", "list_runs", "get_run", "summary", + "confirm_draft", "list_runs", "get_run", "summary", "export_catalog", ] diff --git a/backend/app/domains/products/errors.py b/backend/app/domains/products/errors.py index 310cebe..08beb03 100644 --- a/backend/app/domains/products/errors.py +++ b/backend/app/domains/products/errors.py @@ -35,3 +35,7 @@ class NothingToApply(ProductsError): class RunNotFound(ProductsError): """No such import run for this storefront.""" + + +class EmptyCatalog(ProductsError): + """PUC-9: nothing to export (no products, or none matching the status filter).""" diff --git a/backend/app/domains/products/service.py b/backend/app/domains/products/service.py index 3e0ce96..ba72f5a 100644 --- a/backend/app/domains/products/service.py +++ b/backend/app/domains/products/service.py @@ -7,14 +7,22 @@ writes exactly one row — the import_draft. TEL events per §9.1. from __future__ import annotations import time +from collections.abc import Iterator from datetime import datetime, timezone import psycopg from app.platform import telemetry -from . import codec, diff, repo, validate -from .errors import DraftExpired, DraftNotFound, NothingToApply, PreviewStale, RunNotFound +from . import codec, diff, repo, serialize, validate +from .errors import ( + DraftExpired, + DraftNotFound, + EmptyCatalog, + NothingToApply, + PreviewStale, + RunNotFound, +) def import_validate(conn: psycopg.Connection, storefront_id: int, account_id: int, @@ -218,6 +226,34 @@ def get_run(conn: psycopg.Connection, storefront_id: int, run_id: int) -> dict: return run +def export_catalog( + conn: psycopg.Connection, storefront_id: int, status_filter: str +) -> Iterator[str]: + """Stream the storefront's catalog as canonical CSV (PUC-9; INV-12 codec). + + Read-only: builds the snapshot, then streams the serializer over it. TEL-3 + is emitted once the stream is exhausted, with the product count and elapsed + time. Raises EmptyCatalog before yielding anything if the (filtered) catalog + is empty, so the BFF can answer 409 cleanly with no partial body. + """ + started = time.monotonic() + snapshot = repo.export_catalog(conn, storefront_id, status_filter) + if not snapshot: + raise EmptyCatalog() + + def _stream() -> Iterator[str]: + yield from serialize.catalog_to_csv(snapshot) + telemetry.emit( + "catalog_exported", + storefront_id=storefront_id, + status_filter=status_filter, + product_count=len(snapshot), + duration_ms=int((time.monotonic() - started) * 1000), + ) + + return _stream() + + def summary(conn: psycopg.Connection, storefront_id: int) -> dict: """The products dashboard counts (§6.4).""" return { diff --git a/backend/tests/test_products_export.py b/backend/tests/test_products_export.py index 5ea3fd4..959073b 100644 --- a/backend/tests/test_products_export.py +++ b/backend/tests/test_products_export.py @@ -8,7 +8,7 @@ import psycopg import pytest from app.domains import products -from app.domains.products import repo, serialize +from app.domains.products import repo from app.platform import db CSV = ( @@ -54,3 +54,47 @@ def test_export_status_filter(migrated_conn, merchant): assert [p.handle for p in active] == ["active-mug"] draft = repo.export_catalog(migrated_conn, merchant["storefront_id"], "draft") assert [p.handle for p in draft] == ["draft-tee"] + + +@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_export_streams_canonical_csv(migrated_conn, merchant): + _seed(migrated_conn, merchant) + text = "".join(products.export_catalog(migrated_conn, merchant["storefront_id"], "all")) + rows = list(csv.DictReader(io.StringIO(text))) + assert {r["Handle"] for r in rows} == {"active-mug", "draft-tee"} + assert rows[0]["Handle"] == "active-mug" # sorted by handle + + +def test_export_empty_raises(migrated_conn, merchant): + # No catalog at all → empty. + with pytest.raises(products.EmptyCatalog): + list(products.export_catalog(migrated_conn, merchant["storefront_id"], "all")) + + +def test_export_empty_after_filter_raises(migrated_conn, merchant): + _seed(migrated_conn, merchant) # only active + draft exist + with pytest.raises(products.EmptyCatalog): + list(products.export_catalog(migrated_conn, merchant["storefront_id"], "archived")) + + +def test_tel3_emitted(migrated_conn, merchant, caplog, telemetry_propagation): + _seed(migrated_conn, merchant) + with caplog.at_level(logging.INFO, logger="ecomm.telemetry"): + list(products.export_catalog(migrated_conn, merchant["storefront_id"], "all")) + events = [json.loads(r.message) for r in caplog.records if r.name == "ecomm.telemetry"] + exported = [e for e in events if e["event"] == "catalog_exported"] + assert len(exported) == 1 + assert exported[0]["product_count"] == 2 + assert exported[0]["status_filter"] == "all" + assert "duration_ms" in exported[0]