Simple piping system for Python3. Idea borrowed (of course) from functional languages. Supports async.

pipe.py

from typing import Callable, Any


"""

>>> (P(2) | P(lambda x: x**2) | P(lambda y: y**4))()
256
>>> pipe = (P(2) | P(lambda x: x**2) | P(lambda y: y**4))
>>> pipe.arg
256

>>> import asyncio
>>> async def some_coro(y):
...     await asyncio.sleep(1)
...     return y**2
>>> await (P(2) | P(some_coro))()
4
"""

__all__ = ["Pipe", "P"]


class Pipe:
    """
    Pipe used for piping results from left to right.
    Syntax is pretty much borrowed from functional languages.
    Multilining is done without any "\", because we need to embed Pipes into parenthesis.
    Works pretty much like F#'s "|>" or Bash's "|" operator.
    """
    def __init__(self, arg: Any):
        self.arg: Callable[..., Any] = arg

    def __or__(self, other: "Pipe"):
        if not isinstance(other, Pipe):
            raise TypeError("You need to pipe to Pipe.")
        if not callable(other.arg):
            raise TypeError("Argument must be callable")
        return Pipe(other.arg(self.arg))

    def __call__(self):
        return self.arg


P = Pipe