complete Log->Observe renaming

README.md

@@ -287,7 +287,7 @@ def integrate_pokemon_cards_into_bigquery(
     end_time: datetime.datetime,
 ) -> None:
     (
-        Pipe(PokemonCardSource(start_time, end_time))
+        Pipe(lambda: PokemonCardSource(start_time, end_time))
         # at this point we have a Pipe[List[Dict[str, Any]]]
 
         # Let's say pokemontcg.io rate limits us to 10 calls per second,

kioss/_execution/_core.py

@@ -37,7 +37,7 @@ def __next__(self) -> R:
                     pass
 
 
-class LoggingIteratorWrapper(IteratorWrapper[T]):
+class ObservingIteratorWrapper(IteratorWrapper[T]):
     def __init__(self, iterator: Iterator[T], what: str, colored: bool) -> None:
         super().__init__(iterator)
         self.what = what

kioss/_pipe.py

@@ -246,7 +246,7 @@ def observe(self, what: str = "elements", colored: bool = False) -> "Pipe[T]":
         Returns:
             Pipe[T]: A new Pipe instance with logging capability.
         """
-        return LogPipe(self, what, colored)
+        return ObservePipe(self, what, colored)
 
     def run(
         self,
@@ -274,7 +274,7 @@ def run(
         max_num_error_samples = self._RUN_MAX_NUM_ERROR_SAMPLES
         pipe = self
 
-        if not isinstance(self, LogPipe):
+        if not isinstance(self, ObservePipe):
             pipe = self.observe("output elements")
 
         error_samples: List[Exception] = []
@@ -366,14 +366,14 @@ def _accept(self, visitor: "Visitor[V]") -> V:
         return visitor.visit_do_pipe(self)
 
 
-class LogPipe(Pipe[Y]):
+class ObservePipe(Pipe[Y]):
     def __init__(self, upstream: Pipe[Y], what: str, colored: bool):
         self.upstream: Pipe[Y] = upstream
         self.what = what
         self.colored = colored
 
     def _accept(self, visitor: "Visitor[V]") -> V:
-        return visitor.visit_log_pipe(self)
+        return visitor.visit_observe_pipe(self)
 
 
 class FlattenPipe(Pipe[Y]):

kioss/_visit/_base.py

@@ -44,5 +44,5 @@ def visit_catch_pipe(self, pipe: _pipe.CatchPipe) -> V:
         ...
 
     @abstractmethod
-    def visit_log_pipe(self, pipe: _pipe.LogPipe) -> V:
+    def visit_observe_pipe(self, pipe: _pipe.ObservePipe) -> V:
         ...

kioss/_visit/_explanation.py

@@ -102,7 +102,7 @@ def visit_catch_pipe(self, pipe: _pipe.CatchPipe) -> Any:
         descr = f"exception instances of class in [{', '.join(map(lambda class_: class_.__name__, pipe.classes))}]{', with an additional `when` condition' if pipe.when is not None else ''}"
         return self.visit_any_pipe(pipe, name, descr)
 
-    def visit_log_pipe(self, pipe: _pipe.LogPipe) -> Any:
-        name = "Log"
+    def visit_observe_pipe(self, pipe: _pipe.ObservePipe) -> Any:
+        name = "Observe"
         descr = f"the evolution of the iteration over '{pipe.what}'"
         return self.visit_any_pipe(pipe, name, descr)

kioss/_visit/_iter.py

@@ -64,7 +64,7 @@ def visit_catch_pipe(self, pipe: _pipe.CatchPipe[T]) -> Iterator[T]:
             pipe.upstream._accept(self), *pipe.classes, when=pipe.when
         )
 
-    def visit_log_pipe(self, pipe: _pipe.LogPipe[T]) -> Iterator[T]:
-        return _core.LoggingIteratorWrapper(
+    def visit_observe_pipe(self, pipe: _pipe.ObservePipe[T]) -> Iterator[T]:
+        return _core.ObservingIteratorWrapper(
             pipe.upstream._accept(self), pipe.what, pipe.colored
         )