Files
@ fdb62dd1c641
Branch filter:
Location: NPO-Accounting/conservancy_beancount/conservancy_beancount/plugin/core.py
fdb62dd1c641
9.2 KiB
text/x-python
plugin.core: _RequireLinksPostingMetadataHook can check several metadata.
Extend the base class from checking 1 metadata value to checking N.
This is preparation for RT#10643, letting payables be documented with
invoice or contract.
This does unify error reporting, because now we always report all
type/invalid value errors *plus* a missing error if appropriate.
I think this consistency and thoroughness is appropriate, although
it did require some adjustments to the tests.
Extend the base class from checking 1 metadata value to checking N.
This is preparation for RT#10643, letting payables be documented with
invoice or contract.
This does unify error reporting, because now we always report all
type/invalid value errors *plus* a missing error if appropriate.
I think this consistency and thoroughness is appropriate, although
it did require some adjustments to the tests.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 | """Base classes for plugin checks"""
# 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 abc
import datetime
import re
from .. import config as configmod
from .. import data
from .. import errors as errormod
from typing import (
Any,
Dict,
FrozenSet,
Generic,
Iterable,
Iterator,
Mapping,
Optional,
Sequence,
Type,
TypeVar,
)
from ..beancount_types import (
Account,
Directive,
MetaKey,
MetaValue,
MetaValueEnum,
Transaction,
)
### CONSTANTS
# I expect these will become configurable in the future, which is why I'm
# keeping them outside of a class, but for now constants will do.
DEFAULT_START_DATE: datetime.date = datetime.date(2020, 3, 1)
# The default stop date leaves a little room after so it's easy to test
# dates past the far end of the range.
DEFAULT_STOP_DATE: datetime.date = datetime.date(datetime.MAXYEAR, 1, 1)
### TYPE DEFINITIONS
HookName = str
Entry = TypeVar('Entry', bound=Directive)
class Hook(Generic[Entry], metaclass=abc.ABCMeta):
DIRECTIVE: Type[Directive]
HOOK_GROUPS: FrozenSet[HookName] = frozenset()
def __init__(self, config: configmod.Config) -> None:
pass
# Subclasses that need configuration should override __init__ to check
# and store it.
@abc.abstractmethod
def run(self, entry: Entry) -> errormod.Iter: ...
class TransactionHook(Hook[Transaction]):
DIRECTIVE = Transaction
### HELPER CLASSES
class LessComparable(metaclass=abc.ABCMeta):
@abc.abstractmethod
def __le__(self, other: Any) -> bool: ...
@abc.abstractmethod
def __lt__(self, other: Any) -> bool: ...
CT = TypeVar('CT', bound=LessComparable)
class _GenericRange(Generic[CT]):
"""Convenience class to check whether a value is within a range.
`foo in generic_range` is equivalent to `start <= foo < stop`.
Since we have multiple user-configurable ranges, having the check
encapsulated in an object helps implement the check consistently, and
makes it easier for subclasses to override.
"""
def __init__(self, start: CT, stop: CT) -> None:
self.start = start
self.stop = stop
def __repr__(self) -> str:
return "{clsname}({self.start!r}, {self.stop!r})".format(
clsname=type(self).__name__,
self=self,
)
def __contains__(self, item: CT) -> bool:
return self.start <= item < self.stop
class MetadataEnum:
"""Map acceptable metadata values to their normalized forms.
When a piece of metadata uses a set of allowed values, use this class to
define them. You can also specify aliases that hooks will normalize to
the primary values.
"""
def __init__(self,
key: MetaKey,
standard_values: Iterable[MetaValueEnum],
aliases_map: Optional[Mapping[MetaValueEnum, MetaValueEnum]]=None,
) -> None:
"""Specify allowed values and aliases for this metadata.
Arguments:
* key: The name of the metadata key that uses this enum.
* standard_values: A sequence of strings that enumerate the standard
values for this metadata.
* aliases_map: A mapping of strings to strings. The keys are
additional allowed metadata values. The values are standard values
that each key will evaluate to. The code asserts that all values are
in standard_values.
"""
self.key = key
self._stdvalues = frozenset(standard_values)
self._aliases: Dict[MetaValueEnum, MetaValueEnum] = dict(aliases_map or ())
assert self._stdvalues.issuperset(self._aliases.values())
self._aliases.update((v, v) for v in standard_values)
def __repr__(self) -> str:
return "{}<{}>".format(type(self).__name__, self.key)
def __contains__(self, key: MetaValueEnum) -> bool:
"""Returns true if `key` is a standard value or alias."""
return key in self._aliases
def __getitem__(self, key: MetaValueEnum) -> MetaValueEnum:
"""Return the standard value for `key`.
Raises KeyError if `key` is not a known value or alias.
"""
return self._aliases[key]
def __iter__(self) -> Iterator[MetaValueEnum]:
"""Iterate over standard values."""
return iter(self._stdvalues)
def get(self,
key: MetaValueEnum,
default_key: Optional[MetaValueEnum]=None,
) -> Optional[MetaValueEnum]:
"""Return self[key], or a default fallback if that doesn't exist.
default_key is another key to look up, *not* a default value to return.
This helps ensure you always get a standard value.
"""
try:
return self[key]
except KeyError:
if default_key is None:
return None
else:
return self[default_key]
### HOOK SUBCLASSES
class _PostingHook(TransactionHook, metaclass=abc.ABCMeta):
TXN_DATE_RANGE: _GenericRange = _GenericRange(DEFAULT_START_DATE, DEFAULT_STOP_DATE)
def __init_subclass__(cls) -> None:
cls.HOOK_GROUPS = cls.HOOK_GROUPS.union(['posting'])
def _run_on_txn(self, txn: Transaction) -> bool:
return txn.date in self.TXN_DATE_RANGE
def _run_on_post(self, txn: Transaction, post: data.Posting) -> bool:
return True
def run(self, txn: Transaction) -> errormod.Iter:
if self._run_on_txn(txn):
for post in data.iter_postings(txn):
if self._run_on_post(txn, post):
yield from self.post_run(txn, post)
@abc.abstractmethod
def post_run(self, txn: Transaction, post: data.Posting) -> errormod.Iter: ...
class _NormalizePostingMetadataHook(_PostingHook):
"""Base class to normalize posting metadata from an enum."""
# This class provides basic functionality to filter postings, normalize
# metadata values, and set default values.
METADATA_KEY: MetaKey
VALUES_ENUM: MetadataEnum
def __init_subclass__(cls) -> None:
super().__init_subclass__()
cls.METADATA_KEY = cls.VALUES_ENUM.key
cls.HOOK_GROUPS = cls.HOOK_GROUPS.union(['metadata', cls.METADATA_KEY])
# If the posting does not specify METADATA_KEY, the hook calls
# _default_value to get a default. This method should either return
# a value string from METADATA_ENUM, or else raise InvalidMetadataError.
# This base implementation does the latter.
def _default_value(self, txn: Transaction, post: data.Posting) -> MetaValueEnum:
raise errormod.InvalidMetadataError(txn, self.METADATA_KEY, None, post)
def post_run(self, txn: Transaction, post: data.Posting) -> errormod.Iter:
source_value = post.meta.get(self.METADATA_KEY)
set_value = source_value
error: Optional[errormod.Error] = None
if source_value is None:
try:
set_value = self._default_value(txn, post)
except errormod.Error as error_:
error = error_
else:
try:
set_value = self.VALUES_ENUM[source_value]
except KeyError:
error = errormod.InvalidMetadataError(
txn, self.METADATA_KEY, source_value, post,
)
if error is None:
post.meta[self.METADATA_KEY] = set_value
else:
yield error
class _RequireLinksPostingMetadataHook(_PostingHook):
"""Base class to require that posting metadata include links"""
# This base class confirms that a posting's metadata has one or more links
# under one of the metadata keys listed in CHECKED_METADATA.
# Most subclasses only need to define CHECKED_METADATA and _run_on_post.
CHECKED_METADATA: Sequence[MetaKey]
def __init_subclass__(cls) -> None:
super().__init_subclass__()
cls.HOOK_GROUPS = cls.HOOK_GROUPS.union(cls.CHECKED_METADATA).union('metadata')
def _check_metadata(self,
txn: Transaction,
post: data.Posting,
keys: Sequence[MetaKey],
) -> Iterable[errormod.InvalidMetadataError]:
have_docs = False
for key in keys:
try:
links = post.meta.get_links(key)
except TypeError as error:
yield errormod.InvalidMetadataError(txn, key, post.meta[key], post)
else:
have_docs = have_docs or any(links)
if not have_docs:
yield errormod.InvalidMetadataError(txn, '/'.join(keys), None, post)
def post_run(self, txn: Transaction, post: data.Posting) -> errormod.Iter:
return self._check_metadata(txn, post, self.CHECKED_METADATA)
|