overhaul of paypal handler WIP
Some checks failed
Build and deploy the backend to staging / Build and push image (pull_request) Failing after 56s
Build and deploy the backend to staging / Deploy to staging (pull_request) Has been skipped
Run linting on the backend code / Build (pull_request) Successful in 1m54s
Run testing on the backend code / Build (pull_request) Failing after 2m33s

This commit is contained in:
2025-10-04 17:03:36 +02:00
parent 3bdcdea850
commit f86174bc11
2 changed files with 99 additions and 31 deletions

View File

@@ -1,19 +1,23 @@
from typing import Literal from typing import Literal
import paypalrestsdk import logging
import json
from pydantic import BaseModel from pydantic import BaseModel
from fastapi import HTTPException from fastapi import HTTPException
import logging
import requests import requests
from ..configuration.environment import Environment from ..configuration.environment import Environment
# Model for payment request body # Model for payment request body
class PaymentRequest(BaseModel): class OrderDetails():
user_id: str user_id: str
credit_amount: Literal[10, 50, 100] number_of_credits: Literal[10, 50, 100]
currency: Literal["USD", "EUR", "CHF"] unit_price: float
description: str = "Purchase of credits" amount: int
currency: Literal['USD', 'EUR', 'CHF']
# Payment handler class for managing PayPal payments # Payment handler class for managing PayPal payments
@@ -24,52 +28,110 @@ class PaypalHandler:
password = Environment.paypal_key_sandbox password = Environment.paypal_key_sandbox
def __init__( def __init__(
self, self,
transaction_details: PaymentRequest, order_details: OrderDetails,
sandbox_mode: bool = False sandbox_mode: bool = False
): ):
self.details = transaction_details # Initialize the logger
self.logger = logging.getLogger(__name__) self.logger = logging.getLogger(__name__)
# Payment request parameters
self.order_details = order_details
self.sandbox = sandbox_mode self.sandbox = sandbox_mode
# Only support purchase of credit 'bundles': 10, 50 or 100 credits worth of trip generation # Only support purchase of credit 'bundles': 10, 50 or 100 credits worth of trip generation
def fetch_price(self) -> float: # def fetch_price(self) -> float:
""" # '''
Fetches the price of credits in the specified currency. # Fetches the price of credits in the specified currency.
""" # '''
result = self.supabase.table("prices").select("credit_amount").eq("currency", self.details.currency).single().execute() # result = self.supabase.table('prices').select('credit_amount').eq('currency', self.details.currency).single().execute()
if result.data: # if result.data:
return result.data.get("price") # return result.data.get('price')
else: # else:
self.logger.error(f"Unsupported currency: {self.details.currency}") # self.logger.error(f'Unsupported currency: {self.details.currency}')
return None # return None
def validate(self): def validate(self):
if self.sandbox : if self.sandbox :
validation_url = "https://api-m.sandbox.paypal.com/v1/oauth2/token" validation_url = 'https://api-m.sandbox.paypal.com/v1/oauth2/token'
else : else :
validation_url = "https://api-m.paypal.com/v1/oauth2/token" validation_url = 'https://api-m.paypal.com/v1/oauth2/token'
# payload for the validation request # payload for the validation request
validation_data = {'grant_type': 'client_credentials'} validation_data = {'grant_type': 'client_credentials'}
# pass the request try:
validation_response = requests.post( # pass the request
url=validation_url, validation_response = requests.post(
data=validation_data, url=validation_url,
auth=(self.username, self.password) data=validation_data,
) auth=(self.username, self.password)
)
# TODO: continue here except Exception as exc:
self.logger.error(f'Error while requesting access token: {exc}')
return None
if validation_response.status_code == 201 :
access_token = json.loads(validation_response.text)['access_token']
self.logger.info('Validation step successful. Returning access token.')
return access_token
self.logger.error(f'Error {validation_response.status_code} while requesting access token: {validation_response.text}')
return None
pass def order(
self,
access_token: int
):
def order(self): if self.sandbox :
order_url = 'https://api-m.sandbox.paypal.com/v2/checkout/orders'
else :
order_url = 'https://api-m.paypal.com/v2/checkout/orders'
# payload for the order equest
order_data = {
'intent': 'CAPTURE',
'purchase_units': [
{
'items': [
{
'name': f'{self.order_details.number_of_credits} Credits Pack',
'description': f'Credits for {self.order_details.number_of_credits} trip generations on AnyWay.',
'quantity': self.order_details.amount,
'unit_amount': {
'currency_code': self.order_details.currency,
'value': self.order_details.unit_price
}
}
],
'amount': {
'currency_code': self.order_details.currency,
'value': self.order_details.amount*self.order_details.unit_price,
# 'breakdown': {
# 'item_total': {
# 'currency_code': 'CHF',
# 'value': '5.00'
# }
# }
## what is that for ?
}
}
],
# TODO: add these to anydev website somehow
'application_context': {
'return_url': 'https://anydev.info',
'cancel_url': 'https://anydev.info'
}
}
# TODO continue here
pass pass
@@ -77,6 +139,8 @@ class PaypalHandler:
def capture(self): def capture(self):
pass pass

View File

@@ -0,0 +1,4 @@
"""This module contains the descriptions of items to be purchased in the AnyWay store."""