import json import logging from typing import Literal from datetime import datetime, timedelta import requests from pydantic import BaseModel, Field, field_validator 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 BasketItem(BaseModel): """ Represents a single item in the user's basket. Attributes: id (str): The unique identifier for the item. quantity (int): The number of units of the item. """ id: str quantity: int class Item(BaseModel): """ Represents an item available in the shop. Attributes: id (str): The unique identifier for the item. name (str): The name of the item. description (str): The description of the item. unit_price (float): The unit price of the item. """ id: str name: str description: str unit_price: float unit_credits: int def item_from_sql(item_id: str): """ Fetches an item from the database by its ID. Args: item_id (str): The unique identifier for the item. Returns: Item: The item object retrieved from the database. """ # TODO: Replace with actual SQL fetch logic return Item( id = '12345678', name = 'test_item', description = 'lorem ipsum', unit_price = 0.1, unit_credits = 5 ) 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 @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 or not all(isinstance(i, BasketItem) for i in v): raise ValueError('Basket must contain BasketItem objects') return 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 = item_from_sql(basket_item.id) 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.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): """ 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) } } } } ], # TODO: add these to anydev website 'application_context': { 'return_url': 'https://anydev.info', 'cancel_url': 'https://anydev.info' } } # 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 order_response.raise_for_status() # 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 transactions: # 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 = order_request.user_id, order_id = json.loads(order_response.text)["id"], credits_to_grant = order_request.total_credits) return order_response.json() # 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