import json import logging from typing import Literal from datetime import datetime, timedelta import requests from pydantic import BaseModel, Field from fastapi.responses import RedirectResponse from ..supabase.supabase import SupabaseClient from ..structs.shop import Item, BasketItem from ..configuration.environment import Environment from ..cache import CreditCache, make_credit_cache_key # Intialize the logger logger = logging.getLogger(__name__) # Define the base URL, might move that to toml file BASE_URL_PROD = 'https://api-m.paypal.com' BASE_URL_SANDBOX = 'https://api-m.sandbox.paypal.com' class OrderRequest(BaseModel): """ Represents an order request from the frontend. Attributes: user_id (str): The ID of the user placing the order. basket (list[BasketItem]): List of basket items. currency (str): The currency code for the order. created_at (datetime): Timestamp when the order was created. updated_at (datetime): Timestamp when the order was last updated. items (list[Item]): List of item details loaded from the database. total_price (float): Total price of the order. """ user_id: str basket: list[BasketItem] currency: Literal['CHF', 'EUR', 'USD'] created_at: datetime = Field(default_factory=datetime.now) updated_at: datetime = Field(default_factory=datetime.now) items: list[Item] = Field(default_factory=list) total_price: float = None total_credits: int = None supabase_client: SupabaseClient # @field_validator('basket') def validate_basket(cls, v): """Validates the basket items. Args: v (list): List of basket items. Raises: ValueError: If basket does not contain valid BasketItem objects. Returns: list: The validated basket. """ if not v: raise ValueError("Basket cannot be empty") # Pydantic already converts dict -> BasketItem, so isinstance works if not all(isinstance(i, BasketItem) for i in v): raise ValueError("Basket must contain BasketItem objects") return v def load_items_and_price(self): # This should be automatic upon initialization of the class """ Loads item details from database and calculates the total price as well as the total credits to be granted. """ self.items = [] self.total_price = 0 self.total_credits = 0 for basket_item in self.basket: item = self.supabase_client.get_item(basket_item.item_id, self.currency) self.items.append(item) self.total_price += item.unit_price * basket_item.quantity # increment price self.total_credits += item.unit_credits * basket_item.quantity # increment credit balance def to_paypal_items(self): """ Converts items to the PayPal API item format. Returns: list: List of items formatted for PayPal API. """ item_list = [] for basket_item, item in zip(self.basket, self.items): item_list.append({ 'id': item.item_id, 'name': item.name, 'description': item.description, 'quantity': str(basket_item.quantity), 'unit_amount': { 'currency_code': self.currency, 'value': str(item.unit_price) } }) return item_list # Payment handler class for managing PayPal payments class PaypalClient: """ Handles PayPal payment operations. Attributes: sandbox (bool): Whether to use the sandbox environment. id (str): PayPal client ID. key (str): PayPal client secret. base_url (str): Base URL for PayPal API. _token_cache (dict): Cache for the PayPal OAuth access token. """ _token_cache = { "access_token": None, "expires_at": 0 } def __init__( self, sandbox_mode: bool = False ): """ Initializes the handler. Args: sandbox_mode (bool): Whether to use sandbox credentials. """ self.logger = logging.getLogger(__name__) self.sandbox = sandbox_mode # PayPal keys if sandbox_mode : self.id = Environment.paypal_id_sandbox self.key = Environment.paypal_key_sandbox self.base_url = BASE_URL_SANDBOX else : self.id = Environment.paypal_id_prod self.key = Environment.paypal_key_prod self.base_url = BASE_URL_PROD def _get_access_token(self) -> str | None: """ Gets (and caches) a PayPal access token. Returns: str | None: The access token if successful, None otherwise. """ now = datetime.now() # Check if token is still valid if ( self._token_cache["access_token"] is not None and self._token_cache["expires_at"] > now ): self.logger.info('Returning (cached) access token.') return self._token_cache["access_token"] # Request new token validation_data = {'grant_type': 'client_credentials'} try: # pass the request validation_response = requests.post( url = f'{self.base_url}/v1/oauth2/token', data = validation_data, auth =(self.id, self.key) ) except Exception as exc: self.logger.error(f'Error while requesting access token: {exc}') return None data = validation_response.json() access_token = data.get("access_token") expires_in = int(data.get("expires_in", 3600)) # seconds, default 1 hour # Cache the token and its expiry self._token_cache["access_token"] = access_token self._token_cache["expires_at"] = now + timedelta(seconds=expires_in - 60) # buffer 1 min self.logger.info('Returning (new) access token.') return access_token def order( self, order_request: OrderRequest, return_url_success: str, return_url_failure: str ): """ Creates a new PayPal order. Args: order_request (OrderRequest): The order request. Returns: dict | None: PayPal order response JSON, or None if failed. """ # Fetch details of order from mart database and compute total credits and price order_request.load_items_and_price() # Prepare payload for post request to paypal API order_data = { 'intent': 'CAPTURE', 'purchase_units': [ { 'items': order_request.to_paypal_items(), 'amount': { 'currency_code': order_request.currency, 'value': str(order_request.total_price), 'breakdown': { 'item_total': { 'currency_code': order_request.currency, 'value': str(order_request.total_price) } } } } ], # No redirect from paypal 'application_context': { # 'return_url': f'https://anydev.info/orders/{json.loads(order_response.text)["id"]}/{order_request.user_id}capture', # This returns to backend capture-payment URL # 'cancel_url': return_url_failure 'return_url': 'https://anydev.info/api/paypal/capture', 'cancel_url': return_url_failure } } # Get the access_token: access_token = self._get_access_token() try: order_response = requests.post( url = f'{self.base_url}/v2/checkout/orders', headers = {'Authorization': f'Bearer {access_token}'}, json = order_data, ) # Raise HTTP Exception if request was unsuccessful. except Exception as exc: self.logger.error(f'Error creating PayPal order: {exc}') return None try: order_response.raise_for_status() except: return RedirectResponse(url=return_url_failure) user_id = order_request.user_id order_id = json.loads(order_response.text)["id"] # TODO Now that we have the order ID, we can inscribe the details in sql database using the order id given by paypal # DB for storing the transaction records: # order_id (key): json.loads(order_response.text)["id"] # user_id : order_request.user_id # created_at : order_request.created_at # status : PENDING # basket (json) : OrderDetails.jsonify() # total_price : order_request.total_price # currency : order_request.currency # updated_at : order_request.created_at # Create a cache item for credits to be granted to user CreditCache.set_credits( user_id = user_id, order_id = order_id, credits_to_grant = order_request.total_credits) # return order_response.json() return RedirectResponse(url=f'https://anydev.info/orders/{order_id}/{user_id}capture') # Standalone function to capture a payment def capture(self, user_id: str, order_id: str): """ Captures payment for a PayPal order. Args: order_id (str): The PayPal order ID. Returns: dict | None: PayPal capture response JSON, or None if failed. """ # Get the access_token: access_token = self._get_access_token() try: capture_response = requests.post( url = f'{self.base_url}/v2/checkout/orders/{order_id}/capture', headers = {'Authorization': f'Bearer {access_token}'}, json = {}, ) except Exception as exc: logger.error(f'Error while requesting access token: {exc}') return None # Raise exception if API call failed capture_response.raise_for_status() # print(capture_response.text) # TODO: update status to PAID in sql database # where order_id (key) = order_id # status : 'PAID' # updated_at : datetime.now() # Not sure yet if/how to implement that def cancel(self): pass