Changeset - e3dceb601c00
[Not reviewed]
0 3 0
Brett Smith - 4 years ago 2020-06-11 14:46:06
brettcsmith@brettcsmith.org
filters: Add iter_unique() function.
3 files changed with 21 insertions and 11 deletions:
0 comments (0 inline, 0 general)
conservancy_beancount/filters.py
Show inline comments
...
 
@@ -15,37 +15,43 @@
 
# along with this program.  If not, see <https://www.gnu.org/licenses/>.
 

	
 
import datetime
 
import re
 

	
 
from beancount.core import data as bc_data
 

	
 
from . import data
 
from . import rtutil
 

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

	
 
# Saying Optional works around <https://github.com/python/mypy/issues/8768>.
 
HashT = TypeVar('HashT', bound=Optional[Hashable])
 
Postings = Iterable[data.Posting]
 
Regexp = Union[str, Pattern]
 

	
 
def audit_date(entries: Entries) -> Optional[datetime.date]:
 
    for entry in entries:
 
        if (isinstance(entry, bc_data.Custom)
 
            and entry.type == 'conservancy_beancount_audit'):  # type:ignore[attr-defined]
 
            return entry.date
 
    return None
 

	
 
def filter_meta_equal(postings: Postings, key: MetaKey, value: MetaValue) -> Postings:
 
    for post in postings:
...
 
@@ -63,24 +69,31 @@ def filter_meta_match(postings: Postings, key: MetaKey, regexp: Regexp) -> Posti
 
        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 iter_unique(seq: Iterable[HashT]) -> Iterator[HashT]:
 
    seen: Set[HashT] = set()
 
    for item in seq:
 
        if item not in seen:
 
            seen.add(item)
 
            yield item
 

	
 
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.
conservancy_beancount/reports/accrual.py
Show inline comments
...
 
@@ -198,35 +198,29 @@ class AccrualPostings(core.RelatedPostings):
 

	
 
    def _single_item(self, seq: Iterable[T]) -> Union[T, Sentinel]:
 
        items = iter(seq)
 
        try:
 
            item1 = next(items)
 
        except StopIteration:
 
            all_same = False
 
        else:
 
            all_same = all(item == item1 for item in items)
 
        return item1 if all_same else self.INCONSISTENT
 

	
 
    def entities(self, pred: Callable[[data.Posting], bool]=bool) -> Iterator[MetaValue]:
 
        seen: Set[MetaValue] = set()
 
        for post in self:
 
            if pred(post):
 
                try:
 
                    entity = post.meta['entity']
 
                except KeyError:
 
                    pass
 
                else:
 
                    if entity not in seen:
 
                        yield entity
 
                        seen.add(entity)
 
        return filters.iter_unique(
 
            post.meta['entity']
 
            for post in self
 
            if pred(post) and 'entity' in post.meta
 
        )
 

	
 
    def first_links(self, key: MetaKey, default: Optional[str]=None) -> Iterator[Optional[str]]:
 
        return (post.meta.first_link(key, default) for post in self)
 

	
 
    def make_consistent(self) -> Iterator[Tuple[MetaValue, 'AccrualPostings']]:
 
        account_ok = isinstance(self.account, str)
 
        entity_ok = isinstance(self.entity, str)
 
        # `'/' in self.invoice` is just our heuristic to ensure that the
 
        # invoice metadata is "unique enough," and not just a placeholder
 
        # value like "FIXME". It can be refined if needed.
 
        invoice_ok = isinstance(self.invoice, str) and '/' in self.invoice
 
        if account_ok and entity_ok and invoice_ok:
tests/test_filters.py
Show inline comments
...
 
@@ -174,12 +174,15 @@ def test_audit_date(entry):
 
        testutil.Transaction(postings=[
 
            ('Income:Donations', -10),
 
            ('Assets:Cash', 10),
 
        ]),
 
    ]
 
    if entry is not None:
 
        entries.append(entry)
 
    actual = filters.audit_date(entries)
 
    if entry is None:
 
        assert actual is None
 
    else:
 
        assert actual == entry.date
 

	
 
def test_iter_unique():
 
    assert list(filters.iter_unique('1213231')) == list('123')
0 comments (0 inline, 0 general)