refactor: use iterators for checks

Author: Nytelife26

proselint/command_line.py

@@ -113,7 +113,7 @@ def proselint(paths=None, config=None, version=None,
             except Exception:
                 traceback.print_exc()
                 sys.exit(2)
-        errors = lint(f, config, debug=debug)
+        errors = lint(f, config)
         num_errors += len(errors)
         print_errors(fp, errors, output_json=output_json, compact=compact)
 

proselint/registry/checks/__init__.py

@@ -4,7 +4,10 @@
 
 from __future__ import annotations
 
+from collections.abc import Iterator
 from enum import Enum
+from itertools import chain, islice
+from math import ceil
 from re import RegexFlag
 from typing import TYPE_CHECKING, NamedTuple, Optional
 
@@ -90,61 +93,57 @@ class CheckFlags(NamedTuple):
     ppm_threshold: int = 0
 
     @staticmethod
-    def truncate(results: list[CheckResult], limit: int) -> list[CheckResult]:
+    def truncate(
+        results: Iterator[CheckResult], limit: int
+    ) -> Iterator[CheckResult]:
         """
         Truncate a list of results to a given threshold.
 
         This also notes how many times the check flagged prior to truncation.
         """
-        if limit == 0 or (num_results := len(results)) <= limit:
+        if limit == 0:
             return results
 
-        last_result = results[limit - 1]
-        num_extras = num_results - limit
-        last_message = last_result.message + " Found {} elsewhere.".format(
-            "once" if num_extras == 1 else f"{num_extras} times"
-        )
-
-        return [
-            *results[0 : limit - 1],
-            CheckResult(
-                start_pos=last_result.start_pos,
-                end_pos=last_result.end_pos,
-                check_path=last_result.check_path,
-                message=last_message,
-                replacements=None,
+        return chain(
+            islice(results, limit - 1),
+            (
+                CheckResult(
+                    start_pos=result.start_pos,
+                    end_pos=result.end_pos,
+                    check_path=result.check_path,
+                    message=f"{result.message} Also found elsewhere.",
+                    replacements=result.replacements,
+                )
+                for result in islice(results, 1)
             ),
-        ]
+        )
 
     @staticmethod
     def apply_threshold(
-        results: list[CheckResult], threshold: int, length: int
-    ) -> list[CheckResult]:
+        results: Iterator[CheckResult], threshold: int, length: int
+    ) -> Iterator[CheckResult]:
         """Return an error if the specified PPM `threshold` is surpassed."""
-        if threshold == 0 or length == 0 or (num_results := len(results)) < 2:
-            return []
+        if 0 in {threshold, length}:
+            return results
 
-        length = max(length, 1000)
-        if (ppm := (num_results / length) * 1e6) <= threshold:
-            return []
-        return [
+        req_results = max(ceil((threshold / 1e6) * max(length, 1000)), 2)
+        return (
             CheckResult(
-                start_pos=results[0].start_pos,
-                end_pos=results[0].end_pos,
-                check_path=results[0].check_path,
-                message=results[0].message + f" Reached {ppm:.0f} ppm.",
+                start_pos=result.start_pos,
+                end_pos=result.end_pos,
+                check_path=result.check_path,
+                message=f"{result.message} Surpassed {threshold} ppm.",
                 replacements=None,
             )
-        ]
+            for result in islice(results, req_results - 1, req_results)
+        )
 
     def apply(
-        self, results: list[CheckResult], text_len: int
-    ) -> list[CheckResult]:
-        """Apply the specified flags to a list of `results`."""
+        self, results: Iterator[CheckResult], text_len: int
+    ) -> Iterator[CheckResult]:
+        """Apply the specified flags to an iterator of `results`."""
         return CheckFlags.truncate(
-            CheckFlags.apply_threshold(results, self.ppm_threshold, text_len)
-            if self.ppm_threshold
-            else results,
+            CheckFlags.apply_threshold(results, self.ppm_threshold, text_len),
             self.results_limit,
         )
 
@@ -176,14 +175,14 @@ def matches_partial(self, partial: str) -> bool:
 
         return self.path_segments[:len(partial_segments)] == partial_segments
 
-    def check(self, text: str) -> list[CheckResult]:
+    def check(self, text: str) -> Iterator[CheckResult]:
         """Apply the check over `text`."""
         return (
             self.check_type
             if callable(self.check_type)
             else self.check_type.check
         )(text, self)
 
