diff --git a/backend/service/app.py b/backend/service/app.py index eda35d3..23eeef0 100644 --- a/backend/service/app.py +++ b/backend/service/app.py @@ -45,9 +45,12 @@ from .schemas import ( LineSpacingAnalysisSummary, Project, ProjectSummary, + ProductDetailResponse, + ProductImageRecord, ProductInput, ProductRecord, ProductVersionArchiveResponse, + ProductWordcloudArchiveResponse, Template, WordLocation, ) @@ -1145,6 +1148,73 @@ def _product_archive_service() -> ProductArchiveService: ) +def _version_archive_response(version_id: str) -> ProductVersionArchiveResponse: + row = product_archive_store._fetchone( + "SELECT * FROM product_versions WHERE version_id = ?", (version_id,) + ) + if row is None: + raise HTTPException(status_code=404, detail="version not found") + archives = [ + ProductWordcloudArchiveResponse( + archive_id=item["archive_id"], + product_id=item["product_id"], + version_id=item["version_id"], + archive_path=item["archive_path"], + source_job_id=item["source_job_id"], + source_asset_id=item["source_asset_id"], + db_checksum=item["db_checksum"], + created_at=datetime.fromisoformat(item["created_at"]), + db_path=item["archive_path"], + ) + for item in product_archive_store._fetchall( + """SELECT * FROM product_wordcloud_archives + WHERE version_id = ? ORDER BY created_at, archive_id""", + (version_id,), + ) + ] + return ProductVersionArchiveResponse( + version_id=row["version_id"], + product_id=row["product_id"], + version=row["version"], + metadata=json.loads(row["metadata"] or "{}"), + created_at=datetime.fromisoformat(row["created_at"]), + design_preview_path=str( + PRODUCT_ARCHIVES_DIR / row["product_id"] / row["version_id"] / "design-preview.png" + ), + wordcloud_count=len(archives), + wordcloud_archives=archives, + ) + + +def _product_detail_response(product_id: str) -> ProductDetailResponse: + product = _get_product_or_404(product_id) + images = [ + ProductImageRecord( + image_id=row["image_id"], + product_id=row["product_id"], + version_id=row["version_id"], + image_path=row["image_path"], + image_type=row["image_type"], + is_cover=bool(row["is_cover"]), + created_at=datetime.fromisoformat(row["created_at"]), + ) + for row in product_archive_store._fetchall( + """SELECT * FROM product_images + WHERE product_id = ? ORDER BY created_at, image_id""", + (product_id,), + ) + ] + versions = [ + _version_archive_response(row["version_id"]) + for row in product_archive_store._fetchall( + """SELECT version_id FROM product_versions + WHERE product_id = ? ORDER BY created_at, version_id""", + (product_id,), + ) + ] + return ProductDetailResponse(**product.model_dump(), images=images, versions=versions) + + @app.post("/api/products", response_model=ProductRecord, status_code=201) def create_product(request: Request, product: ProductInput) -> ProductRecord: _require_orders_auth(request) @@ -1160,10 +1230,10 @@ def list_products(request: Request, query: str = Query("")) -> list[ProductRecor return product_archive_store.list_products(query=query) -@app.get("/api/products/{product_id}", response_model=ProductRecord) -def get_product(request: Request, product_id: str) -> ProductRecord: +@app.get("/api/products/{product_id}", response_model=ProductDetailResponse) +def get_product(request: Request, product_id: str) -> ProductDetailResponse: _require_orders_auth(request) - return _get_product_or_404(product_id) + return _product_detail_response(product_id) @app.post( @@ -1200,30 +1270,7 @@ async def create_product_version( except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc - rows = product_archive_store._fetchall( - "SELECT * FROM product_wordcloud_archives WHERE version_id = ? ORDER BY created_at, archive_id", - (version.version_id,), - ) - archives = [ - { - "archive_id": row["archive_id"], - "product_id": row["product_id"], - "version_id": row["version_id"], - "archive_path": row["archive_path"], - "created_at": datetime.fromisoformat(row["created_at"]), - "source_job_id": row["source_job_id"], - "source_asset_id": row["source_asset_id"], - "db_checksum": row["db_checksum"], - "db_path": row["archive_path"], - } - for row in rows - ] - return ProductVersionArchiveResponse( - **version.model_dump(), - design_preview_path=str(PRODUCT_ARCHIVES_DIR / product_id / version.version_id / "design-preview.png"), - wordcloud_count=len(archives), - wordcloud_archives=archives, - ) + return _version_archive_response(version.version_id) @app.delete("/api/products/{product_id}", response_model=ProductRecord) diff --git a/backend/service/schemas.py b/backend/service/schemas.py index 7e81cdc..d8f8459 100644 --- a/backend/service/schemas.py +++ b/backend/service/schemas.py @@ -89,6 +89,11 @@ class ProductVersionArchiveResponse(ProductVersionRecord): wordcloud_archives: list[ProductWordcloudArchiveResponse] = Field(default_factory=list) +class ProductDetailResponse(ProductRecord): + images: list[ProductImageRecord] = Field(default_factory=list) + versions: list[ProductVersionArchiveResponse] = Field(default_factory=list) + + class JobCreateResponse(BaseModel): job_id: str diff --git a/backend/tests/test_product_archive.py b/backend/tests/test_product_archive.py index 920eb08..1da4d58 100644 --- a/backend/tests/test_product_archive.py +++ b/backend/tests/test_product_archive.py @@ -313,6 +313,31 @@ def test_later_archive_version_becomes_current_cover_in_product_contract(product assert detail["cover_image_id"] != first_cover +def test_product_detail_contract_includes_images_and_archive_versions( + product_archive_client, prepared_wordcloud_job +): + product = create_product(product_archive_client) + archived = archive_product_version( + product_archive_client, product["product_id"], prepared_wordcloud_job.document + ) + + detail = product_archive_client.get( + f"/api/products/{product['product_id']}", headers=orders_auth_header() + ).json() + + assert archived.status_code == 201 + assert detail["product_id"] == product["product_id"] + assert len(detail["images"]) == 1 + assert detail["images"][0]["image_type"] == "design_preview" + assert detail["images"][0]["is_cover"] is True + assert len(detail["versions"]) == 1 + version = detail["versions"][0] + assert version["version_id"] == archived.json()["version_id"] + assert version["wordcloud_count"] == 1 + assert version["wordcloud_archives"][0]["source_job_id"] == "job-success" + assert Path(version["design_preview_path"]).name == "design-preview.png" + + def test_product_routes_require_orders_auth(product_archive_client): product = create_product(product_archive_client) requests = [