feat(loyalty): cross-persona page alignment with shared components

Align loyalty pages across admin, merchant, and store personas so each
sees the same page set scoped to their access level. Admin acts as a
superset of merchant with "on behalf" capabilities.

New pages:
- Store: Staff PINs management (CRUD)
- Merchant: Cards, Card Detail, Transactions, Staff PINs (CRUD), Settings (read-only)
- Admin: Merchant Cards, Card Detail, Transactions, PINs (read-only)

Architecture:
- 4 shared Jinja2 partials (cards-list, card-detail, transactions, pins)
- 4 shared JS factory modules parameterized by apiPrefix/scope
- Persona templates are thin wrappers including shared partials
- PinDetailResponse schema for cross-store PIN listings

API: 17 new endpoints (11 merchant, 6 admin on-behalf)
Tests: 38 new integration tests, arch-check green
i18n: ~130 new keys across en/fr/de/lb
Docs: pages-and-navigation.md with full page matrix

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-03-22 19:28:07 +01:00
parent f41f72b86f
commit 6161d69ba2
49 changed files with 4385 additions and 14 deletions

View File

@@ -21,16 +21,23 @@ from app.api.deps import (
from app.core.database import get_db
from app.modules.enums import FrontendType
from app.modules.loyalty.schemas import (
CardDetailResponse,
CardListResponse,
CardResponse,
MerchantSettingsResponse,
MerchantSettingsUpdate,
MerchantStatsResponse,
PinDetailListResponse,
PinDetailResponse,
ProgramCreate,
ProgramListResponse,
ProgramResponse,
ProgramStatsResponse,
ProgramUpdate,
TransactionListResponse,
TransactionResponse,
)
from app.modules.loyalty.services import program_service
from app.modules.loyalty.services import card_service, pin_service, program_service
from app.modules.tenancy.models import User # API-007
logger = logging.getLogger(__name__)
@@ -260,6 +267,202 @@ def update_merchant_settings(
return MerchantSettingsResponse.model_validate(settings)
# =============================================================================
# On-Behalf Card & Transaction Endpoints
# =============================================================================
@router.get("/merchants/{merchant_id}/cards", response_model=CardListResponse)
def list_merchant_cards(
merchant_id: int = Path(..., gt=0),
skip: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=100),
is_active: bool | None = Query(None),
search: str | None = Query(None, max_length=100),
store_id: int | None = Query(None, description="Filter by enrolled store"),
current_user: User = Depends(get_current_admin_api),
db: Session = Depends(get_db),
):
"""List loyalty cards for a merchant (admin on-behalf)."""
cards, total = card_service.list_cards(
db,
merchant_id,
store_id=store_id,
skip=skip,
limit=limit,
is_active=is_active,
search=search,
)
card_responses = []
for card in cards:
program = card.program
response = CardResponse(
id=card.id,
card_number=card.card_number,
customer_id=card.customer_id,
merchant_id=card.merchant_id,
program_id=card.program_id,
enrolled_at_store_id=card.enrolled_at_store_id,
customer_name=card.customer.full_name if card.customer else None,
customer_email=card.customer.email if card.customer else None,
stamp_count=card.stamp_count,
stamps_target=program.stamps_target if program else 0,
stamps_until_reward=max(0, (program.stamps_target if program else 0) - card.stamp_count),
total_stamps_earned=card.total_stamps_earned,
stamps_redeemed=card.stamps_redeemed,
points_balance=card.points_balance,
total_points_earned=card.total_points_earned,
points_redeemed=card.points_redeemed,
is_active=card.is_active,
created_at=card.created_at,
has_google_wallet=bool(card.google_object_id),
has_apple_wallet=bool(card.apple_serial_number),
)
card_responses.append(response)
return CardListResponse(cards=card_responses, total=total)
@router.get("/merchants/{merchant_id}/cards/{card_id}", response_model=CardDetailResponse)
def get_merchant_card(
merchant_id: int = Path(..., gt=0),
card_id: int = Path(..., gt=0),
current_user: User = Depends(get_current_admin_api),
db: Session = Depends(get_db),
):
"""Get detailed loyalty card info (admin on-behalf)."""
card = card_service.get_card(db, card_id)
if not card or card.merchant_id != merchant_id:
from app.modules.loyalty.exceptions import LoyaltyCardNotFoundException
raise LoyaltyCardNotFoundException(str(card_id))
program = card.program
customer = card.customer
return CardDetailResponse(
id=card.id,
card_number=card.card_number,
customer_id=card.customer_id,
merchant_id=card.merchant_id,
program_id=card.program_id,
enrolled_at_store_id=card.enrolled_at_store_id,
customer_name=customer.full_name if customer else None,
customer_email=customer.email if customer else None,
merchant_name=card.merchant.name if card.merchant else None,
qr_code_data=card.qr_code_data or card.card_number,
program_name=program.display_name,
program_type=program.loyalty_type,
reward_description=program.stamps_reward_description,
stamp_count=card.stamp_count,
stamps_target=program.stamps_target,
stamps_until_reward=max(0, program.stamps_target - card.stamp_count),
total_stamps_earned=card.total_stamps_earned,
stamps_redeemed=card.stamps_redeemed,
points_balance=card.points_balance,
total_points_earned=card.total_points_earned,
points_redeemed=card.points_redeemed,
is_active=card.is_active,
created_at=card.created_at,
last_stamp_at=card.last_stamp_at,
last_points_at=card.last_points_at,
last_redemption_at=card.last_redemption_at,
last_activity_at=card.last_activity_at,
has_google_wallet=bool(card.google_object_id),
has_apple_wallet=bool(card.apple_serial_number),
)
@router.get(
"/merchants/{merchant_id}/cards/{card_id}/transactions",
response_model=TransactionListResponse,
)
def get_merchant_card_transactions(
merchant_id: int = Path(..., gt=0),
card_id: int = Path(..., gt=0),
skip: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=100),
current_user: User = Depends(get_current_admin_api),
db: Session = Depends(get_db),
):
"""Get transaction history for a card (admin on-behalf)."""
card = card_service.get_card(db, card_id)
if not card or card.merchant_id != merchant_id:
from app.modules.loyalty.exceptions import LoyaltyCardNotFoundException
raise LoyaltyCardNotFoundException(str(card_id))
transactions, total = card_service.get_card_transactions(
db, card_id, skip=skip, limit=limit
)
return TransactionListResponse(
transactions=[TransactionResponse.model_validate(t) for t in transactions],
total=total,
)
@router.get("/merchants/{merchant_id}/transactions", response_model=TransactionListResponse)
def list_merchant_transactions(
merchant_id: int = Path(..., gt=0),
skip: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=100),
store_id: int | None = Query(None, description="Filter by store location"),
current_user: User = Depends(get_current_admin_api),
db: Session = Depends(get_db),
):
"""List recent transactions for a merchant (admin on-behalf)."""
transactions, total = card_service.get_store_transactions(
db, merchant_id, store_id=store_id, skip=skip, limit=limit
)
tx_responses = []
for t in transactions:
tx = TransactionResponse.model_validate(t)
if t.card and t.card.customer:
tx.customer_name = t.card.customer.full_name
tx_responses.append(tx)
return TransactionListResponse(transactions=tx_responses, total=total)
@router.get("/merchants/{merchant_id}/pins", response_model=PinDetailListResponse)
def list_merchant_pins(
merchant_id: int = Path(..., gt=0),
store_id: int | None = Query(None, description="Filter by store location"),
is_active: bool | None = Query(None),
current_user: User = Depends(get_current_admin_api),
db: Session = Depends(get_db),
):
"""List staff PINs for a merchant (admin read-only)."""
pins = pin_service.list_pins_for_merchant(
db, merchant_id, store_id=store_id, is_active=is_active
)
pin_responses = [PinDetailResponse.model_validate(pin) for pin in pins]
return PinDetailListResponse(pins=pin_responses, total=len(pin_responses))
@router.get("/merchants/{merchant_id}/locations")
def list_merchant_locations(
merchant_id: int = Path(..., gt=0),
current_user: User = Depends(get_current_admin_api),
db: Session = Depends(get_db),
):
"""Get active store locations for a merchant."""
locations = program_service.get_merchant_locations(db, merchant_id)
return [
{
"id": store.id,
"name": store.name,
"code": store.store_code,
}
for store in locations
]
# =============================================================================
# Platform Stats
# =============================================================================

