python-fp-flow

A minimal FP-oriented library for composing function pipelines in Python.

This project provides two utilities:

• Flow: a value wrapper that lets you pipe transformations with the >> operator.
• chain(...): a function composer that combines multiple unary callables into one callable.

The library is intentionally small and focused: it helps make sequential data transformations explicit and readable.

Why this library?

Python already lets you compose transformations, but once a pipeline has many steps, nested calls become harder to parse.

Assume we want to:

1. trim and lowercase text
2. remove punctuation
3. split into words
4. remove common stopwords
5. sort unique words
6. format as a summary string

Shared setup (used by all examples)

import re

STOPWORDS = {"the", "and", "is", "a", "of", "to"}


def strip_punctuation(s: str) -> str:
    return re.sub(r"[^a-z0-9\s]", "", s)


def to_words(s: str) -> list[str]:
    return s.split()


def remove_stopwords(words: list[str]) -> list[str]:
    return [w for w in words if w not in STOPWORDS]


def unique_sorted(words: list[str]) -> list[str]:
    return sorted(set(words))


def summarize(words: list[str]) -> str:
    return f"keywords({len(words)}): {', '.join(words)}"


input_text = "  The Quick, brown fox jumps over the lazy dog!  "

Without this library (nested calls)

result = summarize(
    unique_sorted(
        remove_stopwords(
            to_words(
                strip_punctuation(
                    input_text.strip().lower()
                )
            )
        )
    )
)

With Flow (left-to-right data pipeline)

from flow import Flow

result = (
    Flow(input_text)
    >> str.strip
    >> str.lower
    >> strip_punctuation
    >> to_words
    >> remove_stopwords
    >> unique_sorted
    >> summarize
).value()

With chain (reusable transformation function)

from chain import chain

text_pipeline = chain(
    str.strip,
    str.lower,
    strip_punctuation,
    to_words,
    remove_stopwords,
    unique_sorted,
    summarize,
)

result = text_pipeline(input_text)

Flow and chain reduce cognitive load by making each step explicit and ordered from top-to-bottom / left-to-right, rather than inside-out.

Requirements

• Python >= 3.14

Installation (local development)

This repository is configured for uv.

uv sync

API overview

Flow

• Flow(value): wraps any value.
• .value(): returns the current wrapped value.
• flow >> fn: applies fn to the wrapped value and returns a new Flow.

chain

• chain(fn1, fn2, ..., fnN): returns a new callable that applies each function in sequence.
• Designed for unary callables (Callable[[T], U]).
• Includes typing overloads for better inference across multi-step pipelines.

Running tests

uv run pytest

Current tests cover:

• basic Flow behavior (value, chaining, repr, str)
• chain composition order and exception propagation