NPO-Accounting/oxrlib Changeset - 2d482bb7b045

Changeset - 2d482bb7b045

Parent rev.

Child rev.

[Not reviewed]

0 2 1

Brett Smith - 7 years ago 2017-05-12 17:27:45
brettcsmith@brettcsmith.org

cache: Add CacheWriter class.

3 files changed with 52 insertions and 1 deletions:

oxrlib/cache.py

oxrlib/loaders.py

tests/test_CacheWriter.py

0 comments (0 inline, 0 general)

oxrlib/cache.py

➞

Show inline comments

@@ ... / @@ -24,24 +24,28 @@ class CacheFileBase: @@
     def __exit__(self, exc_type, exc_value, exc_tb):
         try:
             self.open_file.close()
         except OSError as error:
             self._translate_error(error, 'exit')
 class CacheBase:
     ConfigurationError = errors.CacheConfigurationError
     def __init__(self, dir_path, **kwargs):
         self.dir_path = dir_path
         try:
             self.CacheFile = kwargs.pop('file_class')
         except KeyError:
             pass
         self.fn_patterns = {}
         self.setup(**kwargs)
     def setup(self, **kwargs):
         for method_name, pattern in kwargs.items():
             try:
                 is_api_method = getattr(self, method_name).is_api_method
             except AttributeError:
                 is_api_method = False
             if not is_api_method:
                 raise errors.CacheConfigurationError(method_name)
             self.fn_patterns[method_name] = pattern
@@ ... / @@ -53,12 +57,23 @@ class CacheBase: @@
                 fn_pattern = self.fn_patterns[orig_func.__name__]
             except KeyError:
                 raise self.ConfigurationError(orig_func.__name__) from None
             pattern_kwargs = orig_func(self, *args, **kwargs)
             path = self.dir_path / fn_pattern.format(**pattern_kwargs)
             return self.open(path)
         api_method_wrapper.is_api_method = True
         return api_method_wrapper
     @_wrap_api_method
     def historical(self, date, base):
         return {'date': date.isoformat(), 'base': base}
 class WriteCacheFile(CacheFileBase):
     ERRORS_MAP = []
 class CacheWriter(CacheBase):
     CacheFile = WriteCacheFile
     def open(self, path):
         return self.CacheFile(path, 'w')

oxrlib/loaders.py

➞

Show inline comments

@@ ... / @@ -5,28 +5,29 @@ import urllib.request @@
 import urllib.parse
 from . import cache, errors
 class ReadCacheFile(cache.CacheFileBase):
     ERRORS_MAP = [
         (FileNotFoundError, errors.LoaderNoDataError),
         (OSError, errors.LoaderSourceError),
+    ]
 class FileCache(cache.CacheBase):
     CacheFile = ReadCacheFile
     ConfigurationError = errors.CacheLoaderConfigurationError
     def open(self, path):
-        return ReadCacheFile(path)
+        return self.CacheFile(path)
 class OXRAPIRequest:
     DEFAULT_API_ROOT = 'https://openexchangerates.org/api/'
     DEFAULT_RESPONSE_ENCODING = 'utf-8'
     def __init__(self, app_id, api_root=None, *, open_func=urllib.request.urlopen):
         self.api_root = self.DEFAULT_API_ROOT if api_root is None else api_root
         self.app_id = app_id
         self.open_url = open_func
     def _get_response_encoding(self, response, default=None):

tests/test_CacheWriter.py

➞

Show inline comments

 new file 100644
 import datetime
 import io
 import pathlib
 import pytest
 from . import any_date, relpath
 import oxrlib.cache
 import oxrlib.errors
 CACHE_PATH = relpath('writecache')
 HISTORICAL_PATTERN = '{date}_{base}_cache.json'
 class TestCacheFile(oxrlib.cache.WriteCacheFile):
     def __init__(self, path, *args, **kwargs):
         self.path = path
         assert args[0].startswith('w')
     def open(self):
         open_file = io.StringIO()
         open_file.name = self.path.as_posix()
         return open_file
 @pytest.fixture
 def dummycache():
     cache = oxrlib.cache.CacheWriter(CACHE_PATH, file_class=TestCacheFile)
     cache.setup(historical=HISTORICAL_PATTERN)
     return cache
 def test_cache_success(dummycache, any_date):
     base = 'USD'
     expect_name = CACHE_PATH / HISTORICAL_PATTERN.format(date=any_date.isoformat(), base=base)
     with dummycache.historical(any_date, base) as cache_file:
         assert pathlib.Path(cache_file.name) == expect_name

0 comments (0 inline, 0 general)