refactoring memory and disk cache abstractions

bauh/util/cache.py

@@ -0,0 +1,108 @@
+import datetime
+import time
+from threading import Lock, Thread
+from typing import List
+
+from bauh_api.abstract.cache import MemoryCache, MemoryCacheFactory
+
+
+class DefaultMemoryCache(MemoryCache):
+    """
+    A synchronized cache implementation
+    """
+
+    def __init__(self, expiration_time: int):
+        super(DefaultMemoryCache, self).__init__()
+        self.expiration_time = expiration_time
+        self._cache = {}
+        self.lock = Lock()
+
+    def is_enabled(self):
+        return self.expiration_time < 0 or self.expiration_time > 0
+
+    def add(self, key: str, val: object):
+        if key and self.is_enabled():
+            self.lock.acquire()
+            self._add(key, val)
+            self.lock.release()
+
+    def _add(self, key: str, val: object):
+        if key:
+            self._cache[key] = {'val': val, 'expires_at': datetime.datetime.utcnow() + datetime.timedelta(seconds=self.expiration_time) if self.expiration_time > 0 else None}
+
+    def add_non_existing(self, key: str, val: object):
+        if key and self. is_enabled():
+            self.lock.acquire()
+            cur_val = self.get(key)
+
+            if cur_val is None:
+                self._add(key, val)
+
+            self.lock.release()
+
+    def get(self, key: str):
+        if key and self.is_enabled():
+            val = self._cache.get(key)
+
+            if val:
+                expiration = val.get('expires_at')
+
+                if expiration and expiration <= datetime.datetime.utcnow():
+                    self.lock.acquire()
+                    del self._cache[key]
+                    self.lock.release()
+                    return None
+
+                return val['val']
+
+    def delete(self, key):
+        if key and self.is_enabled():
+            if key in self._cache:
+                self.lock.acquire()
+                del self._cache[key]
+                self.lock.release()
+
+    def keys(self):
+        return set(self._cache.keys()) if self.is_enabled() else set()
+
+    def clean_expired(self):
+        if self.is_enabled():
+            for key in self.keys():
+                self.get(key)
+
+
+class CacheCleaner(Thread):
+
+    def __init__(self, check_interval: int = 15):
+        super(CacheCleaner, self).__init__(daemon=True)
+        self.caches = []
+        self.check_interval = check_interval
+
+    def register(self, cache: MemoryCache):
+        if cache.is_enabled():
+            self.caches.append(cache)
+
+    def run(self):
+        if self.caches:
+            while True:
+                for cache in self.caches:
+                    cache.clean_expired()
+
+                time.sleep(self.check_interval)
+
+
+class DefaultMemoryCacheFactory(MemoryCacheFactory):
+
+    def __init__(self, expiration_time: int, cleaner: CacheCleaner):
+        """
+        :param expiration_time: default expiration time for all instantiated caches
+        :param cleaner
+        """
+        super(DefaultMemoryCacheFactory, self).__init__()
+        self.expiration_time = expiration_time
+        self.cleaner = cleaner
+
+    def new(self, expiration: int = None) -> MemoryCache:
+        instance = DefaultMemoryCache(expiration if expiration is not None else self.expiration_time)
+        self.cleaner.register(instance)
+        return instance

bauh/util/disk.py

@@ -0,0 +1,72 @@
+import json
+import os
+from threading import Thread, Lock
+from typing import Type, Dict
+
+from bauh_api.abstract.cache import MemoryCache
+from bauh_api.abstract.disk import DiskCacheLoader, DiskCacheLoaderFactory
+from bauh_api.abstract.model import SoftwarePackage
+
+
+class AsyncDiskCacheLoader(Thread, DiskCacheLoader):
+
+    def __init__(self, enabled: bool, cache_map: Dict[Type[SoftwarePackage], MemoryCache]):
+        super(AsyncDiskCacheLoader, self).__init__(daemon=True)
+        self.apps = []
+        self.stop = False
+        self.lock = Lock()
+        self.cache_map = cache_map
+        self.enabled = enabled
+
+    def fill(self, pkg: SoftwarePackage):
+        """
+        Adds a package which data must be read from the disk to a queue.
+        :param pkg:
+        :return:
+        """
+        if self.enabled:
+            if pkg and pkg.supports_disk_cache():
+                self.lock.acquire()
+                self.apps.append(pkg)
+                self.lock.release()
+
+    def run(self):
+        if self.enabled:
+            while True:
+                if self.apps:
+                    self.lock.acquire()
+                    app = self.apps[0]
+                    del self.apps[0]
+                    self.lock.release()
+                    self._fill_cached_data(app)
+                elif self.stop:
+                    break
+
+    def _fill_cached_data(self, pkg: SoftwarePackage):
+        if self.enabled:
+            if os.path.exists(pkg.get_disk_data_path()):
+                with open(pkg.get_disk_data_path()) as f:
+                    cached_data = json.loads(f.read())
+                    pkg.fill_cached_data(cached_data)
+                    cache = self.cache_map.get(pkg.__class__)\
+
+                    if cache:
+                        cache.add_non_existing(pkg.base_data.id, cached_data)
+
+
+class DefaultDiskCacheLoaderFactory(DiskCacheLoaderFactory):
+
+    def __init__(self, disk_cache_enabled: bool):
+        super(DefaultDiskCacheLoaderFactory, self).__init__()
+        self.disk_cache_enabled = disk_cache_enabled
+        self.cache_map = {}
+
+    def map(self, pkg_type: Type[SoftwarePackage], cache: MemoryCache):
+        if pkg_type:
+            if pkg_type in self.cache_map:
+                raise Exception('{} is already mapped')
+
+            self.cache_map[pkg_type] = cache
+
+    def new(self) -> AsyncDiskCacheLoader:
+        return AsyncDiskCacheLoader(enabled=self.disk_cache_enabled, cache_map=self.cache_map)

bauh/util/memory.py

@@ -1,23 +0,0 @@
-import time
-from threading import Thread
-from typing import List
-
-from bauh_api.util.cache import Cache
-
-
-class CacheCleaner(Thread):
-
-    def __init__(self, caches: List[Cache], check_interval: int = 15):
-        super(CacheCleaner, self).__init__(daemon=True)
-        self.caches = [c for c in caches if c.is_enabled()]
-        self.check_interval = check_interval
-
-    def run(self):
-
-        if self.caches:
-            while True:
-                for cache in self.caches:
-                    cache.clean_expired()
-
-                time.sleep(self.check_interval)
-