noop

.github/workflows/test.yml

@@ -0,0 +1,40 @@
+name: test
+
+permissions:
+  contents: read
+
+on:
+  - pull_request
+  - push
+
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v4
+      - name: Install poetry
+        run: curl -sSL https://install.python-poetry.org | python3 -
+        env:
+          POETRY_VERSION: 1.7.0
+      - name: Add Poetry to path
+        run: echo "${HOME}/.poetry/bin" >> $GITHUB_PATH
+      - name: Set up Python 3.11
+        uses: actions/setup-python@v4
+        with:
+          python-version: "3.11"
+          cache: "poetry"
+      - name: Install Poetry Packages
+        run: |
+          poetry env use "3.11"
+          poetry install --only dev
+      - name: Add venv to path
+        run: echo `poetry env info --path`/bin/ >> $GITHUB_PATH
+
+      - run: ruff check --output-format github .
+      - run: ruff format --check .
+      - run: mypy --check-untyped-defs .
+      - run: |
+          pip install .
+          pytest -s

.gitignore

@@ -0,0 +1,2 @@
+__pycache__
+.coverage

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2023 Bright Network
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1,5 @@
+## `iter_pipe`: Iterable Pipes
+
+Functional pythonic pipelines for iterables
+
+[Documentation](./tests/docs/)

iter_pipes/__init__.py

@@ -0,0 +1 @@
+from .main import *  # noqa

iter_pipes/functional.py

@@ -0,0 +1,190 @@
+from __future__ import annotations
+
+import math
+from collections import deque
+from collections.abc import Callable, Iterable, Iterator
+from functools import partial
+from itertools import count, groupby
+from typing import Any, Literal, TypeGuard, TypeVar, overload
+
+__all__ = [
+    "map",
+    "filter",
+    "for_each",
+    "for_batch",
+    "batch",
+    "fork",
+]
+
+T = TypeVar("T")
+V = TypeVar("V")
+U = TypeVar("U")
+W = TypeVar("W")
+X = TypeVar("X")
+
+
+raw_filter = filter
+
+
+Step = Callable[[Iterable[T]], Iterable[V]]
+
+
+def map(step: Callable[[V], W]) -> Step[V, W]:
+    def f(data: Iterable[V]) -> Iterable[W]:
+        for item in data:
+            yield step(item)
+
+    return f
+
+
+def for_each(step: Callable[[V], Any]) -> Step[V, V]:
+    def f(data: Iterable[V]) -> Iterable[V]:
+        for item in data:
+            step(item)
+            yield item
+
+    return f
+
+
+def for_batch(step: Callable[[list[V]], Any], batch_size: int) -> Step[V, V]:
+    def f(data: Iterable[V]) -> Iterable[V]:
+        for _, batch_iterator in groupby(
+            zip(data, count()),
+            key=lambda x: math.floor(x[1] / batch_size),
+        ):
+            batch = [x[0] for x in batch_iterator]
+            step(batch)
+            yield from batch
+
+    return f
+
+
+def batch(step: Callable[[list[V]], Iterable[U]], batch_size: int) -> Step[V, U]:
+    def f(data: Iterable[V]) -> Iterable[U]:
+        for _, batch_iterator in groupby(
+            zip(data, count()),
+            key=lambda x: math.floor(x[1] / batch_size),
+        ):
+            yield from step([x[0] for x in batch_iterator])
+
+    return f
+
+
+@overload
+def filter(step: Callable[[V], TypeGuard[W]]) -> Step[V, W]:
+    ...
+
+
+@overload
+def filter(step: Callable[[V], bool]) -> Step[V, V]:
+    ...
+
+
+def filter(step: Callable[[V], bool]) -> Step[V, V]:  # type: ignore
+    return partial(raw_filter, step)  # type: ignore
+
+
+@overload
+def fork(
+    step1: Step[T, U] | None,
+    step2: Step[T, V],
+    pick_first: Literal[True],
+    max_inflight: int | None = ...,
+) -> Step[T, U]:
+    ...
+
+
+@overload
+def fork(
+    step1: Step[T, U],
+    step2: Step[T, V],
+    pick_first: Literal[False] | None,
+    max_inflight: int | None = ...,
+) -> Step[T, V | U]:
+    ...
+
+
+@overload
+def fork(
+    step1: Step[T, U],
+    step2: Step[T, V],
+    step3: Step[T, W],
+    pick_first: Literal[False] | None,
+    max_inflight: int | None = ...,
+) -> Step[T, V | U | W]:
+    ...
+
+
+@overload
+def fork(  # noqa: PLR0913
+    step1: Step[T, U],
+    step2: Step[T, V],
+    step3: Step[T, W],
+    step4: Step[T, X],
+    pick_first: Literal[False] | None,
+    max_inflight: int | None = ...,
+) -> Step[T, V | U | W | X]:
+    ...
+
+
+@overload
+def fork(
+    *steps: Step[T, Any] | None,
+    pick_first: Literal[False] | None,
+    max_inflight: int | None = ...,
+) -> Step[T, Any]:
+    ...
+
+
+def fork(  # type: ignore    # noqa: C901
+    *steps: Step[T, Any] | None,
+    max_inflight: int = 1000,
+    pick_first: bool = False,
+) -> Step[T, Any]:
+    def wrapper(iterable: Iterable[T]) -> Iterable[Any]:  # noqa: C901
+        queues: list[deque] = [deque() for _ in steps]
+        it = iter(iterable)
+        paused_iterators: set[int] = set()
+
+        def gen(i: int) -> Iterator[T]:
+            while True:
+                mydeque = queues[i]
+                if not mydeque:  # when the local deque is empty
+                    try:
+                        newval = next(it)  # fetch a new value and
+                    except StopIteration:
+                        return
+                    for d in queues:  # load it to all the deques
+                        d.append(newval)
+
+                    nb_inflights = sum(len(q) for q in queues)
+                    if nb_inflights > max_inflight:
+                        paused_iterators.add(i)
+                        return
+                yield mydeque.popleft()
+
+        iterators = [iter((steps[i] or identity)(gen(i))) for i in range(len(steps))]
+
+        pending_iterators = set(range(len(iterators)))
+
+        while len(pending_iterators):
+            i = max(  # the index of the iterator with the most inflight items
+                pending_iterators,
+                key=lambda i: len(queues[i]),
+            )
+            try:
+                val = next(iterators[i])
+                if not pick_first or i == 0:
+                    yield val
+            except StopIteration:
+                if i in paused_iterators:  # resume the iterator
+                    iterators[i] = iter((steps[i] or identity)(gen(i)))
+                    paused_iterators.remove(i)
+                else:
+                    pending_iterators.remove(i)
+
+    return wrapper
+
+
+def identity(item: W) -> W:
+    return item