# app/modules/loyalty/routes/api/merchant.py """ Loyalty module merchant routes. Merchant portal endpoints for full program CRUD: - Get merchant's loyalty program - 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 get_merchant_for_current_user). Auto-discovered by the route system (merchant.py in routes/api/ triggers registration under /api/v1/merchants/loyalty/*). """ import logging 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 card_service, pin_service, program_service from app.modules.tenancy.models import Merchant logger = logging.getLogger(__name__) ROUTE_CONFIG = { "prefix": "/loyalty", } router = APIRouter() def _build_program_response(program) -> ProgramResponse: """Build a ProgramResponse from a program ORM object.""" response = ProgramResponse.model_validate(program) response.is_stamps_enabled = program.is_stamps_enabled response.is_points_enabled = program.is_points_enabled response.display_name = program.display_name return response # ============================================================================= # Statistics # ============================================================================= @router.get("/stats", response_model=MerchantStatsResponse) def get_stats( merchant: Merchant = Depends(get_merchant_for_current_user), db: Session = Depends(get_db), ): """Get merchant-wide loyalty statistics across all locations.""" stats = program_service.get_merchant_stats(db, merchant.id) return MerchantStatsResponse(**stats) # ============================================================================= # Program CRUD # ============================================================================= @router.get("/program", response_model=ProgramResponse) def get_program( merchant: Merchant = Depends(get_merchant_for_current_user), db: Session = Depends(get_db), ): """Get the merchant's loyalty program.""" program = program_service.require_program_by_merchant(db, merchant.id) return _build_program_response(program) @router.post("/program", response_model=ProgramResponse, status_code=201) def create_program( data: ProgramCreate, merchant: Merchant = Depends(get_merchant_for_current_user), db: Session = Depends(get_db), ): """Create a loyalty program for the merchant.""" program = program_service.create_program(db, merchant.id, data) return _build_program_response(program) @router.patch("/program", response_model=ProgramResponse) def update_program( data: ProgramUpdate, merchant: Merchant = Depends(get_merchant_for_current_user), db: Session = Depends(get_db), ): """Update the merchant's loyalty program.""" program = program_service.require_program_by_merchant(db, merchant.id) program = program_service.update_program(db, program.id, data) return _build_program_response(program) @router.delete("/program", status_code=204) def delete_program( merchant: Merchant = Depends(get_merchant_for_current_user), db: Session = Depends(get_db), ): """Delete the merchant's loyalty 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 ]