Created
August 27, 2020 03:32
-
-
Save gaganjakhotiya/0c3f4ebd8d704dbdf98d8117e76e71f0 to your computer and use it in GitHub Desktop.
In-mem cache
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from collections import OrderedDict | |
from time import time | |
import enum | |
class CacheStrategy(enum.Enum): | |
TTL = "TTL" | |
LRU = "LRU" | |
LFU = "LFU" | |
def get_init_priority(self): | |
if self == CacheStrategy.LFU: return 0 | |
else: return time() | |
def get_updated_priority(self, current_priority): | |
if self == CacheStrategy.TTL: return current_priority | |
elif self == CacheStrategy.LFU: return current_priority + 1 | |
else: return time() | |
class CacheNode(): | |
def __init__(value, priority): | |
self.value = value | |
self.priority = priority | |
def __cmp__(self, other): | |
if self.priority > other.priority: return 1 | |
elif self.priority == other.priority: return 0 | |
else: return -1 | |
class Cache(): | |
def __init__(self, capacity, strategy, options): | |
if not isinstance(capacity, int): | |
raise Exception("Cache capacity is required") | |
if not isinstance(strategy, CacheStrategy): | |
raise Exception("Cache strategy is required") | |
if strategy == CacheStrategy.TTL and not hasattr(options, "ttl"): | |
raise Exception("Cache options is required with ttl property") | |
self.capacity = capacity | |
self.strategy = strategy | |
self.options = options | |
self.cache = OrderedDict(key=lambda x: x[1]) | |
def get(key): | |
if key not in a: | |
return None | |
if self.strategy == CacheStrategy.TTL: | |
if self.cache[key].priority + self.options["ttl"] < time(): | |
return None | |
return self.cache[key] | |
else: | |
node = self.cache.pop(key) | |
node.priority = self.strategy.get_updated_priority(node.priority) | |
self.cache[key] = node | |
return node.value | |
def set(key, value): | |
if len(self.cache) == self.capacity: | |
self.cache.pop(next(iter(self.cache))) | |
self.cache[key] = CacheNode(value, self.get_init_priority()) | |
def remove(key): | |
self.cache.pop(key) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment