Changeset - 8d3d7e7ce4e2
[Not reviewed]
0 3 0
Brett Smith - 4 years ago 2020-06-09 13:04:27
brettcsmith@brettcsmith.org
data: Add part slicing methods to Account.
3 files changed with 150 insertions and 1 deletions:
0 comments (0 inline, 0 general)
conservancy_beancount/data.py
Show inline comments
...
 
@@ -2,64 +2,65 @@
 

	
 
The classes in this module are interface-compatible with Beancount's core data
 
structures, and provide additional business logic that we want to use
 
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
 

	
 
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 position as bc_position
 

	
 
from typing import (
 
    cast,
 
    overload,
 
    Callable,
 
    Hashable,
 
    Iterable,
 
    Iterator,
 
    MutableMapping,
 
    Optional,
 
    Sequence,
 
    TypeVar,
 
    Union,
 
)
 

	
 
from .beancount_types import (
 
    Directive,
 
    MetaKey,
 
    MetaValue,
 
    Posting as BasePosting,
 
    Transaction,
 
)
 

	
 
DecimalCompat = Union[decimal.Decimal, int]
 

	
 
LINK_METADATA = frozenset([
 
    'approval',
 
    'check',
 
    'contract',
 
    'invoice',
 
    'purchase-order',
 
    'receipt',
 
    'rt-id',
 
    'statement',
 
])
 

	
...
 
