trcks.fp.composition

Types and higher order functions for function composition.

Example

Sequentially apply two functions to one input value in three different ways:

>>> def to_length_string(n: int) -> str:
...     return f"Length: {n}"
...
>>> input_ = "Hello, world!"
>>> to_length_string(len(input_))
'Length: 13'
>>> get_length_string = compose((len, to_length_string))
>>> get_length_string(input_)
'Length: 13'
>>> pipe((input_, len, to_length_string))
'Length: 13'

Composable1 = tuple[Callable[[_T0], _T1],] module-attribute

A single function.

Composable2 = tuple[Callable[[_T0], _T1], Callable[[_T1], _T2]] module-attribute

Two compatible functions that can be applied sequentially from first to last.

Composable3 = tuple[Callable[[_T0], _T1], Callable[[_T1], _T2], Callable[[_T2], _T3]] module-attribute

Three compatible functions that can be applied sequentially from first to last.

Composable4 = tuple[Callable[[_T0], _T1], Callable[[_T1], _T2], Callable[[_T2], _T3], Callable[[_T3], _T4]] module-attribute

Four compatible functions that can be applied sequentially from first to last.

Composable5 = tuple[Callable[[_T0], _T1], Callable[[_T1], _T2], Callable[[_T2], _T3], Callable[[_T3], _T4], Callable[[_T4], _T5]] module-attribute

Five compatible functions that can be applied sequentially from first to last.

Composable6 = tuple[Callable[[_T0], _T1], Callable[[_T1], _T2], Callable[[_T2], _T3], Callable[[_T3], _T4], Callable[[_T4], _T5], Callable[[_T5], _T6]] module-attribute

Six compatible functions that can be applied sequentially from first to last.

Composable7 = tuple[Callable[[_T0], _T1], Callable[[_T1], _T2], Callable[[_T2], _T3], Callable[[_T3], _T4], Callable[[_T4], _T5], Callable[[_T5], _T6], Callable[[_T6], _T7]] module-attribute

Seven compatible functions that can be applied sequentially from first to last.

Composable = Composable7[_IN, _T1, _T2, _T3, _T4, _T5, _T6, _OUT] | Composable6[_IN, _T1, _T2, _T3, _T4, _T5, _OUT] | Composable5[_IN, _T1, _T2, _T3, _T4, _OUT] | Composable4[_IN, _T1, _T2, _T3, _OUT] | Composable3[_IN, _T1, _T2, _OUT] | Composable2[_IN, _T1, _OUT] | Composable1[_IN, _OUT] module-attribute

Up to seven compatible functions that can be applied sequentially from first to last.

Pipeline0 = tuple[_T0,] module-attribute

A single value.

Pipeline1 = tuple[_T0, Callable[[_T0], _T1]] module-attribute

A single value followed by a single compatible function that can be applied.

Pipeline2 = tuple[_T0, Callable[[_T0], _T1], Callable[[_T1], _T2]] module-attribute

A single value followed by two compatible functions that can be applied sequentially from first to last.

Pipeline3 = tuple[_T0, Callable[[_T0], _T1], Callable[[_T1], _T2], Callable[[_T2], _T3]] module-attribute

A single value followed by three compatible functions that can be applied sequentially from first to last.

Pipeline4 = tuple[_T0, Callable[[_T0], _T1], Callable[[_T1], _T2], Callable[[_T2], _T3], Callable[[_T3], _T4]] module-attribute

A single value followed by four compatible functions that can be applied sequentially from first to last.

Pipeline5 = tuple[_T0, Callable[[_T0], _T1], Callable[[_T1], _T2], Callable[[_T2], _T3], Callable[[_T3], _T4], Callable[[_T4], _T5]] module-attribute

A single value followed by five compatible functions that can be applied sequentially from first to last.

Pipeline6 = tuple[_T0, Callable[[_T0], _T1], Callable[[_T1], _T2], Callable[[_T2], _T3], Callable[[_T3], _T4], Callable[[_T4], _T5], Callable[[_T5], _T6]] module-attribute

A single value followed by six compatible functions that can be applied sequentially from first to last.

Pipeline7 = tuple[_T0, Callable[[_T0], _T1], Callable[[_T1], _T2], Callable[[_T2], _T3], Callable[[_T3], _T4], Callable[[_T4], _T5], Callable[[_T5], _T6], Callable[[_T6], _T7]] module-attribute

