NPO-Accounting/oxrlib Files · oxrlib/cache.py

Files @ 8dede9d1398c

Branch filter:

Location: NPO-Accounting/oxrlib/oxrlib/cache.py

8dede9d1398c 2.6 KiB text/x-python Show Annotation Show as Raw Download as Raw

Brett Smith

historical: Swap Ledger and Beancount formatters in the class hierarchy.

This makes sense for a couple of reasons:

* The Beancount formatter has "less features" than the Ledger formatter, so
this is a more "logical" organization of the hierarchy anyway. Note how
this eliminates the need for the BeancountFormatter.__init__ override to
turn off Ledger features.

* Any future work will probably be focused on the Beancount formatter, so
this reduces the amount of code you have to understand and hold in your
head to do that.

import functools
import json

from . import errors

class CacheFileBase:
    def __init__(self, path, *args, **kwargs):
        self.path = path
        try:
            self.open_file = path.open(*args, **kwargs)
        except OSError as error:
            self._translate_error(error, 'init')

    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):
        return self.open_file

    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

    def _wrap_api_method(orig_func):
        @functools.wraps(orig_func)
        def api_method_wrapper(self, *args, **kwargs):
            try:
                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')

    def write_json(self, cache_file, thing):
        json.dump(thing, cache_file, indent=2)

    def save_rate(self, rate):
        with self.historical(rate.timestamp.date(), rate.base) as cache_file:
            self.write_json(cache_file, rate.serialize())