Changeset - 01c3b975d811
[Not reviewed]
0 2 0
Brett Smith - 4 years ago 2020-04-12 02:26:01
brettcsmith@brettcsmith.org
data: Fix Amount.__new__.

See the comments for background and rationale.
2 files changed with 14 insertions and 4 deletions:
0 comments (0 inline, 0 general)
conservancy_beancount/data.py
Show inline comments
...
 
@@ -111,48 +111,57 @@ class Account(str):
 
            if self.startswith(prefix) and (
 
                prefix.endswith(self.SEP)
 
                or self == prefix
 
                or self[len(prefix)] == self.SEP
 
            ):
 
                return prefix
 
        return None
 

	
 

	
 
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.
 
    """
 
    __slots__ = ('meta',)
 

	
 
    def __init__(self, source: MutableMapping[MetaKey, MetaValue]) -> None:
 
        self.meta = source
 

	
 
    def __iter__(self) -> Iterator[MetaKey]:
 
        return iter(self.meta)
 

	
 
    def __len__(self) -> int:
 
        return len(self.meta)
 

	
 
    def __getitem__(self, key: MetaKey) -> MetaValue:
 
        return self.meta[key]
 

	
 
    def __setitem__(self, key: MetaKey, value: MetaValue) -> None:
 
        self.meta[key] = value
 

	
...
 
@@ -288,35 +297,35 @@ class Posting(BasePosting):
 

	
 
def balance_of(txn: Transaction,
 
               *preds: Callable[[Account], Optional[bool]],
 
) -> Amount:
 
    """Return the balance of specified postings in a transaction.
 

	
 
    Given a transaction and a series of account predicates, balance_of
 
    returns the balance of the amounts of all postings with accounts that
 
    match any of the predicates.
 

	
 
    balance_of uses the "weight" of each posting, so the return value will
 
    use the currency of the postings' cost when available.
 
    """
 
    match_posts = [post for post in Posting.from_txn(txn)
 
                   if any(pred(post.account) for pred in preds)]
 
    number = decimal.Decimal(0)
 
    if not match_posts:
 
        currency = ''
 
    else:
 
        weights: Sequence[Amount] = [
 
            bc_convert.get_weight(post) for post in match_posts  # type:ignore[no-untyped-call]
 
        ]
 
        number = sum((wt.number for wt in weights), number)
 
        currency = weights[0].currency
 
    return Amount._make((number, currency))
 
    return Amount(number, currency)
 

	
 
@functools.lru_cache()
 
def is_opening_balance_txn(txn: Transaction) -> bool:
 
    opening_equity = balance_of(txn, Account.is_opening_equity)
 
    if not opening_equity.currency:
 
        return False
 
    rest = balance_of(txn, lambda acct: not acct.is_opening_equity())
 
    if not rest.currency:
 
        return False
 
    return abs(opening_equity.number + rest.number) < decimal.Decimal('.01')
tests/testutil.py
Show inline comments
...
 
@@ -53,54 +53,54 @@ def combine_values(*value_seqs):
 
    return itertools.islice(
 
        zip(*(itertools.cycle(seq) for seq in value_seqs)),
 
        stop,
 
    )
 

	
 
def parse_date(s, fmt='%Y-%m-%d'):
 
    return datetime.datetime.strptime(s, fmt).date()
 

	
 
def test_path(s):
 
    if s is None:
 
        return s
 
    s = Path(s)
 
    if not s.is_absolute():
 
        s = TESTS_DIR / s
 
    return s
 

	
 
def Amount(number, currency='USD'):
 
    return bc_amount.Amount(Decimal(number), currency)
 

	
 
def Cost(number, currency='USD', date=FY_MID_DATE, label=None):
 
    return bc_data.Cost(Decimal(number), currency, date, label)
 

	
 
def Posting(account, number,
 
            currency='USD', cost=None, price=None, flag=None,
 
            **meta):
 
            type_=bc_data.Posting, **meta):
 
    if cost is not None:
 
        cost = Cost(*cost)
 
    if meta is None:
 
    if not meta:
 
        meta = None
 
    return bc_data.Posting(
 
    return type_(
 
        account,
 
        Amount(number, currency),
 
        cost,
 
        price,
 
        flag,
 
        meta,
 
    )
 

	
 
LINK_METADATA_STRINGS = {
 
    'Invoices/304321.pdf',
 
    'rt:123/456',
 
    'rt://ticket/234',
 
}
 

	
 
NON_LINK_METADATA_STRINGS = {
 
    '',
 
    ' ',
 
    '     ',
 
}
 

	
 
NON_STRING_METADATA_VALUES = [
 
    Decimal(5),
 
    FY_MID_DATE,
 
    Amount(50),
...
 
@@ -111,48 +111,49 @@ OPENING_EQUITY_ACCOUNTS = itertools.cycle([
 
    'Equity:Funds:Unrestricted',
 
    'Equity:Funds:Restricted',
 
    'Equity:OpeningBalance',
 
])
 

	
 
class Transaction:
 
    def __init__(self,
 
                 date=FY_MID_DATE, flag='*', payee=None,
 
                 narration='', tags=None, links=None, postings=None,
 
                 **meta):
 
        if isinstance(date, str):
 
            date = parse_date(date)
 
        self.date = date
 
        self.flag = flag
 
        self.payee = payee
 
        self.narration = narration
 
        self.tags = set(tags or '')
 
        self.links = set(links or '')
 
        self.postings = []
 
        self.meta = {
 
            'filename': '<test>',
 
            'lineno': 0,
 
        }
 
        self.meta.update(meta)
 
        if postings is not None:
 
            for posting in postings:
 
                self.add_posting(*posting)
 

	
 
    def add_posting(self, arg, *args, **kwargs):
 
        """Add a posting to this transaction. Use any of these forms:
 

	
 
           txn.add_posting(account, number, …, kwarg=value, …)
 
           txn.add_posting(account, number, …, posting_kwargs_dict)
 
           txn.add_posting(posting_object)
 
        """
 
        if kwargs:
 
            posting = Posting(arg, *args, **kwargs)
 
        elif args:
 
            if isinstance(args[-1], dict):
 
                kwargs = args[-1]
 
                args = args[:-1]
 
            posting = Posting(arg, *args, **kwargs)
 
        else:
 
            posting = arg
 
        self.postings.append(posting)
 

	
 
    @classmethod
 
    def opening_balance(cls, acct=None, **txn_meta):
 
        if acct is None:
0 comments (0 inline, 0 general)