-    def check_with_flags(self, text: str) -> list[CheckResult]:
+    def check_with_flags(self, text: str) -> Iterator[CheckResult]:
         """Apply the check over `text`, including specified `CheckFlags`."""
         return self.flags.apply(self.check(text), len(text))

proselint/registry/checks/types.py

@@ -1,7 +1,8 @@
 """Check types."""
 
+from collections.abc import Generator, Iterable, Iterator
 from functools import partial
-from itertools import chain, filterfalse
+from itertools import chain, filterfalse, zip_longest
 from re import (
     Match,
     Pattern,
@@ -10,11 +11,26 @@
     search,
 )
 from re import compile as rcompile
-from typing import Callable, NamedTuple, Union
+from typing import Callable, NamedTuple, TypeVar, Union
 
 from proselint.registry.checks import Check, CheckResult, Padding
 
-CheckFn = Callable[[str, Check], list[CheckResult]]
+CheckFn = Callable[[str, Check], Iterator[CheckResult]]
+
+T = TypeVar("T")
+
+
+def _takewhile_peek(
+    predicate: Callable[[T], bool], iterable: Iterable[T]
+) -> Generator[T]:
+    """Take elements from `iterable` while `predicate` is consecutively true."""
+    prev = True
+    for x in iterable:
+        current = predicate(x)
+        if not prev and not current:
+            break
+        yield x
+        prev = current
 
 
 class Consistency(NamedTuple):
@@ -29,32 +45,41 @@ def process_pair(
         check: Check,
         flag: Union[RegexFlag, int],
         pair: tuple[str, str],
-    ) -> list[CheckResult]:
+    ) -> Iterator[CheckResult]:
         """Check a term pair over `text`."""
-        matches = [list(finditer(term, text, flag)) for term in pair]
+        # Unzip the zip of pair matches while both of the last pair were truthy
+        # Reads the minimum possible elements to generate results
+        matches: tuple[tuple[Union[Match[str], None], ...], ...] = tuple(
+            zip(
+                *_takewhile_peek(
+                    all,
+                    zip_longest(*(finditer(term, text, flag) for term in pair)),
+                )
+            )
+        )
 
-        if not len(matches[0]) and not len(matches[1]):
-            return []
+        if not matches:
+            return iter(())
 
-        idx_minority = len(matches[0]) > len(matches[1])
+        idx_minority = matches[1][-1] is None
         majority_term = pair[not idx_minority]
-        return [
+        return (
             CheckResult(
                 start_pos=m.start() + check.offset[0],
                 end_pos=m.end() + check.offset[1],
                 check_path=check.path,
                 message=check.message.format(majority_term, m.group(0)),
                 replacements=majority_term,
             )
-            for m in matches[idx_minority]
-        ]
+            for m in filter(None, matches[idx_minority])
+        )
 
-    def check(self, text: str, check: Check) -> list[CheckResult]:
+    def check(self, text: str, check: Check) -> Iterator[CheckResult]:
         """Check the consistency of given term pairs in `text`."""
         flag = check.re_flag
         process_pair = partial(Consistency.process_pair, text, check, flag)
 
-        return list(chain.from_iterable(map(process_pair, self.term_pairs)))
+        return chain.from_iterable(map(process_pair, self.term_pairs))
 
 
 class PreferredForms(NamedTuple):
@@ -63,12 +88,12 @@ class PreferredForms(NamedTuple):
     items: dict[str, str]
     padding: Padding = Padding.WORDS_IN_TEXT
 
-    def check(self, text: str, check: Check) -> list[CheckResult]:
+    def check(self, text: str, check: Check) -> Iterator[CheckResult]:
         """Check for terms to be replaced with a preferred form in `text`."""
         offset = self.padding.to_offset_from(check.offset)
         flag = check.re_flag
 
-        return [
+        return (
             CheckResult(
                 start_pos=m.start() + offset[0],
                 end_pos=m.end() + offset[1],
@@ -78,7 +103,7 @@ def check(self, text: str, check: Check) -> list[CheckResult]:
             )
             for original, replacement in self.items.items()
             for m in finditer(self.padding.format(original), text, flag)
-        ]
+        )
 
 
 class PreferredFormsSimple(NamedTuple):
@@ -104,7 +129,7 @@ def map_match(
             replacements=replacement,
         )
 
