Python 编程奇技 | 楠的博客

管道操作符

参考范例

你是否怀念亲爱的管道操作符，比如在其他语言中常见的那样…

1
cat file.txt | grep "hello" | wc -l

1
iris %>% filter(Species == "setosa") %>% select(Sepal.Length, Sepal.Width)

1
import Data.Function ((&))
2

3
[1, 2, 3, 4, 5] & filter odd & map (*2) & sum

其实你也可以在 Python 中拥有它！

实现代码

1
from functools import update_wrapper
2
from typing import Any
3
from collections.abc import Callable
4

5

6
class Pipeline:
7
    def __init__(self, func: Callable | None = None) -> None:
8
        if func is not None:
9
            update_wrapper(self, func)
10
            self.funcs: list[Callable] = [func]
11
        else:
12
            self.funcs: list[Callable] = []
13

14
    def __call__(self, *args: Any, **kwds: Any) -> Any:
15
        """
16
        self(*) => self.funcs[0](self.funcs[1](self.funcs[2](...)))
17
        """
18
        if len(self.funcs) == 0:
19
            return None
20
        res = self.funcs[-1](*args, **kwds)
21
        for func in reversed(self.funcs[:-1]):
22
            try:
23
                res = func(res)
24
            except Exception as e:
25
                print(f"执行 {func} （参数 {res}）时遇到错误:")
26
                raise e
27
        return res
28

29
    def __add__(self, func: Callable | "Pipeline") -> "Pipeline":
30
        """
31
        (f + g)(*) => f(g(*))
32
        """
33
        res = Pipeline()
34
        res.funcs += self.funcs
35
        if isinstance(func, Pipeline):
36
            res.funcs += func.funcs
37
        elif callable(func):
38
            res.funcs.append(func)
39
        else:
40
            raise TypeError(f"不支持的类型 {type(func)}")
41
        return res
42

43
    def __radd__(self, func: Callable | "Pipeline") -> "Pipeline":
44
        res = Pipeline()
45
        if isinstance(func, Pipeline):
46
            res.funcs += func.funcs
47
        elif callable(func):
48
            res.funcs.append(func)
49
        else:
50
            raise TypeError(f"不支持的类型 {type(func)}")
51
        res.funcs += self.funcs
52
        return res
53

54
    def __or__(self, func: Callable | "Pipeline"):
55
        """
56
        * | f | g => g(f(*))
57
        """
58
        return func + self
59

60
    def __ror__(self, func: Callable | "Pipeline" | Any):
61
        if isinstance(func, Callable) or isinstance(func, Pipeline):
62
            return self + func
63
        else:
64
            return self(func)
65

66

67
@Pipeline
68
def foo(a: int) -> list[int]:
69
    return list(range(a))
70

71

72
@Pipeline
73
def bar(b: list[int]) -> int:
74
    return max(b)
75

76

77
# Python 中的管道操作符？！！！
78
if __name__ == "__main__":
79
    func = bar + foo
80
    print(func(10))  # 9
81
    print(10 | foo | bar)  # 9

工作原理

解决方案很简单：

定义一个函数装饰器，其内部保存原始函数（self.funcs 是一个可调用对象列表）
在操作至少一个此类函数时（例如管道操作符 |，在 Python 中视为 or 运算，或者像 f+g <=> f(g(.)) 这样将两个函数“相加”），生成一个新函数
当调用此函数时，它会依次执行所有嵌套的函数
添加一些收尾工作，就大功告成了！

更好的类型提示

查看王奕轩的这个项目，它为类似的管道操作符实现提供了更好的类型提示。

多语句 Lambda 表达式

参考范例

多语句 lambda 表达式在很多语言中都受支持，例如：

1
auto func = [](auto a, auto b)
2
{
3
    a++;
4
    b++;
5
    return a + b;
6
};

1
var func = (int a, int b) =>
2
{
3
    a++;
4
    b++;
5
    return a + b;
6
};

1
func := func(a int, b int) int {
2
    a++
3
    b++
4
    return a + b
5
}

1
const func = (a: number, b: number): number => {
2
  a++;
3
  b++;
4
  return a + b;
5
};

1
const func = (a, b) => {
2
  a++;
3
  b++;
4
  return a + b;
5
};

1
let func = |mut a: i32, mut b: i32| -> i32 {
2
    a += 1;
3
    b += 1;
4
    a + b
5
};

等等！Python 呢？

1
func = lambda a, b: a + b

看起来 Python 只支持单语句 lambda 表达式……是吗？

逗你的啦，你也可以拥有它！

实现代码

1
func = lambda a, b: (
2
    a := a + 1,
3
    b := b + 1,
4
    a + b,
5
)[-1]

工作原理

在 Python 中，元组的求值是顺序执行的，因此你可以使用元组来模拟多语句 lambda 表达式的行为，其中返回值可以通过元素的索引来指定。

注意，为了在元组中赋值，你需要使用 Python 3.8 引入的海象操作符 :=。它会将值赋给左侧的变量，并返回右侧的值，类似于 C/C++ 和许多其他语言中 = 的默认行为。