@@ -92,64 +93,122 @@ class Account(str):
 
    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]
 

	
 

	
 
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.
 
    """
 
    number: decimal.Decimal
 

	
 
    # beancount.core._Amount is the plain namedtuple.
 
    # beancore.core.Amount adds instance methods to it.
 
    # b.c.Amount.__New__ calls `b.c._Amount.__new__`, which confuses type
 
    # checking. See <https://github.com/python/mypy/issues/1279>.
 
    # It works fine if you use super(), which is better practice anyway.
 
    # So we override __new__ just to call _Amount.__new__ this way.
 
    def __new__(cls, number: decimal.Decimal, currency: str) -> 'Amount':
 
        return super(bc_amount.Amount, Amount).__new__(cls, number, currency)
 

	
 

	
 
class Metadata(MutableMapping[MetaKey, MetaValue]):
 
    """Transaction or posting metadata
 

	
 
    This class wraps a Beancount metadata dictionary with additional methods
 
    for common parsing and query tasks.
 
    """
conservancy_beancount/reports/accrual.py
Show inline comments
...
 
@@ -356,65 +356,65 @@ class AgingODS(core.BaseODS[AccrualPostings, Optional[data.Account]]):
 
            odf.style.Style,
 
            name='WideCol',
 
        )
 
        self.style_widecol.setAttribute('family', 'table-column')
 
        self.style_widecol.addElement(odf.style.TableColumnProperties(
 
            columnwidth='1.25in',
 
        ))
 

	
 
    def section_key(self, row: AccrualPostings) -> Optional[data.Account]:
 
        if isinstance(row.account, str):
 
            return row.account
 
        else:
 
            return None
 

	
 
    def start_spreadsheet(self) -> None:
 
        for accrual_type in AccrualAccount:
 
            self.use_sheet(accrual_type.name.title())
 
            for index in range(self.COL_COUNT):
 
                stylename = self.style_widecol if index else ''
 
                self.sheet.addElement(odf.table.TableColumn(stylename=stylename))
 
            self.add_row(*(
 
                self.string_cell(name, stylename=self.style_bold)
 
                for name in self.COLUMNS
 
            ))
 
            self.lock_first_row()
 

	
 
    def start_section(self, key: Optional[data.Account]) -> None:
 
        if key is None:
 
            return
 
        self.age_thresholds = list(AccrualAccount.by_account(key).value.aging_thresholds)
 
        self.age_balances = [core.MutableBalance() for _ in self.age_thresholds]
 
        accrual_date = self.date - datetime.timedelta(days=self.age_thresholds[-1])
 
        acct_parts = key.split(':')
 
        acct_parts = key.slice_parts()
 
        self.use_sheet(acct_parts[1])
 
        self.add_row()
 
        self.add_row(self.string_cell(
 
            f"{' '.join(acct_parts[2:])} {acct_parts[1]} Aging Report"
 
            f" Accrued by {accrual_date.isoformat()} Unpaid by {self.date.isoformat()}",
 
            stylename=self.merge_styles(self.style_bold, self.style_centertext),
 
            numbercolumnsspanned=self.COL_COUNT,
 
        ))
 
        self.add_row()
 

	
 
    def end_section(self, key: Optional[data.Account]) -> None:
 
        if key is None:
 
            return
 
        self.add_row()
 
        text_style = self.merge_styles(self.style_bold, self.style_endtext)
 
        text_span = self.COL_COUNT - 1
 
        for threshold, balance in zip(self.age_thresholds, self.age_balances):
 
            years, days = divmod(threshold, 365)
 
            years_text = f"{years} {'Year' if years == 1 else 'Years'}"
 
            days_text = f"{days} Days"
 
            if years and days:
 
                age_text = f"{years_text} {days_text}"
 
            elif years:
 
                age_text = years_text
 
            else:
 
                age_text = days_text
 
            self.add_row(
 
                self.string_cell(
 
                    f"Total Aged Over {age_text}: ",
 
                    stylename=text_style,
 
                    numbercolumnsspanned=text_span,
 
                ),
tests/test_data_account.py
Show inline comments
...
 
@@ -86,32 +86,122 @@ def test_is_checking(acct_name, expected):
 

	
 
@pytest.mark.parametrize('acct_name,expected', [
 
    ('Assets:Cash', False),
 
    ('Assets:Prepaid:Expenses', False),
 
    ('Assets:Receivable:Accounts', False),
 
    ('Expenses:Other', False),
 
    ('Equity:OpeningBalance', False),
 
    ('Income:Other', False),
 
    ('Liabilities:CreditCard', True),
 
    ('Liabilities:CreditCard:Visa', True),
 
    ('Liabilities:Payable:Accounts', False),
 
    ('Liabilities:UnearnedIncome:Donations', False),
 
])
 
def test_is_credit_card(acct_name, expected):
 
    assert data.Account(acct_name).is_credit_card() == expected
 

	
 
@pytest.mark.parametrize('acct_name,expected', [
 
    ('Assets:Cash', False),
 
    ('Assets:Prepaid:Expenses', False),
 
    ('Assets:Receivable:Accounts', False),
 
    ('Expenses:Other', False),
 
    ('Equity:Funds:Restricted', True),
 
    ('Equity:Funds:Unrestricted', True),
 
    ('Equity:OpeningBalance', True),
 
    ('Equity:Retained:Costs', False),
 
    ('Income:Other', False),
 
    ('Liabilities:CreditCard', False),
 
    ('Liabilities:Payable:Accounts', False),
 
    ('Liabilities:UnearnedIncome:Donations', False),
 
])
 
def test_is_opening_equity(acct_name, expected):
 
    assert data.Account(acct_name).is_opening_equity() == expected
 

	
 
@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_no_args(acct_name):
 
    account = data.Account(acct_name)
 
    assert account.slice_parts() == acct_name.split(':')
 

	
 
@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_index(acct_name):
 
    account = data.Account(acct_name)
 
    parts = acct_name.split(':')
 
    for index, expected in enumerate(parts):
 
        assert account.slice_parts(index) == expected
 
    with pytest.raises(IndexError):
 
        account.slice_parts(index + 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_slice_parts_range(acct_name):
 
    account = data.Account(acct_name)
 
    parts = acct_name.split(':')
 
    for start, stop in zip([0, 0, 1, 1], [2, 3, 2, 3]):
 
        assert account.slice_parts(start, stop) == 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_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])
0 comments (0 inline, 0 general)