A single value followed by seven compatible functions that can be applied sequentially from first to last.

Pipeline = Pipeline7[_T0, _T1, _T2, _T3, _T4, _T5, _T6, _OUT] | Pipeline6[_T0, _T1, _T2, _T3, _T4, _T5, _OUT] | Pipeline5[_T0, _T1, _T2, _T3, _T4, _OUT] | Pipeline4[_T0, _T1, _T2, _T3, _OUT] | Pipeline3[_T0, _T1, _T2, _OUT] | Pipeline2[_T0, _T1, _OUT] | Pipeline1[_T0, _OUT] | Pipeline0[_OUT] module-attribute

A single value followed by up to seven compatible functions that can be applied sequentially from first to last.

compose1(c)

Compose a single function.

Parameters:

• c (Composable1[_T0, _T1]) –

  A single function.

Returns:

• Callable[[_T0], _T1] –

  Function that applies the given function.

Example

>>> get_length = compose1((len,))
>>> get_length("Hello, world!")
13

Source code in src/trcks/fp/composition.py

def compose1(c: Composable1[_T0, _T1]) -> Callable[[_T0], _T1]:
    """Compose a single function.

    Args:
        c: A single function.

    Returns:
        Function that applies the given function.

    Example:
        >>> get_length = compose1((len,))
        >>> get_length("Hello, world!")
        13
    """
    return lambda t0: c[0](t0)  # noqa: PLW0108 # for uniformity with other compose* functions

compose2(c)

Compose two compatible functions from first to last.

Parameters:

• c (Composable2[_T0, _T1, _T2]) –

  Two compatible functions that can be applied sequentially from first to last.

Returns:

• Callable[[_T0], _T2] –

  Function that applies the given functions from first to last.

Example

>>> get_length_string = compose2((len, lambda n: f"Length: {n}"))
>>> get_length_string("Hello, world!")
'Length: 13'

Source code in src/trcks/fp/composition.py

def compose2(c: Composable2[_T0, _T1, _T2]) -> Callable[[_T0], _T2]:
    """Compose two compatible functions from first to last.

    Args:
        c: Two compatible functions that can be applied sequentially from first to last.

    Returns:
        Function that applies the given functions from first to last.

    Example:
        >>> get_length_string = compose2((len, lambda n: f"Length: {n}"))
        >>> get_length_string("Hello, world!")
        'Length: 13'
    """
    return lambda t0: c[1](c[0](t0))

compose3(c)

Compose three compatible functions from first to last.

Parameters:

• c (Composable3[_T0, _T1, _T2, _T3]) –

  Three compatible functions that can be applied sequentially from first to last.

Returns:

• Callable[[_T0], _T3] –

  Function that applies the given functions from first to last.

Example

>>> add_one = lambda n: n + 1
>>> square = lambda n: n * n
>>> to_string = lambda n: f"Result: {n}"
>>> compute = compose3((add_one, square, to_string))
>>> compute(3)
'Result: 16'

Source code in src/trcks/fp/composition.py

def compose3(c: Composable3[_T0, _T1, _T2, _T3]) -> Callable[[_T0], _T3]:
    """Compose three compatible functions from first to last.

    Args:
        c:
            Three compatible functions that can be applied sequentially
            from first to last.

    Returns:
        Function that applies the given functions from first to last.

    Example:
        >>> add_one = lambda n: n + 1
        >>> square = lambda n: n * n
        >>> to_string = lambda n: f"Result: {n}"
        >>> compute = compose3((add_one, square, to_string))
        >>> compute(3)
        'Result: 16'
    """
    return lambda t0: c[2](c[1](c[0](t0)))

compose4(c)

Compose four compatible functions from first to last.

Parameters:

• c (Composable4[_T0, _T1, _T2, _T3, _T4]) –

  Four compatible functions that can be applied sequentially from first to last.

Returns:

• Callable[[_T0], _T4] –

  Function that applies the given functions from first to last.

Example

>>> add_one = lambda n: n + 1
>>> square = lambda n: n * n
>>> halve = lambda n: n / 2
>>> to_string = lambda n: f"Result: {n}"
>>> compute = compose4((add_one, square, halve, to_string))
>>> compute(3)
'Result: 8.0'

Source code in src/trcks/fp/composition.py

