Source code for arpoc.cache
""" Cache Module for OIDCPROXY"""
import collections
import heapq
from dataclasses import dataclass, field
from typing import Any, List
import arpoc.utils
[docs]@dataclass(order=True)
class CacheItem:
""" CacheItem -- A item in the Cache
timestamp: An Unix Timestamp in seconds since beginning of the epoch.
item: The Item
"""
timestamp: int
item: Any = field(compare=False)
[docs]class Cache(collections.UserDict):
""" The cache. On Every action (get, put) a clean up procedure is triggered """
def __init__(self) -> None:
super().__init__({})
self.__valid: List = []
[docs] def expire(self) -> None:
""" Deletes invalid entries from the cache"""
while len(
self.__valid) > 0 and self.__valid[0].timestamp < arpoc.utils.now():
elem = heapq.heappop(self.__valid)
if elem.item in self.data:
del self.data[elem.item]
[docs] def put(self, key: str, data: Any, valid: int) -> None:
""" Inserts the element <data> with <key> into the cache.
Expiration time <valid>.
"""
self.expire()
if key in self.data:
raise Exception
self.data[key] = data
heapq.heappush(self.__valid, CacheItem(valid, key))
def __getitem__(self, key: str) -> Any:
self.expire()
return self.data[key]
[docs] def get(self, key: str, default: Any = None) -> Any:
""" Returns the element in the cache referenced by key.
If no element exists with the key, <default> is returned
"""
try:
return self[key]
except KeyError:
return default