Upload operator.py with huggingface_hub

operator.py

@@ -1,6 +1,7 @@
+import re
 from abc import abstractmethod
 from dataclasses import field
-from typing import Any, Dict, Generator, List, Optional, Union
+from typing import Any, Dict, Generator, List, Optional
 
 from .artifact import Artifact
 from .dataclass import NonPositionalField
@@ -32,8 +33,7 @@ class OperatorError(Exception):
 
 
 class StreamingOperator(Artifact):
-    """
-    Base class for all stream operators in the streaming model.
+    """Base class for all stream operators in the streaming model.
 
     Stream operators are a key component of the streaming model and are responsible for processing continuous data streams.
     They perform operations such as transformations, aggregations, joins, windowing and more on these streams.
@@ -48,8 +48,7 @@ class StreamingOperator(Artifact):
 
     @abstractmethod
     def __call__(self, streams: Optional[MultiStream] = None) -> MultiStream:
-        """
-        Abstract method that performs operations on the stream.
+        """Abstract method that performs operations on the stream.
 
         Args:
             streams (Optional[MultiStream]): The input MultiStream, which can be None.
@@ -60,8 +59,7 @@ class StreamingOperator(Artifact):
 
 
 class StreamSource(StreamingOperator):
-    """
-    A class representing a stream source operator in the streaming system.
+    """A class representing a stream source operator in the streaming system.
 
     A stream source operator is a special type of `StreamingOperator` that generates a data stream without taking any input streams. It serves as the starting point in a stream processing pipeline, providing the initial data that other operators in the pipeline can process.
 
@@ -75,8 +73,7 @@ class StreamSource(StreamingOperator):
 
 
 class SourceOperator(StreamSource):
-    """
-    A class representing a source operator in the streaming system.
+    """A class representing a source operator in the streaming system.
 
     A source operator is responsible for generating the data stream from some source, such as a database or a file. This is the starting point of a stream processing pipeline. The `SourceOperator` class is a type of `StreamSource`, which is a special type of `StreamingOperator` that generates an output stream but does not take any input streams.
 
@@ -99,8 +96,7 @@ class SourceOperator(StreamSource):
 
 
 class StreamInitializerOperator(StreamSource):
-    """
-    A class representing a stream initializer operator in the streaming system.
+    """A class representing a stream initializer operator in the streaming system.
 
     A stream initializer operator is a special type of `StreamSource` that is capable of taking parameters during the stream generation process. This can be useful in situations where the stream generation process needs to be customized or configured based on certain parameters.
 
@@ -123,8 +119,7 @@ class StreamInitializerOperator(StreamSource):
 
 
 class MultiStreamOperator(StreamingOperator):
-    """
-    A class representing a multi-stream operator in the streaming system.
+    """A class representing a multi-stream operator in the streaming system.
 
     A multi-stream operator is a type of `StreamingOperator` that operates on an entire MultiStream object at once. It takes a `MultiStream` as input and produces a `MultiStream` as output. The `process` method should be implemented by subclasses to define the specific operations to be performed on the input `MultiStream`.
     """
@@ -138,9 +133,13 @@ class MultiStreamOperator(StreamingOperator):
                 result.set_caching(self.caching)
             return result
 
-    def _process_multi_stream(self, multi_stream: Optional[MultiStream] = None) -> MultiStream:
+    def _process_multi_stream(
+        self, multi_stream: Optional[MultiStream] = None
+    ) -> MultiStream:
         result = self.process(multi_stream)
-        assert isinstance(result, MultiStream), "MultiStreamOperator must return a MultiStream"
+        assert isinstance(
+            result, MultiStream
+        ), "MultiStreamOperator must return a MultiStream"
         return result
 
     @abstractmethod
@@ -149,30 +148,42 @@ class MultiStreamOperator(StreamingOperator):
 
 
 class SingleStreamOperator(MultiStreamOperator):
-    """
-    A class representing a single-stream operator in the streaming system.
+    """A class representing a single-stream operator in the streaming system.
 
     A single-stream operator is a type of `MultiStreamOperator` that operates on individual `Stream` objects within a `MultiStream`. It iterates through each `Stream` in the `MultiStream` and applies the `process` method. The `process` method should be implemented by subclasses to define the specific operations to be performed on each `Stream`.
     """
 
-    apply_to_streams: List[str] = NonPositionalField(default=None)  # None apply to all streams
-    dont_apply_to_streams: List[str] = NonPositionalField(default_factory=list)
+    apply_to_streams: List[str] = NonPositionalField(
+        default=None
+    )  # None apply to all streams
+    dont_apply_to_streams: List[str] = NonPositionalField(default_factory=None)
 
     def _process_multi_stream(self, multi_stream: MultiStream) -> MultiStream:
         result = {}
         for stream_name, stream in multi_stream.items():
-            stream = self._process_single_stream(stream, stream_name)
-            assert isinstance(stream, Stream), "SingleStreamOperator must return a Stream"
+            if self._is_should_be_processed(stream_name):
+                stream = self._process_single_stream(stream, stream_name)
+            else:
+                stream = stream
+            assert isinstance(
+                stream, Stream
+            ), "SingleStreamOperator must return a Stream"
             result[stream_name] = stream
 
         return MultiStream(result)
 
-    def _process_single_stream(self, stream: Stream, stream_name: str = None) -> Stream:
-        return Stream(self._process_stream, gen_kwargs={"stream": stream, "stream_name": stream_name})
+    def _process_single_stream(
+        self, stream: Stream, stream_name: Optional[str] = None
+    ) -> Stream:
+        return Stream(
+            self._process_stream,
+            gen_kwargs={"stream": stream, "stream_name": stream_name},
+        )
 
-    def is_should_be_processed(self, stream_name):
+    def _is_should_be_processed(self, stream_name):
         if (
             self.apply_to_streams is not None
+            and self.dont_apply_to_streams is not None
             and stream_name in self.apply_to_streams
             and stream_name in self.dont_apply_to_streams
         ):
@@ -182,29 +193,32 @@ class SingleStreamOperator(MultiStreamOperator):
 
         return (
             self.apply_to_streams is None or stream_name in self.apply_to_streams
-        ) and stream_name not in self.dont_apply_to_streams
+        ) and (
+            self.dont_apply_to_streams is None
+            or stream_name not in self.dont_apply_to_streams
+        )
 
-    def _process_stream(self, stream: Stream, stream_name: str = None) -> Generator:
-        if self.is_should_be_processed(stream_name):
-            yield from self.process(stream, stream_name)
-        else:
-            yield from stream
+    def _process_stream(
+        self, stream: Stream, stream_name: Optional[str] = None
+    ) -> Generator:
+        yield from self.process(stream, stream_name)
 
     @abstractmethod
-    def process(self, stream: Stream, stream_name: str = None) -> Generator:
+    def process(self, stream: Stream, stream_name: Optional[str] = None) -> Generator:
         pass
 
 
 class PagedStreamOperator(SingleStreamOperator):
-    """
-    A class representing a paged-stream operator in the streaming system.
+    """A class representing a paged-stream operator in the streaming system.
 
     A paged-stream operator is a type of `SingleStreamOperator` that operates on a page of instances in a `Stream` at a time, where a page is a subset of instances. The `process` method should be implemented by subclasses to define the specific operations to be performed on each page.
     """
 
     page_size: int = 1000
 
-    def _process_stream(self, stream: Stream, stream_name: str = None) -> Generator:
+    def _process_stream(
+        self, stream: Stream, stream_name: Optional[str] = None
+    ) -> Generator:
         page = []
         for instance in stream:
             page.append(instance)
@@ -214,13 +228,12 @@ class PagedStreamOperator(SingleStreamOperator):
         yield from self.process(page, stream_name)
 
     @abstractmethod
-    def process(self, page: List[Dict], stream_name: str = None) -> Generator:
+    def process(self, page: List[Dict], stream_name: Optional[str] = None) -> Generator:
         pass
 
 
 class SingleStreamReducer(StreamingOperator):
-    """
-    A class representing a single-stream reducer in the streaming system.
+    """A class representing a single-stream reducer in the streaming system.
 
     A single-stream reducer is a type of `StreamingOperator` that operates on individual `Stream` objects within a `MultiStream` and reduces each `Stream` to a single output value. The `process` method should be implemented by subclasses to define the specific reduction operation to be performed on each `Stream`.
     """
@@ -239,27 +252,40 @@ class SingleStreamReducer(StreamingOperator):
 
 
 class StreamInstanceOperator(SingleStreamOperator):
