NPO-Accounting/conservancy_beancount Changeset - bd00822b8f43

Changeset - bd00822b8f43

Parent rev.

Child rev.

[Not reviewed]

0 2 0

Brett Smith - 4 years ago 2020-04-24 17:37:35
brettcsmith@brettcsmith.org

reports: Add RelatedPostings.group_by_meta() classmethod.

2 files changed with 55 insertions and 10 deletions:

conservancy_beancount/reports/core.py

tests/test_reports_related_postings.py

0 comments (0 inline, 0 general)

conservancy_beancount/reports/core.py

➞

Show inline comments

@@ ... / @@ -13,24 +13,25 @@ @@
+#
 # 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,
     DefaultDict,
     Dict,
     Iterable,
     Iterator,
     List,
     Mapping,
     Optional,
     Sequence,
     Set,
     Tuple,
     Union,
+)
 from ..beancount_types import (
@@ ... / @@ -84,37 +85,50 @@ class MutableBalance(Balance): @@
 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)
     to organize postings based on some key. See the group_by_meta classmethod
     for an example.
     """
     def __init__(self) -> None:
         self._postings: List[data.Posting] = []
     @classmethod
     def group_by_meta(cls,
                       postings: Iterable[data.Posting],
                       key: MetaKey,
                       default: Optional[MetaValue]=None,
     ) -> Mapping[MetaKey, 'RelatedPostings']:
         """Relate postings by metadata value
         This method takes an iterable of postings and returns a mapping.
         The keys of the mapping are the values of post.meta.get(key, default).
         The values are RelatedPostings instances that contain all the postings
         that had that same metadata value.
         """
         retval: DefaultDict[MetaKey, 'RelatedPostings'] = collections.defaultdict(cls)
         for post in postings:
             retval[post.meta.get(key, default)].add(post)
         retval.default_factory = None
         return retval
     @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:

tests/test_reports_related_postings.py

➞

Show inline comments

@@ ... / @@ -21,27 +21,28 @@ import itertools @@
 from decimal import Decimal
 import pytest
 from . import testutil
 from conservancy_beancount import data
 from conservancy_beancount.reports import core
 def accruals_and_payments(acct, src_acct, dst_acct, start_date, *amounts):
     dates = testutil.date_seq(start_date)
     for amt, currency in amounts:
         post_meta = {'metanumber': amt, 'metacurrency': currency}
         yield testutil.Transaction(date=next(dates), postings=[
             (acct, amt, currency),
             (dst_acct if amt < 0 else src_acct, -amt, currency),
             (acct, amt, currency, post_meta),
             (dst_acct if amt < 0 else src_acct, -amt, currency, post_meta),
         ])
 @pytest.fixture
 def credit_card_cycle():
     return list(accruals_and_payments(
         'Liabilities:CreditCard',
         'Assets:Checking',
         'Expenses:Other',
         datetime.date(2020, 4, 1),
         (-110, 'USD'),
         (110, 'USD'),
         (-120, 'USD'),
@@ ... / @@ -165,12 +166,42 @@ def test_meta_values_all_match_default_given(): @@
 def test_meta_values_many_types():
     expected = {
         datetime.date(2020, 4, 1),
         Decimal(42),
         testutil.Amount(5),
         'rt:42',
+    }
     related = core.RelatedPostings()
     for index, value in enumerate(expected):
         related.add(testutil.Posting('Income:Donations', -index, key=value))
     assert related.meta_values('key') == expected
 def test_group_by_meta_zero():
     assert len(core.RelatedPostings.group_by_meta([], 'metacurrency')) == 0
 def test_group_by_meta_key_error():
     # Make sure the return value doesn't act like a defaultdict.
     with pytest.raises(KeyError):
         core.RelatedPostings.group_by_meta([], 'metakey')['metavalue']
 def test_group_by_meta_one(credit_card_cycle):
     posting = next(post for post in data.Posting.from_entries(credit_card_cycle)
                    if post.account.is_credit_card())
     actual = core.RelatedPostings.group_by_meta([posting], 'metacurrency')
     assert set(actual) == {'USD'}
 def test_group_by_meta_many(two_accruals_three_payments):
     postings = [post for post in data.Posting.from_entries(two_accruals_three_payments)
                 if post.account == 'Assets:Receivable:Accounts']
     actual = core.RelatedPostings.group_by_meta(postings, 'metacurrency')
     assert set(actual) == {'USD', 'EUR'}
     for key, group in actual.items():
         assert 2 <= len(group) <= 3
         assert group.balance().is_zero()
 def test_group_by_meta_many_single_posts(two_accruals_three_payments):
     postings = [post for post in data.Posting.from_entries(two_accruals_three_payments)
                 if post.account == 'Assets:Receivable:Accounts']
     actual = core.RelatedPostings.group_by_meta(postings, 'metanumber')
     assert set(actual) == {post.units.number for post in postings}
     assert len(actual) == len(postings)

0 comments (0 inline, 0 general)