NPO-Accounting/conservancy_beancount Changeset - 4de5df9035f6

Changeset - 4de5df9035f6

Parent rev.

Child rev.

[Not reviewed]

0 2 0

Brett Smith - 3 years ago 2021-02-19 23:54:36
brettcsmith@brettcsmith.org

typing: Upgrade more Posting Iterables to Iterators.

2 files changed with 5 insertions and 4 deletions:

conservancy_beancount/cliutil.py

conservancy_beancount/filters.py

0 comments (0 inline, 0 general)

conservancy_beancount/cliutil.py

➞

Show inline comments

@@ ... / @@ -30,32 +30,33 @@ from pathlib import Path @@
 import rt.exceptions as rt_error
 from . import data
 from . import filters
 from . import rtutil
 from typing import (
     cast,
     Any,
     BinaryIO,
     Callable,
     Container,
     Generic,
     Hashable,
     IO,
     Iterable,
     Iterator,
     NamedTuple,
     NoReturn,
     Optional,
     Sequence,
     TextIO,
     Type,
     TypeVar,
     Union,
+)
 from .beancount_types import (
     MetaKey,
+)
 ET = TypeVar('ET', bound=enum.Enum)
 OutputFile = Union[int, IO]
@@ ... / @@ -270,33 +271,33 @@ class SearchTerm(NamedTuple): @@
                 ticket_id, attachment_id = rt_ids
                 if key is None:
                     if attachment_id is None:
                         key = ticket_default_key
                     else:
                         key = default_key
                 pattern = rtutil.RT.metadata_regexp(
                     ticket_id,
                     attachment_id,
                     first_link_only=key == 'rt-id' and attachment_id is None,
+                )
             if key is None:
                 raise ValueError(f"invalid search term {arg!r}: no metadata key")
             return cls(key, pattern)
         return parse_search_term
-    def filter_postings(self, postings: Iterable[data.Posting]) -> Iterable[data.Posting]:
+    def filter_postings(self, postings: Iterable[data.Posting]) -> Iterator[data.Posting]:
         return filters.filter_meta_match(
             postings, self.meta_key, re.compile(self.pattern),
+        )
 def add_jobs_argument(parser: argparse.ArgumentParser) -> argparse.Action:
     return parser.add_argument(
         '--jobs', '-j',
         metavar='NUM',
         type=jobs_arg,
         default=CPU_COUNT,
         help="""Maximum number of processes to run concurrently.
 Can specify a positive integer or a percentage of CPU cores. Default all cores.
 """)
 def add_loglevel_argument(parser: argparse.ArgumentParser,
                           default: LogLevel=LogLevel.INFO) -> argparse.Action:

conservancy_beancount/filters.py

➞

Show inline comments

@@ ... / @@ -31,49 +31,49 @@ from .beancount_types import ( @@
     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:
+def filter_meta_equal(postings: Postings, key: MetaKey, value: MetaValue) -> Iterator[data.Posting]:
     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:
+def filter_meta_match(postings: Postings, key: MetaKey, regexp: Regexp) -> Iterator[data.Posting]:
     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:
+def filter_for_rt_id(postings: Postings, ticket_id: Union[int, str]) -> Iterator[data.Posting]:
     """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]:

0 comments (0 inline, 0 general)