fixed StreamT annotations, TPorTopicSet can also be optional

Author: tworedz

faust/streams.py

@@ -594,7 +594,7 @@ def enumerate(self, start: int = 0) -> AsyncIterable[Tuple[int, T_co]]:
 
     async def noack_take(
         self, max_: int, within: Seconds
-    ) -> AsyncIterable[Sequence[T_co]]:
+    ) -> AsyncIterable[Sequence[EventT[T_co]]]:
         """
          Buffer n values at a time and yield a list of buffered values.
         :param max_: Max number of messages to receive. When more than this
@@ -606,7 +606,7 @@ async def noack_take(
              the agent is likely to stall and block buffered events for an
              unreasonable length of time(!).
         """
-        buffer: List[T_co] = []
+        buffer: List[EventT[T_co]] = []
         events: List[EventT] = []
         buffer_add = buffer.append
         event_add = events.append
@@ -635,7 +635,7 @@ async def add_to_buffer(value: T) -> T:
 
                 # We want to save events instead of values to allow for manual ack
                 event = self.current_event
-                buffer_add(cast(T_co, event))
+                buffer_add(event)
                 if event is None:
                     raise RuntimeError("Take buffer found current_event is None")
 

faust/types/events.py

@@ -34,7 +34,7 @@ class _SchemaT: ...  # noqa
 class EventT(Generic[T], AsyncContextManager):
     app: _AppT
     key: K
-    value: V
+    value: T
     headers: Mapping
     message: Message
     acked: bool

faust/types/streams.py

@@ -16,7 +16,6 @@
     Tuple,
     TypeVar,
     Union,
-    no_type_check,
 )
 
 from mode import Seconds, ServiceT
@@ -149,22 +148,35 @@ def info(self) -> Mapping[str, Any]: ...
     def clone(self, **kwargs: Any) -> "StreamT": ...
 
     @abc.abstractmethod
-    @no_type_check
-    async def items(self) -> AsyncIterator[Tuple[K, T_co]]: ...
+    def noack(self) -> "StreamT": ...
 
     @abc.abstractmethod
-    @no_type_check
-    async def events(self) -> AsyncIterable[EventT]: ...
+    def items(self) -> AsyncIterator[Tuple[K, T_co]]: ...
 
     @abc.abstractmethod
-    @no_type_check
-    async def take(
+    def events(self) -> AsyncIterable[EventT]: ...
+
+    @abc.abstractmethod
+    def take(self, max_: int, within: Seconds) -> AsyncIterable[Sequence[T_co]]: ...
+
+    @abc.abstractmethod
+    def take_events(
         self, max_: int, within: Seconds
+    ) -> AsyncIterable[Sequence[EventT[T_co]]]: ...
+
+    @abc.abstractmethod
+    def take_with_timestamp(
+        self, max_: int, within: Seconds, timestamp_field_name: str
     ) -> AsyncIterable[Sequence[T_co]]: ...
 
     @abc.abstractmethod
     def enumerate(self, start: int = 0) -> AsyncIterable[Tuple[int, T_co]]: ...
 
+    @abc.abstractmethod
+    def noack_take(
+        self, max_: int, within: Seconds
+    ) -> AsyncIterable[Sequence[EventT[T_co]]]: ...
+
     @abc.abstractmethod
     def through(self, channel: Union[str, ChannelT]) -> "StreamT": ...
 
@@ -180,6 +192,9 @@ def group_by(
         topic: Optional[TopicT] = None,
     ) -> "StreamT": ...
 
+    @abc.abstractmethod
+    def filter(self, fun: Processor[T]) -> "StreamT": ...
+
     @abc.abstractmethod
     def derive_topic(
         self,
@@ -198,9 +213,6 @@ async def throw(self, exc: BaseException) -> None: ...
     @abc.abstractmethod
     def __copy__(self) -> "StreamT": ...
 
-    @abc.abstractmethod
-    def __iter__(self) -> Any: ...
-
     @abc.abstractmethod
     def __next__(self) -> T: ...
 

faust/types/transports.py

@@ -57,7 +57,7 @@ class _AppT: ...  # noqa
 
 #: Argument to Consumer.commit to specify topics/tps to commit.
 TPorTopic = Union[str, TP]
-TPorTopicSet = AbstractSet[TPorTopic]
+TPorTopicSet = Optional[AbstractSet[TPorTopic]]
 
 #: Callback (:keyword:`async def`) called when consumer partitions are revoked.
 PartitionsRevokedCallback = Callable[[Set[TP]], Awaitable[None]]