12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182 |
- #!/usr/bin/env python
- # -*- coding: UTF-8 -*-
-
- import caching
- import logging
- import sys
- import time
-
-
- class test_slow_data(object):
- DATA_VERSION = 0.1
- KEY_ONE = '1'
- KEY_TWO = '2'
- KEY_THREE = 'three'
- KEY_FOUR = 'four'
- KEY_FIVE = 'five'
- KEYS = [KEY_ONE, KEY_TWO, KEY_THREE, KEY_FOUR, KEY_FIVE]
-
- def data_version(self):
- return self.DATA_VERSION
-
- def get(self, key, default=None):
- try:
- return getattr(self, f'__{key}__')()
- except AttributeError:
- return default
-
- def keys(self):
- return self.KEYS
-
- def uid(self):
- return None
-
- def print_n_sleep(self, k):
- sys.stdout.write('slow get executed for %s\n' % k)
- time.sleep(3)
-
- def __1__(self):
- self.print_n_sleep("__1__")
- return 'one'
-
- def __2__(self):
- self.print_n_sleep("__2__")
- return 'two'
-
- def __three__(self):
- self.print_n_sleep("__three__")
- return 'three'
-
- def __four__(self):
- self.print_n_sleep("__four__")
- return 'four'
-
- def __five__(self):
- self.print_n_sleep("__five__")
- return 'five'
-
- class test_slow_data_cached(test_slow_data):
- def __init__(self, *args, **kwargs) -> None:
- super().__init__(*args, **kwargs)
- self._cache = caching.property_cache_json(test_slow_data(), 'cache.json')
- self._cache.add_source_get_keys(self.KEY_THREE)
-
- def get(self, key, default=None):
- return self._cache.get(key, default)
-
-
- if __name__ == "__main__":
- # Logging configuration
- logging.basicConfig(
- format="%(asctime)s: %(levelname)8s - %(name)s - %(message)s",
- level=logging.DEBUG,
- stream=sys.stdout
- )
- # Example
- tm = time.time()
- data = test_slow_data_cached()
- print('Testing property_cache (json):\n--------------------------------')
- for key in data.keys():
- print(data.get(key))
- print('--------------------------------\nThe execution time was %.1fs' % (time.time()-tm))
|