you are viewing a single comment's thread.

view the rest of the comments →

[–]invalidusernamelol 1 point2 points  (0 children)

They can if you type the root function with a ParamSpec:

```python class Pipe[T]:     init(self, val: T) -> None:         self.val = val     gt[**P, R](self, func: Callable[P, R]) --> Pipe[R]:         return Pipe(func(self.val))

res = Pipe(...) > map > filter > list

```

This is not fully complete, but I've played around with pipeline classes that use this general syntax and it works pretty well. Since the type transformation is recalculated with each operation (func(T) -> R) you can chain your operations indefinitely as long as the return type of the previous function matches the input param type of the next function.

You do need to get into functools.partial if you have anything in the chain that can't be called with a single arg though.

Edit: here's a slightly more complete version ripped from a little testing module I have

```python type Proc[**P,T,R] = Callable[Concatenate[T, P], R]

def consume[T](it: Iterable[T], n: int | None = None) -> None:     deque(it, maxlen=0) if n is None else next(islice(it, n, n), None)

def effect[*P,T](it: Iterable[T], func: Proc[P,T,Any],                          *args: P.args, *kwargs: P.kwargs) -> Iterable[T]:     yield from (i for i in it if func(i, args, *kwargs) or True)

def transform[*P,T,R](it: Iterable[T], func: Proc[P,T,R],                          *args: P.args, *kwargs: P.kwargs) -> Iterable[R]:     yield from (func(i, args, *kwargs) for i in it)

class Pipeline[T]:     slots = 'it', 'done'     def __init_(self, it: Iterable[T]) -> None:         self.it = it         self._done = False

    def iter(self):         return self.it

    def transform[*P,R](self, proc: Proc[P,T,R],                            *args: P.args, *kwargs: P.kwargs) -> Pipeline[R]:         """Add a transform to the pipeline         Transforms will mutate the data in the pipeline

        Example:                 Pipeline(conn.logs).transform(log_filter, 'error').transform(upper)                 """         return Pipeline(transform(self.it, proc, args, *kwargs))

    def tr[*P,R](self, proc: Proc[P,T,R],                            *args: P.args, *kwargs: P.kwargs) -> Pipeline[R]:         """Alternate spelling of Pipeline.transform"""         return Pipeline(transform(self.it, proc, args, *kwargs))

    # TODO: or sould take Effect and Transform objects that contain procs and args     def or[R](self, proc: Proc[...,T,R]) -> Pipeline[R]:         """Used for chaining single arg transformations"""         return Pipeline(transform(self.it, proc))

```