A .pipe method for passing the Stream to another function

[mjkanji]

@ebonnal Let me know your thoughts on this suggestion.

It'd be nice if the Stream object itself could be passed to a function using the fluent syntax.

Here's a contrived example. Say I want to say "Hello" five times before proceeding to doing something with the stream. I'd like to do the following:

def greet(stream):
    print("Hello")
    return stream

(
  some_stream
  .pipe(greet)
  .pipe(greet)
  ...
  .pipe(greet)
)

I could certainly do

greet(greet(...(greet(stream)...))

or

res = greet(stream)
res = greet(stream)
...
res = greet(stream)

But both of those defeat the purpose of the fluent syntax.

A more typical use case where I'd appreciate this is if I'm using another package that also has a fluent syntax. This'd allow me to consume the stream downstream without having to break the chain syntax.

For example, when working with data that I eventually want to turn into a Polars DataFrame (which also has method chaining), I have to do:

stream = (
  Stream(range(10))
  map(...)
  ... # other operations
)

my_df = (
  pl.DataFrame(stream)
  .select(...)
  ... # polars operations
)

Instead, I'd love it if I could just do:

(
  Stream(range(10))
  ... # streamable operations
  .pipe(pl.DataFrame)
  ... # polars operations
)

I know I could just wrap the first assignment that creates stream in pl.DataFrame at the start,

(
  pl.DataFrame(
    Stream(range(10))
    map(...)
    ... # other operations
  )
  ... # polars operations
)

But the extra indentation is ugly. Also, when writing code interactively, I need to know beforehand that I'd want to wrap the stream in another function — or backtrack and do so plus add indentation. It's just not a fun experience and "breaks the flow", so to speak.

I could create more complex examples with multiple levels of nesting as I interweave between Streamable and Polars multiple times (each time requires an additional level of indent or intermediate assignment), but it's why I have the contrived example up top.

In essence, the whole change should (ideally) be as follows:

class Stream:
  ...
  def pipe(func: Callable, *args, **kwargs):
    return func(self, *args, **kwargs)

This is inspired by Polars' pipe method, which is really helpful when I want to leave Polars world and do something else, without breaking out of the fluent syntax:
https://docs.pola.rs/api/python/stable/reference/dataframe/api/polars.DataFrame.pipe.html

PS: While this does achieve similar outcomes, this is unrelated to #49, as that was borne out of pains with auto-formatting concerns and also proposed a top-level function, while this is a recommendation for a new method for the Stream object itself.

[ebonnal]

This sounds great @mjkanji 🤝 , super useful indeed, same signature pandas-side, let's go for it!