-    """
-    A class representing a stream instance operator in the streaming system.
+    """A class representing a stream instance operator in the streaming system.
 
     A stream instance operator is a type of `SingleStreamOperator` that operates on individual instances within a `Stream`. It iterates through each instance in the `Stream` and applies the `process` method. The `process` method should be implemented by subclasses to define the specific operations to be performed on each instance.
     """
 
-    def _process_stream(self, stream: Stream, stream_name: str = None) -> Generator:
-        for instance in stream:
-            yield self._process_instance(instance, stream_name)
-
-    def _process_instance(self, instance: Dict[str, Any], stream_name: str = None) -> Dict[str, Any]:
+    def _process_stream(
+        self, stream: Stream, stream_name: Optional[str] = None
+    ) -> Generator:
+        try:
+            _index = None
+            for _index, instance in enumerate(stream):
+                yield self._process_instance(instance, stream_name)
+        except Exception as e:
+            if _index is None:
+                raise e
+            else:
+                raise ValueError(
+                    f"Error processing instance '{_index}' from stream '{stream_name}' in {self.__class__.__name__} due to: {e}"
+                ) from e
+
+    def _process_instance(
+        self, instance: Dict[str, Any], stream_name: Optional[str] = None
+    ) -> Dict[str, Any]:
         return self.process(instance, stream_name)
 
     @abstractmethod
-    def process(self, instance: Dict[str, Any], stream_name: str = None) -> Dict[str, Any]:
+    def process(
+        self, instance: Dict[str, Any], stream_name: Optional[str] = None
+    ) -> Dict[str, Any]:
         pass
 
 
 class StreamInstanceOperatorValidator(StreamInstanceOperator):
-    """
-    A class representing a stream instance operator validator in the streaming system.
+    """A class representing a stream instance operator validator in the streaming system.
 
     A stream instance operator validator is a type of `StreamInstanceOperator` that includes a validation step. It operates on individual instances within a `Stream` and validates the result of processing each instance.
     """
@@ -268,18 +294,24 @@ class StreamInstanceOperatorValidator(StreamInstanceOperator):
     def validate(self, instance):
         pass
 
-    def _process_stream(self, stream: Stream, stream_name: str = None) -> Generator:
+    def _process_stream(
+        self, stream: Stream, stream_name: Optional[str] = None
+    ) -> Generator:
         iterator = iter(stream)
-        first_instance = next(iterator)
+        try:
+            first_instance = next(iterator)
+        except StopIteration as e:
+            raise StopIteration(f"Strem '{stream_name}' is empty") from e
         result = self._process_instance(first_instance, stream_name)
         self.validate(result)
         yield result
-        yield from (self._process_instance(instance, stream_name) for instance in iterator)
+        yield from (
+            self._process_instance(instance, stream_name) for instance in iterator
+        )
 
 
 class InstanceOperator(Artifact):
-    """
-    A class representing an instance operator in the streaming system.
+    """A class representing an instance operator in the streaming system.
 
     An instance operator is a type of `Artifact` that operates on a single instance (represented as a dict) at a time. It takes an instance as input and produces a transformed instance as output.
     """
@@ -293,8 +325,7 @@ class InstanceOperator(Artifact):
 
 
 class BaseFieldOperator(Artifact):
-    """
-    A class representing a field operator in the streaming system.
+    """A class representing a field operator in the streaming system.
 
     A field operator is a type of `Artifact` that operates on a single field within an instance. It takes an instance and a field name as input, processes the field, and updates the field in the instance with the processed value.
     """
@@ -309,33 +340,22 @@ class BaseFieldOperator(Artifact):
         pass
 
 
-class InstanceOperatorWithGlobalAccess(StreamingOperator):
-    """
-    A class representing an instance operator with global access in the streaming system.
+class InstanceOperatorWithMultiStreamAccess(StreamingOperator):
+    """A class representing an instance operator with global access in the streaming system.
 
     An instance operator with global access is a type of `StreamingOperator` that operates on individual instances within a `Stream` and can also access other streams.
     It uses the `accessible_streams` attribute to determine which other streams it has access to.
     In order to make this efficient and to avoid qudratic complexity, it caches the accessible streams by default.
     """
 
-    accessible_streams: Union[MultiStream, List[str]] = None
-    cache_accessible_streams: bool = True
-
     def __call__(self, multi_stream: Optional[MultiStream] = None) -> MultiStream:
         result = {}
 
-        if isinstance(self.accessible_streams, list):
-            # cache the accessible streams:
-            self.accessible_streams = MultiStream(
-                {stream_name: multi_stream[stream_name] for stream_name in self.accessible_streams}
-            )
-
-        if self.cache_accessible_streams:
-            for stream in self.accessible_streams.values():
-                stream.caching = True
-
         for stream_name, stream in multi_stream.items():
-            stream = Stream(self.generator, gen_kwargs={"stream": stream, "multi_stream": self.accessible_streams})
+            stream = Stream(
+                self.generator,
+                gen_kwargs={"stream": stream, "multi_stream": multi_stream},
+            )
             result[stream_name] = stream
 
         return MultiStream(result)
@@ -348,24 +368,44 @@ class InstanceOperatorWithGlobalAccess(StreamingOperator):
         pass
 
 
-class SequntialOperator(MultiStreamOperator):
-    """
-    A class representing a sequential operator in the streaming system.
+class SequentialOperator(MultiStreamOperator):
+    """A class representing a sequential operator in the streaming system.
 
     A sequential operator is a type of `MultiStreamOperator` that applies a sequence of other operators to a `MultiStream`. It maintains a list of `StreamingOperator`s and applies them in order to the `MultiStream`.
     """
 
