Safaricom has made it easy for developers to add M-pesa capabilities into whatever app they might be working on. Therefore in the light of that they have availed the Daraja API version 2.0 for developers to utilize.In brief the Daraja API provides quite a number of endpoints(urls) that one might want to use on their app and of course that depends on the use case, but a quick run through it has some of the following offerings.
- M-pesa Express (STK Push) What this write up is about.
- Customer To Business
- Business To Customer
Those are like the main ones but there is transaction querying, reversals, account balance and others for the full array visit Safaricom APIs documentation
On the back of that light introduction, this write up about M-pesa Express and how to integrate it to a python app. So jumping right in, this guide will to show how easy it is to integrate STK Push to your flask or django app or just basically any python app.
To get started you need to have an account on the Safaricom developer portal right here, once you have an account created, then go ahead and create an app, for that just go to MY APPS. A point to note is that you can mock the APIs right there on the Safaricom Developer portal.
Having said that lets dive in to the guide, we will need a few things from the developer portal copy and we will put them in an environment variables file in the below format. For Consumer Key, Consumer Secret and Pass Key copy them from you test app.
SAF_CONSUMER_KEY=2heA1Q3X-copy-value-from-test-app
SAF_CONSUMER_SECRET=ZubQ-copy-value-from-test-app
SAF_SHORTCODE=174379
SAF_PASS_KEY=bfb279f9aa9-copy-value-from-test-app
SAF_ACCESS_TOKEN_API=https://sandbox.safaricom.co.ke/oauth/v1/generate?grant_type=client_credentials
SAF_STK_PUSH_API=https://sandbox.safaricom.co.ke/mpesa/stkpush/v1/processrequest
SAF_STK_PUSH_QUERY_API=https://sandbox.safaricom.co.ke/mpesa/stkpushquery/v1/query
Now that we have gotten that out of the way, we will make a python class that will handle authorization request and carry out push request, handle transaction querying. I am going to name the file mpesa_handler.py you can name it whatever you want but it is always good to have a descriptive name.
Lets start with the imports they are fairly descriptive and easy to understand. Only the env bit I may want to expound on a little bit, the code is from a Django App I was testing it on, hence the from your_app.settings import env this is just an import of the environment variable from the settings file of my app.
import time
import math
import base64
import requests
from datetime import datetime
from requests.auth import HTTPBasicAuth
from <your_app>.settings import env
Then lets declare a class for handling the calls to mpesa and initialize a number of variables.
class MpesaHandler:
now = None
shortcode = None
consumer_key = None
consumer_secret = None
access_token_url = None
access_token = None
stk_push_url = None
my_callback_url = None
query_status_url = None
timestamp = None
passkey = None
def __init__(self):
""" initializing the variables for use in the class """
self.now = datetime.now()
self.shortcode = env("SAF_SHORTCODE")
self.consumer_key = env("SAF_CONSUMER_KEY")
self.consumer_secret = env("SAF_CONSUMER_SECRET")
self.access_token_url = env("SAF_ACCESS_TOKEN_API")
self.passkey = env("SAF_PASS_KEY")
self.stk_push_url = env("SAF_STK_PUSH_API")
self.query_status_url = env("SAF_STK_PUSH_QUERY_API")
self.my_callback_url = env("CALLBACK_URL")
self.password = self.generate_password()
env("any_given_str") just means get the variables from the .env file. Without wasting time we will create a few functions to handle different operations like one for authorization, another one for handling push stk and another one for querying the transaction status.
- The below function just gets authorization token from Safaricom. It will return a time bound access token which is going to be valid for one hour or 3599secs. If our authorization request is successful we will get a token then we will use that to create a header which will send in every request and also return it to other functions.
def get_mpesa_access_token(self):
""" get access token from safaricom mpesa"""
try:
res = requests.get(
self.access_token_url,auth=HTTPBasicAuth(self.consumer_key, self.consumer_secret),
)
token = res.json()['access_token']
self.headers = {"Authorization": f"Bearer {token}","Content-Type": "application/json" }
except Exception as e:
print(str(e), "error from get access token")
raise e
return token
The next function will be generate password. Every request to the various endpoints will need a password and from the function it is pretty easy to get what is going on.
def generate_password(self):
""" generate a password for the api using shortcode and passkey """
self.timestamp = self.now.strftime("%Y%m%d%H%M%S")
password_str = self.shortcode + self.passkey + self.timestamp
password_bytes = password_str.encode()
return base64.b64encode(password_bytes).decode("utf-8")
The next function is the make_stk_push, this function is where we will combine data from our app, that is the customer entered phone number, amount that the customer is supposed to pay for whatever they are paying for on your app and some other mandatory data to make a json object which is what we will send to Safaricom.
def make_stk_push(self, payload):
""" push payment request to the mpesa no."""
amount = payload['amount']
phone_number = payload['phone_number']
push_data = {
"BusinessShortCode": self.shortcode,
"Password": self.password,
"Timestamp": self.timestamp,
"TransactionType": "CustomerPayBillOnline",
"Amount": math.ceil(float(amount)),
"PartyA": phone_number,
"PartyB": self.shortcode,
"PhoneNumber": phone_number,
"CallBackURL": self.my_callback_url,
"AccountReference": "Whatever you call your app",
"TransactionDesc": "description of the transaction",
}
response = requests.post(self.stk_push_url, json=push_data, headers=self.headers)
response_data = response.json()
return response_data
The stk push will return an object like the one below, which is just a json object telling you that the push request has been successfully accepted for processing (at this point processing is just safaricom forwarding the request to the customer) by Safaricom. Also for failure it will also return a json object giving you a description of why it has failed. I would like you at this point to take note of the "CheckoutRequestID" because it will be very handy in confirming the status of the transaction using the query transaction endpoint.
{
"MerchantRequestID": "29115-34620561-1",
"CheckoutRequestID": "ws_CO_191220191020363925",
"ResponseCode": "0",
"ResponseDescription": "Success. Request accepted for processing",
"CustomerMessage": "Success. Request accepted for processing"
}
And then finally for this guide we will make a query_transaction_status(), this function will help you get the status of the transaction. With that you will be able to give feedback to the customer whether their payment has gone through successfully, while still at this point it is also worth noting that in the make_stk_push there is the "CallBackURL" variable this is a secure url that you have availed on your app that is going to recieve POST data from Safaricom on the status of the transaction, while this works it is also good to use it combination with the query transaction option which is what we are going to write below.
def query_transaction_status(self, checkout_request_id):
""" query the status of the transaction."""
query_data = {
"BusinessShortCode": self.shortcode,
"Password": self.password,
"Timestamp": self.timestamp,
"CheckoutRequestID": checkout_request_id
}
response = requests.post(self.query_status_url, json=query_data, headers=self.headers)
response_data = response.json()
return response_data
So a transaction can fail or be successful that may be due to quite a number of reasons, the customer may cancel the transaction, they might ignore the transaction, their phone might be unreachable and all those reasons may lead to transaction failing and for each option then Safaricom will provide you with details for each option while failure is also very important to know when that occurs or why it did so that you can provide a wholesome experience to the users of your application, I will not delve into the response of a failed transaction for that in full see the documentation, I will only deal with feedback for a successful transaction and the data that you will get from Safaricom in that case and for that then Safaricom will give you a feedback from the transaction in the following format for you to use or process further in your application.
{
"ResponseCode": "0",
"ResponseDescription":"The service request has been accepted successsfully",
"MerchantRequestID":"22205-34066-1",
"CheckoutRequestID":"ws_CO_13012021093521236557",
"ResultCode": "0",
"ResultDesc":"The service request is processed successfully."
}
A caveat there is a lot that I passed over while making this write up, it is not a replacement for the official documentation it is meant to make your life easier, while working with the STK Push API. Having said that our final class is going to look like this, you can improve on it, critique it or use it in your own code.
import time
import math
import base64
import requests
from datetime import datetime
from requests.auth import HTTPBasicAuth
from yourapp.settings import env
class MpesaHandler:
now = None
shortcode = None
consumer_key = None
consumer_secret = None
access_token_url = None
access_token = None
access_token_expiration = None
stk_push_url = None
my_callback_url = None
query_status_url = None
timestamp = None
passkey = None
def __init__(self):
""" initializing payment objects """
self.now = datetime.now()
self.shortcode = env("SAF_SHORTCODE")
self.consumer_key = env("SAF_CONSUMER_KEY")
self.consumer_secret = env("SAF_CONSUMER_SECRET")
self.access_token_url = env("SAF_ACCESS_TOKEN_API")
self.passkey = env("SAF_PASS_KEY")
self.stk_push_url = env("SAF_STK_PUSH_API")
self.query_status_url = env("SAF_STK_PUSH_QUERY_API")
self.my_callback_url = env("CALLBACK_URL")
self.password = self.generate_password()
try:
self.access_token = self.get_mpesa_access_token()
if self.access_token is None:
raise Exception("Request for access token failed")
else:
self.access_token_expiration = time.time() + 3599
except Exception as e:
# log this errors
print(str(e))
def get_mpesa_access_token(self):
""" get access token from safaricom mpesa"""
try:
res = requests.get(
self.access_token_url,auth=HTTPBasicAuth(self.consumer_key, self.consumer_secret),
)
token = res.json()['access_token']
self.headers = {"Authorization": f"Bearer {token}","Content-Type": "application/json" }
except Exception as e:
print(str(e), "error from get access token")
raise e
return token
def generate_password(self):
""" generate a password for the api using shortcode and passkey """
self.timestamp = self.now.strftime("%Y%m%d%H%M%S")
password_str = self.shortcode + self.passkey + self.timestamp
password_bytes = password_str.encode()
return base64.b64encode(password_bytes).decode("utf-8")
def make_stk_push(self, payload):
""" push payment request to the mpesa no."""
amount = payload['amount']
phone_number = payload['phone_number']
push_data = {
"BusinessShortCode": self.shortcode,
"Password": self.password,
"Timestamp": self.timestamp,
"TransactionType": "CustomerPayBillOnline",
"Amount": math.ceil(float(amount)),
"PartyA": phone_number,
"PartyB": self.shortcode,
"PhoneNumber": phone_number,
"CallBackURL": self.my_callback_url,
"AccountReference": "Journaling Therapy",
"TransactionDesc": "journaling transaction test",
}
response = requests.post(self.stk_push_url, json=push_data, headers=self.headers)
response_data = response.json()
return response_data
def query_transaction_status(self, checkout_request_id):
""" query the status of the transaction."""
query_data = {
"BusinessShortCode": self.shortcode,
"Password": self.password,
"Timestamp": self.timestamp,
"CheckoutRequestID": checkout_request_id
}
response = requests.post(self.query_status_url, json=query_data, headers=self.headers)
response_data = response.json()
return response_data
So to utilize this class anywhere in your code you just need to instantiate a class object like so
mpesa_handler = MpesaHandler()
Then use the instance object to access the class methods like so
response = mpesa_handler.make_stk_push(your_payload)
That is it for now, the write has been a little verbose than I intended for it to be,however I would not have it any other way. Hope it helps some one out and makes it a little bit better to work with the push api.
Thank you.
Oldest comments (0)