333 lines
13 KiB
Python
333 lines
13 KiB
Python
"""
|
|
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
|
|
) -> 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
|
|
|
|
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)
|
|
|
|
# Create payment - pass parameters directly as keyword arguments
|
|
result = self.client.payments.create(
|
|
source_id=source_id,
|
|
idempotency_key=idempotency_key,
|
|
amount_money={
|
|
'amount': amount_in_pence,
|
|
'currency': 'GBP'
|
|
},
|
|
location_id=self.location_id,
|
|
customer_id=customer_id if customer_id else None,
|
|
reference_id=reference_id if reference_id else None,
|
|
note=note if note else None
|
|
)
|
|
|
|
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()
|