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?
Asynchrische Filterfunktion mit generischer Typisierung ⇐ Python
-
- Similar Topics
- Replies
- Views
- Last post