Files
@ fdd067b10e48
Branch filter:
Location: NPO-Accounting/conservancy_beancount/conservancy_beancount/reports/core.py
fdd067b10e48
4.4 KiB
text/x-python
tests: Add sample configuration file.
This should've been included in 7f45788.
This should've been included in 7f45788.
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 | """core.py - Common data classes for reporting functionality"""
# 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 collections
from decimal import Decimal
from .. import data
from typing import (
overload,
Dict,
Iterable,
Iterator,
List,
Mapping,
Optional,
Sequence,
Set,
Tuple,
Union,
)
from ..beancount_types import (
MetaKey,
MetaValue,
)
class Balance(Mapping[str, data.Amount]):
"""A collection of amounts mapped by currency
Each key is a Beancount currency string, and each value represents the
balance in that currency.
"""
__slots__ = ('_currency_map',)
def __init__(self,
source: Union[Iterable[Tuple[str, data.Amount]],
Mapping[str, data.Amount]]=(),
) -> None:
if isinstance(source, Mapping):
source = source.items()
self._currency_map = {
currency: amount.number for currency, amount in source
}
def __repr__(self) -> str:
return f"{type(self).__name__}({self._currency_map!r})"
def __getitem__(self, key: str) -> data.Amount:
return data.Amount(self._currency_map[key], key)
def __iter__(self) -> Iterator[str]:
return iter(self._currency_map)
def __len__(self) -> int:
return len(self._currency_map)
def is_zero(self) -> bool:
return all(number == 0 for number in self._currency_map.values())
class MutableBalance(Balance):
__slots__ = ()
def add_amount(self, amount: data.Amount) -> None:
try:
self._currency_map[amount.currency] += amount.number
except KeyError:
self._currency_map[amount.currency] = amount.number
class RelatedPostings(Sequence[data.Posting]):
"""Collect and query related postings
This class provides common functionality for collecting related postings
and running queries on them: iterating over them, tallying their balance,
etc.
This class doesn't know anything about how the postings are related. That's
entirely up to the caller.
A common pattern is to use this class with collections.defaultdict
to organize postings based on some key::
report = collections.defaultdict(RelatedPostings)
for txn in transactions:
for post in Posting.from_txn(txn):
if should_report(post):
key = post_key(post)
report[key].add(post)
"""
def __init__(self) -> None:
self._postings: List[data.Posting] = []
@overload
def __getitem__(self, index: int) -> data.Posting: ...
@overload
def __getitem__(self, s: slice) -> Sequence[data.Posting]: ...
def __getitem__(self,
index: Union[int, slice],
) -> Union[data.Posting, Sequence[data.Posting]]:
if isinstance(index, slice):
raise NotImplementedError("RelatedPostings[slice]")
else:
return self._postings[index]
def __len__(self) -> int:
return len(self._postings)
def add(self, post: data.Posting) -> None:
self._postings.append(post)
def clear(self) -> None:
self._postings.clear()
def iter_with_balance(self) -> Iterable[Tuple[data.Posting, Balance]]:
balance = MutableBalance()
for post in self:
balance.add_amount(post.units)
yield post, balance
def balance(self) -> Balance:
for _, balance in self.iter_with_balance():
pass
try:
return balance
except NameError:
return Balance()
def meta_values(self,
key: MetaKey,
default: Optional[MetaValue]=None,
) -> Set[Optional[MetaValue]]:
return {post.meta.get(key, default) for post in self}
|