Files @ 5601ece2ac43
Branch filter:

Location: NPO-Accounting/conservancy_beancount/conservancy_beancount/filters.py

Brett Smith
tests: books.Loader tests do more bounds checking.
"""filters.py - Common filters for postings"""
# 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 re

from beancount.core import data as bc_data

from . import data
from . import rtutil

from typing import (
    cast,
    Iterable,
    Optional,
    Pattern,
    Union,
)
from .beancount_types import (
    Directive,
    Entries,
    MetaKey,
    MetaValue,
    Transaction,
)

Postings = Iterable[data.Posting]
Regexp = Union[str, Pattern]

def filter_meta_equal(postings: Postings, key: MetaKey, value: MetaValue) -> Postings:
    for post in postings:
        try:
            if post.meta[key] == value:
                yield post
        except KeyError:
            pass

def filter_meta_match(postings: Postings, key: MetaKey, regexp: Regexp) -> Postings:
    for post in postings:
        try:
            if re.search(regexp, post.meta[key]):
                yield post
        except (KeyError, TypeError):
            pass

def filter_for_rt_id(postings: Postings, ticket_id: Union[int, str]) -> Postings:
    """Filter postings with a primary RT ticket

    This functions yields postings where the *first* rt-id matches the given
    ticket number.
    """
    regexp = rtutil.RT.metadata_regexp(ticket_id, first_link_only=True)
    return filter_meta_match(postings, 'rt-id', regexp)

def remove_opening_balance_txn(entries: Entries) -> Optional[Transaction]:
    """Remove an opening balance transaction from entries returned by Beancount

    Returns the removed transaction if found, or None if not.
    Note that it modifies the ``entries`` argument in-place.

    This function is useful for tools like accrual-report that are more
    focused on finding and reporting related transactions than providing
    total account balances, etc. Since the opening balance transaction does not
    provide the same metadata documentation as typical transactions, it's
    typically easiest to filter it out before cross-referencing transactions by
    metadata.

    Note that this function only removes a single transaction, because that's
    fastest for the common case.
    """
    for index, entry in enumerate(entries):
        if isinstance(entry, bc_data.Transaction):
            entry = cast(Transaction, entry)
            if data.is_opening_balance_txn(entry):
                break
    else:
        return None
    del entries[index]
    return entry