Changeset - cf2833ee201e
[Not reviewed]
0 2 0
Brett Smith - 4 years ago 2020-06-16 19:10:19
brettcsmith@brettcsmith.org
plugin: Load user configuration file.
2 files changed with 14 insertions and 1 deletions:
0 comments (0 inline, 0 general)
conservancy_beancount/plugin/__init__.py
Show inline comments
 
"""Beancount plugin entry point for Conservancy"""
 
# Copyright © 2020  Brett Smith
 
#
 
# This program is free software: you can redistribute it and/or modify
 
# it under the terms of the GNU Affero General Public License as published by
 
# the Free Software Foundation, either version 3 of the License, or
 
# (at your option) any later version.
 
#
 
# This program is distributed in the hope that it will be useful,
 
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
# GNU Affero General Public License for more details.
 
#
 
# You should have received a copy of the GNU Affero General Public License
 
# along with this program.  If not, see <https://www.gnu.org/licenses/>.
 

	
 
import importlib
 
import logging
 

	
 
import beancount.core.data as bc_data
 

	
 
from typing import (
 
    AbstractSet,
 
    Any,
 
    Dict,
 
    Iterator,
 
    List,
 
    Optional,
 
    Set,
 
    Tuple,
 
    Type,
 
)
 
from ..beancount_types import (
 
    ALL_DIRECTIVES,
 
    Directive,
 
    Entries,
 
    Errors,
 
    OptionsMap,
 
)
 
from .. import config as configmod
 
from .core import (
 
    Hook,
 
    HookName,
 
)
 
from ..errors import (
 
    ConfigurationError,
 
    Error,
 
)
 

	
 
__plugins__ = ['run']
 

	
 
logger = logging.getLogger('conservancy_beancount.plugin')
 

	
 
class HookRegistry:
 
    INCLUDED_HOOKS: Dict[str, Optional[List[str]]] = {
 
        '.meta_approval': None,
 
        '.meta_entity': None,
 
        '.meta_expense_allocation': None,
 
        '.meta_income_type': None,
 
        '.meta_invoice': None,
 
        # Enforcing this hook would be premature as of May 2020.  --brett
 
        # '.meta_payable_documentation': None,
 
        '.meta_paypal_id': ['MetaPayPalID'],
 
        '.meta_project': None,
 
        '.meta_receipt': None,
 
        '.meta_receivable_documentation': None,
 
        '.meta_repo_links': None,
 
        '.meta_rt_links': ['MetaRTLinks'],
 
        '.meta_tax_implication': None,
 
    }
 

	
 
    def __init__(self) -> None:
 
        self.group_name_map: Dict[HookName, Set[Type[Hook]]] = {
 
            t.__name__: set() for t in ALL_DIRECTIVES
 
        }
 
        self.group_name_map['all'] = set()
 

	
 
    def add_hook(self, hook_cls: Type[Hook]) -> Type[Hook]:
 
        self.group_name_map['all'].add(hook_cls)
 
        self.group_name_map[hook_cls.DIRECTIVE.__name__].add(hook_cls)
 
        for key in hook_cls.HOOK_GROUPS:
 
            self.group_name_map.setdefault(key, set()).add(hook_cls)
 
        return hook_cls  # to allow use as a decorator
 

	
 
    def import_hooks(self,
 
                     mod_name: str,
 
                     *hook_names: str,
 
                     package: Optional[str]=None,
 
    ) -> None:
 
        if not hook_names:
 
            _, _, hook_name = mod_name.rpartition('.')
 
            hook_names = (hook_name.title().replace('_', ''),)
 
        module = importlib.import_module(mod_name, package)
 
        for hook_name in hook_names:
 
            self.add_hook(getattr(module, hook_name))
 

	
 
    def load_included_hooks(self) -> None:
 
        for mod_name, hook_names in self.INCLUDED_HOOKS.items():
 
            self.import_hooks(mod_name, *(hook_names or []), package=self.__module__)
 

	
 
    def group_by_directive(self, config_str: str='') -> Iterator[Tuple[HookName, Type[Hook]]]:
 
        config_str = config_str.strip()
 
        if not config_str:
 
            config_str = 'all'
 
        elif config_str.startswith('-'):
 
            config_str = 'all ' + config_str
 
        available_hooks: Set[Type[Hook]] = set()
 
        for token in config_str.split():
 
            if token.startswith('-'):
 
                update_available = available_hooks.difference_update
 
                key = token[1:]
 
            else:
 
                update_available = available_hooks.update
 
                key = token
 
            try:
 
                update_set = self.group_name_map[key]
 
            except KeyError:
 
                raise ValueError("configuration refers to unknown hooks {!r}".format(key)) from None
 
            else:
 
                update_available(update_set)
 
        for directive in ALL_DIRECTIVES:
 
            key = directive.__name__
 
            for hook in self.group_name_map[key] & available_hooks:
 
                yield key, hook
 

	
 

	
 
