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
...
 
@@ -24,9 +24,13 @@ from . import rtutil
 

	
 
from typing import (
 
    cast,
 
    Hashable,
 
    Iterable,
 
    Iterator,
 
    Optional,
 
    Pattern,
 
    Set,
 
    TypeVar,
 
    Union,
 
)
 
from .beancount_types import (
...
 
@@ -37,6 +41,8 @@ from .beancount_types import (
 
    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]
 

	
...
 
@@ -72,6 +78,13 @@ def filter_for_rt_id(postings: Postings, ticket_id: Union[int, str]) -> Postings
 
    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
 

	
conservancy_beancount/reports/accrual.py
Show inline comments
...
 
@@ -207,17 +207,11 @@ class AccrualPostings(core.RelatedPostings):
 
        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)
tests/test_filters.py
Show inline comments
...
 
@@ -183,3 +183,6 @@ def test_audit_date(entry):
 
        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)