Files @ d145e2273483
Branch filter:

Location: NPO-Accounting/conservancy_beancount/conservancy_beancount/plugin/__init__.py

Brett Smith
test_plugin_run: Simplify testing strategy.

Avoid keeping state in the hook classes/instances.
"""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 beancount.core.data as bc_data

__plugins__ = ['run']

class HookRegistry:
    DIRECTIVES = frozenset([
        *(cls.__name__ for cls in bc_data.ALL_DIRECTIVES),
        'Posting',
    ])

    @classmethod
    def group_by_directive(cls, hooks_seq):
        hooks_map = {key: [] for key in cls.DIRECTIVES}
        for hook in hooks_seq:
            for key in cls.DIRECTIVES & hook.HOOK_GROUPS:
                hooks_map[key].append(hook)
        return hooks_map


def run(entries, options_map, config):
    errors = []
    hooks = HookRegistry.group_by_directive(config)
    for entry in entries:
        entry_type = type(entry).__name__
        for hook in hooks[entry_type]:
            errors.extend(hook.run(entry))
        if entry_type == 'Transaction':
            for post in entry.postings:
                for hook in hooks['Posting']:
                    errors.extend(hook.run(entry, post))
    return entries, errors