def run(
 
        entries: Entries,
 
        options_map: OptionsMap,
 
        config: str='',
 
        hook_registry: Optional[HookRegistry]=None,
 
) -> Tuple[Entries, Errors]:
 
    if hook_registry is None:
 
        hook_registry = HookRegistry()
 
        hook_registry.load_included_hooks()
 
    errors: Errors = []
 
    hooks: Dict[HookName, List[Hook]] = {
 
        # mypy thinks NamedTuples don't have __name__ but they do at runtime.
 
        t.__name__: [] for t in bc_data.ALL_DIRECTIVES  # type:ignore[attr-defined]
 
    }
 
    user_config = configmod.Config()
 
    try:
 
        user_config.load_file()
 
    except OSError as error:
 
        logger.debug("error reading configuration file %s: %s",
 
                     error.filename, error.strerror, exc_info=True)
 
        errors.append(ConfigurationError(
 
            f"error reading configuration file {error.filename}: {error.strerror}",
 
            source={'filename': error.filename},
 
        ))
 
    for key, hook_type in hook_registry.group_by_directive(config):
 
        try:
 
            hook = hook_type(user_config)
 
        except Error as error:
 
            errors.append(error)
 
        else:
 
            hooks[key].append(hook)
 
    for entry in entries:
 
        entry_type = type(entry).__name__
 
        for hook in hooks[entry_type]:
 
            errors.extend(hook.run(entry))
 
    return entries, errors
 

	
setup.py
Show inline comments
 
#!/usr/bin/env python3
 

	
 
from setuptools import setup
 

	
 
setup(
 
    name='conservancy_beancount',
 
    description="Plugin, library, and reports for reading Conservancy's books",
 
    version='1.2.2',
 
    version='1.2.3',
 
    author='Software Freedom Conservancy',
 
    author_email='info@sfconservancy.org',
 
    license='GNU AGPLv3+',
 

	
 
    install_requires=[
 
        'babel>=2.6',  # Debian:python3-babel
 
        'beancount>=2.2',  # Debian:beancount
 
        # 1.4.1 crashes when trying to save some documents.
 
        'odfpy>=1.4.0,!=1.4.1',  # Debian:python3-odf
 
        'PyYAML>=3.0',  # Debian:python3-yaml
 
        'regex',  # Debian:python3-regex
 
        'rt>=2.0',
 
    ],
 
    setup_requires=[
 
        'pytest-mypy',
 
        'pytest-runner',  # Debian:python3-pytest-runner
 
    ],
 
    tests_require=[
 
        'mypy>=0.770',  # Debian:python3-mypy
 
        'pytest',  # Debian:python3-pytest
 
    ],
 

	
 
    packages=[
 
        'conservancy_beancount',
 
        'conservancy_beancount.plugin',
 
        'conservancy_beancount.reports',
 
    ],
 
    entry_points={
 
        'console_scripts': [
 
            'accrual-report = conservancy_beancount.reports.accrual:entry_point',
 
            'ledger-report = conservancy_beancount.reports.ledger:entry_point',
 
        ],
 
    },
 
)
0 comments (0 inline, 0 general)