Utils: Add Cache class to manage cache

- Add pytz to requirements
parent 7b8ff999
Provides useful stuff, generally!
from datetime import datetime
from collections import OrderedDict
from typing import Callable
from typing import Optional
import json
from pytz import timezone
class LimitedSizeDict(OrderedDict):
LimitedSizeDict pops items from the first if the size of dictionary exceeds
the specified limit.
def __init__(self, *args, **kwargs):
self.size_limit = kwargs.pop('size_limit', None)
super(LimitedSizeDict, self).__init__(*args, **kwargs)
def __setitem__(self, *args, **kwargs):
super(LimitedSizeDict, self).__setitem__(*args, **kwargs)
def _check_size_limit(self):
if self.size_limit is not None:
while self.size_limit < len(self):
class Cache:
A class to manage cache with IGitt and any other external application.
The cache mechanism should be able to process raw JSON data. The response
data from external requests is stored in the cache with the API URL of the
associated IGitt object as the key and the data is stored along with its
entity tag header, which is used alongside `If-None-Match` header for
further queries using conditional requests. When an incoming webhook is
received, the timestamp of reception is cached and any queries later on the
same URL use the `If-Modified-Since` HTTP Header and the reception time
using conditional requests, since the ETag Header is no longer valid. Note
that conditional requests do not add up to rate limits on APIs.
To use IGitt's caching mechanism for external request management, simply
add the following code to your application before using IGitt.
>>> from IGitt.Utils import Cache
>>> Cache.use(read_from, write_to)
For further details follow the specific method documentation below.
__mem_store = LimitedSizeDict(size_limit=10 ** 6) # a million entries
_get = __mem_store.__getitem__
_set = __mem_store.__setitem__
def use(cls, read_from: Callable, write_to: Callable):
Connects the cache read, write functions to Cache class.
:param read_from:
The method to be called to fetch data from cache. It should be able
to receive only one parameter, key, which is used to identify an
entry uniquely in the cache.
:param write_to:
The method to be called to write data to cache. It should be able
to receive two parameters, key (used to identify the entry in
cache) and the item to be stored in cache, in the specified
respective order.
cls._get = read_from
cls._set = write_to
def validate(cls, item: dict) -> dict:
Checks if the given item has valid data and adds missing fields with
default values. Expected fields are as follows.
:type fromWebhook: bool
:type data: object
:type links: dict
:type lastUpdated: str (formatted as '%a, %d %m %Y %H:%M:%S %Z')
:type entityTag: str or None
:return: The item dictionary after validation without any missing
fields. Also removes any additional unrelated fields from
the given dictionary.
:raises: TypeError, if the field does not match the expected type.
ValueError, if the type is correct, but the value is
if 'data' not in item:
item['data'] = {}
if 'links' not in item:
item['links'] = {}
elif not isinstance(item['links'], dict):
raise TypeError("'links' field should be a dictionary, not {}"
if 'lastFetched' not in item:
item['lastFetched'] = datetime.now(
timezone('GMT')).strftime('%a, %d %m %Y %H:%M:%S %Z')
# check if the datetime format is correct and raises an exception
# if it is invalid. TypeError, if item['lastFetched'] is not a
# string and ValueError, if the format doesn't match the expected.
'%a, %d %m %Y %H:%M:%S %Z')
if 'entityTag' not in item:
item['entityTag'] = None
elif (not isinstance(item['entityTag'], str) and
item['entityTag'] != None):
raise TypeError(
"'entityTag' field should either be a string or None, not {}"
if 'fromWebhook' not in item:
item['fromWebhook'] = False
elif not isinstance(item['fromWebhook'], bool):
raise TypeError("'fromWebhook' field should be a bool, not {}"
# drop any other extra fields in the dictionary
fields = {'fromWebhook', 'entityTag', 'lastFetched', 'links', 'data'}
item = {k: v for k, v in item.items() if k in fields}
return item
def get(cls, key) -> Optional[dict]:
Retrieves the entry from cache if present, otherwise None.
return cls.validate(json.loads(cls._get(key)))
except (KeyError, TypeError):
return None
def set(cls, key, item):
Stores the entry in cache.
item = cls.validate(item)
cls._set(key, json.dumps(item))
def update(cls, key, new_value):
Updates the existing entry with new data, if present, otherwise creates
a new entry in cache.
cls.set(key, {**(cls.get(key) or {}), **new_value})
class PossiblyIncompleteDict:
......@@ -6,3 +6,4 @@ cryptography~=2.1.4
from tests import IGittTestCase
from IGitt.Utils import Cache
from IGitt.Utils import LimitedSizeDict
class CacheTestCase(IGittTestCase):
def test_LimitedSizeDict(self):
store = LimitedSizeDict(size_limit=10)
for i in range(100):
store[i] = i + 1
# assert that even after inserting large data, the dict only retains
# latest 10 entries
self.assertEqual(len(store), 10)
def test_cache_validation_entityTag(self):
with self.assertRaises(TypeError):
Cache.validate({'entityTag': 10})
def test_cache_validation_links(self):
with self.assertRaises(TypeError):
Cache.validate({'links': 10})
def test_cache_validation_lastFetched(self):
with self.assertRaises(ValueError):
Cache.validate({'lastFetched': 'Tue, 10 March 2021'})
with self.assertRaises(TypeError):
Cache.validate({'lastFetched': None})
def test_cache_validation_fromWebhook(self):
with self.assertRaises(TypeError):
Cache.validate({'fromWebhook': None})
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment