Add filterfalse to itertools

asyncstdlib/__init__.py

@@ -25,6 +25,7 @@
     chain,
     compress,
     dropwhile,
+    filterfalse,
     islice,
     takewhile,
     starmap,
@@ -71,6 +72,7 @@
     "chain",
     "compress",
     "dropwhile",
+    "filterfalse",
     "takewhile",
     "islice",
     "starmap",

asyncstdlib/itertools.py

@@ -242,6 +242,27 @@ async def dropwhile(
             yield item
 
 
+async def filterfalse(
+    predicate: Union[Callable[[T], bool], Callable[[T], Awaitable[bool]], None],
+    iterable: AnyIterable[T],
+) -> AsyncIterator[T]:
+    """
+    Yield items from ``iterable`` for which ``predicate(item)`` is false.
+
+    If ``predicate`` is ``None``, return items which are false.
+
+    Lazily iterates over ``iterable``, yielding only items for which
+    ``predicate`` of the current item is false.
+    """
+    async with ScopedIter(iterable) as async_iter:
+        if predicate is None:
+            predicate = bool
+        predicate = _awaitify(predicate)
+        async for item in async_iter:
+            if not await predicate(item):
+                yield item
+
+
 async def islice(iterable: AnyIterable[T], *args: Optional[int]) -> AsyncIterator[T]:
     """
     An :term:`asynchronous iterator` over items from ``iterable`` in a :py:class:`slice`

docs/source/api/itertools.rst

@@ -43,6 +43,9 @@ Iterator filtering
 .. autofunction:: dropwhile(predicate: (T) → (await) bool, iterable: (async) iter T)
     :async-for: :T
 
+.. autofunction:: filterfalse(predicate: None | (T) → (await) bool, iterable: (async) iter T)
+    :async-for: :T
+
 .. autofunction:: takewhile(predicate: (T) → (await) bool, iterable: (async) iter T)
     :async-for: :T
 

unittests/test_itertools.py

@@ -178,6 +178,35 @@ async def test_dropwhile(iterable, predicate):
     )
 
 
+filterfalse_cases = (
+    (lambda x: True, [0, 1] * 5),
+    (lambda x: False, [0, 1] * 5),
+    (lambda x: x, [0, 1] * 5),
+    (lambda x: x < 5, range(20)),
+    (lambda x: x > 5, range(20)),
+)
+
+
+@pytest.mark.parametrize("predicate, iterable", filterfalse_cases)
+@sync
+async def test_filterfalse(predicate, iterable):
+    expected = list(itertools.filterfalse(predicate, iterable))
+    assert await a.list(a.filterfalse(predicate, iterable)) == expected
+    assert await a.list(a.filterfalse(awaitify(predicate), iterable)) == expected
+    assert await a.list(a.filterfalse(predicate, asyncify(iterable))) == expected
+    assert (
+        await a.list(a.filterfalse(awaitify(predicate), asyncify(iterable))) == expected
+    )
+
+
+@pytest.mark.parametrize("predicate, iterable", filterfalse_cases)
+@sync
+async def test_filterfalse_predicate_none(predicate, iterable):
+    expected = list(itertools.filterfalse(None, iterable))
+    assert await a.list(a.filterfalse(None, iterable)) == expected
+    assert await a.list(a.filterfalse(None, asyncify(iterable))) == expected
+
+
 @pytest.mark.parametrize("iterable, predicate", droptakewhile_cases)
 @sync
 async def test_takewhile(iterable, predicate):