""" AUTHOR: Khushal P Soonderji DATE: Friday, 13th Dec., 2024 OBJECTIVE: To handle all SMS 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, AsyncMongoStorage # Data models: from models.core.user import CoreUserInfoModel from models.core.auth_token import CoreAuthTokenModel from models.core.payment import CorePaymentModel, PaymentEvent from models.api.finstitutions.payments.request import ( PaymentRequestOneResult ) # Payment Clients: from utils_v2.payments.safaricom.models.auth import MPesaExpressAuthorization from utils_v2.payments.safaricom.controllers.m_pesa_express import SafaricomMPesaExpress # To work with MongoDB: from bson import ObjectId from pymongo import InsertOne # To work with datatypes: from typing import Literal, List, Dict, Any, Union # To make API calls: import httpx # For asynchronous activities: import asyncio # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** CLASSES *** # ***** **** # ***************************************************************************************************************** class PaymentController: # ┏┓┓ ┓┏ # ┃ ┃┏┓┏┏ ┃┃┏┓┏┓┏ # ┗┛┗┗┻┛┛ ┗┛┗┻┛ ┛ pass # ┓┏ ┓ # ┣┫┏┓┃┏┓┏┓┏┓┏ # ┛┗┗ ┗┣┛┗ ┛ ┛ # ┛ pass # ┏┓ ┓ # ┣┫┓┏╋┣┓ # ┛┗┗┻┗┛┗ @staticmethod async def set_token_direct( db_conn: AsyncMySQL, mongo_conn: AsyncMongo, auth_token: CoreAuthTokenModel, session_token: str = None ) -> bool: # Start by assuming failure: success = False # Get a token id: token_key = await current_app.core_auth_token_controller.get_token_key( db_conn = db_conn, mongo_conn = mongo_conn, auth_token = auth_token, token_notes = {}, session_token = session_token ) # Immediately save the details against that token id: success = await current_app.core_auth_token_controller.set_token( db_conn = db_conn, mongo_conn = mongo_conn, token_key = token_key, auth_token = auth_token, token_notes = {}, session_token = session_token ) # Done here: return success @staticmethod async def get_token( mongo_conn: AsyncMongo, token_key: ObjectId | str = None, ) -> CoreAuthTokenModel | None: # Simply call the core model: return await current_app.core_auth_token_controller.get_token( mongo_conn = mongo_conn, token_key = token_key ) # ┳┓ ┏┓ # ┣┫┏┓┏┓┓┏┏┓┏╋ ┃┃┏┓┓┏┏┳┓┏┓┏┓╋┏ # ┛┗┗ ┗┫┗┻┗ ┛┗ ┣┛┗┻┗┫┛┗┗┗ ┛┗┗┛ # ┗ ┛ # @staticmethod # async def __request_from_safaricom_m_pesa_express( # http_client: httpx.AsyncClient, # auth_token: CoreAuthTokenModel, # payment: Union[SafaricomMPesaExpressRequest], # callback_url: str # ) -> PaymentRequestOneResult: # # # Start by assuming failure: # request_result = PaymentRequestOneResult() # # # Create the client: # client = SafaricomMPesaExpress( # auth = MPesaExpressAuthorization( # consumerKey = auth_token.auth["consumerKey"], # consumerSecret = auth_token.auth["consumerSecret"], # businessShortCode = auth_token.auth["businessShortCode"], # appPasskey = auth_token.auth["appPasskey"] # ), # http_client = http_client # ) # # # Make the payment request: # client_response = await client.request_payment( # amount = payment.amount, # party_a = payment.partyA, # type = payment.transactionType, # reference = payment.accountReference, # description = payment.transactionDescription, # callback_url = callback_url, # payer_no = payment.phoneNo, # party_b = payment.partyB # ) # # # Construct the payment details: # payment_details = CorePaymentModel( # user = None, # lastEventTs = date_time.get_current_utc_date_time(as_string = False), # lastPaymentStatus = "initiated" if client_response.success else "initFailed", # tokenId = auth_token.authTokenId, # amount = payment.amount, # curencyCode = "KES", # # metadata = payment. # ) # # # Done here: # request_result.success = client_response.success # request_result.message = client_response.message # request_result.message = client_response.message # return request_result # # async def request_payment( # self, # mongo_conn: AsyncMongo, # http_client: httpx.AsyncClient, # auth_token: CoreAuthTokenModel, # payment: Union[SafaricomMPesaExpressRequest], # callback_url: str # ) -> PaymentRequestOneResult: # # # Start by assuming failure: # request_result = PaymentRequestOneResult() # # # Now we route the message to the appropriate client: # match auth_token.client: # case "safaricomMPesaExpress": # request_result = await self.__request_from_safaricom_m_pesa_express( # http_client = http_client, # auth_token = auth_token, # payment = payment, # callback_url = callback_url # ) # case _: # request_result.message = f"invalid client {auth_token.client}" # # # Done here: # return request_result # ┓ • ┓ ┏┓ ┏┓ # ┃ ┓┏╋ ┏┓┏┓┏┫ ┃┓┏┓╋ ┃┃┏┓┓┏┏┳┓┏┓┏┓╋┏ # ┗┛┗┛┗ ┗┻┛┗┗┻ ┗┛┗ ┗ ┣┛┗┻┗┫┛┗┗┗ ┛┗┗┛ # ┛ # These are simply for retrieving payment records. # You need to already have them saved to the database. @staticmethod async def list_payments( mongo_conn: AsyncMongo, token_ids: List[ObjectId | str], limit: int = 100, skip: int = 0, additional_filter: dict = None ) -> List[CorePaymentModel] | None: # Regardless of what additional filter is provided from outside, # we add a payment-selecting filter here: if additional_filter is None: additional_filter = {} additional_filter["serviceType"] = "paymentGateway" # Simply call the core model: return await current_app.core_payment_controller.get_payment_previews( mongo_conn = mongo_conn, token_ids = token_ids, limit = limit, skip = skip, additional_filter = additional_filter ) @staticmethod async def get_payment( mongo_conn: AsyncMongo, token_id: ObjectId | str, payment_id: ObjectId | str ) -> CorePaymentModel | None: # Simply call the core model: return await current_app.core_payment_controller.get_payment( mongo_conn = mongo_conn, token_id = token_id, payment_id = payment_id ) # ┳┳ ┓ # ┃┃┏┓┏┫┏┓╋┏┓ # ┗┛┣┛┗┻┗┻┗┗ # ┛ @staticmethod async def add_event( mongo_conn: AsyncMongo, payment_id: ObjectId | str, event: PaymentEvent ) -> bool: # Simply call the core model: return await current_app.core_payment_controller.add_event( mongo_conn = mongo_conn, payment_id = payment_id, event = event ) @staticmethod async def update_tags( mongo_conn: AsyncMongo, token_id: ObjectId | str, payment_id: ObjectId | str, unset_tags: List[str] = None, set_tags: List[str] = None ) -> bool: # Simply call the core model: return await current_app.core_payment_controller.update_tags( mongo_conn = mongo_conn, token_id = token_id, payment_id = payment_id, unset_tags = unset_tags, set_tags = set_tags ) # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": pass # from utils_v2.string import json # # file_options = [ # r"/home/developer/Downloads/recursive parts parse - 20241210.json", # r"/home/developer/Downloads/recursive parts parse (no attachment) - 20241210.json", # ] # # raw_mail_json = json.from_file(file_options[1]) # print("FROM FILE:", json.to_string(raw_mail_json["payload"])) # print("\n\n---------\n\n") # mail_controller = MailController() # print(json.to_string(mail_controller.drop_attachments(raw_mail_json["payload"])))