Changeset - df0c3546fd5c
[Not reviewed]
0 2 0
Brett Smith - 4 years ago 2020-07-16 14:39:31
brettcsmith@brettcsmith.org
data: Add Account.load_from_books convenience classmethod.
2 files changed with 25 insertions and 0 deletions:
0 comments (0 inline, 0 general)
conservancy_beancount/data.py
Show inline comments
...
 
@@ -67,256 +67,261 @@ LINK_METADATA = frozenset([
 
    '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__ = ()
 

	
 
    ACCOUNT_RE: Pattern
 
    SEP = bc_account.sep
 
    _meta_map: MutableMapping[str, AccountMeta] = {}
 
    _options_map: OptionsMap
 

	
 
    @classmethod
 
    def load_options_map(cls, options_map: OptionsMap) -> None:
 
        cls._options_map = options_map
 
        roots: Sequence[str] = bc_options.get_account_types(options_map)
 
        cls.ACCOUNT_RE = re.compile(
 
            r'^(?:{})(?:{}[A-Z0-9][-A-Za-z0-9]*)+$'.format(
 
                '|'.join(roots), cls.SEP,
 
            ))
 

	
 
    @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]
 

	
 
    @classmethod
 
    def load_from_books(cls, entries: Iterable[Directive], options_map: OptionsMap) -> None:
 
        cls.load_options_map(options_map)
 
        cls.load_openings_and_closings(entries)
 

	
 
    @classmethod
 
    def is_account(cls, s: str) -> bool:
 
        return cls.ACCOUNT_RE.fullmatch(s) is not None
 

	
 
    @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
 
        care which account string is matched.
 

	
 
        An account is considered to be under itself:
 

	
 
          Account('Expenses:Tax').is_under('Expenses:Tax') # returns 'Expenses:Tax'
 

	
 
        To do a "strictly under" search, end your search strings with colons:
 

	
 
          Account('Expenses:Tax').is_under('Expenses:Tax:') # returns None
 
          Account('Expenses:Tax').is_under('Expenses:') # returns 'Expenses:'
 

	
 
        This method does check that all the account boundaries match:
 

	
 
          Account('Expenses:Tax').is_under('Exp') # returns None
 
        """
 
        for prefix in acct_seq:
 
            if self.startswith(prefix) and (
 
                prefix.endswith(self.SEP)
 
                or self == prefix
 
                or self[len(prefix)] == self.SEP
 
            ):
 
                return prefix
 
        return None
 

	
 
    def _find_part_slice(self, index: int) -> slice:
 
        if index < 0:
 
            raise ValueError(f"bad part index {index!r}")
 
        start = 0
 
        for _ in range(index):
 
            try:
 
                start = self.index(self.SEP, start) + 1
 
            except ValueError:
 
                raise IndexError("part index {index!r} out of range") from None
 
        try:
 
            stop = self.index(self.SEP, start + 1)
 
        except ValueError:
 
            stop = len(self)
 
        return slice(start, stop)
 

	
 
    def count_parts(self) -> int:
 
        return self.count(self.SEP) + 1
 

	
 
    @overload
 
    def slice_parts(self, start: None=None, stop: None=None) -> Sequence[str]: ...
 

	
 
    @overload
 
    def slice_parts(self, start: slice, stop: None=None) -> Sequence[str]: ...
 

	
 
    @overload
 
    def slice_parts(self, start: int, stop: int) -> Sequence[str]: ...
 

	
 
    @overload
 
    def slice_parts(self, start: int, stop: None=None) -> str: ...
 

	
 
    def slice_parts(self,
 
                    start: Optional[Union[int, slice]]=None,
 
                    stop: Optional[int]=None,
 
    ) -> Sequence[str]:
 
        """Slice the account parts like they were a list
 

	
 
        Given a single index, return that part of the account name as a string.
 
        Otherwise, return a list of part names sliced according to the arguments.
 
        """
 
        if start is None:
 
            part_slice = slice(None)
 
        elif isinstance(start, slice):
 
            part_slice = start
 
        elif stop is None:
 
            return self[self._find_part_slice(start)]
 
        else:
 
            part_slice = slice(start, stop)
 
        return self.split(self.SEP)[part_slice]
 

	
 
    def root_part(self, count: int=1) -> str:
 
        """Return the first part(s) of the account name as a string"""
 
        try:
 
            stop = self._find_part_slice(count - 1).stop
 
        except IndexError:
 
            return self
 
        else:
 
            return self[:stop]
 
Account.load_options_map(bc_options.OPTIONS_DEFAULTS)
 

	
 

	
 
class Amount(bc_amount.Amount):
 
    """Beancount amount after processing
 

	
 
    Beancount's native Amount class declares number to be Optional[Decimal],
 
    because the number is None when Beancount first parses a posting that does
 
    not have an amount, because the user wants it to be automatically balanced.
 

	
 
    As part of the loading process, Beancount replaces those None numbers
 
    with the calculated amount, so it will always be a Decimal. This class
 
    overrides the type declaration accordingly, so the type checker knows
 
    that our code doesn't have to consider the possibility that number is
 
    None.
 
    """
tests/test_data_account.py
Show inline comments
...
 
@@ -188,128 +188,148 @@ def test_slice_parts_range(acct_name):
 

	
 
@pytest.mark.parametrize('acct_name', [
 
    'Assets:Cash',
 
    'Assets:Receivable:Accounts',
 
    'Expenses:Other',
 
    'Equity:Funds:Restricted',
 
    'Income:Other',
 
    'Liabilities:CreditCard',
 
    'Liabilities:Payable:Accounts',
 
])
 
def test_slice_parts_slice(acct_name):
 
    account = data.Account(acct_name)
 
    parts = acct_name.split(':')
 
    for start, stop in zip([0, 0, 1, 1], [2, 3, 2, 3]):
 
        sl = slice(start, stop)
 
        assert account.slice_parts(sl) == parts[start:stop]
 

	
 
@pytest.mark.parametrize('acct_name', [
 
    'Assets:Cash',
 
    'Assets:Receivable:Accounts',
 
    'Expenses:Other',
 
    '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])
 

	
 
@pytest.mark.parametrize('account_s', [
 
    'Assets:Bank:Checking',
 
    'Equity:Funds:Restricted',
 
    'Expenses:Other',
 
    'Income:Donations',
 
    'Liabilities:CreditCard:Visa',
 
])
 
def test_is_account(account_s):
 
    assert data.Account.is_account(account_s)
 

	
 
@pytest.mark.parametrize('account_s', [
 
    'Assets:Bank:12-345',
 
    'Equity:Funds:Restricted',
 
    'Expenses:Other',
 
    'Income:Donations',
 
    'Liabilities:CreditCard:Visa0123',
 
])
 
def test_is_account(clean_account_meta, account_s):
 
    assert data.Account.is_account(account_s)
 

	
 
@pytest.mark.parametrize('account_s', [
 
    'Assets:checking',
 
    'Assets::Cash',
 
    'Equity',
 
    'Liabilities:Credit Card',
 
    'income:Donations',
 
    'Expenses:Banking_Fees',
 
    'Revenue:Grants',
 
])
 
def test_is_not_account(clean_account_meta, account_s):
 
    assert not data.Account.is_account(account_s)
 

	
 
@pytest.mark.parametrize('account_s,expected', [
 
    ('Revenue:Donations', True),
 
    ('Costs:Other', True),
 
    ('Income:Donations', False),
 
    ('Expenses:Other', False),
 
])
 
def test_is_account_respects_configured_roots(clean_account_meta, account_s, expected):
 
    config = bc_options.OPTIONS_DEFAULTS.copy()
 
    config['name_expenses'] = 'Costs'
 
    config['name_income'] = 'Revenue'
 
    data.Account.load_options_map(config)
 
    assert data.Account.is_account(account_s) == expected
 

	
 
def test_load_from_books(clean_account_meta):
 
    entries = [
 
        Open({'lineno': 310}, Date(2001, 1, 1), 'Assets:Bank:Checking', ['USD'], None),
 
        Open({'lineno': 315}, Date(2001, 2, 1), 'Revenue:Donations', None, Booking.STRICT),
 
        testutil.Transaction(date=Date(2001, 2, 10), postings=[
 
            ('Revenue:Donations', -10),
 
            ('Assets:Bank:Checking', 10),
 
        ]),
 
        Close({'lineno': 320}, Date(2001, 3, 1), 'Assets:Bank:Checking'),
 
    ]
 
    config = bc_options.OPTIONS_DEFAULTS.copy()
 
    config['name_expenses'] = 'Costs'
 
    config['name_income'] = 'Revenue'
 
    data.Account.load_from_books(entries, config)
 
    for post in entries[2].postings:
 
        assert data.Account.is_account(post.account)
 
    check_meta = data.Account(entries[0].account).meta
 
    assert check_meta.open_date == entries[0].date
 
    assert check_meta.close_date == entries[-1].date
0 comments (0 inline, 0 general)