feat: return full product detail with images and versions
This commit is contained in:
+74
-27
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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 = [
|
||||
|
||||
Reference in New Issue
Block a user