Skip to content

Pipe

Bases: Protocol


              flowchart TD
              pyochain.abc._mixins.Pipe[Pipe]

              

              click pyochain.abc._mixins.Pipe href "" "pyochain.abc._mixins.Pipe"
            

Mixin class providing the pipe method for fluent chaining.

Source code in pyochain/abc/_mixins.pyi
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
class Pipe(Protocol):
    """Mixin class providing the `pipe` method for fluent chaining."""
    def pipe[**P, R](
        self,
        func: Callable[Concatenate[Self, P], R],
        *args: P.args,
        **kwargs: P.kwargs,
    ) -> R:
        """Convert `Self` to `R`.

        This method allows to pipe the instance into an object or function that can convert `Self` into another type.

        Conceptually, this allow to do `x.pipe(f)` instead of `f(x)`, hence keeping a fluent chaining style.

        Args:
            func (Callable[Concatenate[Self, P], R]): Function for conversion.
            *args (P.args): Positional arguments to pass to **func**.
            **kwargs (P.kwargs): Keyword arguments to pass to **func**.

        Returns:
            R: The converted value.

        Example:
            ```python
            from pyochain import Seq, Result, Ok, Err
            from collections.abc import Sequence

            def check_data(data: Sequence[int]) -> Result[Sequence[int], str]:
                if len(data) == 0:
                    return Err("Empty data")
                return Ok(data)

            def handle_result(res: Result[Sequence[int], str]) -> str:
                match res:
                    case Ok(data):
                        return f"Data is valid: {data}"
                    case Err(err):
                        return f"Data is invalid: {err}"

            x = Seq(1, 2, 3).pipe(check_data).pipe(handle_result)
            assert x == "Data is valid: Seq(1, 2, 3)"
            ```
        """

pipe(func, *args, **kwargs)

Convert Self to R.

This method allows to pipe the instance into an object or function that can convert Self into another type.

Conceptually, this allow to do x.pipe(f) instead of f(x), hence keeping a fluent chaining style.

Parameters:

Name Type Description Default
func Callable[Concatenate[Self, P], R]

Function for conversion.

required
*args P.args

Positional arguments to pass to func.

()
**kwargs P.kwargs

Keyword arguments to pass to func.

{}

Returns:

Name Type Description
R R

The converted value.

Example
from pyochain import Seq, Result, Ok, Err
from collections.abc import Sequence

def check_data(data: Sequence[int]) -> Result[Sequence[int], str]:
    if len(data) == 0:
        return Err("Empty data")
    return Ok(data)

def handle_result(res: Result[Sequence[int], str]) -> str:
    match res:
        case Ok(data):
            return f"Data is valid: {data}"
        case Err(err):
            return f"Data is invalid: {err}"

x = Seq(1, 2, 3).pipe(check_data).pipe(handle_result)
assert x == "Data is valid: Seq(1, 2, 3)"
Source code in pyochain/abc/_mixins.pyi
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def pipe[**P, R](
    self,
    func: Callable[Concatenate[Self, P], R],
    *args: P.args,
    **kwargs: P.kwargs,
) -> R:
    """Convert `Self` to `R`.

    This method allows to pipe the instance into an object or function that can convert `Self` into another type.

    Conceptually, this allow to do `x.pipe(f)` instead of `f(x)`, hence keeping a fluent chaining style.

    Args:
        func (Callable[Concatenate[Self, P], R]): Function for conversion.
        *args (P.args): Positional arguments to pass to **func**.
        **kwargs (P.kwargs): Keyword arguments to pass to **func**.

    Returns:
        R: The converted value.

    Example:
        ```python
        from pyochain import Seq, Result, Ok, Err
        from collections.abc import Sequence

        def check_data(data: Sequence[int]) -> Result[Sequence[int], str]:
            if len(data) == 0:
                return Err("Empty data")
            return Ok(data)

        def handle_result(res: Result[Sequence[int], str]) -> str:
            match res:
                case Ok(data):
                    return f"Data is valid: {data}"
                case Err(err):
                    return f"Data is invalid: {err}"

        x = Seq(1, 2, 3).pipe(check_data).pipe(handle_result)
        assert x == "Data is valid: Seq(1, 2, 3)"
        ```
    """