KabakiAntony

import os  # os.environ.get('variable_name')
import time
import math
import base64
import requests
from datetime import datetime
from requests.auth import HTTPBasicAuth
from yourapp.settings import env  # environment for Django or use the os option too

KabakiAntony

def query_transaction_status(self, checkout_request_id):
    query_data = {
        "BusinessShortCode": self.shortcode,
        "Password": self.password,
        "Timestamp": self.timestamp,
        "CheckoutRequestID": checkout_request_id
    }

    response = requests.post(
        self.query_status_url,

KabakiAntony

def generate_password(self):
    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")

KabakiAntony

def get_mpesa_access_token(self):
    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}",

KabakiAntony

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

KabakiAntony

# mpesa_handler.py

import time
import math
import base64
import requests
from datetime import datetime
from requests.auth import HTTPBasicAuth
from <your_app>.settings import env

KabakiAntony

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

KabakiAntony

import logging

# Create a formatter
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')

# Create a file handler and set the formatter
file_handler = logging.FileHandler('example.log')
file_handler.setFormatter(formatter)

# Create a logger and add the file handler

KabakiAntony

import requests
from threading import Thread

class PDFDownloader(Thread):
   def __init__(self, url, filename):
       super().__init__()
       self.url = url
       self.filename = filename

   def run(self):

KabakiAntony

import time
from threading import Thread

def add_two_numbers(a, b):
   return a + b

def simulate_api_call(name):
   print(f'{name} is sleeping for 3 seconds')
   time.sleep(3)
   print(f"{name} is done sleeping")