add experimental 'print' function, replace 'update' with 'peek', allo…

…w skipping assignment for peek, print, and produce

quixstreams/dataframe/dataframe.py

@@ -241,29 +241,32 @@ def func(d: dict, state: State):
             )
         return self.__dataframe_clone__(stream=stream)
 
+    def print(self, metadata: bool = False) -> Self:
+        return self.peek(_print, metadata=metadata)
+
     @overload
-    def update(self, func: UpdateCallback) -> Self: ...
+    def peek(self, func: UpdateCallback) -> Self: ...
 
     @overload
-    def update(
+    def peek(
         self, func: UpdateWithMetadataCallback, *, metadata: Literal[True]
     ) -> Self: ...
 
     @overload
-    def update(
+    def peek(
         self, func: UpdateCallbackStateful, *, stateful: Literal[True]
     ) -> Self: ...
 
     @overload
-    def update(
+    def peek(
         self,
         func: UpdateWithMetadataCallbackStateful,
         *,
         stateful: Literal[True],
         metadata: Literal[True],
     ) -> Self: ...
 
-    def update(
+    def peek(
         self,
         func: Union[
             UpdateCallback,
@@ -327,7 +330,7 @@ def func(values: list, state: State):
                 cast(Union[UpdateCallback, UpdateWithMetadataCallback], func),
                 metadata=metadata,
             )
-        return self.__dataframe_clone__(stream=stream)
+        return self.__dataframe_inplace__(stream)
 
     @overload
     def filter(self, func: FilterCallback) -> Self: ...
@@ -570,7 +573,7 @@ def to_topic(
             By default, the current message key will be used.
 
         """
-        return self.update(
+        return self.peek(
             lambda value, orig_key, timestamp, headers: self._produce(
                 topic=topic,
                 value=value,
@@ -927,6 +930,7 @@ def _produce(
         row = Row(
             value=value, key=key, timestamp=timestamp, context=ctx, headers=headers
         )
+        print("PRODUCE!!!")
         self._producer.produce_row(row=row, topic=topic, key=key, timestamp=timestamp)
 
     def _register_store(self):
@@ -960,6 +964,10 @@ def _groupby_key(
         else:
             raise TypeError("group_by 'key' must be callable or string (column name)")
 
+    def __dataframe_inplace__(self, stream: Stream):
+        self._stream = stream
+        return self
+
     def __dataframe_clone__(
         self,
         stream: Optional[Stream] = None,
@@ -1071,6 +1079,15 @@ def wrapper(
     return wrapper
 
 
+import pprint
+
+_PRINT_ARGS = ["value", "key", "timestamp", "headers"]
+
+
+def _print(*args):
+    pprint.pprint({_PRINT_ARGS[i]: args[i] for i in range(len(args))}, indent=2)
+
+
 def _as_stateful(
     func: Union[
         ApplyWithMetadataCallbackStateful,