trcks.fp.composition

Types and higher-order functions for function composition.

Attributes:

• Composable1 (TypeAlias) –

  A single function.

• Composable2 (TypeAlias) –

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

• Composable3 (TypeAlias) –

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

• Composable4 (TypeAlias) –

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

• Composable5 (TypeAlias) –

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

• Composable6 (TypeAlias) –

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

• Composable7 (TypeAlias) –

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

• Pipeline0 (TypeAlias) –

  A single value.

• Pipeline1 (TypeAlias) –

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

• Pipeline2 (TypeAlias) –

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

• Pipeline3 (TypeAlias) –

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

• Pipeline4 (TypeAlias) –

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

• Pipeline5 (TypeAlias) –

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

• Pipeline6 (TypeAlias) –

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

• Pipeline7 (TypeAlias) –

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

Examples:

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

>>> from trcks.fp.composition import compose, pipe
>>> 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'

The first function passed to trcks.fp.composition.compose may accept multiple arguments:

>>> from trcks.fp.composition import compose
>>> def repeat(text: str, times: int = 2) -> str:
...     return text * times
...
>>> get_repeated_length = compose(repeat, len)
>>> get_repeated_length("Hi")
4
>>> get_repeated_length("Hi", 3)
6

compose(callable1: Callable[_P0, _T1]) -> Callable[_P0, _T1]

compose(callable1: Callable[_P0, _T1], callable2: Callable[[_T1], _T2]) -> Callable[_P0, _T2]

compose(callable1: Callable[_P0, _T1], callable2: Callable[[_T1], _T2], callable3: Callable[[_T2], _T3]) -> Callable[_P0, _T3]

compose(callable1: Callable[_P0, _T1], callable2: Callable[[_T1], _T2], callable3: Callable[[_T2], _T3], callable4: Callable[[_T3], _T4]) -> Callable[_P0, _T4]

compose(callable1: Callable[_P0, _T1], callable2: Callable[[_T1], _T2], callable3: Callable[[_T2], _T3], callable4: Callable[[_T3], _T4], callable5: Callable[[_T4], _T5]) -> Callable[_P0, _T5]

compose(callable1: Callable[_P0, _T1], callable2: Callable[[_T1], _T2], callable3: Callable[[_T2], _T3], callable4: Callable[[_T3], _T4], callable5: Callable[[_T4], _T5], callable6: Callable[[_T5], _T6]) -> Callable[_P0, _T6]

compose(callable1: Callable[_P0, _T1], callable2: Callable[[_T1], _T2], callable3: Callable[[_T2], _T3], callable4: Callable[[_T3], _T4], callable5: Callable[[_T4], _T5], callable6: Callable[[_T5], _T6], callable7: Callable[[_T6], _T7]) -> Callable[_P0, _T7]

>>> from trcks.fp.composition import compose
>>> def to_length_string(n: int) -> str:
...     return f"Length: {n}"
...
>>> get_length_string = compose(len, to_length_string)
>>> get_length_string("Hello, world!")
'Length: 13'

>>> from trcks.fp.composition import compose
>>> def repeat(text: str, times: int = 2) -> str:
...     return text * times
...
>>> get_repeated_length = compose(repeat, len)
>>> get_repeated_length("Hi")
4
>>> get_repeated_length("Hi", 3)
6

190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234

def compose(  # type: ignore[explicit-any]
    callable1: Callable[_P0, Any],  # pyrefly: ignore[explicit-any]
    /,
    *callables: Callable[[Any], Any],  # pyrefly: ignore[explicit-any]
) -> Callable[_P0, Any]:  # pyrefly: ignore[explicit-any]
    """Concatenate compatible functions from first to last.

    Args:
        callable1: First function.
        callables: Zero to six additional compatible functions.

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

    Examples:
        Sequentially apply two compatible functions to one input value:

        >>> from trcks.fp.composition import compose
        >>> def to_length_string(n: int) -> str:
        ...     return f"Length: {n}"
        ...
        >>> get_length_string = compose(len, to_length_string)
        >>> get_length_string("Hello, world!")
        'Length: 13'

        The first function may accept multiple arguments:

        >>> from trcks.fp.composition import compose
        >>> def repeat(text: str, times: int = 2) -> str:
        ...     return text * times
        ...
        >>> get_repeated_length = compose(repeat, len)
        >>> get_repeated_length("Hi")
        4
        >>> get_repeated_length("Hi", 3)
        6
    """

    def composed(*args: _P0.args, **kwargs: _P0.kwargs) -> Any:  # type: ignore[explicit-any]  # pyrefly: ignore[explicit-any]  # noqa: ANN401
        output = callable1(*args, **kwargs)
        for callable_ in callables:
            output = callable_(output)
        return output

    return composed

pipe(input_: _T0) -> _T0

pipe(input_: _T0, callable1: Callable[[_T0], _T1]) -> _T1

pipe(input_: _T0, callable1: Callable[[_T0], _T1], callable2: Callable[[_T1], _T2]) -> _T2

pipe(input_: _T0, callable1: Callable[[_T0], _T1], callable2: Callable[[_T1], _T2], callable3: Callable[[_T2], _T3]) -> _T3

pipe(input_: _T0, callable1: Callable[[_T0], _T1], callable2: Callable[[_T1], _T2], callable3: Callable[[_T2], _T3], callable4: Callable[[_T3], _T4]) -> _T4

pipe(input_: _T0, callable1: Callable[[_T0], _T1], callable2: Callable[[_T1], _T2], callable3: Callable[[_T2], _T3], callable4: Callable[[_T3], _T4], callable5: Callable[[_T4], _T5]) -> _T5

pipe(input_: _T0, callable1: Callable[[_T0], _T1], callable2: Callable[[_T1], _T2], callable3: Callable[[_T2], _T3], callable4: Callable[[_T3], _T4], callable5: Callable[[_T4], _T5], callable6: Callable[[_T5], _T6]) -> _T6

pipe(input_: _T0, callable1: Callable[[_T0], _T1], callable2: Callable[[_T1], _T2], callable3: Callable[[_T2], _T3], callable4: Callable[[_T3], _T4], callable5: Callable[[_T4], _T5], callable6: Callable[[_T5], _T6], callable7: Callable[[_T6], _T7]) -> _T7

>>> from trcks.fp.composition import pipe
>>> def to_length_string(n: int) -> str:
...     return f"Length: {n}"
...
>>> pipe("Hello, world!", len, to_length_string)
'Length: 13'

311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336

def pipe(input_: Any, /, *callables: Callable[[Any], Any]) -> Any:  # type: ignore[explicit-any]  # pyrefly: ignore[explicit-any]
    """Evaluate a pipeline consisting of a starting value and compatible functions.

    Args:
        input_: Starting value of the pipeline.
        callables:
            Zero to seven compatible functions
            that can be applied to `input_` sequentially from first to last.

    Returns:
        Result of sequentially applying the given functions to the starting value.

    Examples:
        Sequentially apply two compatible functions to one input value:

        >>> from trcks.fp.composition import pipe
        >>> def to_length_string(n: int) -> str:
        ...     return f"Length: {n}"
        ...
        >>> pipe("Hello, world!", len, to_length_string)
        'Length: 13'
    """
    output = input_
    for callable_ in callables:
        output = callable_(output)
    return output