feat(modules): create cart, catalog, and checkout e-commerce modules

Phase 3 of storefront restructure plan - create dedicated modules for
e-commerce functionality:

- cart: Shopping cart management with storefront API routes
  - CartItem model with cents-based pricing
  - CartService for cart operations
  - Storefront routes for cart CRUD operations

- catalog: Product catalog browsing for customers
  - CatalogService for public product queries
  - Storefront routes for product listing/search/details

- checkout: Order creation from cart (placeholder)
  - CheckoutService stub for future cart-to-order conversion
  - Schemas for checkout flow

These modules separate e-commerce concerns from core platform
concerns (customer auth), enabling non-commerce platforms.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
2026-01-29 22:53:35 +01:00
parent 3e86d4b58b
commit 0845555413
32 changed files with 1748 additions and 3 deletions

View File

@@ -0,0 +1,9 @@
# app/modules/catalog/routes/api/__init__.py
"""Catalog module API routes."""
from app.modules.catalog.routes.api.storefront import router as storefront_router
# Tag for OpenAPI documentation
STOREFRONT_TAG = "Catalog (Storefront)"
__all__ = ["storefront_router", "STOREFRONT_TAG"]

View File

@@ -0,0 +1,170 @@
# app/modules/catalog/routes/api/storefront.py
"""
Catalog Module - Storefront API Routes
Public endpoints for browsing product catalog in storefront.
Uses vendor from middleware context (VendorContextMiddleware).
No authentication required.
Vendor Context: require_vendor_context() - detects vendor from URL/subdomain/domain
"""
import logging
from fastapi import APIRouter, Depends, Path, Query, Request
from sqlalchemy.orm import Session
from app.core.database import get_db
from app.modules.catalog.services import catalog_service
from app.modules.catalog.schemas import (
ProductDetailResponse,
ProductListResponse,
ProductResponse,
)
from middleware.vendor_context import require_vendor_context
from models.database.vendor import Vendor
router = APIRouter()
logger = logging.getLogger(__name__)
@router.get("/products", response_model=ProductListResponse) # public
def get_product_catalog(
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=1000),
search: str | None = Query(None, description="Search products by name"),
is_featured: bool | None = Query(None, description="Filter by featured products"),
vendor: Vendor = Depends(require_vendor_context()),
db: Session = Depends(get_db),
):
"""
Get product catalog for current vendor.
Vendor is automatically determined from request context (URL/subdomain/domain).
Only returns active products visible to customers.
No authentication required.
Query Parameters:
- skip: Number of products to skip (pagination)
- limit: Maximum number of products to return
- search: Search query for product name/description
- is_featured: Filter by featured products only
"""
logger.debug(
f"[CATALOG_STOREFRONT] get_product_catalog for vendor: {vendor.subdomain}",
extra={
"vendor_id": vendor.id,
"vendor_code": vendor.subdomain,
"skip": skip,
"limit": limit,
"search": search,
"is_featured": is_featured,
},
)
# Get only active products for public view
products, total = catalog_service.get_catalog_products(
db=db,
vendor_id=vendor.id,
skip=skip,
limit=limit,
is_featured=is_featured,
)
return ProductListResponse(
products=[ProductResponse.model_validate(p) for p in products],
total=total,
skip=skip,
limit=limit,
)
@router.get("/products/{product_id}", response_model=ProductDetailResponse) # public
def get_product_details(
product_id: int = Path(..., description="Product ID", gt=0),
vendor: Vendor = Depends(require_vendor_context()),
db: Session = Depends(get_db),
):
"""
Get detailed product information for customers.
Vendor is automatically determined from request context (URL/subdomain/domain).
No authentication required.
Path Parameters:
- product_id: ID of the product to retrieve
"""
logger.debug(
f"[CATALOG_STOREFRONT] get_product_details for product {product_id}",
extra={
"vendor_id": vendor.id,
"vendor_code": vendor.subdomain,
"product_id": product_id,
},
)
product = catalog_service.get_product(
db=db, vendor_id=vendor.id, product_id=product_id
)
# Check if product is active
if not product.is_active:
from app.exceptions import ProductNotActiveException
raise ProductNotActiveException(str(product_id))
return ProductDetailResponse.model_validate(product)
@router.get("/products/search", response_model=ProductListResponse) # public
def search_products(
request: Request,
q: str = Query(..., min_length=1, description="Search query"),
skip: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=100),
vendor: Vendor = Depends(require_vendor_context()),
db: Session = Depends(get_db),
):
"""
Search products in current vendor's catalog.
Searches in product names, descriptions, SKUs, brands, and GTINs.
Vendor is automatically determined from request context (URL/subdomain/domain).
No authentication required.
Query Parameters:
- q: Search query string (minimum 1 character)
- skip: Number of results to skip (pagination)
- limit: Maximum number of results to return
"""
# Get preferred language from request (via middleware or default)
language = getattr(request.state, "language", "en")
logger.debug(
f"[CATALOG_STOREFRONT] search_products: '{q}'",
extra={
"vendor_id": vendor.id,
"vendor_code": vendor.subdomain,
"query": q,
"skip": skip,
"limit": limit,
"language": language,
},
)
# Search products using the service
products, total = catalog_service.search_products(
db=db,
vendor_id=vendor.id,
query=q,
skip=skip,
limit=limit,
language=language,
)
return ProductListResponse(
products=[ProductResponse.model_validate(p) for p in products],
total=total,
skip=skip,
limit=limit,
)