NPO-Accounting/conservancy_beancount Changeset - 6a7815090ca0

Changeset - 6a7815090ca0

Parent rev.

Child rev.

[Not reviewed]

0 4 1

Brett Smith - 4 years ago 2020-07-15 14:27:05
brettcsmith@brettcsmith.org

data: Add AccountMeta class.

5 files changed with 301 insertions and 2 deletions:

conservancy_beancount/beancount_types.py

conservancy_beancount/data.py

104

tests/test_data_account.py

tests/test_data_account_meta.py

113

tests/testutil.py

0 comments (0 inline, 0 general)

conservancy_beancount/beancount_types.py

➞

Show inline comments

@@ ... / @@ -17,57 +17,71 @@ @@
 import abc
 import datetime
 import beancount.core.data as bc_data
 from typing import (
     TYPE_CHECKING,
     Any,
     FrozenSet,
     Iterable,
     List,
     Mapping,
     NamedTuple,
     Optional,
     Set,
     Tuple,
     Type,
     Union,
+)
 if TYPE_CHECKING:
     from . import errors as errormod
 Account = bc_data.Account
 Currency = bc_data.Currency
 Meta = bc_data.Meta
 MetaKey = str
 MetaValue = Any
 MetaValueEnum = str
 Posting = bc_data.Posting
 class Directive(NamedTuple):
-    meta: bc_data.Meta
     meta: Meta
     date: datetime.date
 class Close(Directive):
     account: str
 class Error(NamedTuple):
     source: Mapping[MetaKey, MetaValue]
     message: str
     entry: Optional[Directive]
 class Open(Directive):
     account: str
     # Beancount's own type declarations say these aren't Optional, but in
     # practice these fields are both None if not defined in the directive.
     currencies: Optional[List[Currency]]
     booking: Optional[bc_data.Booking]
 class Transaction(Directive):
     flag: bc_data.Flag
     payee: Optional[str]
     narration: str
     tags: Set[str]
     links: Set[str]
     postings: List[Posting]
 ALL_DIRECTIVES: FrozenSet[Type[Directive]] = frozenset([
     Transaction,
 ])
 Entries = List[Directive]
 Errors = List[Union[Error, 'errormod.Error']]
 OptionsMap = Mapping[str, Any]
 LoadResult = Tuple[Entries, Errors, OptionsMap]

conservancy_beancount/data.py

➞

Show inline comments

