WIP: Functional style pipes

.flake8

@@ -2,3 +2,4 @@
 max-line-length = 88
 max-complexity = 10
 ignore = W391,F401
+exclude=*.pyi

Makefile

@@ -2,4 +2,4 @@ format:
 	poetry run black pipe_framework && poetry run isort pipe_framework && docformatter -r --in-place --force-wrap pipe_framework/*
 
 lint:
-	poetry run flake8 pipe_framework && poetry run isort --check-only
+	poetry run flake8 pipe_framework && poetry run isort --check-only .

pipe_framework/core/__init__.py

@@ -0,0 +1 @@
+# TODO: provide way to pass data from store with underscore notation. e.g.: _.reason

pipe_framework/core/base.py

@@ -0,0 +1,131 @@
+from __future__ import annotations
+
+from typing import (
+    Callable,
+    Generic,
+    Iterable,
+    Iterator,
+    NamedTuple,
+    Protocol,
+    Tuple,
+    TypeVar,
+)
+
+from core.runner import run_simple
+
+StateType = TypeVar("StateType")
+DataType = TypeVar("DataType", covariant=True)
+# Runner = Callable[["Pipe"[StateType]], StateType]
+
+
+class Runner(Generic[StateType], Protocol):
+    def __call__(self, pipe: "Pipe"[StateType]) -> StateType:
+        ...
+
+
+class Hooks(NamedTuple):
+    before: Callable
+    after: Callable
+    interrupt: Callable[..., bool]
+
+
+class Step(Generic[StateType], Protocol):
+    def __call__(self, state: StateType) -> StateType:
+        ...
+
+
+class Pipe(Generic[StateType], Iterable, Protocol):
+    hooks: Hooks
+    state: StateType
+    steps: Tuple[Step[StateType], ...]
+    run: Runner[StateType]
+
+    def __init__(
+        self,
+        steps: Tuple[Step[StateType], ...],
+        state: StateType,
+        hooks: Hooks,
+        runner: Runner[StateType],
+    ) -> None:
+        ...
+
+    def __call__(self) -> StateType:
+        ...
+
+    def __iter__(self) -> Iterator[Step[StateType]]:
+        ...
+
+    def __len__(self) -> int:
+        ...
+
+    def add(self, step: Step) -> "Pipe":
+        ...
+
+    def get_state(self) -> StateType:
+        ...
+
+    def set_state(self, state: StateType) -> "Pipe":
+        ...
+
+    def set_runner(self, runner: Runner[StateType]) -> "Pipe":
+        ...
+
+    def get_runner(self) -> Runner[StateType]:
+        ...
+
+    def handle_interrupt(self) -> bool:
+        ...
+
+
+class SimplePipe(Generic[StateType], Pipe[StateType]):
+    def __init__(
+        self,
+        steps: Tuple[Step[StateType], ...],
+        state: StateType,
+        hooks,
+        runner: Runner[StateType] = run_simple,
+    ):
+        self.steps = steps
+        self.set_runner(runner)
+        self.set_hooks(hooks)
+        self.state = state
+
+    def set_steps(self, steps):
+        map(self.add, steps)
+
+        return self
+
+    def get_state(self):
+        return self.state
+
+    def set_state(self, state: StateType):
+        self.state = state
+
+    def add(self, step):
+        self.steps = self.steps + (step,)
+
+        return self
+
+    def set_runner(self, runner: Runner[StateType]):
+        self.run = runner
+
+    def set_hooks(self, hooks):
+        self.hooks = hooks
+
+        return self
+
+    def get_runner(self):
+        return self.run
+
+    def handle_interrupt(self):
+        # TODO: figure out how to handle this
+        return self.hooks.interrupt()
+
+    def __iter__(self):
+        return iter(self.steps)
+
+    def __len__(self):
+        return len(self.steps)
+
+    def __call__(self):
+        return self.run(self)

pipe_framework/core/decorators.py

@@ -0,0 +1,15 @@
+from functools import wraps
+from typing import Callable
+
+
+def step(fn: Callable) -> Callable:
+    """Decorator to convert function to step.
+
+    :param step_fn: function to convert :return: pipe step
+    """
+
+    @wraps(fn)
+    def step(*args, **kwargs):
+        return fn(*args, **kwargs)
+
+    return step

pipe_framework/core/factories.py

@@ -0,0 +1,27 @@
+from typing import Generic, Tuple, TypeVar
+
+from pipe_framework.core.base import Hooks, SimplePipe, Step
+
+
+def create_hooks(*, before=None, after=None, interrupt=None):
+    bypass = lambda x: x
+
+    return Hooks(
+        before=before or bypass,
+        after=after or bypass,
+        interrupt=interrupt or (lambda s: False),
+    )
+
+
+T = TypeVar("T")
+
+
+def create_simple_pipe(
+    callables: Tuple[Step[T], ...], state: T, hooks: Hooks | None = None
+) -> SimplePipe[T]:
+    if hooks is None:
+        hooks = create_hooks()
+
+    _pipe = SimplePipe(callables, state, hooks)
+
+    return _pipe

pipe_framework/core/runner.py

@@ -0,0 +1,29 @@
+from typing import TypeVar
+
+from core.base import Runner, SimplePipe
+
+T = TypeVar("T")
+
+
+class __run_simple_factory(Runner[T]):
+    """Run the pipe.
+
+    :param pipe: The pipe to run. :param hooks: The hooks to wrap the
+    pipe with. :return: The result of the pipe.
+    """
+
+    def __call__(self, pipe: SimplePipe[T]) -> T:
+        # get initial state for pipe
+        pipe.set_state(pipe.hooks.before(pipe.get_state()))
+
+        for step in pipe:
+            result = step(pipe.get_state())
+            pipe.set_state(result)
+
+            if pipe.hooks.interrupt(pipe.get_state()):
+                break
+
+        return pipe.hooks.after(pipe.get_state())
+
+
+run_simple = __run_simple_factory()

pipe_framework/core/tests/conftest.py

@@ -0,0 +1,46 @@
+from typing import List, TypedDict
+
+import pytest
+from core.base import SimplePipe
+
+from pipe_framework.core.factories import create_simple_pipe
+
+
+class TestState(TypedDict):
+    x: int
+    y: int
+    operations: List[str]
+    result: int
+
+
+def callable_one(state: TestState) -> TestState:
+    state["x"] = 1
+    state["y"] = 1
+    state["operations"] = ["add"]
+
+    return state
+
+
+def callable_two(state):
+    if "add" in state["operations"]:
+        state["result"] = state["x"] + state["y"]
+
+    return state
+
+
+test_state: TestState = {"x": 0, "y": 0, "operations": [], "result": 0}
+
+adder_pipe = create_simple_pipe((callable_one, callable_two), state=test_state)
+adder_pipe_with_hooks = create_simple_pipe(
+    (callable_one, callable_two), state=test_state
+)
+
+
+@pytest.fixture(scope="module")
+def pipe_instance() -> SimplePipe[TestState]:
+    return adder_pipe
+
+
+@pytest.fixture(scope="module")
+def expected_result():
+    return {"x": 1, "y": 1, "operations": ["add"], "result": 2}

pipe_framework/core/tests/test_simple.py

@@ -0,0 +1,8 @@
+from core.base import SimplePipe
+from core.tests.conftest import TestState
+
+
+def test_simple(pipe_instance: SimplePipe[TestState], expected_result):
+    result = pipe_instance()
+
+    assert result == expected_result