pypeliner.py

# Pypeliner for Python 3
#
# Usage
#
# [PROLOGUE]
# _ | <Value>
# <Value> | _
#
# [INTERMEDIATE OPERATIONS]
# (pipeline) \
# | "Mapping Function" # applying mappings
#
# [TERMINAL OPERATIONS]
# (pipeline) \
# > "Terminal Operation"
#

import pathlib
from typing import Any, Callable, Generic, TypeVar

T = TypeVar("T")
R = TypeVar("R")
S = TypeVar("S")


class StreamPipeline(Generic[T]):

    def __init__(self, x: T) -> None:
        self.x = x

    def __or__(self, mapper) -> "StreamPipeline":
        """
        Executes "map" opeation.
        """
        return self.__class__(mapper(self.x))

    def __gt__(self, consumer) -> "StreamPipeline":
        """
        Executes "peek" operation.
        """
        consumer(self.x)
        return self

    def __le__(self, right) -> T:
        """
        Unwrap the value
        """
        return self.x


class Prologue:

    def __or__(self, right) -> StreamPipeline:
        return StreamPipeline(right)

    def __call__(self, *args):
        return self

    def __le__(self, right):
        return right.x


_ = Prologue()

if __name__ == "__main__":
    (_
     | "hello"
     | str.upper
     | (lambda x: x.replace("ELL", "ell"))
     ) > print

    hello = _ <= (_
                  | "hello"
                  | str.upper
                  | (lambda x: x.replace("ELL", "ell"))
                  )
    print(hello)

    (_
     | range(0, 100)
     | (lambda c: map(lambda x: x + 1, c))
     | (lambda c: map(lambda x: x * 100, c))
     | list
     ) > print