khushalps/api_internal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Resetting utils subtree.

Browse Source
This commit is contained in:
Khushal P Soonderji
2025-06-14 11:44:29 +05:30
parent d1c0ce24f4
commit dbcc5b8da5
217 changed files with 0 additions and 149788 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils_v2/payments/safaricom/controllers/m_pesa_express.py
-410
View File
@@ -1,410 +0,0 @@
"""
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.string import regex
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 not in [200]:
api_json = await api_response.get_json()
api_response.message = f"{api_json['errorCode']} -> {api_json['errorMessage']}"
# If the call failed:
else:
api_json = await api_response.get_json()
success = True if str(api_json.get("ResponseCode")) == "0" else False
api_response.message = api_json.get("ResponseDescription", "N/A")
api_response.data = api_json
api_response.success = success
api_response.referenceId = str(api_json["CheckoutRequestID"])
# 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 = "254700123007",
type = "CustomerPayBillOnline",
reference = "TestTransactionTXN12345678",
description = "Some description about the payment reason...",
# callback_url = r"https://api.thecaoffice.com/converse/test/callback",
callback_url = r"https://v2.api.bicree.com/user/callback/test",
payer_no = "254700123007",
)
# 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())