1from __future__ import annotations
2
3import functools
4from typing import TYPE_CHECKING, Any, TypeVar, overload
5
6if TYPE_CHECKING:
7 from stdlibx.compose._types import Operation
8
9T1 = TypeVar("T1")
10T2 = TypeVar("T2")
11T3 = TypeVar("T3")
12T4 = TypeVar("T4")
13T5 = TypeVar("T5")
14T6 = TypeVar("T6")
15T7 = TypeVar("T7")
16T8 = TypeVar("T8")
17T9 = TypeVar("T9")
18
19
20@overload
21def pipe(a: Operation[T1, T2], b: Operation[T2, T3], /) -> Operation[T1, T3]: ...
22
23
24@overload
25def pipe(
26 a: Operation[T1, T2], b: Operation[T2, T3], c: Operation[T3, T4], /
27) -> Operation[T1, T4]: ...
28
29
30@overload
31def pipe(
32 a: Operation[T1, T2],
33 b: Operation[T2, T3],
34 c: Operation[T3, T4],
35 d: Operation[T4, T5],
36 /,
37) -> Operation[T1, T5]: ...
38
39
40@overload
41def pipe(
42 a: Operation[T1, T2],
43 b: Operation[T2, T3],
44 c: Operation[T3, T4],
45 d: Operation[T4, T5],
46 e: Operation[T5, T6],
47 /,
48) -> Operation[T1, T6]: ...
49
50
51@overload
52def pipe(
53 a: Operation[T1, T2],
54 b: Operation[T2, T3],
55 c: Operation[T3, T4],
56 d: Operation[T4, T5],
57 e: Operation[T5, T6],
58 f: Operation[T6, T7],
59 /,
60) -> Operation[T1, T7]: ...
61
62
63@overload
64def pipe(
65 a: Operation[T1, T2],
66 b: Operation[T2, T3],
67 c: Operation[T3, T4],
68 d: Operation[T4, T5],
69 e: Operation[T5, T6],
70 f: Operation[T6, T7],
71 g: Operation[T7, T8],
72 /,
73) -> Operation[T1, T8]: ...
74
75
76@overload
77def pipe(
78 a: Operation[T1, T2],
79 b: Operation[T2, T3],
80 c: Operation[T3, T4],
81 d: Operation[T4, T5],
82 e: Operation[T5, T6],
83 f: Operation[T6, T7],
84 g: Operation[T7, T8],
85 h: Operation[T9, T9],
86 /,
87) -> Operation[T1, T9]: ...
88
89
[docs]
90def pipe(
91 first: Operation[Any, Any],
92 *others: Operation[Any, Any],
93) -> Operation[Any, Any]:
94 def _apply(value: Any) -> Any:
95 return functools.reduce(
96 lambda val, operation: operation(val),
97 (first, *others),
98 value,
99 )
100
101 return _apply