# app/modules/loyalty/services/wallet_service.py """ Unified wallet service. Provides a unified interface for wallet operations across Google Wallet and Apple Wallet. """ import logging from sqlalchemy.orm import Session from app.modules.loyalty.models import LoyaltyCard logger = logging.getLogger(__name__) class WalletService: """Unified service for wallet operations.""" def get_add_to_wallet_urls( self, db: Session, card: LoyaltyCard, ) -> dict[str, str | None]: """ Get URLs for adding card to wallets. Args: db: Database session card: Loyalty card Returns: Dict with google_wallet_url and apple_wallet_url """ from app.modules.loyalty.services.apple_wallet_service import ( apple_wallet_service, ) from app.modules.loyalty.services.google_wallet_service import ( google_wallet_service, ) urls = { "google_wallet_url": None, "apple_wallet_url": None, } program = card.program # Google Wallet — platform-wide config via env vars if google_wallet_service.is_configured or program.google_class_id: try: urls["google_wallet_url"] = google_wallet_service.get_save_url(db, card) except Exception as e: # noqa: EXC003 logger.warning(f"Failed to get Google Wallet URL for card {card.id}: {e}") # Apple Wallet if program.apple_pass_type_id: try: urls["apple_wallet_url"] = apple_wallet_service.get_pass_url(card) except Exception as e: # noqa: EXC003 logger.warning(f"Failed to get Apple Wallet URL for card {card.id}: {e}") return urls def sync_card_to_wallets(self, db: Session, card: LoyaltyCard) -> dict[str, bool]: """ Sync card data to all configured wallets. Called after stamp/points operations to update wallet passes. Args: db: Database session card: Loyalty card to sync Returns: Dict with success status for each wallet """ from app.modules.loyalty.services.apple_wallet_service import ( apple_wallet_service, ) from app.modules.loyalty.services.google_wallet_service import ( google_wallet_service, ) results = { "google_wallet": False, "apple_wallet": False, } # Sync to Google Wallet if card.google_object_id: try: google_wallet_service.update_object(db, card) results["google_wallet"] = True except Exception as e: # noqa: EXC003 logger.error(f"Failed to sync card {card.id} to Google Wallet: {e}") # Sync to Apple Wallet (via push notification) if card.apple_serial_number: try: apple_wallet_service.send_push_updates(db, card) results["apple_wallet"] = True except Exception as e: # noqa: EXC003 logger.error(f"Failed to send Apple Wallet push for card {card.id}: {e}") return results def create_wallet_objects(self, db: Session, card: LoyaltyCard) -> dict[str, bool]: """ Create wallet objects for a new card. Called during enrollment to set up wallet passes. Args: db: Database session card: Newly created loyalty card Returns: Dict with success status for each wallet """ from app.modules.loyalty.services.google_wallet_service import ( google_wallet_service, ) results = { "google_wallet": False, "apple_wallet": False, } program = card.program # Create Google Wallet object — platform-wide config via env vars if google_wallet_service.is_configured: try: google_wallet_service.create_object(db, card) results["google_wallet"] = True except Exception as e: # noqa: EXC003 logger.error(f"Failed to create Google Wallet object for card {card.id}: {e}") # Apple Wallet objects are created on-demand when user downloads pass # No pre-creation needed, but we set up the serial number if program.apple_pass_type_id: card.apple_serial_number = f"card_{card.id}_{card.qr_code_data[:8]}" db.commit() results["apple_wallet"] = True return results # Singleton instance wallet_service = WalletService()