I'm trying to implement the python built in filter but async, seem like an easy task right? async def simulated_data() -> AsyncIterator[int|None]: for i in [1, None,3,5]: yield i async def afilter[T](predicate, iterable): async for item in iterable: if predicate is not none and predicate(item): yield item b = afilter(None, simulated_data()) # or just this! b = (it async for it in iter if iter is not None) Even a comprehension does the trick But... waait... what about typing? the type of b still shows "AsyncGenerator[int, None]" but it can´t be None. I tried with TypeGuard... no luck, then I went to the original filter function, because this problem is solved already there right? class filter(Generic[_T]): @overload def __new__(cls, function: None, iterable: Iterable[_T | None], /) -> Self: ... @overload def __new__(cls, function: Callable[[_S], TypeGuard[_T]], iterable: Iterable[_S], /) -> Self: ... @overload def __new__(cls, function: Callable[[_S], TypeIs[_T]], iterable: Iterable[_S], /) -> Self: ... @overload def __new__(cls, function: Callable[[_T], Any], iterable: Iterable[_T], /) -> Self: ... def __iter__(self) -> Self: ... def __next__(self) -> _T: ... 0.o, well it seems filter is not even a function is a generic class, at this point the task doesn´t look so easy, anyone has the solution (with generic types) by any chance? Continue reading...