View File

@@ -7,6 +7,11 @@ Merchant portal endpoints for full program CRUD:
- Create a loyalty program
- Update the loyalty program
- Delete the loyalty program
- List/view loyalty cards
- Transaction history
- Staff PIN management
- Merchant settings (read-only)
- Store locations
Authentication: Authorization header (API-only, no cookies for CSRF safety).
The user must own at least one active merchant (validated by
@@ -18,18 +23,33 @@ registration under /api/v1/merchants/loyalty/*).
import logging
from fastapi import APIRouter, Depends
from fastapi import APIRouter, Depends, Path, Query
from sqlalchemy.orm import Session
from app.api.deps import get_merchant_for_current_user
from app.core.database import get_db
from app.modules.loyalty.exceptions import (
LoyaltyCardNotFoundException,
StaffPinNotFoundException,
)
from app.modules.loyalty.schemas import (
CardDetailResponse,
CardListResponse,
CardResponse,
MerchantSettingsResponse,
PinCreate,
PinDetailListResponse,
PinDetailResponse,
PinResponse,
PinUpdate,
ProgramCreate,
ProgramResponse,
ProgramUpdate,
TransactionListResponse,
TransactionResponse,
)
from app.modules.loyalty.schemas.program import MerchantStatsResponse
from app.modules.loyalty.services import program_service
from app.modules.loyalty.services import card_service, pin_service, program_service
from app.modules.tenancy.models import Merchant
logger = logging.getLogger(__name__)
@@ -112,3 +132,284 @@ def delete_program(
program = program_service.require_program_by_merchant(db, merchant.id)
program_service.delete_program(db, program.id)
logger.info(f"Merchant {merchant.id} ({merchant.name}) deleted loyalty program")
# =============================================================================
# Cards
# =============================================================================
@router.get("/cards", response_model=CardListResponse)
def list_cards(
skip: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=100),
is_active: bool | None = Query(None),
search: str | None = Query(None, max_length=100),
store_id: int | None = Query(None),
merchant: Merchant = Depends(get_merchant_for_current_user),
db: Session = Depends(get_db),
):
"""List loyalty cards for the merchant."""
program = program_service.require_program_by_merchant(db, merchant.id)
cards, total = card_service.list_cards(
db,
merchant.id,
store_id=store_id,
skip=skip,
limit=limit,
is_active=is_active,
search=search,
)
card_responses = []
for card in cards:
customer = card.customer
response = CardResponse(
id=card.id,
card_number=card.card_number,
customer_id=card.customer_id,
merchant_id=card.merchant_id,
program_id=card.program_id,
enrolled_at_store_id=card.enrolled_at_store_id,
customer_name=customer.full_name if customer else None,
customer_email=customer.email if customer else None,
stamp_count=card.stamp_count,
stamps_target=program.stamps_target,
stamps_until_reward=max(0, program.stamps_target - card.stamp_count),
total_stamps_earned=card.total_stamps_earned,
stamps_redeemed=card.stamps_redeemed,
points_balance=card.points_balance,
total_points_earned=card.total_points_earned,
points_redeemed=card.points_redeemed,
is_active=card.is_active,
created_at=card.created_at,
has_google_wallet=bool(card.google_object_id),
has_apple_wallet=bool(card.apple_serial_number),
)
card_responses.append(response)
return CardListResponse(cards=card_responses, total=total)
@router.get("/cards/{card_id}", response_model=CardDetailResponse)
def get_card_detail(
card_id: int = Path(..., gt=0),
merchant: Merchant = Depends(get_merchant_for_current_user),
db: Session = Depends(get_db),
):
"""Get detailed loyalty card info by ID."""
card = card_service.get_card(db, card_id)
if not card or card.merchant_id != merchant.id:
raise LoyaltyCardNotFoundException(str(card_id))
program = card.program
customer = card.customer
return CardDetailResponse(
id=card.id,
card_number=card.card_number,
customer_id=card.customer_id,
merchant_id=card.merchant_id,
program_id=card.program_id,
enrolled_at_store_id=card.enrolled_at_store_id,
customer_name=customer.full_name if customer else None,
customer_email=customer.email if customer else None,
merchant_name=card.merchant.name if card.merchant else None,
qr_code_data=card.qr_code_data or card.card_number,
program_name=program.display_name,
program_type=program.loyalty_type,
reward_description=program.stamps_reward_description,
stamp_count=card.stamp_count,
stamps_target=program.stamps_target,
stamps_until_reward=max(0, program.stamps_target - card.stamp_count),
total_stamps_earned=card.total_stamps_earned,
stamps_redeemed=card.stamps_redeemed,
points_balance=card.points_balance,
total_points_earned=card.total_points_earned,
points_redeemed=card.points_redeemed,
is_active=card.is_active,
created_at=card.created_at,
last_stamp_at=card.last_stamp_at,
last_points_at=card.last_points_at,
last_redemption_at=card.last_redemption_at,
last_activity_at=card.last_activity_at,
has_google_wallet=bool(card.google_object_id),
has_apple_wallet=bool(card.apple_serial_number),
)
@router.get("/cards/{card_id}/transactions", response_model=TransactionListResponse)
def get_card_transactions(
card_id: int = Path(..., gt=0),
skip: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=100),
merchant: Merchant = Depends(get_merchant_for_current_user),
db: Session = Depends(get_db),
):
"""Get transaction history for a card."""
card = card_service.get_card(db, card_id)
if not card or card.merchant_id != merchant.id:
raise LoyaltyCardNotFoundException(str(card_id))
transactions, total = card_service.get_card_transactions(
db, card_id, skip=skip, limit=limit
)
return TransactionListResponse(
transactions=[TransactionResponse.model_validate(t) for t in transactions],
total=total,
)
# =============================================================================
# Transactions
# =============================================================================
@router.get("/transactions", response_model=TransactionListResponse)
def list_transactions(
skip: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=100),
store_id: int | None = Query(None),
merchant: Merchant = Depends(get_merchant_for_current_user),
db: Session = Depends(get_db),
):
"""Merchant-wide transaction feed."""
transactions, total = card_service.get_store_transactions(
db, merchant.id, store_id=store_id, skip=skip, limit=limit
)
tx_responses = []
for t in transactions:
tx = TransactionResponse.model_validate(t)
if t.card and t.card.customer:
tx.customer_name = t.card.customer.full_name
tx_responses.append(tx)
return TransactionListResponse(transactions=tx_responses, total=total)
# =============================================================================
# Staff PINs
# =============================================================================
def _validate_pin_ownership(db: Session, pin_id: int, merchant_id: int):
"""Validate that a PIN belongs to the merchant. Returns the pin."""
pin = pin_service.require_pin(db, pin_id)
if pin.merchant_id != merchant_id:
raise StaffPinNotFoundException(str(pin_id))
return pin
@router.get("/pins", response_model=PinDetailListResponse)
def list_pins(
store_id: int | None = Query(None),
is_active: bool | None = Query(None),
merchant: Merchant = Depends(get_merchant_for_current_user),
db: Session = Depends(get_db),
):
"""List PINs across all stores for the merchant."""
pins = pin_service.list_pins_for_merchant(
db, merchant.id, store_id=store_id, is_active=is_active
)
pin_responses = []
for pin in pins:
resp = PinDetailResponse.model_validate(pin)
resp.store_id = pin.store_id
resp.store_name = pin.store.name if pin.store else None
pin_responses.append(resp)
return PinDetailListResponse(pins=pin_responses, total=len(pin_responses))
@router.post("/pins", response_model=PinResponse, status_code=201)
def create_pin(
data: PinCreate,
store_id: int = Query(..., gt=0),
merchant: Merchant = Depends(get_merchant_for_current_user),
db: Session = Depends(get_db),
):
"""Create a new staff PIN."""
# Validate store belongs to merchant
locations = program_service.get_merchant_locations(db, merchant.id)
store_ids = [loc.id for loc in locations]
if store_id not in store_ids:
from app.modules.tenancy.exceptions import StoreNotFoundException
raise StoreNotFoundException(str(store_id), identifier_type="id")
program = program_service.require_program_by_merchant(db, merchant.id)
pin = pin_service.create_pin(db, program.id, store_id, data)
return PinResponse.model_validate(pin)
@router.patch("/pins/{pin_id}", response_model=PinResponse)
def update_pin(
data: PinUpdate,
pin_id: int = Path(..., gt=0),
merchant: Merchant = Depends(get_merchant_for_current_user),
db: Session = Depends(get_db),
):
"""Update a staff PIN."""
_validate_pin_ownership(db, pin_id, merchant.id)
pin = pin_service.update_pin(db, pin_id, data)
return PinResponse.model_validate(pin)
@router.delete("/pins/{pin_id}", status_code=204)
def delete_pin(
pin_id: int = Path(..., gt=0),
merchant: Merchant = Depends(get_merchant_for_current_user),
db: Session = Depends(get_db),
):
"""Delete a staff PIN."""
_validate_pin_ownership(db, pin_id, merchant.id)
pin_service.delete_pin(db, pin_id)
@router.post("/pins/{pin_id}/unlock", response_model=PinResponse)
def unlock_pin(
pin_id: int = Path(..., gt=0),
merchant: Merchant = Depends(get_merchant_for_current_user),
db: Session = Depends(get_db),
):
"""Unlock a locked staff PIN."""
_validate_pin_ownership(db, pin_id, merchant.id)
pin = pin_service.unlock_pin(db, pin_id)
return PinResponse.model_validate(pin)
# =============================================================================
# Settings
# =============================================================================
@router.get("/settings", response_model=MerchantSettingsResponse)
def get_settings(
merchant: Merchant = Depends(get_merchant_for_current_user),
db: Session = Depends(get_db),
):
"""Get read-only merchant loyalty settings."""
settings = program_service.get_or_create_merchant_settings(db, merchant.id)
return MerchantSettingsResponse.model_validate(settings)
# =============================================================================
# Locations
# =============================================================================
@router.get("/locations")
def list_locations(
merchant: Merchant = Depends(get_merchant_for_current_user),
db: Session = Depends(get_db),
):
"""List merchant stores (for filter dropdowns)."""
locations = program_service.get_merchant_locations(db, merchant.id)
return [
{"id": loc.id, "name": loc.name, "code": loc.store_code}
for loc in locations
]