kscheidecker
e41d3f5e3a
Some checks failed
Build and deploy the backend to staging / Build and push image (pull_request) Successful in 1m34s
Run linting on the backend code / Build (pull_request) Failing after 18s
Run testing on the backend code / Build (pull_request) Failing after 38s
Build and deploy the backend to staging / Deploy to staging (pull_request) Successful in 1m5s
71 lines
2.3 KiB
Python
71 lines
2.3 KiB
Python
from typing import Literal
|
|
import paypalrestsdk
|
|
from pydantic import BaseModel
|
|
from fastapi import HTTPException
|
|
import logging
|
|
|
|
|
|
# Model for payment request body
|
|
class PaymentRequest(BaseModel):
|
|
user_id: str
|
|
credit_amount: Literal[10, 50, 100]
|
|
currency: Literal["USD", "EUR", "CHF"]
|
|
description: str = "Purchase of credits"
|
|
|
|
|
|
# Payment handler class for managing PayPal payments
|
|
class PaymentHandler:
|
|
|
|
payment_id: str
|
|
|
|
def __init__(self, transaction_details: PaymentRequest):
|
|
self.details = transaction_details
|
|
self.logger = logging.getLogger(__name__)
|
|
|
|
# Only support purchase of credit 'bundles': 10, 50 or 100 credits worth of trip generation
|
|
def fetch_price(self) -> float:
|
|
"""
|
|
Fetches the price of credits in the specified currency.
|
|
"""
|
|
result = self.supabase.table("prices").select("credit_amount").eq("currency", self.details.currency).single().execute()
|
|
if result.data:
|
|
return result.data.get("price")
|
|
else:
|
|
self.logger.error(f"Unsupported currency: {self.details.currency}")
|
|
return None
|
|
|
|
def create_paypal_payment(self) -> str:
|
|
"""
|
|
Creates a PayPal payment and returns the approval URL.
|
|
"""
|
|
price = self.fetch_price()
|
|
payment = paypalrestsdk.Payment({
|
|
"intent": "sale",
|
|
"payer": {
|
|
"payment_method": "paypal"
|
|
},
|
|
"transactions": [{
|
|
"amount": {
|
|
"total": f"{price:.2f}",
|
|
"currency": self.details.currency
|
|
},
|
|
"description": self.details.description
|
|
}],
|
|
"redirect_urls": {
|
|
"return_url": "http://localhost:8000/payment/success",
|
|
"cancel_url": "http://localhost:8000/payment/cancel"
|
|
}
|
|
})
|
|
|
|
if payment.create():
|
|
self.logger.info("Payment created successfully")
|
|
self.payment_id = payment.id
|
|
|
|
# Get the approval URL and return it for the user to approve
|
|
for link in payment.links:
|
|
if link.rel == "approval_url":
|
|
return link.href
|
|
else:
|
|
self.logger.error(f"Failed to create payment: {payment.error}")
|
|
raise HTTPException(status_code=500, detail="Payment creation failed")
|