def compose4(c: Composable4[_T0, _T1, _T2, _T3, _T4]) -> Callable[[_T0], _T4]:
    """Compose four compatible functions from first to last.

    Args:
        c:
            Four compatible functions that can be applied sequentially
            from first to last.

    Returns:
        Function that applies the given functions from first to last.

    Example:
        >>> add_one = lambda n: n + 1
        >>> square = lambda n: n * n
        >>> halve = lambda n: n / 2
        >>> to_string = lambda n: f"Result: {n}"
        >>> compute = compose4((add_one, square, halve, to_string))
        >>> compute(3)
        'Result: 8.0'
    """
    return lambda t0: c[3](c[2](c[1](c[0](t0))))

compose5(c)

Compose five compatible functions from first to last.

Parameters:

• c (Composable5[_T0, _T1, _T2, _T3, _T4, _T5]) –

  Five compatible functions that can be applied sequentially from first to last.

Returns:

• Callable[[_T0], _T5] –

  Function that applies the given functions from first to last.

Example

>>> add_one = lambda n: n + 1
>>> square = lambda n: n * n
>>> halve = lambda n: n / 2
>>> to_string = lambda n: f"Result: {n}"
>>> exclaim = lambda s: s + "!"
>>> compute = compose5((add_one, square, halve, to_string, exclaim))
>>> compute(3)
'Result: 8.0!'

Source code in src/trcks/fp/composition.py

def compose5(c: Composable5[_T0, _T1, _T2, _T3, _T4, _T5]) -> Callable[[_T0], _T5]:
    """Compose five compatible functions from first to last.

    Args:
        c:
            Five compatible functions that can be applied sequentially
            from first to last.

    Returns:
        Function that applies the given functions from first to last.

    Example:
        >>> add_one = lambda n: n + 1
        >>> square = lambda n: n * n
        >>> halve = lambda n: n / 2
        >>> to_string = lambda n: f"Result: {n}"
        >>> exclaim = lambda s: s + "!"
        >>> compute = compose5((add_one, square, halve, to_string, exclaim))
        >>> compute(3)
        'Result: 8.0!'
    """
    return lambda t0: c[4](c[3](c[2](c[1](c[0](t0)))))

compose6(c)

Compose six compatible functions from first to last.

Parameters:

• c (Composable6[_T0, _T1, _T2, _T3, _T4, _T5, _T6]) –

  Six compatible functions that can be applied sequentially from first to last.

Returns:

• Callable[[_T0], _T6] –

  Function that applies the given functions from first to last.

Example

>>> add_one = lambda n: n + 1
>>> square = lambda n: n * n
>>> halve = lambda n: n / 2
>>> to_string = lambda n: f"Result: {n}"
>>> exclaim = lambda s: s + "!"
>>> to_list = lambda s: [s]
>>> compute = compose6((add_one, square, halve, to_string, exclaim, to_list))
>>> compute(3)
['Result: 8.0!']

Source code in src/trcks/fp/composition.py

def compose6(c: Composable6[_T0, _T1, _T2, _T3, _T4, _T5, _T6]) -> Callable[[_T0], _T6]:
    """Compose six compatible functions from first to last.

    Args:
        c: Six compatible functions that can be applied sequentially from first to last.

    Returns:
        Function that applies the given functions from first to last.

    Example:
        >>> add_one = lambda n: n + 1
        >>> square = lambda n: n * n
        >>> halve = lambda n: n / 2
        >>> to_string = lambda n: f"Result: {n}"
        >>> exclaim = lambda s: s + "!"
        >>> to_list = lambda s: [s]
        >>> compute = compose6((add_one, square, halve, to_string, exclaim, to_list))
        >>> compute(3)
        ['Result: 8.0!']
    """
    return lambda t0: c[5](c[4](c[3](c[2](c[1](c[0](t0))))))

compose7(c)

Compose seven compatible functions from first to last.

Parameters:

• c (Composable7[_T0, _T1, _T2, _T3, _T4, _T5, _T6, _T7]) –

  Seven compatible functions that can be applied sequentially from first to last.

Returns:

• Callable[[_T0], _T7] –

  Function that applies the given functions from first to last.

Example

>>> add_one = lambda n: n + 1
>>> square = lambda n: n * n
>>> halve = lambda n: n / 2
>>> to_string = lambda n: f"Result: {n}"
>>> exclaim = lambda s: s + "!"
>>> to_list = lambda s: [s]
>>> wrap_in_dict = lambda lst: {"result": lst}
>>> compute = compose7(
...     (add_one, square, halve, to_string, exclaim, to_list, wrap_in_dict)
... )
>>> compute(3)
{'result': ['Result: 8.0!']}

