khushalps
6d1f731ff7
git-subtree-dir: utils_v2 git-subtree-split: d593c1ec43b43c6881c40cbef6ae7c170dc76776
270 lines
9.7 KiB
Python
270 lines
9.7 KiB
Python
"""
|
|
|
|
AUTHOR:
|
|
|
|
Khushal P Soonderji
|
|
|
|
DATE:
|
|
|
|
Wednesday, 30th Oct., 2024.
|
|
|
|
OBJECTIVE:
|
|
|
|
To provide data model(s) for describing the data needed for M-Pesa Express transactions.
|
|
|
|
REFERENCES:
|
|
|
|
1. Documentation: https://developer.safaricom.co.ke/APIs/Authorization
|
|
|
|
DOWNLOADS:
|
|
|
|
N/A
|
|
|
|
"""
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** IMPORT ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# To make sibling directories accessible for imports:
|
|
import sys
|
|
sys.path.append(".")
|
|
sys.path.append("..")
|
|
|
|
# For making data behaviour_models:
|
|
from pydantic import BaseModel, Field, field_validator, model_validator, AwareDatetime
|
|
from typing import Optional, Literal, Union, Dict, List, Any
|
|
|
|
# My utils:
|
|
from utils_v2.string import json
|
|
from utils_v2.string import regex
|
|
from utils_v2.date_time import date_time
|
|
|
|
# To work with date and time:
|
|
import datetime
|
|
import dateparser
|
|
|
|
# To make API calls:
|
|
import httpx
|
|
|
|
# For asynchronous activities:
|
|
import asyncio
|
|
|
|
# Misc:
|
|
import base64
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MACROS / ONE-TIME INIT ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# M-Pesa URLs:
|
|
M_PESA_EXPRESS_AUTH_API_URL = r"https://api.safaricom.co.ke/oauth/v1/generate"
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** VARIABLES ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** FUNCTIONS ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
class MPesaExpressAuthorization(BaseModel):
|
|
|
|
# Pass the following values from outside.
|
|
# These will set the model up.
|
|
|
|
consumerKey: str = Field(
|
|
description = "the app's consumer key given by safaricom; found in 'my apps'",
|
|
frozen = True
|
|
)
|
|
|
|
consumerSecret: str = Field(
|
|
description = "the app's consumer secret given by safaricom; found in 'my apps'",
|
|
frozen = True
|
|
)
|
|
|
|
businessShortCode: str = Field(
|
|
description = "your app's business short code; found in 'my apps'",
|
|
frozen = True
|
|
)
|
|
|
|
appPasskey: str = Field(
|
|
description = "your app's passkey; taken from human representative",
|
|
frozen = True
|
|
)
|
|
|
|
callbackUrl: str | None = Field(
|
|
description = "the url for m-pesa to inform you about a successful or failed payment",
|
|
frozen = True,
|
|
pattern = regex.REGEX_HTTPS_URL,
|
|
default = None
|
|
)
|
|
|
|
# Don't pass these values from outside,
|
|
# these will be set and used internally.
|
|
|
|
accessToken: str | None = Field(
|
|
description = "the token received after authorization",
|
|
frozen = False,
|
|
default = None
|
|
)
|
|
|
|
expiresAt: AwareDatetime = Field(
|
|
description = "the time (utc) at which the token will expire",
|
|
frozen = False,
|
|
default = date_time.as_if_timezone(
|
|
date_time.parse_date_time(0),
|
|
timezone = date_time.TIMEZONE_UTC
|
|
)
|
|
)
|
|
|
|
lastRefresh: datetime.datetime | None = Field(
|
|
description = "the time at which the token was last refreshed",
|
|
frozen = False,
|
|
default = date_time.as_if_timezone(
|
|
date_time.parse_date_time(0),
|
|
timezone = date_time.TIMEZONE_UTC
|
|
)
|
|
)
|
|
|
|
# ┏┓ ┏•
|
|
# ┃ ┏┓┏┓╋┓┏┓
|
|
# ┗┛┗┛┛┗┛┗┗┫
|
|
# ┛
|
|
|
|
class Config:
|
|
extra = "forbid"
|
|
|
|
# ┏┓ •
|
|
# ┃┃┏┓┏┓┏┓┏┓┏┓╋┓┏┓┏
|
|
# ┣┛┛ ┗┛┣┛┗ ┛ ┗┗┗ ┛
|
|
# ┛
|
|
|
|
@property
|
|
def expired(self) -> bool:
|
|
return True if date_time.get_current_utc_date_time() >= self.expiresAt else False
|
|
|
|
@property
|
|
def ttl(self):
|
|
ttl = (self.expiresAt - date_time.get_current_utc_date_time()).total_seconds()
|
|
return max(0, ttl)
|
|
|
|
# ┏┓ ┏┓
|
|
# ┃ ┓┏┏╋┏┓┏┳┓ ┣ ┓┏┏┓┏┏
|
|
# ┗┛┗┻┛┗┗┛┛┗┗ ┻ ┗┻┛┗┗┛
|
|
|
|
async def refresh(
|
|
self,
|
|
http_client: httpx.AsyncClient = None,
|
|
force_refresh: bool = False
|
|
) -> bool:
|
|
|
|
# Start by assuming failure:
|
|
token_refreshed = False
|
|
|
|
# If the token is stale or a refresh is being forced:
|
|
if self.expired or force_refresh:
|
|
|
|
# Prepare the inputs:
|
|
url = M_PESA_EXPRESS_AUTH_API_URL
|
|
key = base64.b64encode(f"{self.consumerKey}:{self.consumerSecret}".encode()).decode()
|
|
input_headers = {"Authorization": "Basic " + key}
|
|
input_params = {"grant_type": "client_credentials"}
|
|
|
|
# Make the API call:
|
|
if http_client:
|
|
api_response = await http_client.get(
|
|
url = url,
|
|
headers = input_headers,
|
|
params = input_params
|
|
)
|
|
else:
|
|
async with httpx.AsyncClient() as http_client:
|
|
api_response = await http_client.get(
|
|
url = url,
|
|
headers = input_headers,
|
|
params = input_params
|
|
)
|
|
|
|
# If the API call was successful:
|
|
if api_response.status_code in [200]:
|
|
api_json = api_response.json()
|
|
self.accessToken = api_json.get("access_token")
|
|
self.lastRefresh = date_time.get_current_utc_date_time()
|
|
self.expiresAt = self.lastRefresh + datetime.timedelta(seconds = int(0.95 * float(api_json.get("expires_in"))))
|
|
token_refreshed = True
|
|
|
|
# Done here:
|
|
return token_refreshed
|
|
|
|
async def get_access_token(
|
|
self,
|
|
http_client: httpx.AsyncClient = None,
|
|
force_refresh: bool = False
|
|
) -> str | None:
|
|
|
|
"""
|
|
If you just pick the 'accessToken' variable, you may get a stale token. Using this method will ensure that the
|
|
token is refreshed, if needed, before being handed to you.
|
|
:param http_client: A preset HTTP client to use. If not given, one will be created inside and used to make the
|
|
API call. Giving it from outside could save time in setting up a new client. You may start passing this when
|
|
you notice delays.
|
|
:param force_refresh: Use this to force a token refresh even if it hasn't expired yet.
|
|
:return: Either a valid access token or None if the refreshing fails.
|
|
"""
|
|
|
|
await self.refresh(http_client = http_client, force_refresh = force_refresh)
|
|
return self.accessToken
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MAIN PROGRAM ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
async def main():
|
|
|
|
auth = MPesaExpressAuthorization(
|
|
consumerKey = "kFiHZ3G1vCqxkQfHgMZzPvkPd5ilsJD3",
|
|
consumerSecret = "NIp2mp1V0cSEQ63G",
|
|
businessShortCode = "4092041",
|
|
appPasskey = "cf5c0f05298e63b4039c60e3fd12c2f72e1adac840d3dbd68c88a33b43dbef82",
|
|
# callbackUrl = "https://www.something.otherthing.com/my/callback/path?with=params"
|
|
)
|
|
|
|
print("AUTH:", auth.model_dump_json(indent = 4))
|
|
print("EXPIRED:", auth.expired)
|
|
print("TTL:", auth.ttl)
|
|
|
|
print("\n\nRefreshing...\n\n")
|
|
await auth.get_access_token()
|
|
|
|
print("AUTH:", auth.model_dump_json(indent = 4))
|
|
print("EXPIRED:", auth.expired)
|
|
print("TTL:", auth.ttl)
|
|
|
|
|
|
asyncio.run(main())
|