""" Square Payment Service Handles Square payment processing including payment creation, verification, and webhooks """ from square.client import Square, SquareEnvironment from typing import Dict, Optional import uuid from datetime import datetime from ..core.config import settings class SquareService: """Service for handling Square payment processing""" def __init__(self): """Initialize Square client""" # Configure Square client based on environment environment = ( SquareEnvironment.SANDBOX if settings.SQUARE_ENVIRONMENT.lower() == 'sandbox' else SquareEnvironment.PRODUCTION ) self.client = Square( token=settings.SQUARE_ACCESS_TOKEN, environment=environment ) self.location_id = settings.SQUARE_LOCATION_ID async def create_payment( self, amount_money: float, source_id: str, idempotency_key: Optional[str] = None, customer_id: Optional[str] = None, reference_id: Optional[str] = None, note: Optional[str] = None, billing_details: Optional[Dict] = None ) -> Dict: """ Create a payment using Square Args: amount_money: Amount in the currency's smallest denomination (e.g., cents for GBP) source_id: Payment source ID from the Web Payments SDK idempotency_key: Unique key to prevent duplicate payments customer_id: Optional Square customer ID reference_id: Optional reference ID for internal tracking note: Optional note about the payment billing_details: Optional billing address and cardholder name for AVS Returns: Dict with payment result including payment_id, status, and details """ try: # Generate idempotency key if not provided if not idempotency_key: idempotency_key = str(uuid.uuid4()) # Convert amount to money object (Square expects amount in smallest currency unit) # For GBP, this is pence amount_in_pence = int(amount_money * 100) # Build payment parameters payment_params = { 'source_id': source_id, 'idempotency_key': idempotency_key, 'amount_money': { 'amount': amount_in_pence, 'currency': 'GBP' }, 'location_id': self.location_id } # Add billing address for AVS if provided if billing_details: # Add buyer email if available if billing_details.get('email'): payment_params['buyer_email_address'] = billing_details.get('email') # Build billing address for AVS billing_address = {} if billing_details.get('address_line_1'): billing_address['address_line_1'] = billing_details.get('address_line_1') if billing_details.get('address_line_2'): billing_address['address_line_2'] = billing_details.get('address_line_2') if billing_details.get('city'): billing_address['locality'] = billing_details.get('city') if billing_details.get('postal_code'): billing_address['postal_code'] = billing_details.get('postal_code') if billing_details.get('country'): billing_address['country'] = billing_details.get('country') if billing_address: payment_params['billing_address'] = billing_address # Add optional parameters if customer_id: payment_params['customer_id'] = customer_id if reference_id: payment_params['reference_id'] = reference_id if note: payment_params['note'] = note # Create payment - pass parameters directly as keyword arguments result = self.client.payments.create(**payment_params) if result.errors: # Payment failed - extract user-friendly error messages error_messages = [] for error in result.errors: code = error.code if hasattr(error, 'code') else None detail = error.detail if hasattr(error, 'detail') else str(error) # Map Square error codes to user-friendly messages if code == 'GENERIC_DECLINE': error_messages.append('Your card was declined. Please try a different payment method.') elif code == 'CVV_FAILURE': error_messages.append('The CVV code is invalid. Please check and try again.') elif code == 'INVALID_CARD': error_messages.append('The card information is invalid. Please check your card details.') elif code == 'CARD_DECLINED': error_messages.append('Your card was declined. Please contact your bank or try a different card.') elif code == 'INSUFFICIENT_FUNDS': error_messages.append('Insufficient funds. Please try a different payment method.') elif code == 'INVALID_EXPIRATION': error_messages.append('The card expiration date is invalid.') elif code == 'ADDRESS_VERIFICATION_FAILURE': error_messages.append('Address verification failed. Please check your billing address.') else: # For other errors, use the detail message but clean it up if 'Authorization error' in detail: error_messages.append('Payment authorization failed. Please try a different card.') else: error_messages.append(detail) return { 'success': False, 'errors': error_messages if error_messages else ['Payment processing failed. Please try again.'] } # Payment succeeded payment = result.payment return { 'success': True, 'payment_id': payment.id, 'status': payment.status, 'amount': amount_money, 'currency': 'GBP', 'created_at': payment.created_at, 'receipt_url': payment.receipt_url if hasattr(payment, 'receipt_url') else None, 'reference_id': reference_id } except Exception as e: # Handle Square API exceptions error_message = str(e) # Check if this is a Square API error with response body if hasattr(e, 'errors') and e.errors: # Extract user-friendly messages from Square error friendly_errors = [] for error in e.errors: code = error.get('code') if isinstance(error, dict) else getattr(error, 'code', None) if code == 'GENERIC_DECLINE': friendly_errors.append('Your card was declined. Please try a different payment method.') elif code == 'INSUFFICIENT_FUNDS': friendly_errors.append('Insufficient funds. Please try a different payment method.') elif code == 'CVV_FAILURE': friendly_errors.append('The CVV code is invalid. Please check and try again.') elif code == 'CARD_DECLINED': friendly_errors.append('Your card was declined. Please contact your bank or try a different card.') else: detail = error.get('detail') if isinstance(error, dict) else getattr(error, 'detail', str(error)) if 'Authorization error' in str(detail): friendly_errors.append('Payment authorization failed. Please try a different card.') else: friendly_errors.append(str(detail)) return { 'success': False, 'errors': friendly_errors if friendly_errors else ['Payment processing failed. Please try again.'] } # Generic error fallback return { 'success': False, 'errors': ['Payment processing failed. Please try again or contact support.'] } async def get_payment(self, payment_id: str) -> Dict: """ Retrieve payment details from Square Args: payment_id: Square payment ID Returns: Dict with payment details """ try: result = self.client.payments.get(payment_id) if result.errors: return { 'success': False, 'errors': [error.detail if hasattr(error, 'detail') else str(error) for error in result.errors] } payment = result.payment amount_money = payment.amount_money return { 'success': True, 'payment_id': payment.id, 'status': payment.status, 'amount': amount_money.amount / 100, # Convert pence to pounds 'currency': amount_money.currency, 'created_at': payment.created_at, 'receipt_url': payment.receipt_url if hasattr(payment, 'receipt_url') else None, 'reference_id': payment.reference_id if hasattr(payment, 'reference_id') else None } except Exception as e: return { 'success': False, 'errors': [str(e)] } async def refund_payment( self, payment_id: str, amount_money: Optional[float] = None, reason: Optional[str] = None, idempotency_key: Optional[str] = None ) -> Dict: """ Refund a payment Args: payment_id: Square payment ID to refund amount_money: Amount to refund (None for full refund) reason: Reason for the refund idempotency_key: Unique key to prevent duplicate refunds Returns: Dict with refund result """ try: if not idempotency_key: idempotency_key = str(uuid.uuid4()) # Prepare parameters amount_in_pence = int(amount_money * 100) if amount_money else None # Create refund - pass parameters directly as keyword arguments if amount_in_pence: result = self.client.refunds.refund_payment( idempotency_key=idempotency_key, amount_money={ 'amount': amount_in_pence, 'currency': 'GBP' }, payment_id=payment_id, reason=reason if reason else None ) else: # Full refund - get payment amount first payment_result = await self.get_payment(payment_id) if not payment_result.get('success'): return payment_result amount_in_pence = int(payment_result['amount'] * 100) result = self.client.refunds.refund_payment( idempotency_key=idempotency_key, amount_money={ 'amount': amount_in_pence, 'currency': 'GBP' }, payment_id=payment_id, reason=reason if reason else None ) if result.errors: return { 'success': False, 'errors': [error.detail if hasattr(error, 'detail') else str(error) for error in result.errors] } refund = result.refund return { 'success': True, 'refund_id': refund.id, 'status': refund.status, 'payment_id': payment_id } except Exception as e: return { 'success': False, 'errors': [str(e)] } async def create_customer( self, email: str, given_name: str, family_name: str, phone_number: Optional[str] = None, address: Optional[Dict] = None, idempotency_key: Optional[str] = None ) -> Dict: """ Create a Square customer for future payments Args: email: Customer email given_name: Customer first name family_name: Customer last name phone_number: Optional phone number address: Optional address dict idempotency_key: Unique key Returns: Dict with customer details """ try: if not idempotency_key: idempotency_key = str(uuid.uuid4()) # Create customer - pass parameters directly as keyword arguments result = self.client.customers.create( idempotency_key=idempotency_key, email_address=email, given_name=given_name, family_name=family_name, phone_number=phone_number if phone_number else None, address=address if address else None ) if result.errors: return { 'success': False, 'errors': [error.detail if hasattr(error, 'detail') else str(error) for error in result.errors] } customer = result.customer return { 'success': True, 'customer_id': customer.id, 'email': customer.email_address if hasattr(customer, 'email_address') else None } except Exception as e: return { 'success': False, 'errors': [str(e)] } # Singleton instance square_service = SquareService()