""" Razorpay Payment Service - Modular, plug-and-play payment integration. This module provides a complete Razorpay integration that can be easily moved to another project with minimal changes. Features: - Create Razorpay orders for credit purchases - Verify payment signatures (HMAC SHA256) - Process webhook events - Configurable credit packages Usage: from services.razorpay_service import RazorpayService, CREDIT_PACKAGES # Initialize service = RazorpayService() # Uses environment variables # Or with explicit credentials service = RazorpayService( key_id="rzp_test_xxx", key_secret="secret", webhook_secret="webhook_secret" ) # Create order order = service.create_order( amount_paise=9900, transaction_id="txn_abc123", notes={"user_id": "user123"} ) # Verify payment is_valid = service.verify_payment_signature( order_id="order_xxx", payment_id="pay_xxx", signature="signature" ) Environment Variables: RAZORPAY_KEY_ID: Your Razorpay Key ID RAZORPAY_KEY_SECRET: Your Razorpay Key Secret RAZORPAY_WEBHOOK_SECRET: Webhook secret for signature verification """ import os import hmac import hashlib import logging from typing import Optional, Dict, Any, List from dataclasses import dataclass from enum import Enum import razorpay logger = logging.getLogger(__name__) # ============================================================================= # Credit Packages Configuration # ============================================================================= @dataclass class CreditPackage: """Represents a purchasable credit package.""" id: str name: str credits: int amount_paise: int # Amount in paise (INR × 100) currency: str = "INR" @property def amount_rupees(self) -> float: """Get amount in rupees.""" return self.amount_paise / 100 def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for API responses.""" return { "id": self.id, "name": self.name, "credits": self.credits, "amount_paise": self.amount_paise, "amount_rupees": self.amount_rupees, "currency": self.currency } # Available credit packages CREDIT_PACKAGES: Dict[str, CreditPackage] = { "starter": CreditPackage( id="starter", name="Starter Pack", credits=100, amount_paise=9900 # ₹99 ), "standard": CreditPackage( id="standard", name="Standard Pack", credits=500, amount_paise=44900 # ₹449 ), "pro": CreditPackage( id="pro", name="Pro Pack", credits=1000, amount_paise=79900 # ₹799 ) } def get_package(package_id: str) -> Optional[CreditPackage]: """Get a credit package by ID.""" return CREDIT_PACKAGES.get(package_id.lower()) def list_packages() -> List[Dict[str, Any]]: """List all available credit packages.""" return [pkg.to_dict() for pkg in CREDIT_PACKAGES.values()] # ============================================================================= # Payment Status Enum # ============================================================================= class PaymentStatus(str, Enum): """Payment transaction statuses.""" CREATED = "created" AUTHORIZED = "authorized" CAPTURED = "captured" PAID = "paid" FAILED = "failed" REFUNDED = "refunded" # ============================================================================= # Razorpay Service # ============================================================================= class RazorpayServiceError(Exception): """Base exception for Razorpay service errors.""" pass class RazorpayConfigError(RazorpayServiceError): """Raised when Razorpay is not properly configured.""" pass class RazorpayOrderError(RazorpayServiceError): """Raised when order creation fails.""" pass class RazorpayVerificationError(RazorpayServiceError): """Raised when payment verification fails.""" pass class RazorpayService: """ Modular Razorpay payment service. This service handles all Razorpay interactions including: - Order creation - Payment signature verification - Webhook signature verification The service can be initialized with explicit credentials or will automatically use environment variables. """ def __init__( self, key_id: Optional[str] = None, key_secret: Optional[str] = None, webhook_secret: Optional[str] = None ): """ Initialize Razorpay service. Args: key_id: Razorpay Key ID (or RAZORPAY_KEY_ID env var) key_secret: Razorpay Key Secret (or RAZORPAY_KEY_SECRET env var) webhook_secret: Webhook secret (or RAZORPAY_WEBHOOK_SECRET env var) """ self.key_id = key_id or os.getenv("RAZORPAY_KEY_ID") self.key_secret = key_secret or os.getenv("RAZORPAY_KEY_SECRET") self.webhook_secret = webhook_secret or os.getenv("RAZORPAY_WEBHOOK_SECRET") if not self.key_id or not self.key_secret: raise RazorpayConfigError( "Razorpay credentials not configured. " "Set RAZORPAY_KEY_ID and RAZORPAY_KEY_SECRET environment variables." ) # Initialize Razorpay client self._client = razorpay.Client(auth=(self.key_id, self.key_secret)) logger.info("Razorpay service initialized") @property def is_configured(self) -> bool: """Check if the service is properly configured.""" return bool(self.key_id and self.key_secret) def create_order( self, amount_paise: int, transaction_id: str, currency: str = "INR", notes: Optional[Dict[str, str]] = None ) -> Dict[str, Any]: """ Create a Razorpay order for payment. Args: amount_paise: Amount in paise (₹1 = 100 paise) transaction_id: Your internal transaction/receipt ID currency: Currency code (default: INR) notes: Optional notes to attach to the order Returns: Razorpay order response containing order_id Raises: RazorpayOrderError: If order creation fails """ try: order_data = { "amount": amount_paise, "currency": currency, "receipt": transaction_id, "notes": notes or {} } order = self._client.order.create(data=order_data) logger.info( f"Created Razorpay order: {order['id']} " f"for amount: {amount_paise} paise, " f"receipt: {transaction_id}" ) return order except razorpay.errors.BadRequestError as e: logger.error(f"Razorpay order creation failed (bad request): {e}") raise RazorpayOrderError(f"Invalid order data: {e}") except razorpay.errors.ServerError as e: logger.error(f"Razorpay server error: {e}") raise RazorpayOrderError(f"Razorpay server error: {e}") except Exception as e: logger.error(f"Unexpected error creating Razorpay order: {e}") raise RazorpayOrderError(f"Failed to create order: {e}") def verify_payment_signature( self, order_id: str, payment_id: str, signature: str ) -> bool: """ Verify Razorpay payment signature. This MUST be called after receiving payment confirmation from the client to ensure the payment is authentic. Args: order_id: Razorpay order ID payment_id: Razorpay payment ID signature: Razorpay signature from checkout response Returns: True if signature is valid Raises: RazorpayVerificationError: If signature verification fails """ try: # Generate expected signature message = f"{order_id}|{payment_id}" expected_signature = hmac.new( self.key_secret.encode('utf-8'), message.encode('utf-8'), hashlib.sha256 ).hexdigest() # Constant-time comparison to prevent timing attacks is_valid = hmac.compare_digest(expected_signature, signature) if is_valid: logger.info(f"Payment signature verified: order={order_id}, payment={payment_id}") else: logger.warning(f"Invalid payment signature: order={order_id}, payment={payment_id}") return is_valid except Exception as e: logger.error(f"Error verifying payment signature: {e}") raise RazorpayVerificationError(f"Signature verification failed: {e}") def verify_webhook_signature( self, body: bytes, signature: str ) -> bool: """ Verify Razorpay webhook signature. Use this to authenticate incoming webhook requests. Args: body: Raw request body bytes signature: X-Razorpay-Signature header value Returns: True if webhook signature is valid """ if not self.webhook_secret: logger.warning("Webhook secret not configured, skipping verification") return False try: expected_signature = hmac.new( self.webhook_secret.encode('utf-8'), body, hashlib.sha256 ).hexdigest() is_valid = hmac.compare_digest(expected_signature, signature) if not is_valid: logger.warning("Invalid webhook signature received") return is_valid except Exception as e: logger.error(f"Error verifying webhook signature: {e}") return False def fetch_payment(self, payment_id: str) -> Dict[str, Any]: """ Fetch payment details from Razorpay. Args: payment_id: Razorpay payment ID Returns: Payment details from Razorpay """ try: payment = self._client.payment.fetch(payment_id) return payment except Exception as e: logger.error(f"Error fetching payment {payment_id}: {e}") raise RazorpayServiceError(f"Failed to fetch payment: {e}") def fetch_order(self, order_id: str) -> Dict[str, Any]: """ Fetch order details from Razorpay. Args: order_id: Razorpay order ID Returns: Order details from Razorpay """ try: order = self._client.order.fetch(order_id) return order except Exception as e: logger.error(f"Error fetching order {order_id}: {e}") raise RazorpayServiceError(f"Failed to fetch order: {e}") # ============================================================================= # Module-level convenience functions # ============================================================================= _service_instance: Optional[RazorpayService] = None def get_razorpay_service() -> RazorpayService: """ Get or create a singleton Razorpay service instance. Returns: RazorpayService instance Raises: RazorpayConfigError: If Razorpay is not configured """ global _service_instance if _service_instance is None: _service_instance = RazorpayService() return _service_instance def is_razorpay_configured() -> bool: """Check if Razorpay credentials are configured in environment.""" return bool( os.getenv("RAZORPAY_KEY_ID") and os.getenv("RAZORPAY_KEY_SECRET") )