-    def check(self, text: str, check: Check) -> list[CheckResult]:
+    def check(self, text: str, check: Check) -> Iterator[CheckResult]:
         """Check for terms to be replaced with a preferred form in `text`."""
         offset = self.padding.to_offset_from(check.offset)
         flag = check.re_flag
@@ -116,7 +141,33 @@ def check(self, text: str, check: Check) -> list[CheckResult]:
             else next(iter(self.items))
         )
 
-        return list(map(map_match, finditer(pattern, text, flag)))
+        return map(map_match, finditer(pattern, text, flag))
+
+
+def _process_existence(
+    pattern: str,
+    exceptions: tuple[str, ...],
+    offset: tuple[int, int],
+    text: str,
+    check: Check,
+) -> Iterator[CheckResult]:
+    """Match against `pattern` respecting `offset` in `text`."""
+    flag = check.re_flag
+
+    return (
+        CheckResult(
+            start_pos=m.start() + offset[0],
+            end_pos=m.end() + offset[1],
+            check_path=check.path,
+            message=check.message.format(m.group(0).strip()),
+            replacements=None,
+        )
+        for m in finditer(pattern, text, flag)
+        if not any(
+            search(exception, m.group(0).strip(), flag)
+            for exception in exceptions
+        )
+    )
 
 
 class Existence(NamedTuple):
@@ -126,31 +177,17 @@ class Existence(NamedTuple):
     padding: Padding = Padding.WORDS_IN_TEXT
     exceptions: tuple[str, ...] = ()
 
-    def check(self, text: str, check: Check) -> list[CheckResult]:
+    def check(self, text: str, check: Check) -> Iterator[CheckResult]:
         """Check for the existence of terms in `text`."""
         offset = self.padding.to_offset_from(check.offset)
-        flag = check.re_flag
 
         pattern = self.padding.format(
             Padding.SAFE_JOIN.format("|".join(self.items))
             if len(self.items) > 1
             else next(iter(self.items))
         )
 
-        return [
-            CheckResult(
-                start_pos=m.start() + offset[0],
-                end_pos=m.end() + offset[1],
-                check_path=check.path,
-                message=check.message.format(m.group(0).strip()),
-                replacements=None,
-            )
-            for m in finditer(pattern, text, flag)
-            if not any(
-                search(exception, m.group(0).strip(), flag)
-                for exception in self.exceptions
-            )
-        ]
+        return _process_existence(pattern, self.exceptions, offset, text, check)
 
 
 class ExistenceSimple(NamedTuple):
@@ -159,24 +196,11 @@ class ExistenceSimple(NamedTuple):
     pattern: str
     exceptions: tuple[str, ...] = ()
 
-    def check(self, text: str, check: Check) -> list[CheckResult]:
+    def check(self, text: str, check: Check) -> Iterator[CheckResult]:
         """Check for the existence of a single pattern in `text`."""
-        flag = check.re_flag
-
-        return [
-            CheckResult(
-                start_pos=m.start() + check.offset[0],
-                end_pos=m.end() + check.offset[1],
-                check_path=check.path,
-                message=check.message.format(m.group(0).strip()),
-                replacements=None,
-            )
-            for m in finditer(self.pattern, text, flag)
-            if not any(
-                search(exception, m.group(0).strip(), flag)
-                for exception in self.exceptions
-            )
-        ]
+        return _process_existence(
+            self.pattern, self.exceptions, check.offset, text, check
+        )
 
 
 _DEFAULT_TOKENIZER = rcompile(
@@ -198,15 +222,15 @@ def _allowed_match(
         m_text = match.group(0)
         return (m_text.lower() if ignore_case else m_text) in allowed
 
-    def check(self, text: str, check: Check) -> list[CheckResult]:
+    def check(self, text: str, check: Check) -> Iterator[CheckResult]:
         """Check for words in `text` that are not `allowed`."""
         allowed_match = partial(
             ReverseExistence._allowed_match,
             self.allowed,
             ignore_case=check.ignore_case,
         )
 
-        return [
+        return (
             CheckResult(
                 start_pos=m.start() + check.offset[0] + 1,
                 end_pos=m.end() + check.offset[1],
@@ -215,7 +239,7 @@ def check(self, text: str, check: Check) -> list[CheckResult]:
                 replacements=None,
             )
             for m in filterfalse(allowed_match, self.TOKENIZER.finditer(text))
-        ]
+        )
 
 
 CheckType = Union[