Files @ 08073f752bc8
Branch filter:

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

Brett Smith
Configuration: Parse more conversion arguments for historical subcommand.
import argparse
import configparser
import datetime
import decimal
import os.path
import pathlib

from . import loaders

HOME_PATH = pathlib.Path(os.path.expanduser('~'))
CONFFILE_SEED = """
[Historical]
base=USD
"""

def currency_code(s):
    if not ((len(s) == 3) and s.isalpha()):
        raise ValueError("bad currency code: {!r}".format(s))
    return s.upper()

def date_from(fmt_s):
    def date_from_fmt(s):
        return datetime.datetime.strptime(s, fmt_s).date
    return date_from_fmt

class Configuration:
    DEFAULT_CONFIG_PATH = pathlib.Path(HOME_PATH, '.config', 'oxrlib.ini')
    PREPOSITIONS = frozenset(['in', 'to', 'into'])

    def __init__(self, arglist):
        argparser = self._build_argparser()
        self.error = argparser.error
        self.args = argparser.parse_args(arglist)

        if self.args.config_file is None:
            self.args.config_file = [self.DEFAULT_CONFIG_PATH]
        self.conffile = self._build_conffile()
        conffile_paths = [path.as_posix() for path in self.args.config_file]
        read_files = self.conffile.read(conffile_paths)
        for expected_path, read_path in zip(conffile_paths, read_files):
            if read_path != expected_path:
                self.error("failed to read configuration file {!r}".format(expected_path))

        try:
            post_hook = getattr(self, '_post_hook_' + self.args.command)
        except AttributeError:
            pass
        else:
            post_hook()

    def _build_argparser(self):
        prog_parser = argparse.ArgumentParser()
        prog_parser.add_argument(
            '--config-file', '-c',
            action='append', type=pathlib.Path,
            help="Path of a configuration file to read",
        )
        subparsers = prog_parser.add_subparsers()

        hist_parser = subparsers.add_parser('historical', aliases=['hist'])
        hist_parser.set_defaults(
            command='historical',
            amount=None,
            from_currency=None,
        )
        hist_parser.add_argument(
            '--base',
            type=currency_code,
            help="Base currency (default USD)",
        )
        hist_parser.add_argument(
            'date',
            type=date_from('%Y-%m-%d'), metavar='YYYY-MM-DD',
        )
        hist_parser.add_argument(
            'remainder',
            nargs=argparse.REMAINDER,
        )

        return prog_parser

    def _build_conffile(self):
        conffile = configparser.ConfigParser()
        conffile.read_string(CONFFILE_SEED)
        return conffile

    def _convert_or_error(self, argtype, s_value, argname=None, typename=None):
        try:
            return argtype(s_value)
        except (decimal.InvalidOperation, TypeError, ValueError):
            errmsg = []
            if argname:
                errmsg.append("argument {}".format(argname))
            if typename is None:
                typename = argtype.__name__.replace('_', ' ')
            errmsg.append("invalid {} value".format(typename))
            errmsg.append(repr(s_value))
            self.error(': '.join(errmsg))

    def _post_hook_historical(self):
        if self.args.base is None:
            self.args.base = self.conffile.get('Historical', 'base')
        self.args.to_currency = self.args.base
        remain_len = len(self.args.remainder)
        if (remain_len > 3) and (self.args.remainder[2].lower() in self.PREPOSITIONS):
            del self.args.remainder[2]
            remain_len -= 1
        if remain_len == 0:
            pass
        elif remain_len == 1:
            self.args.from_currency = self._convert_or_error(
                currency_code, self.args.remainder[0])
        elif remain_len < 4:
            self.args.amount = self._convert_or_error(
                decimal.Decimal, self.args.remainder[0])
            self.args.from_currency = self._convert_or_error(
                currency_code, self.args.remainder[1])
            if remain_len == 3:
                self.args.to_currency = self._convert_or_error(
                    currency_code, self.args.remainder[2])
        else:
            self.error("too many arguments")

    def _build_cache_loader(self):
        kwargs = dict(self.conffile.items('Cache'))
        try:
            kwargs['dir_path'] = kwargs.pop('directory')
        except KeyError:
            pass
        return loaders.FileCache(**kwargs)

    def _build_oxrapi_loader(self):
        kwargs = dict(self.conffile.items('OXR'))
        return loaders.OXRAPIRequest(**kwargs)

    def get_loaders(self):
        loader_chain = loaders.LoaderChain()
        for build_func in [
                self._build_cache_loader,
                self._build_oxrapi_loader,
        ]:
            try:
                loader = build_func()
            except (TypeError, ValueError, configparser.NoSectionError):
                pass
            else:
                loader_chain.add_loader(loader)
        return loader_chain