In-Memory Cache in Python

A cache is the data storage layer, usually transient in nature that could reside in memory, disk, or even on a different server. It provides fast access to data, avoids repeated computation, and significantly improves application performance.

This blog demonstrates the custom implementation of in-memory cache with each key-value pair having its own expiry. In-memory cache can be helpful in scenarios where data needs to be accessed frequently, it will save a few network calls, and hence response time is significantly small. In python, We can use the built-in dictionary to cache the data.

To use and maintain same memory space, we'll use object of a singleton class:

inmemory.py

import time

class InMemory:
	__instance = None
	
	@staticmethod
	def getInstance():
		if InMemory.__instance == None:
				InMemory()

		return InMemory.__instance

	def __init__(self):
		self.in_memory_cache = {}
		InMemory.__instance = self

	def set(self, key, value, expiry_in_secs):
		record = {
			"key":key, 
			"value":value,
			"expires_at": self.__set_expiry(expiry_in_secs)
			}
		self.in_memory_cache[key] = record

	def get(self, key):
		record = self.__get_record(key)
		val = self.__get_value(record, key)if record else None
		return val

	def multi_get(self, list_of_keys):
		response = {}
		for key in list_of_keys:
			record = self.__get_record(key)
			val = self.__get_value(record, key)if record else None
			response[key] = val
		return response

	def delete(self, key):
		record = self.__get_record(key)
		if record:
			self.in_memory_cache.pop(key,None)
		return

	def __get_value(self, record, key):
		if self.__is_expired(record['expires_at']):
			self.in_memory_cache.pop(key)
			return None

		return record.get('value')

	def __get_record(self,key):
		return self.in_memory_cache.get(key,None)

	def __is_expired(self, entry_expiry):
		return int(entry_expiry) < current_timestamp_in_secs()

	def __set_expiry(self, ttl_in_secs):
		ttl_in_secs = int(ttl_in_secs)

		if ttl_in_secs is None or ttl_in_secs < 0:
			ttl_in_secs = 0

		return int(current_timestamp_in_secs() + int(ttl_in_secs))

# Get current epoch timestamp in seconds
def current_timestamp_in_secs():
 return int(time.time())

test.py

from time import sleep
from inmemory import InMemory

# Get instance to set data into cache
instance1 = InMemory.getInstance()

instance1.set('Bob','Alice',50)
instance1.set('Peter','Spiderman',5)
instance1.set('Apple','Macbook',10)

# Get instance to get data from cache
instance2 = InMemory.getInstance()

print(instance2.get('Bob'))
print(instance2.multi_get(['Peter','Apple','Key_doesnt_exists_in_cache']))

# Waiting for Key to be expired:
sleep(7)
print(instance2.get('Peter'))

sleep(5)
print(instance2.get('Apple'))

output

Alice
{'Peter': 'Spiderman', 'Apple': 'Macbook', 'Key_doesnt_exists_in_cache': None}
None
None

Note: In the above implementation, the expired key is popped only when it is being accessed. This can be memory inefficient.  Instead of simple dictionary, other custom cache like LRU, LFU, TTL, etc can also be used as per the use case.

Make sure to checkout cachetools for some advanced custom-implemented caches.

Kartik Kapgate

Kartik Kapgate