Page 1 of 1

Asynchrische Filterfunktion mit generischer Typisierung

Posted: 15 Feb 2025, 10:11
by Anonymous
Ich versuche, den in Filter aber asynchronisierten Python -integrierten Python zu implementieren.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)
< /code>
Sogar ein Verständnis macht der Trick: D < /p>
Aber ... Wait ... was ist mit dem Tippen? Die Art von B zeigt immer noch "AsyncGenerator [int, keine]", aber es kann keine sein. Funktion, weil dieses Problem bereits gelöst ist, oder? < /p>
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: ...
< /code>
0.o, es scheint, dass Filter nicht einmal eine Funktion ist eine generische Klasse. Zu diesem Zeitpunkt sieht die Aufgabe nicht so einfach aus, dass jeder die Lösung hat (mit generischen Typen) Zufällig?