""" AUTHOR: Khushal P Soonderji DATE: Saturday, 28th Dec., 2024 OBJECTIVE: To handle all payments-related behaviour from one place. REFERENCES: N/A DOWNLOADS: N/A """ # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # To make sibling directories accessible for imports: import sys sys.path.append(".") sys.path.append("..") # For Quart: from quart import current_app # My async utils: from utils_v2.string import json from utils_v2.date_time import date_time from utils_v2.database.async_mysql_v2 import AsyncMySQL from utils_v2.database.async_mongo_v2 import AsyncMongo from utils_v2.cache.async_redis_cache_v2 import AsyncRedisCache # Controllers: from controllers_v2.core.auth_token import CoreAuthTokenController # Models: from models.core.user import CoreUserInfoModel from models.core.auth_token import CoreAuthTokenModel from models.core.payment import CorePaymentModel, PaymentEvent, CustomerDetails from models.api.finstitutions.payments.request import ( PGPaymentRequestData, PaymentRequestOneResult, PaymentCallbackResult ) # To work with MongoDB: from bson import ObjectId # To work with datatypes: from typing import List, Any # To make HTTP requests: import httpx # To make abstract classes: from abc import ABC, abstractmethod # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** CLASSES *** # ***** **** # ***************************************************************************************************************** class PaymentsController(CoreAuthTokenController, ABC): # ┏┓┓ ┓┏ # ┃ ┃┏┓┏┏ ┃┃┏┓┏┓┏ # ┗┛┗┗┻┛┛ ┗┛┗┻┛ ┛ # For MongoDB: PAYMENTS_COLLECTION = "_payments" # ┏┓ # ┃ ┏┓┏┓┏╋┏┓┓┏┏╋┏┓┏┓ # ┗┛┗┛┛┗┛┗┛ ┗┻┗┗┗┛┛ def __init__( self, cache: AsyncRedisCache = None, http_client: httpx.AsyncClient = None, alert_url: str = None, base_filter: dict = None, debug: bool = True, debug_prefix: str = "Payments (C) | ", debug_only_errors: bool = True ): """ This is the foundational controller for all trading/stockbroking services. This is built on top of the authorization model, and, in turn, the individual stockbroking clients should be built on top of this. :param cache: The object to use for caching results from database calls. :param http_client: The HTTP client :param base_filter: The basic filter that will be applied to all fetching/updating queries. WARNING: THE BASE FILTER WILL ALWAYS BE APPLIED AUTOMATICALLY. SET THIS UP WISELY. :param debug: Whether, or not, you would like to print debugging messages: :param debug_prefix: The prefix to print with the debugging messages. :param debug_only_errors: Whether you would like to print only error messages or all messages. :return: None. """ # Declare the service type: this_service_type = "paymentGateway" # Prepare base filter: this_filter = {} for k, v in (base_filter or {}).items(): this_filter[k] = v this_filter["serviceType"] = this_service_type # Invoke the parent's constructor: CoreAuthTokenController.__init__( self, cache = cache, alert_url = alert_url, http_client = http_client, base_filter = this_filter, debug = debug, debug_prefix = debug_prefix, debug_only_errors = debug_only_errors ) # Init a variable in a parent: self._service_type = this_service_type # ┓ • ┏┓ # ┃ ┓┏╋ ┃┃┏┓┓┏┏┳┓┏┓┏┓╋┏ # ┗┛┗┛┗ ┣┛┗┻┗┫┛┗┗┗ ┛┗┗┛ # ┛ # These are simply for retrieving payment records. # You need to already have them saved to the database. async def count_payments( self, mongo_data_conn: AsyncMongo, token_ids: List[ObjectId | str], additional_filter: dict = None ) -> int: """ Just counts the no. of payment records that match a given set of conditions. :param mongo_data_conn: The instance of the database connector to use for the operation. :param token_ids: The token ids of the accounts from which these payment details must be fetched. :param additional_filter: Any addition filters to use. :return: The no. of payment records that match the given conditions. """ # Prepare the filter: if not isinstance(token_ids, list): token_ids = [token_ids] token_ids = [ObjectId(t) for t in token_ids] filter_json = {"tokenId": {"$in": token_ids}} if additional_filter: for k, v in additional_filter.items(): filter_json[k] = v # Get the count of the documents that match the criteria: count = await mongo_data_conn.count( collection = self.PAYMENTS_COLLECTION, filter = filter_json, raise_exception = True ) # Done here: return count async def list_payments( self, mongo_data_conn: AsyncMongo, token_ids: List[ObjectId | str], limit: int = 100, skip: int = 0, additional_filter: dict = None ) -> List[CorePaymentModel] | None: """ Fetches many payment details in one call, but just their previews. :param mongo_data_conn: The instance of the database connector to use for the operation. :param token_ids: The token ids of the accounts from which these messages must be fetched. :param limit: The max. no. of payment details to retrieve in this call. :param skip: The no. of initial payment details to skip. Useful for pagination. :param additional_filter: Any addition filters to use. :return: The list of payments (as the payments model). This list can be empty. """ # Prepare the filter: if not isinstance(token_ids, list): token_ids = [token_ids] token_ids = [ObjectId(t) for t in token_ids] filter_json = {"tokenId": {"$in": token_ids}} if additional_filter: for k, v in additional_filter.items(): filter_json[k] = v # We fetch the messages that are identified by a specific token id, # with the specified fetching limits, while enforcing the sorting condition: records = await mongo_data_conn.find_many( collection = self.PAYMENTS_COLLECTION, filter = filter_json, projection = {"events": False}, limit = limit, skip = skip, sort = {"ts": -1}, raise_exception = True ) # Convert the fetched records to instances of the data model and return: for record in records: record["events"] = [] return [CorePaymentModel(**record) for record in records] async def get_payment( self, mongo_data_conn: AsyncMongo, payment_id: ObjectId | str ) -> CorePaymentModel | None: """ Gets one payment detail if you know its payment id. :param mongo_data_conn: The instance of the database connector to use for the operation. :param payment_id: The id of the payment detail that needs to be read. :return: The contents of that one payment detail in a structured format. """ # We fetch the whole payload of that one message: record = await mongo_data_conn.find_one( collection = self.PAYMENTS_COLLECTION, filter = {"_id": ObjectId(payment_id)}, raise_exception = True ) # If no such message was found: if record is None: return None # If a record was found, # we return it as our data model: return CorePaymentModel(**record) async def get_payment_by_client_reference_id( self, mongo_data_conn: AsyncMongo, client_reference_id: str ) -> CorePaymentModel | None: """ Gets one payment detail if you know its payment id. :param mongo_data_conn: The instance of the database connector to use for the operation. :param client_reference_id: The way the client identifies this payment. You need to pass this only on the first event. Typically, when you initiate the payment request. :return: The contents of that one payment detail in a structured format. """ # We fetch the whole payload of that one message: record = await mongo_data_conn.find_one( collection = self.PAYMENTS_COLLECTION, filter = {"clientPaymentReferenceId": client_reference_id}, raise_exception = True ) # If no such message was found: if record is None: return None # If a record was found, # we return it as our data model: return CorePaymentModel(**record) # ┳┳ ┓ ┏┓ # ┃┃┏┓┏┫┏┓╋┏┓ ┃┃┏┓┓┏┏┳┓┏┓┏┓╋┏ # ┗┛┣┛┗┻┗┻┗┗ ┣┛┗┻┗┫┛┗┗┗ ┛┗┗┛ # ┛ ┛ # We don't support updating payments themselves, # but we will allow updating fields like tags, adding events, etc. async def add_receipt( self, sql_conn: AsyncMySQL, mongo_data_conn: AsyncMongo, payment: CorePaymentModel, session_token: str = None ) -> bool: """ The API endpoint that initiates the payment request will send details for calling this procedure in the 'metadata' field. We take those values and invoke the procedure to add a receipt in the database's records. :param sql_conn: The database connection to use to perform this activity. :param mongo_data_conn: The database connection to use to perform this activity. :param payment: The record of the payment whose details need to be added as a receipt. :param session_token: The session token of the user to identify whose request caused this receipt update. :return: True is the receipt was added, else False. """ # Call the database's procedure: sql_json = await self.call_procedure( sql_conn = sql_conn, proc_name = "accounting_receipt_add", proc_args = ( payment.metadata.idClient, # .................................... p_client_id date_time.get_current_ist_date_time().strftime("%Y-%m-%d"), # ... p_date payment.client, # ............................................... p_source payment.amount, # ............................................... p_amount payment.metadata.reference, # ................................... p_reference payment.metadata.inAccount, # ................................... p_in_account payment.metadata.idUser, # ...................................... p_user_id payment.metadata.documentUrl, # ................................. p_document_url json.to_string(payment.metadata.againstReference) # ............. p_notes ), retry_count = 1, backoff_seconds = 0.5, backoff_multiplier = 1.1, session_token = session_token, alert_on_all_failures = True ) # Return the response: if sql_json["status"] == 1: return True else: return False async def add_sales_transaction( self, sql_conn: AsyncMySQL, mongo_data_conn: AsyncMongo, payment: CorePaymentModel, session_token: str = None ) -> bool: """ ??? Fuck it. Not worth breaking your head over. :param sql_conn: The database connection to use to perform this activity. :param mongo_data_conn: The database connection to use to perform this activity. :param payment: The record of the payment whose details need to be added as a receipt. :param session_token: The session token of the user to identify whose request caused this receipt update. :return: True is the receipt was added, else False. """ # If this is not an against-sale kind of transaction: if not payment.metadata.reference.strip().lower().find("against sale") >= 0: return True # If this is against some sale: success = False api_input_json = { "transactionDate": date_time.get_current_ist_date_time().strftime("%Y-%m-%d"), "clientId": payment.metadata.idClient, "clientName": payment.metadata.clientName, "clientAddress": None, "deliveryAddress": None, "total": payment.amount, "documentUrl": None, "invoiceNo": None, "deliveryBoyId": None, "deliveryBoyName": None, "lineItems": payment.metadata.againstReference } print("ADD SALES TXN INPUT JSON:", json.to_string(api_input_json)) api_response = await current_app.http_client.post( url = "https://api.thecaoffice.com/account/sales/transaction/add", headers = {"X-Session-Token": session_token}, json = api_input_json ) try: api_json = api_response.json() except: api_json = {} print(f"ADD SALES TXN ({api_response.status_code}):", api_response.content) print("ADD SALES TXN:", json.to_string(api_json)) if api_response.is_success: success = True # Done here: return success async def add_event_by_payment_id( self, mongo_data_conn: AsyncMongo, payment_id: ObjectId | str, event: PaymentEvent, client_reference_id: str = None ) -> bool: """ Add an event to an existing record of a payment detail. :param mongo_data_conn: The instance of the database connector to use for the operation. :param payment_id: The id of the payment detail that needs to be read. :param event: The event that occurred. This will typically be generated by the third-party client. :param client_reference_id: The way the client identifies this payment. You need to pass this only on the first event. Typically, when you initiate the payment request. :return: True if successfully noted, else False. """ # Prepare the update document: update_json = { "$push": { "events": event.model_dump() }, "$set": { "lastEventTs": event.eventTs, "lastEventMessage": event.message, "lastPaymentStatus": event.paymentStatus, } } if client_reference_id: update_json["$set"]["clientPaymentReferenceId"] = client_reference_id # Try to update the existing record: return await mongo_data_conn.update_one( collection = self.PAYMENTS_COLLECTION, filter = {"_id": ObjectId(payment_id)}, update = update_json, upsert = False, raise_exception = True ) async def add_event_by_client_reference_id( self, mongo_data_conn: AsyncMongo, client_reference_id: str, event: PaymentEvent, ) -> bool: """ Add an event to an existing record of a payment detail. :param mongo_data_conn: The instance of the database connector to use for the operation. :param event: The event that occurred. This will typically be generated by the third-party client. :param client_reference_id: The way the client identifies this payment. You need to pass this only on the first event. Typically, when you initiate the payment request. :return: True if successfully noted, else False. """ # Prepare the update document: update_json = { "$push": { "events": event.model_dump() }, "$set": { "lastEventTs": event.eventTs, "lastEventMessage": event.message, "lastPaymentStatus": event.paymentStatus } } if client_reference_id: update_json["$set"]["clientPaymentReferenceId"] = client_reference_id # Try to update the existing record: return await mongo_data_conn.update_one( collection = self.PAYMENTS_COLLECTION, filter = {"clientPaymentReferenceId": client_reference_id}, update = update_json, upsert = False, raise_exception = True ) async def update_tags( self, mongo_data_conn: AsyncMongo, payment_id: ObjectId | str, unset_tags: List[str] = None, set_tags: List[str] = None ) -> bool: """ Updates the tags on one payment. The tags to remove are processed first, the ones to add are processed later. :param mongo_data_conn: The instance of the database connector to use for the operation. :param payment_id: The id of the payment detail that needs to be read. :param unset_tags: The tags to remove from the payment record. :param set_tags: The tags to add to the payment record. :return: True if the update was successful, else False. """ # Update the tags: return await mongo_data_conn.update_one( collection = self.PAYMENTS_COLLECTION, filter = {"_id": ObjectId(payment_id)}, update = [{ "$set": { "tags": { "$let": { "vars": { "removed_tags": { "$setDifference": [ "$tags", unset_tags ] } }, "in": { "$setUnion": [ "$$removed_tags", set_tags ] } } } } }], raise_exception = True ) # ┳┓ ┓ ┏┓ # ┃┃┏┓┃┏┓╋┏┓ ┃┃┏┓┓┏┏┳┓┏┓┏┓╋┏ # ┻┛┗ ┗┗ ┗┗ ┣┛┗┻┗┫┛┗┗┗ ┛┗┗┛ # ┛ # No support whatsoever for deleting payment records! # ┳┓ ┏┓ # ┣┫┏┓┏┓┓┏┏┓┏╋ ┃┃┏┓┓┏┏┳┓┏┓┏┓╋┏ # ┛┗┗ ┗┫┗┻┗ ┛┗ ┣┛┗┻┗┫┛┗┗┗ ┛┗┗┛ # ┗ ┛ async def init_payment( self, mongo_data_conn: AsyncMongo, auth_token: CoreAuthTokenModel, user_info: CoreUserInfoModel, payment_request: PGPaymentRequestData, tags: List[Any] ) -> ObjectId | None: """ Do this before you hit the third-party client's service when requesting payments. This creates a validated payment record in the database which can then be used as reference for successive updates. :param mongo_data_conn: The database connection to use to perform this activity. :param auth_token: The token that has to be used to fetch the data. :param user_info: The user of your platform, NOT THE PAYING PARTY. :param payment_request: The payment request that came in through the API call. :param tags: And initial tags to apply to this payment's records that you know you will need for filtering. :return: The id of the MongoDB document that will hold the full record of this payment. """ # Model the payment's request. This ensures we're validating the inputs. payment_model = CorePaymentModel( user = user_info, customer = CustomerDetails( name = payment_request.customerName, contactNo = payment_request.customerNo, payerNo = payment_request.payerNo, email = payment_request.email ), lastEventMessage = "Payment Request Queued", lastPaymentStatus = "queued", tokenId = auth_token.authTokenId, amount = payment_request.amount, currencyCode = payment_request.currencyCode, metadata = payment_request.metadata.model_dump(), tags = list(set(payment_request.tags + (tags or []))), serviceType = "paymentGateway", client = auth_token.client, clientPaymentReferenceId = None, events = [] ) # Simply insert the document and return the id: return await mongo_data_conn.insert_one( collection = self.PAYMENTS_COLLECTION, document = payment_model.model_dump(), raise_exception = True ) @abstractmethod async def request_payment( self, sql_conn: AsyncMySQL, mongo_data_conn: AsyncMongo, auth_token: CoreAuthTokenModel, user_info: CoreUserInfoModel, payment_request: PGPaymentRequestData, ) -> PaymentRequestOneResult: """ To request payment from someone through a payment gateway. :param sql_conn: Only needed when a payment gateway says that the credentials are invalid and the account needs to be disabled. :param mongo_data_conn: The database connection to use to perform this activity. :param auth_token: The token that has to be used to fetch the data. :param user_info: The info. of your user, so that you can identify who requested the payment. :param payment_request: The data that came in with the APi call. :return: The structured response form the payment gateway. """ pass @abstractmethod async def handle_payment_callback( self, sql_conn: AsyncMySQL, mongo_data_conn: AsyncMongo, inbound_data: dict, inbound_headers: dict ) -> PaymentCallbackResult: """ Whenever the payment gateway sends an update about a requested payment, we use this method to update our records as per the specification of the third-party payment gateway. :param sql_conn: Only needed when a payment gateway says that the credentials are invalid and the account needs to be disabled. :param mongo_data_conn: The database connection to use to perform this activity. :param inbound_data: The data sent by the payment gateway in their update. :param inbound_headers: The headers sent by the payment gateway in their update. :return: A structured response about the process of updating the payment event. """ # There are 3 main steps here: # 01. Add the payment's update event to the main record, # 02. Add the receipt (if payment successful), # 03. Disable the payment auth-token if the gateway says that the credentials are invalid. pass # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": pass