Refreshing and caching

Implementing the Refresh Mechanism

The process for refreshing an expired access token is identical to the initial authentication process. You must encode your ConsumerKey and ConsumerSecret using Base64 encoding. The ConsumerKey and ConsumerSecret are unique identifiers and passwords provided by Safaricom when you register your application, used to identify your application to the Daraja API. Base64 encoding is a method of converting binary data (like your key and secret) into an ASCII string format that can be safely transmitted over the internet, often used for authentication credentials.

Base64 Encoding Logic: The Daraja API expects the Authorization header for the token generation request to be formatted as Basic <Base64_Encoded_String>. The string that needs to be encoded is a combination of your ConsumerKey and ConsumerSecret, separated by a colon: ConsumerKey:ConsumerSecret.

• Example: If your ConsumerKey is ABC and your ConsumerSecret is XYZ, the string to be encoded is ABC:XYZ.
• Encoding: In Python, you would perform this encoding using the base64 library. The code base64.b64encode("ABC:XYZ".encode()).decode() first converts the string to bytes (.encode()) and then applies Base64 encoding, finally decoding the result back into a standard string.

💡 What is happening here: To get a new token, your application sends a special request to the Daraja API. This request includes your application's unique ID and secret, which are first combined and then scrambled using Base64 encoding for secure transmission.

Caching the Token

Caching is the process of storing data (in this case, the access token and its expiry information) in a temporary location. This avoids the need to repeatedly fetch the same data, which saves time and network resources.

In-Memory Caching: For a single-server application, storing the token in a global variable or as an attribute of a class instance is often sufficient. This is known as in-memory caching, where data is held directly in the application's active memory. However, if you restart your server, this in-memory cache is wiped clean, and your application will need to re-authenticate.

Distributed Caching: For applications running on multiple servers (e.g., deployed in a Kubernetes cluster, which is a system for automating deployment, scaling, and management of containerized applications), you must use a distributed cache. A distributed cache is a system that allows multiple application instances to share the same cached data. A popular choice for this is Redis, an open-source, in-memory data store. Using a distributed cache ensures that if one server instance refreshes the token, all other server instances can immediately use the new token without each needing to perform their own authentication request.

Synchronization and Concurrency: The "Thundering Herd" Problem

In a high-traffic application, you might have many parts of your code (e.g., 50 different threads or processes) trying to send payments at the exact same time. If the access token expires, all 50 of these threads might detect the expiry simultaneously. This could lead to all 50 threads attempting to call the Daraja authentication endpoint at once to get a new token. This scenario is known as the "Thundering Herd" problem. It can overwhelm the authentication server, cause delays, and potentially lead to your application being rate-limited (blocked from making too many requests in a short period).

The Solution: Locking To prevent the "Thundering Herd" problem, you must implement a Mutex (Mutual Exclusion) Lock. A Mutex Lock is a programming construct that ensures only one thread or process can access a shared resource (like the token refresh logic) at any given time.

Here's how it works:

1. Thread 1 detects an expired token and attempts to acquire the lock. It succeeds and begins the token refresh process.
2. Thread 2 (and all subsequent threads) also detects the expired token and tries to acquire the lock. It fails because Thread 1 holds the lock, so Thread 2 waits patiently.
3. Thread 1 finishes the refresh process, updates the cached token with the new one, and then releases the lock.
4. Thread 2 (now able to acquire the lock) immediately re-checks the cache. It sees that the token has already been updated by Thread 1 and is now valid. Thread 2 then proceeds with its original task without needing to make a redundant network call to refresh the token.

This locking mechanism ensures that even if many parts of your application need a new token, only one actual network request is made to the Daraja authentication server, making the process efficient and preventing overload.

💡 What is happening here: When many parts of your application simultaneously need a new token, a "Thundering Herd" can overwhelm the Daraja API. A special "lock" ensures that only one part of your application refreshes the token at a time, preventing duplicate requests and system overload.

Code Implementation Example (Python)

The following DarajaAuthenticator class encapsulates these concepts into a production-ready structure. This class manages the token's lifecycle, including proactive checks, reactive retries, and concurrency handling.

import requests
import base64
import time
import threading
import os
 
class DarajaAuthenticator:
    def __init__(self, consumer_key, consumer_secret, auth_url, api_base_url):
        self.consumer_key = consumer_key
        self.consumer_secret = consumer_secret
        self.auth_url = auth_url
        self.api_base_url = api_base_url
        self._access_token = None
        self._expiry_timestamp = 0
        self._lock = threading.Lock()
        self._buffer = 120 # 2-minute buffer
 
    def _encode_credentials(self):
        creds = f"{self.consumer_key}:{self.consumer_secret}"
        return base64.b64encode(creds.encode()).decode()
 
    def get_access_token(self):
        # Quick check without lock
        if self._access_token and time.time() < (self._expiry_timestamp - self._buffer):
            return self._access_token
 
        with self._lock:
            # Re-check after acquiring lock
            if self._access_token and time.time() < (self._expiry_timestamp - self._buffer):
                return self._access_token
            
            # Perform refresh
            headers = {"Authorization": f"Basic {self._encode_credentials()}"}
            response = requests.get(self.auth_url, headers=headers, params={"grant_type": "client_credentials"})
            response.raise_for_status()
            data = response.json()
            
            self._access_token = data["access_token"]
            self._expiry_timestamp = time.time() + int(data["expires_in"])
            return self._access_token
 
    def make_api_request(self, endpoint, method="POST", json_data=None):
        token = self.get_access_token()
        headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
        
        url = f"{self.api_base_url}{endpoint}"
        response = requests.request(method, url, json=json_data, headers=headers)
        
        if response.status_code == 401:
            # Reactive fallback: Clear token and retry
            with self._lock:
                self._access_token = None
            return self.make_api_request(endpoint, method, json_data)
            
        return response