Samir Boulahtit
1b8a40f1ff
All checks were successful
- Add centralized _is_noqa_suppressed() to BaseValidator with normalization (accepts both SEC001 and SEC-001 formats for ruff compatibility) - Wire noqa support into all 21 security and 18 performance check functions - Add ruff external config for SEC/PERF/MOD/EXC codes in pyproject.toml - Convert all 280 Python noqa comments to dashless format (ruff-compatible) - Add site/ to IGNORE_PATTERNS (excludes mkdocs build output) - Suppress 152 false positive findings (test passwords, seed data, validator self-references, Apple Wallet SHA1, etc.) - Security: 79 errors → 0, 60 warnings → 0 - Performance: 80 warnings → 77 (3 test script suppressions) - Add proposal doc with noqa inventory and remaining findings recommendations Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
366 lines
12 KiB
Python
366 lines
12 KiB
Python
# app/modules/loyalty/services/google_wallet_service.py
|
|
"""
|
|
Google Wallet service.
|
|
|
|
Handles Google Wallet integration including:
|
|
- Creating LoyaltyClass for programs
|
|
- Creating LoyaltyObject for cards
|
|
- Updating objects on balance changes
|
|
- Generating "Add to Wallet" URLs
|
|
"""
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.modules.loyalty.config import config
|
|
from app.modules.loyalty.exceptions import (
|
|
GoogleWalletNotConfiguredException,
|
|
WalletIntegrationException,
|
|
)
|
|
from app.modules.loyalty.models import LoyaltyCard, LoyaltyProgram
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class GoogleWalletService:
|
|
"""Service for Google Wallet integration."""
|
|
|
|
def __init__(self):
|
|
"""Initialize the Google Wallet service."""
|
|
self._credentials = None
|
|
self._http_client = None
|
|
|
|
@property
|
|
def is_configured(self) -> bool:
|
|
"""Check if Google Wallet is configured."""
|
|
return bool(config.google_issuer_id and config.google_service_account_json)
|
|
|
|
def _get_credentials(self):
|
|
"""Get Google service account credentials."""
|
|
if self._credentials:
|
|
return self._credentials
|
|
|
|
if not config.google_service_account_json:
|
|
raise GoogleWalletNotConfiguredException()
|
|
|
|
try:
|
|
from google.oauth2 import service_account
|
|
|
|
scopes = ["https://www.googleapis.com/auth/wallet_object.issuer"]
|
|
|
|
self._credentials = service_account.Credentials.from_service_account_file(
|
|
config.google_service_account_json,
|
|
scopes=scopes,
|
|
)
|
|
return self._credentials
|
|
except (ValueError, OSError) as e:
|
|
logger.error(f"Failed to load Google credentials: {e}")
|
|
raise WalletIntegrationException("google", str(e))
|
|
|
|
def _get_http_client(self):
|
|
"""Get authenticated HTTP client."""
|
|
if self._http_client:
|
|
return self._http_client
|
|
|
|
try:
|
|
from google.auth.transport.requests import AuthorizedSession
|
|
|
|
credentials = self._get_credentials()
|
|
self._http_client = AuthorizedSession(credentials)
|
|
return self._http_client
|
|
except Exception as e: # noqa: EXC003
|
|
logger.error(f"Failed to create Google HTTP client: {e}")
|
|
raise WalletIntegrationException("google", str(e))
|
|
|
|
# =========================================================================
|
|
# LoyaltyClass Operations (Program-level)
|
|
# =========================================================================
|
|
|
|
def create_class(self, db: Session, program: LoyaltyProgram) -> str:
|
|
"""
|
|
Create a LoyaltyClass for a loyalty program.
|
|
|
|
Args:
|
|
db: Database session
|
|
program: Loyalty program
|
|
|
|
Returns:
|
|
Google Wallet class ID
|
|
"""
|
|
if not self.is_configured:
|
|
raise GoogleWalletNotConfiguredException()
|
|
|
|
issuer_id = config.google_issuer_id
|
|
class_id = f"{issuer_id}.loyalty_program_{program.id}"
|
|
|
|
class_data = {
|
|
"id": class_id,
|
|
"issuerId": issuer_id,
|
|
"reviewStatus": "UNDER_REVIEW",
|
|
"programName": program.display_name,
|
|
"programLogo": {
|
|
"sourceUri": {
|
|
"uri": program.logo_url or "https://via.placeholder.com/100",
|
|
},
|
|
},
|
|
"hexBackgroundColor": program.card_color,
|
|
"localizedProgramName": {
|
|
"defaultValue": {
|
|
"language": "en",
|
|
"value": program.display_name,
|
|
},
|
|
},
|
|
}
|
|
|
|
# Add hero image if configured
|
|
if program.hero_image_url:
|
|
class_data["heroImage"] = {
|
|
"sourceUri": {"uri": program.hero_image_url},
|
|
}
|
|
|
|
try:
|
|
http = self._get_http_client()
|
|
response = http.post(
|
|
"https://walletobjects.googleapis.com/walletobjects/v1/loyaltyClass",
|
|
json=class_data,
|
|
)
|
|
|
|
if response.status_code in (200, 201):
|
|
# Update program with class ID
|
|
program.google_class_id = class_id
|
|
db.commit()
|
|
|
|
logger.info(f"Created Google Wallet class {class_id} for program {program.id}")
|
|
return class_id
|
|
if response.status_code == 409:
|
|
# Class already exists
|
|
program.google_class_id = class_id
|
|
db.commit()
|
|
return class_id
|
|
error = response.json() if response.text else {}
|
|
raise WalletIntegrationException(
|
|
"google",
|
|
f"Failed to create class: {response.status_code} - {error}",
|
|
)
|
|
except WalletIntegrationException:
|
|
raise
|
|
except Exception as e: # noqa: EXC003
|
|
logger.error(f"Failed to create Google Wallet class: {e}")
|
|
raise WalletIntegrationException("google", str(e))
|
|
|
|
def update_class(self, db: Session, program: LoyaltyProgram) -> None:
|
|
"""Update a LoyaltyClass when program settings change."""
|
|
if not program.google_class_id:
|
|
return
|
|
|
|
class_data = {
|
|
"programName": program.display_name,
|
|
"hexBackgroundColor": program.card_color,
|
|
}
|
|
|
|
if program.logo_url:
|
|
class_data["programLogo"] = {
|
|
"sourceUri": {"uri": program.logo_url},
|
|
}
|
|
|
|
try:
|
|
http = self._get_http_client()
|
|
response = http.patch(
|
|
f"https://walletobjects.googleapis.com/walletobjects/v1/loyaltyClass/{program.google_class_id}",
|
|
json=class_data,
|
|
)
|
|
|
|
if response.status_code not in (200, 201):
|
|
logger.warning(
|
|
f"Failed to update Google Wallet class {program.google_class_id}: "
|
|
f"{response.status_code}"
|
|
)
|
|
except Exception as e: # noqa: EXC003
|
|
logger.error(f"Failed to update Google Wallet class: {e}")
|
|
|
|
# =========================================================================
|
|
# LoyaltyObject Operations (Card-level)
|
|
# =========================================================================
|
|
|
|
def create_object(self, db: Session, card: LoyaltyCard) -> str:
|
|
"""
|
|
Create a LoyaltyObject for a loyalty card.
|
|
|
|
Args:
|
|
db: Database session
|
|
card: Loyalty card
|
|
|
|
Returns:
|
|
Google Wallet object ID
|
|
"""
|
|
if not self.is_configured:
|
|
raise GoogleWalletNotConfiguredException()
|
|
|
|
program = card.program
|
|
if not program.google_class_id:
|
|
# Create class first
|
|
self.create_class(db, program)
|
|
|
|
issuer_id = config.google_issuer_id
|
|
object_id = f"{issuer_id}.loyalty_card_{card.id}"
|
|
|
|
object_data = self._build_object_data(card, object_id)
|
|
|
|
try:
|
|
http = self._get_http_client()
|
|
response = http.post(
|
|
"https://walletobjects.googleapis.com/walletobjects/v1/loyaltyObject",
|
|
json=object_data,
|
|
)
|
|
|
|
if response.status_code in (200, 201):
|
|
card.google_object_id = object_id
|
|
db.commit()
|
|
|
|
logger.info(f"Created Google Wallet object {object_id} for card {card.id}")
|
|
return object_id
|
|
if response.status_code == 409:
|
|
# Object already exists
|
|
card.google_object_id = object_id
|
|
db.commit()
|
|
return object_id
|
|
error = response.json() if response.text else {}
|
|
raise WalletIntegrationException(
|
|
"google",
|
|
f"Failed to create object: {response.status_code} - {error}",
|
|
)
|
|
except WalletIntegrationException:
|
|
raise
|
|
except Exception as e: # noqa: EXC003
|
|
logger.error(f"Failed to create Google Wallet object: {e}")
|
|
raise WalletIntegrationException("google", str(e))
|
|
|
|
def update_object(self, db: Session, card: LoyaltyCard) -> None:
|
|
"""Update a LoyaltyObject when card balance changes."""
|
|
if not card.google_object_id:
|
|
return
|
|
|
|
object_data = self._build_object_data(card, card.google_object_id)
|
|
|
|
try:
|
|
http = self._get_http_client()
|
|
response = http.patch(
|
|
f"https://walletobjects.googleapis.com/walletobjects/v1/loyaltyObject/{card.google_object_id}",
|
|
json=object_data,
|
|
)
|
|
|
|
if response.status_code in (200, 201):
|
|
logger.debug(f"Updated Google Wallet object for card {card.id}")
|
|
else:
|
|
logger.warning(
|
|
f"Failed to update Google Wallet object {card.google_object_id}: "
|
|
f"{response.status_code}"
|
|
)
|
|
except Exception as e: # noqa: EXC003
|
|
logger.error(f"Failed to update Google Wallet object: {e}")
|
|
|
|
def _build_object_data(self, card: LoyaltyCard, object_id: str) -> dict[str, Any]:
|
|
"""Build the LoyaltyObject data structure."""
|
|
program = card.program
|
|
|
|
object_data = {
|
|
"id": object_id,
|
|
"classId": program.google_class_id,
|
|
"state": "ACTIVE" if card.is_active else "INACTIVE",
|
|
"accountId": card.card_number,
|
|
"accountName": card.card_number,
|
|
"barcode": {
|
|
"type": "CODE_128",
|
|
"value": card.card_number.replace("-", ""),
|
|
"alternateText": card.card_number,
|
|
},
|
|
}
|
|
|
|
# Add loyalty points (stamps as points for display)
|
|
if program.is_stamps_enabled:
|
|
object_data["loyaltyPoints"] = {
|
|
"label": "Stamps",
|
|
"balance": {
|
|
"int": card.stamp_count,
|
|
},
|
|
}
|
|
# Add secondary points showing target
|
|
object_data["secondaryLoyaltyPoints"] = {
|
|
"label": f"of {program.stamps_target}",
|
|
"balance": {
|
|
"int": program.stamps_target,
|
|
},
|
|
}
|
|
elif program.is_points_enabled:
|
|
object_data["loyaltyPoints"] = {
|
|
"label": "Points",
|
|
"balance": {
|
|
"int": card.points_balance,
|
|
},
|
|
}
|
|
|
|
return object_data
|
|
|
|
# =========================================================================
|
|
# Save URL Generation
|
|
# =========================================================================
|
|
|
|
def get_save_url(self, db: Session, card: LoyaltyCard) -> str:
|
|
"""
|
|
Get the "Add to Google Wallet" URL for a card.
|
|
|
|
Args:
|
|
db: Database session
|
|
card: Loyalty card
|
|
|
|
Returns:
|
|
URL for adding pass to Google Wallet
|
|
"""
|
|
if not self.is_configured:
|
|
raise GoogleWalletNotConfiguredException()
|
|
|
|
# Ensure object exists
|
|
if not card.google_object_id:
|
|
self.create_object(db, card)
|
|
|
|
# Generate JWT for save link
|
|
try:
|
|
from datetime import datetime, timedelta
|
|
|
|
import jwt
|
|
|
|
credentials = self._get_credentials()
|
|
|
|
claims = {
|
|
"iss": credentials.service_account_email,
|
|
"aud": "google",
|
|
"origins": [],
|
|
"typ": "savetowallet",
|
|
"payload": {
|
|
"loyaltyObjects": [{"id": card.google_object_id}],
|
|
},
|
|
"iat": datetime.utcnow(),
|
|
"exp": datetime.utcnow() + timedelta(hours=1),
|
|
}
|
|
|
|
# Sign with service account private key
|
|
token = jwt.encode(
|
|
claims,
|
|
credentials._signer._key,
|
|
algorithm="RS256",
|
|
)
|
|
|
|
card.google_object_jwt = token
|
|
db.commit()
|
|
|
|
return f"https://pay.google.com/gp/v/save/{token}"
|
|
except Exception as e: # noqa: EXC003
|
|
logger.error(f"Failed to generate Google Wallet save URL: {e}")
|
|
raise WalletIntegrationException("google", str(e))
|
|
|
|
|
|
# Singleton instance
|
|
google_wallet_service = GoogleWalletService()
|