@@ ... / @@ -7,97 +7,201 @@ throughout Conservancy tools. @@
 # 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
 import datetime
 import decimal
 import functools
 import re
 from beancount.core import account as bc_account
 from beancount.core import amount as bc_amount
 from beancount.core import convert as bc_convert
 from beancount.core import data as bc_data
 from beancount.core import position as bc_position
 from typing import (
     cast,
     overload,
     Callable,
     Hashable,
     Iterable,
     Iterator,
     MutableMapping,
     Optional,
     Sequence,
     TypeVar,
     Union,
+)
 from .beancount_types import (
     Close,
     Currency,
     Directive,
     Meta,
     MetaKey,
     MetaValue,
     Open,
     Posting as BasePosting,
     Transaction,
+)
 DecimalCompat = Union[decimal.Decimal, int]
 LINK_METADATA = frozenset([
     'approval',
     'bank-statement',
     'check',
     'contract',
     'invoice',
     'purchase-order',
     'receipt',
     'rt-id',
     'statement',
     'tax-statement',
 ])
 class AccountMeta(MutableMapping[MetaKey, MetaValue]):
     """Access account metadata
     This class provides a consistent interface to all the metadata provided by
     Beancount's ``open`` and ``close`` directives: open and close dates,
     used currencies, booking method, and the metadata associated with each.
     For convenience, you can use this class as a Mapping to access the ``open``
     directive's metadata directly.
     """
     __slots__ = ('_opening', '_closing')
     def __init__(self, opening: Open, closing: Optional[Close]=None) -> None:
         self._opening = opening
         self._closing: Optional[Close] = None
         if closing is not None:
             self.add_closing(closing)
     def add_closing(self, closing: Close) -> None:
         if self._closing is not None and self._closing is not closing:
             raise ValueError(f"{self.account} already closed by {self._closing!r}")
         elif closing.account != self.account:
             raise ValueError(f"cannot close {self.account} with {closing.account}")
         elif closing.date < self.open_date:
             raise ValueError(f"close date {closing.date} predates open date {self.open_date}")
         else:
             self._closing = closing
     def __iter__(self) -> Iterator[MetaKey]:
         return iter(self._opening.meta)
     def __len__(self) -> int:
         return len(self._opening.meta)
     def __getitem__(self, key: MetaKey) -> MetaValue:
         return self._opening.meta[key]
     def __setitem__(self, key: MetaKey, value: MetaValue) -> None:
         self._opening.meta[key] = value
     def __delitem__(self, key: MetaKey) -> None:
         del self._opening.meta[key]
     @property
     def account(self) -> 'Account':
         return Account(self._opening.account)
     @property
     def booking(self) -> Optional[bc_data.Booking]:
         return self._opening.booking
     @property
     def close_date(self) -> Optional[datetime.date]:
         return None if self._closing is None else self._closing.date
     @property
     def close_meta(self) -> Optional[Meta]:
         return None if self._closing is None else self._closing.meta
     @property
     def currencies(self) -> Optional[Sequence[Currency]]:
         return self._opening.currencies
     @property
     def open_date(self) -> datetime.date:
         return self._opening.date
     @property
     def open_meta(self) -> Meta:
         return self._opening.meta
 class Account(str):
     """Account name string
     This is a string that names an account, like Assets:Bank:Checking
     or Income:Donations. This class provides additional methods for common
     account name parsing and queries.
     """
     __slots__ = ()
     SEP = bc_account.sep
     _meta_map: MutableMapping[str, AccountMeta] = {}
     @classmethod
     def load_opening(cls, opening: Open) -> None:
         cls._meta_map[opening.account] = AccountMeta(opening)
     @classmethod
     def load_closing(cls, closing: Close) -> None:
         try:
             cls._meta_map[closing.account].add_closing(closing)
         except KeyError:
             raise ValueError(
                 f"tried to load {closing.account} close directive before open",
             ) from None
     @classmethod
     def load_openings_and_closings(cls, entries: Iterable[Directive]) -> None:
         for entry in entries:
             # type ignores because Beancount's directives aren't type-checkable.
             if isinstance(entry, bc_data.Open):
                 cls.load_opening(entry)  # type:ignore[arg-type]
             elif isinstance(entry, bc_data.Close):
                 cls.load_closing(entry)  # type:ignore[arg-type]
     @property
     def meta(self) -> AccountMeta:
         return self._meta_map[self]
     def is_cash_equivalent(self) -> bool:
         return (
             self.is_under('Assets:') is not None
             and self.is_under('Assets:Prepaid', 'Assets:Receivable') is None
+        )
     def is_checking(self) -> bool:
         return self.is_cash_equivalent() and ':Check' in self
     def is_credit_card(self) -> bool:
         return self.is_under('Liabilities:CreditCard') is not None
     def is_opening_equity(self) -> bool:
         return self.is_under('Equity:Funds', 'Equity:OpeningBalance') is not None
     def is_under(self, *acct_seq: str) -> Optional[str]:
         """Return a match if this account is "under" a part of the hierarchy
         Pass in any number of account name strings as arguments. If this
         account is under one of those strings in the account hierarchy, the
         first matching string will be returned. Otherwise, None is returned.
         You can use the return value of this method as a boolean if you don't

tests/test_data_account.py

➞

Show inline comments

 """Test Account class"""
 # 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 pytest
 from . import testutil
 from datetime import date as Date
 from beancount.core.data import Open, Close, Booking
 from conservancy_beancount import data
 clean_account_meta = pytest.fixture()(testutil.clean_account_meta)
 def check_account_meta(acct_meta, opening, closing=None):
     if isinstance(acct_meta, str):
         acct_meta = data.Account(acct_meta).meta
     assert acct_meta == opening.meta
     assert acct_meta.account == opening.account
     assert acct_meta.booking == opening.booking
     assert acct_meta.currencies == opening.currencies
     assert acct_meta.open_date == opening.date
     assert acct_meta.open_meta == opening.meta
     if closing is None:
         assert acct_meta.close_date is None
         assert acct_meta.close_meta is None
     else:
         assert acct_meta.close_date == closing.date
         assert acct_meta.close_meta == closing.meta
 @pytest.mark.parametrize('acct_name,under_arg,expected', [
     ('Expenses:Tax:Sales', 'Expenses:Tax:Sales:', False),
     ('Expenses:Tax:Sales', 'Expenses:Tax:Sales', True),
     ('Expenses:Tax:Sales', 'Expenses:Tax:', True),
     ('Expenses:Tax:Sales', 'Expenses:Tax', True),
     ('Expenses:Tax:Sales', 'Expenses:', True),
     ('Expenses:Tax:Sales', 'Expenses', True),
     ('Expenses:Tax:Sales', 'Expense', False),
     ('Expenses:Tax:Sales', 'Equity:', False),
     ('Expenses:Tax:Sales', 'Equity', False),
 ])
 def test_is_under_one_arg(acct_name, under_arg, expected):
     expected = under_arg if expected else None
     assert data.Account(acct_name).is_under(under_arg) == expected
 @pytest.mark.parametrize('acct_name,expected', [
     ('Assets:Cash', None),
     ('Assets:Checking', None),
     ('Assets:Prepaid:Expenses', 'Assets:Prepaid'),
     ('Assets:Receivable:Accounts', 'Assets:Receivable'),
 ])
 def test_is_under_multi_arg(acct_name, expected):
     assert expected == data.Account(acct_name).is_under(
         'Assets:Prepaid', 'Assets:Receivable',
@@ ... / @@ -184,24 +208,62 @@ def test_slice_parts_slice(acct_name): @@
     'Equity:Funds:Restricted',
     'Income:Other',
     'Liabilities:CreditCard',
     'Liabilities:Payable:Accounts',
 ])
 def test_count_parts(acct_name):
     account = data.Account(acct_name)
     assert account.count_parts() == acct_name.count(':') + 1
 @pytest.mark.parametrize('acct_name', [
     'Assets:Cash',
     'Assets:Receivable:Accounts',
     'Expenses:Other',
     'Equity:Funds:Restricted',
     'Income:Other',
     'Liabilities:CreditCard',
     'Liabilities:Payable:Accounts',
 ])
 def test_root_part(acct_name):
     account = data.Account(acct_name)
     parts = acct_name.split(':')
     assert account.root_part() == parts[0]
     assert account.root_part(1) == parts[0]
     assert account.root_part(2) == ':'.join(parts[:2])
 def test_load_opening(clean_account_meta):
     opening = Open({'lineno': 210}, Date(2010, 2, 1), 'Assets:Cash', None, None)
     data.Account.load_opening(opening)
     check_account_meta('Assets:Cash', opening)
 def test_load_closing(clean_account_meta):
     name = 'Assets:Checking'
     opening = Open({'lineno': 230}, Date(2010, 10, 1), name, None, None)
     closing = Close({'lineno': 235}, Date(2010, 11, 1), name)
     data.Account.load_opening(opening)
     data.Account.load_closing(closing)
     check_account_meta(name, opening, closing)
 def test_load_closing_without_opening(clean_account_meta):
     closing = Close({'lineno': 245}, Date(2010, 3, 1), 'Assets:Cash')
     with pytest.raises(ValueError):
         data.Account.load_closing(closing)
 def test_load_openings_and_closings(clean_account_meta):
     entries = [
         Open({'lineno': 1, 'income-type': 'Donations'},
              Date(2000, 3, 1), 'Income:Donations', None, None),
         Open({'lineno': 2},
              Date(2000, 3, 1), 'Income:Other', None, None),
         Open({'lineno': 3, 'asset-type': 'Cash equivalent'},
              Date(2000, 4, 1), 'Assets:Checking', ['USD', 'EUR'], Booking.STRICT),
         testutil.Transaction(date=Date(2000, 4, 10), postings=[
             ('Income:Donations', -10),
             ('Assets:Checking', 10),
         ]),
         Close({'lineno': 30, 'why': 'Changed banks'},
               Date(2000, 5, 1), 'Assets:Checking')
+    ]
     data.Account.load_openings_and_closings(iter(entries))
     check_account_meta('Income:Donations', entries[0])
     check_account_meta('Income:Other', entries[1])
     check_account_meta('Assets:Checking', entries[2], entries[-1])

tests/test_data_account_meta.py

➞

Show inline comments

 new file 100644
 """Test AccountMeta class"""
 # 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 itertools
 import pytest
 from datetime import date as Date
 from beancount.core.data import Open, Close, Booking
 from conservancy_beancount import data
 def test_attributes():
     open_date = Date(2019, 6, 1)
     name = 'Assets:Bank:Checking'
     currencies = ['USD', 'EUR']
     booking = Booking.STRICT
     actual = data.AccountMeta(Open({}, open_date, name, list(currencies), booking))
     assert actual.open_date == open_date
     assert actual.account == name
     assert isinstance(actual.account, data.Account)
     assert actual.currencies == currencies
     assert actual.booking == booking
 def test_mapping():
     src_meta = {'filename': 'maptest', 'lineno': 10}
     actual = data.AccountMeta(Open(
         src_meta.copy(), Date(2019, 6, 1), 'Income:Donations', None, None,
     ))
     assert len(actual) == 2
     assert set(actual) == set(src_meta)  # Test __iter__
     for key, expected in src_meta.items():
         assert actual[key] == expected
 def test_close_attributes_without_closing():
     actual = data.AccountMeta(Open(
         {}, Date(2019, 6, 1), 'Assets:Cash', None, None,
     ))
     assert actual.close_date is None
     assert actual.close_meta is None
 def test_close_at_init():
     src_meta = {'filename': 'initclose', 'lineno': 50}
     close_date = Date(2020, 6, 1)
     name = 'Assets:Bank:MoneyMarket'
     actual = data.AccountMeta(
         Open({}, Date(2019, 6, 1), name, None, None),
         Close(src_meta.copy(), close_date, name),
+    )
     assert actual.close_date == close_date
     assert actual.close_meta == src_meta
 def test_add_closing():
     src_meta = {'filename': 'laterclose', 'lineno': 65}
     close_date = Date(2020, 1, 1)
     name = 'Assets:Bank:EUR'
     actual = data.AccountMeta(Open({}, Date(2019, 6, 1), name, None, None))
     assert actual.close_date is None
     assert actual.close_meta is None
     actual.add_closing(Close(src_meta.copy(), close_date, name))
     assert actual.close_date == close_date
     assert actual.close_meta == src_meta
 def test_add_closing_already_inited():
     name = 'Assets:Bank:Savings'
     actual = data.AccountMeta(
         Open({}, Date(2019, 6, 1), name, None, None),
         Close({}, Date(2019, 7, 1), name),
+    )
     with pytest.raises(ValueError):
         actual.add_closing(Close({}, Date(2019, 8, 1), name))
 def test_add_closing_called_twice():
     name = 'Assets:Bank:FX'
     actual = data.AccountMeta(Open({}, Date(2019, 6, 1), name, None, None))
     actual.add_closing(Close({}, Date(2019, 7, 1), name))
     with pytest.raises(ValueError):
         actual.add_closing(Close({}, Date(2019, 8, 1), name))
 @pytest.mark.parametrize('close_date,close_name', [
     (Date(2020, 6, 1), 'Income:Grants'),  # Account name doesn't match
     (Date(2010, 6, 1), 'Income:Donations'),  # Close predates Open
 ])
 def test_bad_closing_at_init(close_date, close_name):
     with pytest.raises(ValueError):
         data.AccountMeta(
             Open({}, Date(2019, 6, 1), 'Income:Donations', None, None),
             Close({}, close_date, close_name),
+        )
 @pytest.mark.parametrize('close_date,close_name', [
     (Date(2020, 6, 1), 'Income:Grants'),  # Account name doesn't match
     (Date(2010, 6, 1), 'Income:Donations'),  # Close predates Open
 ])
 def test_add_closing_wrong_account(close_date, close_name):
     actual = data.AccountMeta(
         Open({}, Date(2019, 6, 1), 'Income:Donations', None, None),
+    )
     with pytest.raises(ValueError):
         actual.add_closing(Close({}, close_date, close_name))

tests/testutil.py

➞

Show inline comments

@@ ... / @@ -9,57 +9,63 @@ @@
 # 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 datetime
 import itertools
 import re
 import beancount.core.amount as bc_amount
 import beancount.core.data as bc_data
 import beancount.loader as bc_loader
 import odf.element
 import odf.opendocument
 import odf.table
 from decimal import Decimal
 from pathlib import Path
 from typing import Any, Optional, NamedTuple
 from conservancy_beancount import books, rtutil
+from conservancy_beancount import books, data, rtutil
 EXTREME_FUTURE_DATE = datetime.date(datetime.MAXYEAR, 12, 30)
 FUTURE_DATE = datetime.date.today() + datetime.timedelta(days=365 * 99)
 FY_START_DATE = datetime.date(2020, 3, 1)
 FY_MID_DATE = datetime.date(2020, 9, 1)
 PAST_DATE = datetime.date(2000, 1, 1)
 TESTS_DIR = Path(__file__).parent
 # This function is primarily used as a fixture, but different test files use
 # it with different scopes. Typical usage looks like:
 #   clean_account_meta = pytest.fixture([options])(testutil.clean_account_meta)
 def clean_account_meta():
     data.Account._meta_map.clear()
 def _ods_cell_value_type(cell):
     assert cell.tagName == 'table:table-cell'
     return cell.getAttribute('valuetype')
 def _ods_cell_value(cell):
     value_type = cell.getAttribute('valuetype')
     if value_type == 'currency' or value_type == 'float':
         return Decimal(cell.getAttribute('value'))
     elif value_type == 'date':
         return datetime.datetime.strptime(
             cell.getAttribute('datevalue'), '%Y-%m-%d',
         ).date()
     else:
         return cell.getAttribute('value')
 def _ods_elem_text(elem):
     if isinstance(elem, odf.element.Text):
         return elem.data
     else:
         return '\0'.join(_ods_elem_text(child) for child in elem.childNodes)
 odf.element.Element.value_type = property(_ods_cell_value_type)
 odf.element.Element.value = property(_ods_cell_value)
 odf.element.Element.text = property(_ods_elem_text)

0 comments (0 inline, 0 general)