from collections.abc import AsyncIterator
from typing import Protocol, TypeVar
StreamType = TypeVar("StreamType", covariant=True)
[docs]
class Stream(Protocol[StreamType]):
[docs]
async def next(self) -> StreamType: ...
[docs]
def __aiter__(self) -> AsyncIterator:
return self
[docs]
async def __anext__(self) -> StreamType:
return await self.next()
[docs]
class StreamReader(Protocol[StreamType]):
[docs]
async def read(self) -> StreamType: ...
[docs]
class StreamSource(Protocol[StreamType]):
[docs]
async def stream(self) -> Stream[StreamType]: ...
[docs]
class StreamWithIterator(Stream[StreamType]):
_stream: AsyncIterator[StreamType]
def __init__(self, stream: AsyncIterator[StreamType]):
self._stream = stream
[docs]
async def next(self) -> StreamType:
return await self._stream.__anext__()
[docs]
def __aiter__(self):
return self._stream
[docs]
async def __anext__(self) -> StreamType:
return await self._stream.__anext__()