diff --git a/api/blueprints/finstitutions/payments/auth.py b/api/blueprints/finstitutions/payments/auth_v2.py similarity index 95% rename from api/blueprints/finstitutions/payments/auth.py rename to api/blueprints/finstitutions/payments/auth_v2.py index 9d04e81..4e6e271 100644 --- a/api/blueprints/finstitutions/payments/auth.py +++ b/api/blueprints/finstitutions/payments/auth_v2.py @@ -6,7 +6,7 @@ DATE: - Friday, 13th Dec., 2024 + Saturday, 28th Dec., 2024 OBJECTIVE: @@ -170,9 +170,9 @@ async def authorize_payment_gateway( if inbound_data.client == "safaricomMPesaExpress": - success = await current_app.payment_controller.set_token_direct( - db_conn = current_app.sql_writer, - mongo_conn = current_app.data_mongo, + success = await current_app.payments_controller.set_token_direct( + sql_conn = current_app.sql_writer, + mongo_data_conn = current_app.data_mongo, auth_token = CoreAuthTokenModel( serviceType = "paymentGateway", client = inbound_data.client, @@ -185,6 +185,9 @@ async def authorize_payment_gateway( status = "active", syncFreq = 60 ), + token_notes = {"businessShortCode": inbound_data.auth.businessShortCode}, + display_name = inbound_data.auth.businessShortCode, + display_picture = None, session_token = inbound_headers["X-Session-Token"] ) diff --git a/api/blueprints/finstitutions/payments/callback.py b/api/blueprints/finstitutions/payments/callback_v2.py similarity index 64% rename from api/blueprints/finstitutions/payments/callback.py rename to api/blueprints/finstitutions/payments/callback_v2.py index 600956a..6091ed6 100644 --- a/api/blueprints/finstitutions/payments/callback.py +++ b/api/blueprints/finstitutions/payments/callback_v2.py @@ -113,7 +113,7 @@ def init(blueprint_setup_state): # --------------------------------------------------------------------------------------------------------------------- -@pg_callback_bp.route("/callback/safaricom/mpesaexpress", methods = ["POST"]) +@pg_callback_bp.route("/callback/safaricom/mpesaexpress", methods = ["GET", "POST"]) @set_api_version(api_version = "1.0.0") @read_input(sanitize_headers = False, sanitize_data = False) @log_request_to_mongo( @@ -144,82 +144,14 @@ async def safaricom_m_pesa_express_callback( :return: A standard response structure. """ - # payment_info = await current_app.payment_controller.get_payment_internal( - # mongo_conn = current_app.data_mongo, - # payment_id = ObjectId(payment_id) - # ) - # print("PAYMENT INFO:", json.to_string(payment_info.model_dump(), default=str)) - # ┏┓ ┓ ┓ ┏┓ # ┣┫┏┫┏┫ ┣ ┓┏┏┓┏┓╋ # ┛┗┗┻┗┻ ┗┛┗┛┗ ┛┗┗ - # Map out the documented codes provided by the payment gateway. - # URL: https://developer.safaricom.co.ke/APIs/MpesaExpressSimulate - code_map = { - 0: { - "status": "settled", - "message": "Payment successful :)" - }, # ... Success - 1037: { - "status": "failed", - "message": "The payment gateway could not reach your customer." - }, # ... DS Timeout. User could not be reached. - 1025: { - "status": "failed", - "message": "There was a system error in the payment gateway (1025)." - }, # ... System error while trying to send the push request. - 9999: { - "status": "failed", - "message": "There was a system error in the payment gateway (9999)." - }, # ... System error while trying to send the push request. - 1032: { - "status": "rejected", - "message": "Your customer declined the payment request." - }, # ... Request Cancelled by the user. - 1: { - "status": "failed", - "message": "Your customer has insufficient balance." - }, # ... The user has insufficient balance. - 2001: { - "status": "failed", - "message": "The payment gateway says your credentials are invalid." - }, # ... Invalid credentials of the initiator. - 1019: { - "status": "failed", - "message": "The transaction expired before your customer processed it." - }, # ... Transaction expired. - 1001: { - "status": "failed", - "message": "Your customer is already in the middle of some transaction on the payment gateway." - }, # ... The payer is already making some transaction. - } - - # Figure out which of the above codes is relevant to you: - pg_reference_id = inbound_data["Body"]["stkCallback"]["CheckoutRequestID"] - pg_result_code = int(inbound_data["Body"]["stkCallback"]["ResultCode"]) - pg_result_desc = inbound_data["Body"]["stkCallback"]["ResultDesc"] - relevant_code = code_map.get( - pg_result_code, - { - "status": "unknown", - "message": f"Unknown code '{pg_result_code}' from the payment gateway. PG: '{pg_result_desc}'" - } - ) - - # For now, we just insert the event into the record: - event_note_success = await current_app.payment_controller.add_event_by_client_reference_id( - mongo_conn = current_app.data_mongo, - event = PaymentEvent( - paymentStatus = relevant_code["status"], - message = relevant_code["message"], - initByPG = True, - ipAddr = inbound_headers["Remote-IP"], - httpCode = None, - headers = inbound_headers, - payload = inbound_data - ), - client_reference_id = pg_reference_id + client_response = await current_app.safaricom_mpesa_express_controller.handle_payment_callback( + mongo_data_conn = current_app.data_mongo, + inbound_data = inbound_data, + inbound_headers = inbound_headers ) # ┳┓ @@ -229,8 +161,9 @@ async def safaricom_m_pesa_express_callback( # Done here: return ResponseModel( - status_code = StatusCodes.OK if event_note_success else StatusCodes.FAILED, - http_code = HttpCodes.SUCCESS if event_note_success else HttpCodes.INTERNAL_SERVER_ERROR + status_code = StatusCodes.OK if client_response.success else StatusCodes.FAILED, + http_code = HttpCodes.SUCCESS if client_response.success else HttpCodes.INTERNAL_SERVER_ERROR, + message = client_response.message ) diff --git a/api/blueprints/finstitutions/payments/get.py b/api/blueprints/finstitutions/payments/get_v2.py similarity index 97% rename from api/blueprints/finstitutions/payments/get.py rename to api/blueprints/finstitutions/payments/get_v2.py index fb39c80..9afa644 100644 --- a/api/blueprints/finstitutions/payments/get.py +++ b/api/blueprints/finstitutions/payments/get_v2.py @@ -6,7 +6,7 @@ DATE: - Wednesday, 18th Dec., 2024 + Saturday, 28th Dec., 2024 OBJECTIVE: @@ -177,8 +177,8 @@ async def get_one_payment_record( # ┻ ┗ ┗┗┛┗ ┛┗┗ ┗┗┛┛ ┗┻ # Get the payment record: - record = await current_app.payment_controller.get_payment( - mongo_conn = current_app.data_mongo, + record = await current_app.payments_controller.get_payment( + mongo_data_conn = current_app.data_mongo, payment_id = inbound_data.paymentId ) @@ -189,7 +189,7 @@ async def get_one_payment_record( # We check if the token that was used to fetch the mail is owned by this user: if not await token_check.is_authorized( - mongo_conn = current_app.data_mongo, + mongo_data_conn = current_app.data_mongo, user_info = CoreUserInfoModel(**kwargs["session_info"]), token_ids = [record.tokenId] ): return ResponseModel( diff --git a/api/blueprints/finstitutions/payments/list.py b/api/blueprints/finstitutions/payments/list_v2.py similarity index 93% rename from api/blueprints/finstitutions/payments/list.py rename to api/blueprints/finstitutions/payments/list_v2.py index 056728c..ce635e2 100644 --- a/api/blueprints/finstitutions/payments/list.py +++ b/api/blueprints/finstitutions/payments/list_v2.py @@ -6,7 +6,7 @@ DATE: - Tuesday, 17th Dec., 2024 + Saturday, 28th Dec., 2024 OBJECTIVE: @@ -161,30 +161,27 @@ async def payment_list( message = "invalid session" ) - # ┏┓ ┓• ┳┳┓ •┓ - # ┣ ┏┓┃┓┏╋ ┃┃┃┏┓┓┃┏ - # ┗┛┛┗┗┗┛┗ ┛ ┗┗┻┗┗┛ + # ┏┓ ┓• ┏┓ + # ┣ ┏┓┃┓┏╋ ┃┃┏┓┓┏┏┳┓┏┓┏┓╋┏ + # ┗┛┛┗┗┗┛┗ ┣┛┗┻┗┫┛┗┗┗ ┛┗┗┛ + # ┛ # Get the token ids from the token keys: - auth_tokens = await current_app.payment_controller.get_tokens_from_keys( - mongo_conn = current_app.data_mongo, + auth_tokens = await current_app.payments_controller.get_tokens_from_keys( + mongo_data_conn = current_app.data_mongo, token_keys = inbound_data.tokenKeys ) token_ids = [t.authTokenId for t in auth_tokens] - print("TOKEN IDS:", json.to_string(token_ids, default = str)) - # Build the additional filter: additional_filter = {} if inbound_data.tags: additional_filter["tags"] = {"$in": inbound_data.tags} if inbound_data.paymentStatus: additional_filter["lastPaymentStatus"] = {"$in": inbound_data.paymentStatus} additional_filter = additional_filter or None - print("ADDFIL:", json.to_string(additional_filter)) - # Get the mails: - payment_records = await current_app.payment_controller.list_payments( - mongo_conn = current_app.data_mongo, + payment_records = await current_app.payments_controller.list_payments( + mongo_data_conn = current_app.data_mongo, token_ids = token_ids, limit = inbound_data.count, skip = inbound_data.fromCount, diff --git a/api/blueprints/finstitutions/payments/request.py b/api/blueprints/finstitutions/payments/request_v2.py similarity index 89% rename from api/blueprints/finstitutions/payments/request.py rename to api/blueprints/finstitutions/payments/request_v2.py index 1263750..f9b7939 100644 --- a/api/blueprints/finstitutions/payments/request.py +++ b/api/blueprints/finstitutions/payments/request_v2.py @@ -6,7 +6,7 @@ DATE: - Monday, 16th Dec., 2024 + Saturday, 28th Dec., 2024 OBJECTIVE: @@ -161,29 +161,36 @@ async def request_payment( http_code = HttpCodes.UNAUTHORIZED ) - # ┳┓ ┏┓ - # ┣┫┏┓┏┓┓┏┏┓┏╋ ┃┃┏┓┓┏┏┳┓┏┓┏┓╋ - # ┛┗┗ ┗┫┗┻┗ ┛┗ ┣┛┗┻┗┫┛┗┗┗ ┛┗┗ - # ┗ ┛ - # Get the token from the token key: - auth_token = await current_app.payment_controller.get_token_from_key( - mongo_conn = current_app.data_mongo, + auth_token = await current_app.payments_controller.get_token_from_key( + mongo_data_conn = current_app.data_mongo, token_key = inbound_data.tokenKey ) if auth_token is None: return ResponseModel( status_code = StatusCodes.FAILED, http_code = HttpCodes.UNAUTHORIZED, - message = f"no such token key" + message = f"No such token key." ) - # Make the request with the fetched token: - response = await current_app.payment_controller.request_payment( - mongo_conn = current_app.data_mongo, - http_client = current_app.http_client, - auth_token = auth_token, - user_info = CoreUserInfoModel(**kwargs["session_info"]), - payment_request = inbound_data + # ┳┓ ┏┓ + # ┣┫┏┓┏┓┓┏┏┓┏╋ ┃┃┏┓┓┏┏┳┓┏┓┏┓╋ + # ┛┗┗ ┗┫┗┻┗ ┛┗ ┣┛┗┻┗┫┛┗┗┗ ┛┗┗ + # ┗ ┛ + + # For Safaricom's M-Pesa Express client: + if auth_token.client == "safaricomMPesaExpress": + response = await current_app.safaricom_mpesa_express_controller.request_payment( + mongo_data_conn = current_app.data_mongo, + auth_token = auth_token, + user_info = CoreUserInfoModel(**kwargs["session_info"]), + payment_request = inbound_data + ) + + # Invalid client: + else: return ResponseModel( + status_code = StatusCodes.FAILED, + http_code = HttpCodes.BAD_REQUEST, + message = f"Invalid/unimplemented client '{auth_token.client}'" ) # ┳┓ diff --git a/api/blueprints/finstitutions/payments/tags.py b/api/blueprints/finstitutions/payments/tags_v2.py similarity index 93% rename from api/blueprints/finstitutions/payments/tags.py rename to api/blueprints/finstitutions/payments/tags_v2.py index 1ed5d6e..688f979 100644 --- a/api/blueprints/finstitutions/payments/tags.py +++ b/api/blueprints/finstitutions/payments/tags_v2.py @@ -6,7 +6,7 @@ DATE: - Wednesday, 18th Dec., 2024 + Saturday, 28th Dec., 2024 OBJECTIVE: @@ -176,10 +176,15 @@ async def update_payment_record_tags( # ┻ ┗ ┗┗┛┗ ┛┗┗ ┗┗┛┛ ┗┻ # Get the payment record: - record = await current_app.payment_controller.get_payment( - mongo_conn = current_app.data_mongo, + record = await current_app.payments_controller.get_payment( + mongo_data_conn = current_app.data_mongo, payment_id = inbound_data.paymentId ) + if not record: return ResponseModel( + status_code = StatusCodes.FAILED, + http_code = HttpCodes.NOT_FOUND, + message = "No matching payment record found." + ) # ┏┓ ┓ • ┏┓┓ ┓ # ┃┃┓┏┏┏┓┏┓┏┓┏┣┓┓┏┓ ┃ ┣┓┏┓┏┃┏ @@ -188,13 +193,13 @@ async def update_payment_record_tags( # We check if the token that was used to fetch the mail is owned by this user: if not await token_check.is_authorized( - mongo_conn = current_app.data_mongo, + mongo_data_conn = current_app.data_mongo, user_info = CoreUserInfoModel(**kwargs["session_info"]), token_ids = [record.tokenId] ): return ResponseModel( status_code = StatusCodes.FAILED, http_code = HttpCodes.UNAUTHORIZED, - message = "The record does not belong to this user." + message = "The payment record does not belong to this user." ) # ┳┳ ┓ ┳┓ ┓ @@ -203,8 +208,8 @@ async def update_payment_record_tags( # ┛ # Update the record: - success = await current_app.payment_controller.update_tags( - mongo_conn = current_app.data_mongo, + success = await current_app.payments_controller.update_tags( + mongo_data_conn = current_app.data_mongo, payment_id = inbound_data.paymentId, unset_tags = inbound_data.unsetTags, set_tags = inbound_data.setTags diff --git a/api/helpers/user/token_check.py b/api/helpers/user/token_check.py index 08647b0..6be94e4 100644 --- a/api/helpers/user/token_check.py +++ b/api/helpers/user/token_check.py @@ -88,14 +88,14 @@ from models.core.user import CoreUserInfoModel async def is_authorized( - mongo_conn: AsyncMongo, + mongo_data_conn: AsyncMongo, user_info: CoreUserInfoModel | dict, token_ids: ObjectId | str | List[ObjectId | str] ) -> bool: """ To verify if the token id is owned by the user (who will be initially identified from his session token). - :param mongo_conn: The instance of the MongoDB connector to use to perform this action. + :param mongo_data_conn: The instance of the MongoDB connector to use to perform this action. :param user_info: The user trying to access some service. :param token_ids: One or more token ids that the user is claiming to own. :return: True if the user is authorized to use this token, else False. @@ -113,7 +113,7 @@ async def is_authorized( # Fetch the auth tokens from the database: # Retrieve the document of the token id: auth_tokens = await current_app.core_auth_token_controller.get_tokens_from_ids( - mongo_data_conn = mongo_conn, + mongo_data_conn = mongo_data_conn, token_ids = [ObjectId(t) for t in token_ids] ) @@ -131,21 +131,21 @@ async def is_authorized( async def is_not_authorized( - mongo_conn: AsyncMongo, + mongo_data_conn: AsyncMongo, user_info: CoreUserInfoModel, token_ids: ObjectId | str | List[ObjectId | str] ) -> bool: """ Just a wrapper around the above function to improve readability. - :param mongo_conn: The instance of the MongoDB connector to use to perform this action. + :param mongo_data_conn: The instance of the MongoDB connector to use to perform this action. :param user_info: The user trying to access some service. :param token_ids: One or more token ids that the user is claiming to own. :return: True if the user is authorized to use this token, else False. """ authorized = await is_authorized( - mongo_conn = mongo_conn, + mongo_data_conn = mongo_data_conn, user_info = user_info, token_ids = token_ids ) diff --git a/api/main.py b/api/main.py index 62643ba..fc04967 100644 --- a/api/main.py +++ b/api/main.py @@ -71,7 +71,7 @@ from controllers.core.payment import CorePaymentController # API Controller Models: from controllers.api.mail import MailController # from controllers.api.sms import SMSController -from controllers.api.payment import PaymentController +# from controllers.api.payment import PaymentController # Controllers V2: from controllers_v2.core.auth_token import CoreAuthTokenController @@ -82,6 +82,9 @@ from controllers_v2.message.sms.savvy_bulk_sms_kenya import SavvyBulkSMSKenyaCon # --- from controllers_v2.finstitutions.trading.all_trading import AllTradingController from controllers_v2.finstitutions.trading.zerodha_kite import ZerodhaKiteTradingController +# --- +from controllers_v2.finstitutions.payments.all_payments import AllPaymentsController +from controllers_v2.finstitutions.payments.safaricom_mpesa_express import SafaricomMPesaExpressPaymentsController # To make REST API calls: import httpx @@ -112,12 +115,12 @@ from api.blueprints.sms.tags import sms_update_tags_bp from api.blueprints.software.auth import sw_auth_bp # Finstitutions / Payment Blueprints: -from api.blueprints.finstitutions.payments.auth import pg_auth_bp -from api.blueprints.finstitutions.payments.request import pg_request_bp -from api.blueprints.finstitutions.payments.callback import pg_callback_bp -from api.blueprints.finstitutions.payments.list import pg_list_bp -from api.blueprints.finstitutions.payments.get import pg_get_bp -from api.blueprints.finstitutions.payments.tags import pg_tags_update_bp +from api.blueprints.finstitutions.payments.auth_v2 import pg_auth_bp +from api.blueprints.finstitutions.payments.request_v2 import pg_request_bp +from api.blueprints.finstitutions.payments.callback_v2 import pg_callback_bp +from api.blueprints.finstitutions.payments.list_v2 import pg_list_bp +from api.blueprints.finstitutions.payments.get_v2 import pg_get_bp +from api.blueprints.finstitutions.payments.tags_v2 import pg_tags_update_bp # Finstitutions / Trading Blueprints: from api.blueprints.finstitutions.trading.oauth.request import trading_oauth_request_bp @@ -401,14 +404,14 @@ async def app_startup(**kwargs): debug_prefix = "Message (CM) | ", debug_only_errors = True ) - current_app.core_payment_controller = CorePaymentController( - cache = current_app.module_cache, - alert_url = current_app.script_data["alerts"]["url"], - http_client = current_app.http_client, - debug = enable_debugging, - debug_prefix = "Pymnt. (CM) | ", - debug_only_errors = True - ) + # current_app.core_payment_controller = CorePaymentController( + # cache = current_app.module_cache, + # alert_url = current_app.script_data["alerts"]["url"], + # http_client = current_app.http_client, + # debug = enable_debugging, + # debug_prefix = "Pymnt. (CM) | ", + # debug_only_errors = True + # ) # ┏┓┏┓┳ ┏┓ ┓┓ # ┣┫┃┃┃ ┃ ┏┓┏┓╋┏┓┏┓┃┃┏┓┏┓┏ @@ -416,7 +419,7 @@ async def app_startup(**kwargs): current_app.mail_controller = MailController() # current_app.sms_controller = SMSController() - current_app.payment_controller = PaymentController() + # current_app.payment_controller = PaymentController() # ┏┓ ┓┓ ┓┏┏┓ # ┃ ┏┓┏┓╋┏┓┏┓┃┃┏┓┏┓┏ ┃┃┏┛ @@ -464,6 +467,20 @@ async def app_startup(**kwargs): debug = enable_debugging ) + # Finstitutions / Payments Controllers: + current_app.payments_controller = AllPaymentsController( + cache = current_app.module_cache, + http_client = current_app.http_client, + alert_url = current_app.script_data["alerts"]["url"], + debug = enable_debugging + ) + current_app.safaricom_mpesa_express_controller = SafaricomMPesaExpressPaymentsController( + cache = current_app.module_cache, + http_client = current_app.http_client, + alert_url = current_app.script_data["alerts"]["url"], + debug = enable_debugging + ) + # ┏┓ ┓ ┏┓┓• # ┃ ┏┓┏┓┏┓┏┓┏╋┏┓┏┓┏ ┏┓┏┓┏┫ ┃ ┃┓┏┓┏┓╋┏ # ┗┛┗┛┛┗┛┗┗ ┗┗┗┛┛ ┛ ┗┻┛┗┗┻ ┗┛┗┗┗ ┛┗┗┛ diff --git a/controllers_v2/finstitutions/payments/__init__.py b/controllers_v2/finstitutions/payments/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/controllers_v2/finstitutions/payments/all_payments.py b/controllers_v2/finstitutions/payments/all_payments.py new file mode 100644 index 0000000..ba90138 --- /dev/null +++ b/controllers_v2/finstitutions/payments/all_payments.py @@ -0,0 +1,186 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Saturday, 28th Dec., 2024 + + OBJECTIVE: + + To handle all common actions related to payments from one place. This includes cases where you don't yet know + the third-party client or it doesn't matter who the third-party client is. For example, take those cases when + you just need to fetch one record about a payment. + + REFERENCES: + + N/A + + DOWNLOADS: + + N/A + +""" + + +# ***************************************************************************************************************** +# ***** **** +# *** IMPORT *** +# ***** **** +# ***************************************************************************************************************** + + +# To make sibling directories accessible for imports: +import sys +sys.path.append(".") +sys.path.append("..") + +# My async utils: +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.finstitutions.payments.base import PaymentsController + +# Models: +from models.core.user import CoreUserInfoModel +from models.core.auth_token import CoreAuthTokenModel +from models.api.finstitutions.payments.request import PGPaymentRequestData, PaymentRequestOneResult + +# To work with datatypes: +from typing import List, Any + +# To make HTTP requests: +import httpx + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** CLASSES *** +# ***** **** +# ***************************************************************************************************************** + + +class AllPaymentsController(PaymentsController): + + # ┏┓ + # ┃ ┏┓┏┓┏╋┏┓┓┏┏╋┏┓┏┓ + # ┗┛┗┛┛┗┛┗┛ ┗┻┗┗┗┛┛ + + def __init__( + self, + cache: AsyncRedisCache = None, + http_client: httpx.AsyncClient = None, + alert_url: str = None, + debug: bool = True, + debug_prefix: str = "All Payments (C) | ", + debug_only_errors: bool = True + ): + + """ + This is the foundational controller for all trading services. Use this for any smaller common tasks where you + may not know the exact client beforehand. + :param cache: The object to use for caching results from database calls. + :param http_client: The HTTP client + :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. + """ + + # Invoke the parent's constructor: + super().__init__( + cache = cache, + alert_url = alert_url, + http_client = http_client, + base_filter = None, + debug = debug, + debug_prefix = debug_prefix, + debug_only_errors = debug_only_errors + ) + + # ┳┓ ┏┓ + # ┣┫┏┓┏┓┓┏┏┓┏╋ ┃┃┏┓┓┏┏┳┓┏┓┏┓╋┏ + # ┛┗┗ ┗┫┗┻┗ ┛┗ ┣┛┗┻┗┫┛┗┗┗ ┛┗┗┛ + # ┗ ┛ + + async def request_payment( + self, + mongo_data_conn: AsyncMongo, + auth_token: CoreAuthTokenModel, + user_info: CoreUserInfoModel, + payment_request: PGPaymentRequestData, + ) -> PaymentRequestOneResult: + + """ + To request payment from someone through a payment gateway. + :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. + """ + + raise NotImplementedError + + async def handle_payment_callback( + self, + mongo_data_conn: AsyncMongo, + inbound_data: dict + ): + + """ + 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 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. + + :return: ?? + """ + + raise NotImplementedError + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + pass diff --git a/controllers_v2/finstitutions/payments/base.py b/controllers_v2/finstitutions/payments/base.py new file mode 100644 index 0000000..92fe020 --- /dev/null +++ b/controllers_v2/finstitutions/payments/base.py @@ -0,0 +1,515 @@ +""" + + 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("..") + +# My async utils: +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 + +# 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) + + # ┳┳ ┓ ┏┓ + # ┃┃┏┓┏┫┏┓╋┏┓ ┃┃┏┓┓┏┏┳┓┏┓┏┓╋┏ + # ┗┛┣┛┗┻┗┻┗┗ ┣┛┗┻┗┫┛┗┗┗ ┛┗┗┛ + # ┛ ┛ + + # We don't support updating payments themselves, + # but we will allow updating fields like tags, adding events, etc. + + 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, + mongo_data_conn: AsyncMongo, + auth_token: CoreAuthTokenModel, + user_info: CoreUserInfoModel, + payment_request: PGPaymentRequestData, + ) -> PaymentRequestOneResult: + + """ + To request payment from someone through a payment gateway. + :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, + mongo_data_conn: AsyncMongo, + inbound_data: dict, + inbound_headers: dict + ) -> None: + + """ + 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 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: ?? + """ + + pass + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + pass diff --git a/controllers_v2/finstitutions/payments/safaricom_mpesa_express.py b/controllers_v2/finstitutions/payments/safaricom_mpesa_express.py new file mode 100644 index 0000000..bd845f5 --- /dev/null +++ b/controllers_v2/finstitutions/payments/safaricom_mpesa_express.py @@ -0,0 +1,356 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Saturday, 28th Dec., 2024 + + OBJECTIVE: + + To handle all interactions with Safaricom's M-Pesa Express payment gateway from one place. + + REFERENCES: + + 01. Official Documentation: https://developer.safaricom.co.ke/APIs/MpesaExpressSimulate + + DOWNLOADS: + + N/A + +""" + + +# ***************************************************************************************************************** +# ***** **** +# *** IMPORT *** +# ***** **** +# ***************************************************************************************************************** + + +# To make sibling directories accessible for imports: +import sys +sys.path.append(".") +sys.path.append("..") + +# My async utils: +from utils_v2.string import json +from utils_v2.string import regex +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.finstitutions.payments.base import PaymentsController + +# 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 +) + +# Payment Client: +from utils_v2.payments.safaricom.models.auth import MPesaExpressAuthorization +from utils_v2.payments.safaricom.controllers.m_pesa_express import SafaricomMPesaExpress + +# To work with datatypes: +from typing import List, Any + +# To make HTTP requests: +import httpx + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** CLASSES *** +# ***** **** +# ***************************************************************************************************************** + + +class SafaricomMPesaExpressPaymentsController(PaymentsController): + + # ┏┓ + # ┃ ┏┓┏┓┏╋┏┓┓┏┏╋┏┓┏┓ + # ┗┛┗┛┛┗┛┗┛ ┗┻┗┗┗┛┛ + + def __init__( + self, + cache: AsyncRedisCache = None, + http_client: httpx.AsyncClient = None, + alert_url: str = None, + debug: bool = True, + debug_prefix: str = "Sfrcm. M-Pesa Exp. (C) | ", + debug_only_errors: bool = True + ): + + """ + This is the foundational controller for all trading services. Use this for any smaller common tasks where you + may not know the exact client beforehand. + :param cache: The object to use for caching results from database calls. + :param http_client: The HTTP client + :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. + """ + + # Invoke the parent's constructor: + super().__init__( + cache = cache, + alert_url = alert_url, + http_client = http_client, + base_filter = None, + debug = debug, + debug_prefix = debug_prefix, + debug_only_errors = debug_only_errors + ) + + # ┓┏ ┓ + # ┣┫┏┓┃┏┓┏┓┏┓┏ + # ┛┗┗ ┗┣┛┗ ┛ ┛ + # ┛ + + @staticmethod + def clean_phone_no(value: str) -> str: + + """ + To clean-up input Kenyan phone nos. + :param value: The phone no. to clean, provided as a string. + :return: The cleaned phone no. + """ + + return regex.replace( + text = value, + pattern = r"[^0-9]", + substitute_text = "" + ) + + # ┳┓ ┏┓ + # ┣┫┏┓┏┓┓┏┏┓┏╋ ┃┃┏┓┓┏┏┳┓┏┓┏┓╋┏ + # ┛┗┗ ┗┫┗┻┗ ┛┗ ┣┛┗┻┗┫┛┗┗┗ ┛┗┗┛ + # ┗ ┛ + + async def request_payment( + self, + mongo_data_conn: AsyncMongo, + auth_token: CoreAuthTokenModel, + user_info: CoreUserInfoModel, + payment_request: PGPaymentRequestData + ) -> PaymentRequestOneResult: + + """ + To request payment from someone through a payment gateway. + :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. + """ + + # Start by assuming failure: + result = PaymentRequestOneResult() + + # Initialize the payment request by creating a placeholder record in the database: + payment_id = await self.init_payment( + mongo_data_conn = mongo_data_conn, + auth_token = auth_token, + user_info = user_info, + payment_request = payment_request, + tags = ["Payment", "Safaricom", "M-Pesa Express", "Kenya"] + ) + + # Initialize the third-party 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 = self._http_client + ) + + # Make the payment request: + client_response = await client.request_payment( + amount = payment_request.amount, + party_a = self.clean_phone_no( + payment_request.customerNo + ) if isinstance(payment_request.customerNo, str) else payment_request.customerNo, + type = "CustomerPayBillOnline", + reference = str(payment_id), + description = payment_request.description, + callback_url = f"https://api.thecaoffice.com/finstitutions/payments/callback/safaricom/mpesaexpress", + payer_no = self.clean_phone_no( + payment_request.payerNo + ) if isinstance(payment_request.payerNo, str) else payment_request.payerNo, + party_b = auth_token.auth["businessShortCode"] + ) + + # Add this event to the payment's document: + event_note_success = await self.add_event_by_payment_id( + mongo_data_conn = mongo_data_conn, + payment_id = payment_id, + event = PaymentEvent( + paymentStatus = "initiated" if client_response.success else "initFailed", + message = f"PG: {client_response.message}", + initByPG = False, + httpCode = client_response.httpCode, + headers = await client_response.get_headers(), + payload = await client_response.get_json() + ), + client_reference_id = client_response.referenceId + ) + + # Done here: + result.success = client_response.success and event_note_success + result.message = " ".join([ + "Payment requested successfully." if client_response.success + else f"Payment request FAILED (PG: '{client_response.message}').", + " " if event_note_success + else "Event noting FAILED.", + ]).strip() + return result + + async def handle_payment_callback( + self, + 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 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. + """ + + # Start by assuming failure: + result = PaymentCallbackResult() + + # Map out the documented codes provided by the payment gateway. + # URL: https://developer.safaricom.co.ke/APIs/MpesaExpressSimulate + code_map = { + 0: { + "status": "settled", + "message": "Payment successful :)" + }, # ... Success + 1037: { + "status": "failed", + "message": "The payment gateway could not reach your customer." + }, # ... DS Timeout. User could not be reached. + 1025: { + "status": "failed", + "message": "There was a system error in the payment gateway (1025)." + }, # ... System error while trying to send the push request. + 9999: { + "status": "failed", + "message": "There was a system error in the payment gateway (9999)." + }, # ... System error while trying to send the push request. + 1032: { + "status": "rejected", + "message": "Your customer declined the payment request." + }, # ... Request Cancelled by the user. + 1: { + "status": "failed", + "message": "Your customer has insufficient balance." + }, # ... The user has insufficient balance. + 2001: { + "status": "failed", + "message": "The payment gateway says your credentials are invalid." + }, # ... Invalid credentials of the initiator. + 1019: { + "status": "failed", + "message": "The transaction expired before your customer processed it." + }, # ... Transaction expired. + 1001: { + "status": "failed", + "message": "Your customer is already in the middle of some transaction on the payment gateway." + }, # ... The payer is already making some transaction. + } + + # Figure out which of the above codes is relevant to you: + pg_reference_id = inbound_data["Body"]["stkCallback"]["CheckoutRequestID"] + pg_result_code = int(inbound_data["Body"]["stkCallback"]["ResultCode"]) + pg_result_desc = inbound_data["Body"]["stkCallback"]["ResultDesc"] + relevant_code = code_map.get( + pg_result_code, + { + "status": "unknown", + "message": f"Unknown code '{pg_result_code}' from the payment gateway. PG: '{pg_result_desc}'" + } + ) + + # For now, we just insert the event into the record: + event_note_success = await self.add_event_by_client_reference_id( + mongo_data_conn = mongo_data_conn, + event = PaymentEvent( + paymentStatus = relevant_code["status"], + message = relevant_code["message"], + initByPG = True, + ipAddr = inbound_headers["Remote-IP"], + httpCode = None, + headers = inbound_headers, + payload = inbound_data + ), + client_reference_id = pg_reference_id + ) + + # Done here: + if event_note_success: + result.success = True + result.message = f"Payment event noted." + else: + result.success = False + result.message = f"Payment event NOT noted." + return result + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + pass diff --git a/models/api/finstitutions/payments/request.py b/models/api/finstitutions/payments/request.py index 56d8be3..7db3496 100644 --- a/models/api/finstitutions/payments/request.py +++ b/models/api/finstitutions/payments/request.py @@ -230,7 +230,31 @@ class PGPaymentRequestData(BaseModel): class PaymentRequestOneResult(BaseModel): success: bool = Field( - description = "whether, or not, the sms was successfully sent", + description = "whether, or not, the payment was successfully requested", + default = False + ) + + message: str | None = Field( + description = "a brief message to summarize the result of the process", + default = None + ) + + # ┏┓ ┏• + # ┃ ┏┓┏┓╋┓┏┓ + # ┗┛┗┛┛┗┛┗┗┫ + # ┛ + + class Config: + extra = "forbid" + + +# --------------------------------------------------------------------------------------------------------------------- + + +class PaymentCallbackResult(BaseModel): + + success: bool = Field( + description = "whether, or not, the payment event was noted", default = False ) diff --git a/models/core/payment.py b/models/core/payment.py index cfb8197..8197e12 100644 --- a/models/core/payment.py +++ b/models/core/payment.py @@ -216,7 +216,8 @@ class CorePaymentModel(BaseModel): description = "the id of the document in mongodb that holds this information", frozen = True, default = None, - alias = "_id" + alias = "_id", + exclude = True ) user: CoreUserInfoModel = Field( diff --git a/models/finstitutions/trading/ticks.py b/models/finstitutions/trading/ticks.py index 0929b2c..162924d 100644 --- a/models/finstitutions/trading/ticks.py +++ b/models/finstitutions/trading/ticks.py @@ -155,6 +155,10 @@ class TradingTick(BaseModel): description = "the symbol of the instrument" ) + name: str = Field( + description = "the name of the co./underlying" + ) + exchange: Literal["NSE", "NFO", "BSE", "BFO", "MCX", "CDS", "BCD"] = Field( description = "the exchange on which this instrument is traded", frozen = True @@ -357,6 +361,7 @@ class TradingTick(BaseModel): "segment": self.segment, "type": self.type, "symbol": self.symbol, + "name": self.name, "expiry": date_time.to_timezone( self.expiryTs, timezone = self.expiryTz @@ -398,6 +403,7 @@ class TradingTick(BaseModel): modelled_ticks.append( TradingTick( symbol = tick_lookup["symbol"], + name = tick_lookup["name"], exchange = tick_lookup["exchange"], exchangeToken = tick_lookup["exchangeToken"], broker = "zerodhaKite", diff --git a/playground/socketio/kafka_passthrough.py b/playground/socketio/kafka_passthrough.py new file mode 100644 index 0000000..98062df --- /dev/null +++ b/playground/socketio/kafka_passthrough.py @@ -0,0 +1,154 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Saturday, 28th Dec., 2024 + + OBJECTIVE: + + To provide a quick way to test out passthrough messages over Kafka. + + REFERENCES: + + N/A + + DOWNLOADS: + + N/A + +""" + + +# ***************************************************************************************************************** +# ***** **** +# *** IMPORT *** +# ***** **** +# ***************************************************************************************************************** + + +# To make sibling directories accessible for imports: +import sys +sys.path.append(".") +sys.path.append("..") + +# System-level: +import os + +# Utils: +from utils_v2.string import json +from utils_v2.system import files +from utils_v2.date_time import date_time +from utils_v2.queue.async_kafka import ProducerKafka, ConsumerKafka, get_ssl_context + +# For async activities: +import asyncio + +# Common: +from shared import constants + +# For random choices: +import random + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + # Define the test params: + TOPIC = "socket-io-bcast" + BOOTSTRAP_SERVERS = "del.ditscentre.in:9092" + cwd = files.get_cwd() + pdir = files.get_parent_directory(cwd, depth = 2) + SSL_CONTEXT = ssl_context = get_ssl_context( + ca_file = os.path.join(pdir, "creds", "kafka", "cert_authority.pem"), + cert_file = os.path.join(pdir, "creds", "kafka", "fullchain.pem"), + key_file = os.path.join(pdir, "creds", "kafka", "privkey.pem") + ) + + async def keep_producing(): + + # Create the producer: + my_producer = ProducerKafka( + topic = TOPIC, + bootstrap_servers = BOOTSTRAP_SERVERS, + security_protocol = "SSL", + ssl_context = SSL_CONTEXT + ) + + # Create a message for the producer to produce: + strategy_message = { + "to": "FC9O3N75Ax7B1v4NAAAD", + "event": "Strategy", + "namespace": "/finstitutions/trading", + "data": { + "message": "Your strategy ABC says buy XYZ.", + "playSound": True + } + } + + corporate_action_message = { + "to": "FC9O3N75Ax7B1v4NAAAD", + "event": "Corporate Action", + "namespace": "/finstitutions/trading", + "data": { + "message": "Stock ABC has a corporate action tomorrow.", + "playSound": True, + "date": "29-Dec-2024", + "action": "Extraordinary Board Meeting" + } + } + + # Keep sending the message in intervals: + while True: + producer_message = random.choice([ + strategy_message, + corporate_action_message + ]) + success = await my_producer.produce(producer_message) + print("PRODUCED:", success) + await asyncio.sleep(1.0) + + async def main(): + await asyncio.gather(*[keep_producing()]) + + + asyncio.run(main()) diff --git a/playground/socketio/to_kafka.py b/playground/socketio/to_kafka.py index ab4c6ea..cc267c3 100644 --- a/playground/socketio/to_kafka.py +++ b/playground/socketio/to_kafka.py @@ -130,6 +130,8 @@ def to_kafka(tick: TradingTick) -> bool: success = False summary = tick.summary + print(json.to_string(summary)) + print(json.to_string(tick.model_dump(), default=str)) summary["messageType"] = "ticks" success = kafka_producer.produce(value = summary) if not success: @@ -203,8 +205,15 @@ def main(): # instruments += kite.instruments(exchange = "BCD") # # Pick the instruments of interest: - instruments = [TradingSymbol.from_zerodha_kite(i) for i in instruments[:1000]] - # instruments = [TradingSymbol.from_zerodha_kite(i) for i in instruments if i["tradingsymbol"] in symbols_of_interest] + # instruments = [TradingSymbol.from_zerodha_kite(i) for i in instruments[:1000]] + # instruments = [ + # TradingSymbol.from_zerodha_kite(i) for i in instruments + # if i["tradingsymbol"] in symbols_of_interest or i["name"] in symbols_of_interest + # ] + instruments = [ + TradingSymbol.from_zerodha_kite(i) for i in instruments + if i["instrument_token"] in [109760007] + ] # Create the lookup: for i in instruments: diff --git a/wsio/finstitutions/trading/main.py b/wsio/finstitutions/trading/main.py index 307ced4..ff44ad3 100644 --- a/wsio/finstitutions/trading/main.py +++ b/wsio/finstitutions/trading/main.py @@ -106,8 +106,11 @@ EVENT_TICKS = "ticks" # ***************************************************************************************************************** +# For locking user-noting operations: +lock = asyncio.Semaphore(1) + # Session-awareness: -pass +connected_clients = {} # Script-local: flags = { @@ -123,13 +126,20 @@ flags = { @sio.on(event = EVENT_CONNECT, namespace = NAMESPACE_MODULE) -async def handle_connect(sid, environ): +async def handle_connect(sid, environ) -> bool: # Start the common background processes: if not flags.get("initDone"): flags["initDone"] = True asyncio.create_task(init()) + # Note down user changes: + async with lock: + connected_clients[sid] = { + "user": None, + "rooms": [] + } + # Allow/reject requests: printer(sid) print("ENVIRON:", json.to_string(environ, default = str)) @@ -140,7 +150,7 @@ async def handle_connect(sid, environ): @sio.on(event = EVENT_DISCONNECT, namespace = NAMESPACE_MODULE) -async def handle_disconnect(sid, reason): +async def handle_disconnect(sid, reason) -> None: printer(sid, reason) @@ -148,7 +158,7 @@ async def handle_disconnect(sid, reason): @sio.on(event = EVENT_ECHO, namespace = NAMESPACE_MODULE) -async def handle_echo(sid, data): +async def handle_echo(sid, data) -> None: """ For testing. This is a quick way to check if the module is up. @@ -168,6 +178,88 @@ async def handle_echo(sid, data): # --------------------------------------------------------------------------------------------------------------------- +async def send_passthrough( + to: str | List[str], + event: str, + namespace: str, + data: dict | list +) -> None: + + """ + To send out the arbitrary passthrough message + :param to: The recipient of the message. This can be set to the 'sid' of a client to address only that client, or to + any custom room created by the application to address all the clients in that room, or to a list of custom + room names. If null, the event is broadcasted to all connected clients. + :param event: Any name for the event that the recipients are listening to. The strings 'connect', 'disconnect', and + 'message' are reserved. Everything else is fair game. + :param namespace: The namespace (path) to send the data to. + :param data: The data to send to the target recipients. + :return: None. + """ + + # Send out the event: + try: await sio.emit( + event = event, + data = data, + to = to, + namespace = namespace + ) + except Exception as exception: + printer(exception) + + +# --------------------------------------------------------------------------------------------------------------------- + + +async def passthrough_from_kafka( + consumer: ConsumerKafka, + fetch_count: int = 100, + fetch_timeout: float = 1.0 +) -> None: + + """ + This function must run in the background forever and just keep listening for any passthrough messages from the + backend. The backend message must give the following kind of JSON: + { + "to": , + "event": , + "namespace": , + "data": + } + :param consumer: The preconfigured Kafka consumer that can listen for ticks in asynchronous mode. + :param fetch_count: How many messages to consume in one go. + :param fetch_timeout: How long to wait (in seconds) while consuming messages from Kafka. + :return: None + """ + + # Do the next part infinitely: + while True: + + # Get messages form Kafka: + messages = await consumer.consume( + count = fetch_count, + timeout = fetch_timeout + ) + + # If there are no updates to give: + if not messages: continue + + # Each message is a passthrough to be sent to the connected clients: + tasks = [ + send_passthrough( + to = message["value"].get("to", None), + event = message["value"].get("event", None), + namespace = message["value"].get("namespace", "/"), + data = message["value"].get("data", {}) + ) for message in messages + ] + results = await asyncio.gather(*tasks) + printer(len(messages)) + + +# --------------------------------------------------------------------------------------------------------------------- + + async def send_ticks(ticks: List[dict]): """ @@ -238,6 +330,14 @@ async def init(): # Start consuming ticks in the background: cwd = files.get_cwd() parent_dir = cwd + ssl_context = get_ssl_context( + ca_file = "/etc/ssl/dbu/ca.pem", + cert_file = "/etc/ssl/dbu/fullchain.pem", + key_file = "/etc/ssl/dbu/privkey.pem" + # ca_file = os.path.join(parent_dir, "creds", "kafka", "cert_authority.pem"), + # cert_file = os.path.join(parent_dir, "creds", "kafka", "fullchain.pem"), + # key_file = os.path.join(parent_dir, "creds", "kafka", "privkey.pem") + ) sio.start_background_task( ticks_from_kafka, consumer = ConsumerKafka( @@ -245,19 +345,27 @@ async def init(): # group_id = f"{SERVER_HOSTNAME}_tickers", bootstrap_servers = "del.ditscentre.in:9092", security_protocol = "SSL", - ssl_context = get_ssl_context( - ca_file = "/etc/ssl/dbu/ca.pem", - cert_file = "/etc/ssl/dbu/fullchain.pem", - key_file = "/etc/ssl/dbu/privkey.pem" - # ca_file = os.path.join(parent_dir, "creds", "kafka", "cert_authority.pem"), - # cert_file = os.path.join(parent_dir, "creds", "kafka", "fullchain.pem"), - # key_file = os.path.join(parent_dir, "creds", "kafka", "privkey.pem") - ), + ssl_context = ssl_context, auto_offset_reset = "latest" ), fetch_count = 1_250, fetch_timeout = 1.0 ) + sio.start_background_task( + passthrough_from_kafka, + consumer = ConsumerKafka( + topic = "socket-io-bcast", + # group_id = f"{SERVER_HOSTNAME}_tickers", + bootstrap_servers = "del.ditscentre.in:9092", + security_protocol = "SSL", + ssl_context = ssl_context, + auto_offset_reset = "latest" + ), + fetch_count = 100, + fetch_timeout = 1.0 + ) + + printer("Initialized.") # *****************************************************************************************************************