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

Files @ ae3e4617d31e

Branch filter:

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

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

Brett Smith

historical: Always format rates with the same precision.

When we format a rate as a price, we don't know how much precision
is "enough" to do the conversion, because we don't know what's
being converted to. As a result, we may (=will almost certainly)
end up formatting the rate with different precision on the cost
date vs. the price date, and that causes Beancount/Ledger to fail
to make the connection between them.

Using a constant of 6 is enough to make the current test for
"enough" precision pass, so just do that for now. This might need
further refinement in the future.

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())