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 flow(source: T1, a: Operation[T1, T2], /) -> T2: ...
22
23
24@overload
25def flow(source: T1, a: Operation[T1, T2], b: Operation[T2, T3], /) -> T3: ...
26
27
28@overload
29def flow(
30 source: T1, a: Operation[T1, T2], b: Operation[T2, T3], c: Operation[T3, T4], /
31) -> T4: ...
32
33
34@overload
35def flow(
36 source: T1,
37 a: Operation[T1, T2],
38 b: Operation[T2, T3],
39 c: Operation[T3, T4],
40 d: Operation[T4, T5],
41 /,
42) -> T5: ...
43
44
45@overload
46def flow(
47 source: T1,
48 a: Operation[T1, T2],
49 b: Operation[T2, T3],
50 c: Operation[T3, T4],
51 d: Operation[T4, T5],
52 e: Operation[T5, T6],
53 /,
54) -> T6: ...
55
56
57@overload
58def flow(
59 source: T1,
60 a: Operation[T1, T2],
61 b: Operation[T2, T3],
62 c: Operation[T3, T4],
63 d: Operation[T4, T5],
64 e: Operation[T5, T6],
65 f: Operation[T6, T7],
66 /,
67) -> T7: ...
68
69
70@overload
71def flow(
72 source: T1,
73 a: Operation[T1, T2],
74 b: Operation[T2, T3],
75 c: Operation[T3, T4],
76 d: Operation[T4, T5],
77 e: Operation[T5, T6],
78 f: Operation[T6, T7],
79 g: Operation[T7, T8],
80 /,
81) -> T8: ...
82
83
84@overload
85def flow(
86 source: T1,
87 a: Operation[T1, T2],
88 b: Operation[T2, T3],
89 c: Operation[T3, T4],
90 d: Operation[T4, T5],
91 e: Operation[T5, T6],
92 f: Operation[T6, T7],
93 g: Operation[T7, T8],
94 h: Operation[T8, T9],
95 /,
96) -> T9: ...
97
98
[docs]
99def flow(source: Any, *operations: Operation[Any, Any]) -> Any:
100 return functools.reduce(lambda val, operator: operator(val), operations, source)