Source code in src/trcks/fp/composition.py

def compose7(
    c: Composable7[_T0, _T1, _T2, _T3, _T4, _T5, _T6, _T7],
) -> Callable[[_T0], _T7]:
    """Compose seven compatible functions from first to last.

    Args:
        c:
            Seven compatible functions that can be applied sequentially
            from first to last.

    Returns:
        Function that applies the given functions from first to last.

    Example:
        >>> add_one = lambda n: n + 1
        >>> square = lambda n: n * n
        >>> halve = lambda n: n / 2
        >>> to_string = lambda n: f"Result: {n}"
        >>> exclaim = lambda s: s + "!"
        >>> to_list = lambda s: [s]
        >>> wrap_in_dict = lambda lst: {"result": lst}
        >>> compute = compose7(
        ...     (add_one, square, halve, to_string, exclaim, to_list, wrap_in_dict)
        ... )
        >>> compute(3)
        {'result': ['Result: 8.0!']}
    """
    return lambda t0: c[6](c[5](c[4](c[3](c[2](c[1](c[0](t0)))))))

compose(c)

Compose a tuple of compatible functions from first to last.

Parameters:

• c (Composable[_IN, _T1, _T2, _T3, _T4, _T5, _T6, _OUT]) –

  Compatible functions that can be applied sequentially from first to last.

Returns:

• Callable[[_IN], _OUT] –

  Function that applies the given functions from first to last.

Example

>>> get_length_string = compose((len, lambda n: f"Length: {n}"))
>>> get_length_string("Hello, world!")
'Length: 13'

Source code in src/trcks/fp/composition.py

def compose(  # noqa: PLR0911
    c: Composable[_IN, _T1, _T2, _T3, _T4, _T5, _T6, _OUT],
) -> Callable[[_IN], _OUT]:
    """Compose a tuple of compatible functions from first to last.

    Args:
        c: Compatible functions that can be applied sequentially from first to last.

    Returns:
        Function that applies the given functions from first to last.

    Example:
        >>> get_length_string = compose((len, lambda n: f"Length: {n}"))
        >>> get_length_string("Hello, world!")
        'Length: 13'
    """
    match c:
        case (_,):
            return compose1(c)
        case (_, _):
            return compose2(c)
        case (_, _, _):
            return compose3(c)
        case (_, _, _, _):
            return compose4(c)
        case (_, _, _, _, _):
            return compose5(c)
        case (_, _, _, _, _, _):
            return compose6(c)
        case (_, _, _, _, _, _, _):
            return compose7(c)
        case _:  # pragma: no cover
            assert_type(c, Never)  # type: ignore[unreachable]  # pyright: ignore[reportUnreachable]
            msg = f"{type(c).__name__!r} is not a valid Composable"
            raise TypeError(msg)

pipe(p)

Evaluate a Pipeline.

Parameters:

• p (Pipeline[_T0, _T1, _T2, _T3, _T4, _T5, _T6, _OUT]) –

  Single value followed by up to seven compatible functions that can be applied sequentially from first to last.

Returns:

• _OUT –

  Result of sequentially applying the given functions from first to last to the given value.

Example

>>> pipe(("Hello, world!", len, lambda n: f"Length: {n}"))
'Length: 13'

Source code in src/trcks/fp/composition.py

def pipe(p: Pipeline[_T0, _T1, _T2, _T3, _T4, _T5, _T6, _OUT]) -> _OUT:
    """Evaluate a `Pipeline`.

    Args:
        p:
            Single value followed by up to seven compatible functions
            that can be applied sequentially from first to last.

    Returns:
        Result of sequentially applying the given functions from first to last
            to the given value.

    Example:
        >>> pipe(("Hello, world!", len, lambda n: f"Length: {n}"))
        'Length: 13'
    """
    match p:
        case (value,):
            return value
        case (value, _, *_):
            composable = p[1:]
            return compose(composable)(value)
        case _:  # pragma: no cover
            assert_type(p, Never)  # type: ignore[unreachable]  # pyright: ignore[reportUnreachable]  # pyrefly: ignore [assert-type]
            msg = f"{type(p).__name__!r} is not a valid Pipeline"
            raise TypeError(msg)