feat(products): status-filtered export_catalog snapshot query (PUC-9)

This commit is contained in:
2026-06-11 22:01:11 -07:00
parent fce0b5eaed
commit 5a97b9dc59
2 changed files with 73 additions and 0 deletions
+17
View File
@@ -104,6 +104,23 @@ def load_catalog(conn: psycopg.Connection, storefront_id: int) -> dict[str, Cata
return catalog
_EXPORT_STATUSES = ("all", "active", "draft", "archived")
def export_catalog(
conn: psycopg.Connection, storefront_id: int, status_filter: str
) -> list[CatalogProduct]:
"""The storefront's catalog as an ordered snapshot list, optionally filtered
by product status (PUC-9). Reuses load_catalog's snapshot builder; the
catalog fits in memory (≤5k rows, INV-18). Ordered by handle for a stable,
deterministic export."""
catalog = load_catalog(conn, storefront_id)
products = sorted(catalog.values(), key=lambda p: p.handle)
if status_filter and status_filter != "all":
products = [p for p in products if p.fields.get("status") == status_filter]
return products
def product_count(conn: psycopg.Connection, storefront_id: int) -> int:
return conn.execute(
"SELECT count(*) FROM product WHERE storefront_id = %s", (storefront_id,)