feat(loyalty): transaction categories (what was sold)
Some checks failed
CI / ruff (push) Successful in 20s
CI / validate (push) Has been cancelled
CI / dependency-scanning (push) Has been cancelled
CI / docs (push) Has been cancelled
CI / deploy (push) Has been cancelled
CI / pytest (push) Has been cancelled

Merchants can configure per-store product categories (e.g., Men,
Women, Accessories, Kids) that sellers select when entering loyalty
transactions. Enables per-category sales analytics.

Backend:
- New model: StoreTransactionCategory (store-scoped, max 10 per store)
- Migration loyalty_007: creates table + adds category_id FK on
  loyalty_transactions
- New category_service.py with CRUD + validation
- New schemas/category.py (Create, Update, Response, ListResponse)
- Admin CRUD: GET/POST/PATCH/DELETE /admin/loyalty/stores/{id}/categories
- Store CRUD: GET/POST/PATCH/DELETE /store/loyalty/categories
- Stamp/Points request schemas accept optional category_id
- Stamp/Points services pass category_id to transaction creation
- TransactionResponse includes category_id + category_name

342 tests pass.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-04-19 12:23:17 +02:00
parent cd4f83f2cb
commit 1cf9fea40a
14 changed files with 508 additions and 0 deletions

View File

@@ -497,6 +497,79 @@ def get_platform_stats(
return program_service.get_platform_stats(db)
# =============================================================================
# Transaction Categories (admin manages on behalf of stores)
# =============================================================================
@router.get("/stores/{store_id}/categories")
def list_store_categories(
store_id: int = Path(..., gt=0),
current_user: User = Depends(get_current_admin_api),
db: Session = Depends(get_db),
):
"""List transaction categories for a store."""
from app.modules.loyalty.schemas.category import (
CategoryListResponse,
CategoryResponse,
)
from app.modules.loyalty.services.category_service import category_service
categories = category_service.list_categories(db, store_id)
return CategoryListResponse(
categories=[CategoryResponse.model_validate(c) for c in categories],
total=len(categories),
)
@router.post("/stores/{store_id}/categories", status_code=201)
def create_store_category(
data: dict,
store_id: int = Path(..., gt=0),
current_user: User = Depends(get_current_admin_api),
db: Session = Depends(get_db),
):
"""Create a transaction category for a store."""
from app.modules.loyalty.schemas.category import CategoryCreate, CategoryResponse
from app.modules.loyalty.services.category_service import category_service
category = category_service.create_category(
db, store_id, CategoryCreate(**data)
)
return CategoryResponse.model_validate(category)
@router.patch("/stores/{store_id}/categories/{category_id}")
def update_store_category(
data: dict,
store_id: int = Path(..., gt=0),
category_id: int = Path(..., gt=0),
current_user: User = Depends(get_current_admin_api),
db: Session = Depends(get_db),
):
"""Update a transaction category for a store."""
from app.modules.loyalty.schemas.category import CategoryResponse, CategoryUpdate
from app.modules.loyalty.services.category_service import category_service
category = category_service.update_category(
db, category_id, store_id, CategoryUpdate(**data)
)
return CategoryResponse.model_validate(category)
@router.delete("/stores/{store_id}/categories/{category_id}", status_code=204)
def delete_store_category(
store_id: int = Path(..., gt=0),
category_id: int = Path(..., gt=0),
current_user: User = Depends(get_current_admin_api),
db: Session = Depends(get_db),
):
"""Delete a transaction category for a store."""
from app.modules.loyalty.services.category_service import category_service
category_service.delete_category(db, category_id, store_id)
# =============================================================================
# Advanced Analytics
# =============================================================================

View File

@@ -244,6 +244,84 @@ def get_revenue_attribution(
)
# =============================================================================
# Transaction Categories
# =============================================================================
@router.get("/categories")
def list_categories(
current_user: User = Depends(get_current_store_api),
db: Session = Depends(get_db),
):
"""List transaction categories for this store."""
from app.modules.loyalty.schemas.category import (
CategoryListResponse,
CategoryResponse,
)
from app.modules.loyalty.services.category_service import category_service
categories = category_service.list_categories(db, current_user.token_store_id)
return CategoryListResponse(
categories=[CategoryResponse.model_validate(c) for c in categories],
total=len(categories),
)
@router.post("/categories", status_code=201)
def create_category(
data: dict,
current_user: User = Depends(get_current_store_api),
db: Session = Depends(get_db),
):
"""Create a transaction category for this store (merchant_owner only)."""
if current_user.role != "merchant_owner":
raise AuthorizationException("Only merchant owners can manage categories")
from app.modules.loyalty.schemas.category import CategoryCreate, CategoryResponse
from app.modules.loyalty.services.category_service import category_service
category = category_service.create_category(
db, current_user.token_store_id, CategoryCreate(**data)
)
return CategoryResponse.model_validate(category)
@router.patch("/categories/{category_id}")
def update_category(
category_id: int,
data: dict,
current_user: User = Depends(get_current_store_api),
db: Session = Depends(get_db),
):
"""Update a transaction category (merchant_owner only)."""
if current_user.role != "merchant_owner":
raise AuthorizationException("Only merchant owners can manage categories")
from app.modules.loyalty.schemas.category import CategoryResponse, CategoryUpdate
from app.modules.loyalty.services.category_service import category_service
category = category_service.update_category(
db, category_id, current_user.token_store_id, CategoryUpdate(**data)
)
return CategoryResponse.model_validate(category)
@router.delete("/categories/{category_id}", status_code=204)
def delete_category(
category_id: int,
current_user: User = Depends(get_current_store_api),
db: Session = Depends(get_db),
):
"""Delete a transaction category (merchant_owner only)."""
if current_user.role != "merchant_owner":
raise AuthorizationException("Only merchant owners can manage categories")
from app.modules.loyalty.services.category_service import category_service
category_service.delete_category(db, category_id, current_user.token_store_id)
# =============================================================================
# Staff PINs
# =============================================================================
@@ -684,6 +762,7 @@ def add_stamp(
qr_code=data.qr_code,
card_number=data.card_number,
staff_pin=data.staff_pin,
category_id=data.category_id,
ip_address=ip,
user_agent=user_agent,
notes=data.notes,
@@ -774,6 +853,7 @@ def earn_points(
purchase_amount_cents=data.purchase_amount_cents,
order_reference=data.order_reference,
staff_pin=data.staff_pin,
category_id=data.category_id,
ip_address=ip,
user_agent=user_agent,
notes=data.notes,