+    max_steps = None
+
     steps: List[StreamingOperator] = field(default_factory=list)
 
+    def num_steps(self) -> int:
+        return len(self.steps)
+
+    def set_max_steps(self, max_steps):
+        assert (
+            max_steps <= self.num_steps()
+        ), f"Max steps requested ({max_steps}) is larger than defined steps {self.num_steps()}"
+        assert max_steps >= 1, f"Max steps requested ({max_steps}) is less than 1"
+        self.max_steps = max_steps
+
+    def get_last_step_description(self):
+        last_step = (
+            self.max_steps - 1 if self.max_steps is not None else len(self.steps) - 1
+        )
+        description = str(self.steps[last_step])
+        return re.sub(r"\w+=None, ", "", description)
+
+    def _get_max_steps(self):
+        return self.max_steps if self.max_steps is not None else len(self.steps)
+
     def process(self, multi_stream: Optional[MultiStream] = None) -> MultiStream:
-        for operator in self.steps:
+        for operator in self.steps[0 : self._get_max_steps()]:
             multi_stream = operator(multi_stream)
         return multi_stream
 
 
-class SourceSequntialOperator(SequntialOperator):
-    """
-    A class representing a source sequential operator in the streaming system.
+class SourceSequentialOperator(SequentialOperator):
+    """A class representing a source sequential operator in the streaming system.
 
     A source sequential operator is a type of `SequntialOperator` that starts with a source operator. The first operator in its list of steps is a `StreamSource`, which generates the initial `MultiStream` that the other operators then process.
     """
@@ -374,15 +414,17 @@ class SourceSequntialOperator(SequntialOperator):
         return super().__call__()
 
     def process(self, multi_stream: Optional[MultiStream] = None) -> MultiStream:
+        assert (
+            self.num_steps() > 0
+        ), "Calling process on a SourceSequentialOperator without any steps"
         multi_stream = self.steps[0]()
-        for operator in self.steps[1:]:
+        for operator in self.steps[1 : self._get_max_steps()]:
             multi_stream = operator(multi_stream)
         return multi_stream
 
 
-class SequntialOperatorInitilizer(SequntialOperator):
-    """
-    A class representing a sequential operator initializer in the streaming system.
+class SequentialOperatorInitilizer(SequentialOperator):
+    """A class representing a sequential operator initializer in the streaming system.
 
     A sequential operator initializer is a type of `SequntialOperator` that starts with a stream initializer operator. The first operator in its list of steps is a `StreamInitializerOperator`, which generates the initial `MultiStream` based on the provided arguments and keyword arguments.
     """
@@ -392,10 +434,14 @@ class SequntialOperatorInitilizer(SequntialOperator):
             return self.process(*args, **kwargs)
 
     def process(self, *args, **kwargs) -> MultiStream:
+        assert (
+            self.num_steps() > 0
+        ), "Calling process on a SequentialOperatorInitilizer without any steps"
+
         assert isinstance(
             self.steps[0], StreamInitializerOperator
-        ), "The first step in a SequntialOperatorInitilizer must be a StreamInitializerOperator"
+        ), "The first step in a SequentialOperatorInitilizer must be a StreamInitializerOperator"
         multi_stream = self.steps[0](*args, **kwargs)
-        for operator in self.steps[1:]:
+        for operator in self.steps[1 : self._get_max_steps()]:
             multi_stream = operator(multi_stream)
         return multi_stream