""" AUTHOR: Khushal P Soonderji DATE: Saturday, 7th Dec., 2024 OBJECTIVE: To make payments requests from M-PESA Express. REFERENCES: 1. Simulator & Docs: https://developer.safaricom.co.ke/APIs/MpesaExpressSimulate 2. Postman Collection: https://api.postman.com/collections/4395533-1a8f1c81-0502-4f9d-8699-d45551834b7d?access_key=PMAT-01J8R72MBSHP5CJ4J9Q46TG6G9 DOWNLOADS: N/A """ # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # To make sibling directories accessible for imports: import sys sys.path.append(".") sys.path.append("..") # System-level activities: import io # My utils: from utils_v2.string import json from utils_v2.date_time import date_time # Data models: from utils_v2.payments.safaricom.models.auth import MPesaExpressAuthorization from utils_v2.payments.safaricom.models.api_call import MPesaExpressApiResponse # To make REST-ful requests: import httpx # To work with datatypes: from typing import Literal # To work with date and time: import datetime # Misc: import base64 # For debugging: from icecream import IceCreamDebugger import inspect # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** CLASSES *** # ***** **** # ***************************************************************************************************************** class SafaricomMPesaExpress: def __init__( self, auth: MPesaExpressAuthorization, http_client: httpx.AsyncClient = None, debug = True, debug_prefix = "M-Pesa Exp | ", debug_only_errors = True ): # Prepare the debugging utility: self.__debug_prefix = debug_prefix self.__printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True) if not debug: self.__printer.disable() self.__debug_only_errors = debug_only_errors # Accept the input configuration: self.__auth = auth # Accept/create an HTTP client to work with: if http_client: self.__http_client = http_client else: self.__http_client = httpx.AsyncClient( limits = httpx.Limits( max_connections = 100, # ............ Maximum number of connections allowed in the pool. max_keepalive_connections = 50, # ... Maximum number of connections that can be kept alive. ), timeout = httpx.Timeout( pool = 120.0, # .... Time to wait for a free connection from the pool. connect = 5.0, # ... Time to wait for establishing a connection to the server. write = 10.0, # .... Time to wait for sending data. read = 120.0 # ..... Time to wait for receiving data. ) ) def enable_debug(self): self.__printer.enable() def disable_debug(self): self.__printer.disable() def debug_only_errors(self): self.__debug_only_errors = True def debug_everything(self): self.__debug_only_errors = False # ┏┓┏┓┳ ┏┓ ┓┓• # ┣┫┃┃┃ ┃ ┏┓┃┃┓┏┓┏┓ # ┛┗┣┛┻ ┗┛┗┻┗┗┗┛┗┗┫ # ┛ async def __get( self, url: str, headers: dict = None, params: dict = None ) -> MPesaExpressApiResponse: """ To call an API using the GET method. :param url: The URL to call. :param headers: The headers to pass. :param params: The params to send in the query string itself. :return: A structured response that includes the raw response, the exception (if any), and so on. """ # Prepare the structure of the response: api_response = MPesaExpressApiResponse( action = inspect.stack()[1].function, url = url, method = "GET" ) try: # Make the API call: response = await self.__http_client.get( url = url, headers = headers, params = params ) # Note down the results: api_response.response = response api_response.httpCode = response.status_code api_response.message = response.reason_phrase # If something goes wrong: except Exception as exception: api_response.exception = exception api_response.message = str(exception) self.__printer(exception, api_response.url, api_response.method, headers, params) # Done here: return api_response async def __post( self, url: str, headers: dict = None, json: dict = None, data: dict = None, params: dict = None, content: str | bytes = None, files: dict = None ) -> MPesaExpressApiResponse: """ To call an API using the POST method. :param url: The URL to call. :param headers: The headers to pass. :param json: The params to send in the JSON body. :param data: The params to send in the form-data in the body. :param params: The params to send in the query string itself. :param content: The raw content to be sent in the body (typically as an octet-stream). :param files: Any file that you may want to send. :return: A structured response that includes the raw response, the exception (if any), and so on. """ # Prepare the structure of the response: api_response = MPesaExpressApiResponse( action = inspect.stack()[1].function, url = url, method = "POST" ) try: # Make the API call: response = await self.__http_client.post( url = url, headers = headers, json = json, data = data, params = params, content = content, files = files ) # Note down the results: api_response.response = response api_response.httpCode = response.status_code api_response.message = response.reason_phrase # If something goes wrong: except Exception as exception: api_response.exception = exception api_response.message = str(exception) self.__printer(exception, api_response.url, api_response.method, headers, json, data) # Done here: return api_response # ┓┏ ┓ # ┣┫┏┓┃┏┓┏┓┏┓┏ # ┛┗┗ ┗┣┛┗ ┛ ┛ # ┛ def generate_password( self, timestamp: str ) -> str: """ The password is a combination of the short code, the app's passkey, and the timestamp in base64 encoded string. DOCUMENTATION: 1. https://developer.safaricom.co.ke/APIs/Authorization :param timestamp: The time (YYYYMMDDHHmmss) at which the request is being made. :return: The base-64 encoded string that has to be used as the password. """ if not self.__debug_only_errors: self.__printer("Generating Password") if timestamp is None: timestamp = "" if self.__auth.appPasskey is None: self.__auth.appPasskey = "" if self.__auth.businessShortCode is None: self.__auth.businessShortCode = "" return base64.b64encode((self.__auth.businessShortCode + self.__auth.appPasskey + timestamp).encode()).decode() # ┏┓ # ┃┃┏┓┓┏┏┳┓┏┓┏┓╋┏ # ┣┛┗┻┗┫┛┗┗┗ ┛┗┗┛ # ┛ async def request_payment( self, amount: float | int, party_a: str, type: Literal["CustomerPayBillOnline", "CustomerBuyGoodsOnline"], reference: str, description: str, callback_url: str = None, payer_no: str = None, party_b: str = None, ) -> MPesaExpressApiResponse: """ To request a payment from a user. When this method is called, the user's phone will immediately receive a flash message with the payment details and payment options. The user then gets to choose his action. DOCUMENTATION: 1. https://developer.safaricom.co.ke/APIs/MpesaExpressSimulate :param amount: The amount in Kenyan Shillings that the user must pay. :param party_a: The phone no. that will make the payment. Needs to be a valid Safaricom no. of the format 2547xxxxxxxx and must be registered with M-Pesa. :param type: "CustomerPayBillOnline" for PayBill nos. and "CustomerBuyGoodsOnline" for Till nos. :param reference: A reference id from your system (not Safaricom's system) for you to identify this transaction. This value will be displayed to the paying customer. Can be max. of 12 characters long. :param description: A description about the payment. Can be max. of 13 characters long. :param callback_url: The URL that will receive a webhook callback when the customer either pays or declines the payment request. If not provided, the callback URL from the auth details will be used. :param payer_no: The phone no. that shall receive the payment prompt. If not provided, the value of 'party_a' will be copied here. :param party_b: The organization that receives the funds. If not provided, the Business Short Code from the auth details will be used. :return: """ # Prepare the inputs: request_ts = datetime.datetime.now().strftime("%Y%m%d%H%M%S") access_token = await self.__auth.get_access_token(http_client = self.__http_client, force_refresh = False) input_headers = {"Authorization": f"Bearer {access_token}"} input_json = { "BusinessShortCode": self.__auth.businessShortCode, "Password": self.generate_password(timestamp = request_ts), "Timestamp": request_ts, "TransactionType": type, "PartyA": party_a, "PhoneNumber": payer_no or party_a, "Amount": str(int(amount)), "PartyB": party_b or self.__auth.businessShortCode, "CallBackURL": callback_url or self.__auth.callbackUrl, "AccountReference": reference, "TransactionDesc": description[:13] if len(description) > 13 else description } # Make the API call: api_response = await self.__post( url = r"https://api.safaricom.co.ke/mpesa/stkpush/v1/processrequest", headers = input_headers, json = input_json ) # If the call failed: if api_response.httpCode == 400: api_json = await api_response.get_json() api_response.message = f"{api_json['errorCode']} -> {api_json['errorMessage']}" # If the call failed: if api_response.httpCode in [200]: api_json = await api_response.get_json() api_response.message = api_json.get("ResponseDescription", "N/A") api_response.data = api_json api_response.success = True if str(api_json.get("ResponseCode")) == "0" else False # Done here: return api_response # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": import asyncio m_pesa_auth = MPesaExpressAuthorization( consumerKey = "kFiHZ3G1vCqxkQfHgMZzPvkPd5ilsJD3", consumerSecret = "NIp2mp1V0cSEQ63G", businessShortCode = "4092041", appPasskey = "cf5c0f05298e63b4039c60e3fd12c2f72e1adac840d3dbd68c88a33b43dbef82", callbackUrl = None ) my_m_pesa = SafaricomMPesaExpress( auth = m_pesa_auth ) async def main(): # Make the request: response = await my_m_pesa.request_payment( amount = 1.00, party_a = "254748877373", type = "CustomerPayBillOnline", reference = "TestTransactionTXN12345678", description = "Some description about the payment reason...", callback_url = r"https://api.thecaoffice.com/converse/test/callback", ) # Show the response: print("SUMMARY:", response.to_markdown(), "\n\n---\n\n") if response.success: print("DATA:", json.to_string(response.data, default = str)) else: print("JSON:", json.to_string(await response.get_json(), default = str)) asyncio.run(main())