new file 100644
import functools
from . import errors
class CacheFileBase:
def __init__(self, path, *args, **kwargs):
self.path = path
self.open = functools.partial(path.open, *args, **kwargs)
def _translate_error(self, error, when):
for orig_type, mapped_type in self.ERRORS_MAP:
if isinstance(error, orig_type):
raise mapped_type(self.path) from error
raise error
def __enter__(self):
try:
self.open_file = self.open()
except OSError as error:
self._translate_error(error, 'enter')
else:
return self.open_file
def __exit__(self, exc_type, exc_value, exc_tb):
self.open_file.close()
self._translate_error(error, 'exit')
class CacheBase:
ConfigurationError = errors.CacheConfigurationError
def __init__(self, dir_path, **kwargs):
self.dir_path = dir_path
self.fn_patterns = {}
self.setup(**kwargs)
def setup(self, **kwargs):
for method_name, pattern in kwargs.items():
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
def _wrap_api_method(orig_func):
@functools.wraps(orig_func)
def api_method_wrapper(self, *args, **kwargs):
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}