在我编写从 HBase 读取并分析数据的代码时候，遇到了一个问题——多重循环嵌套。
众所周知，HBase 本身是没有二级索引的，所有的数据分析都只能靠扫描 rowkey 读出一整个列表的数据，然后再自己提炼加工。而提炼加工的过程就是不断地循环数据列表，对每个数据项进行筛选、修改的过程。
这让我写的很痛苦，我不得不反复不断地嵌套循环。于是我想起了管道，然后我自己实现了一下 Python 没有在语言级别支持的管道运算。
下面这段代码，F 首先继承了 partial，这使得 F(map, lambda x: x*x) 的返回值是一个新的函数。与此同时 F 对象又被实现了 __ror__ 这个魔法方法，这样一来就实现了 other | F(...) 这个运算，在这个运算里我们把左侧的值作为该函数剩余的参数传递给函数并调用它获取返回结果。
from functools import partial
class F(partial):
def __ror__(self, other):
return self(other)
但这个 F 有一个缺点，它只能把左侧的值作为一个整体传递给被绑定的函数，不能传递多个参数。那么再实现一个 FF，使用 Python 解参语法来自动把可迭代对象变成多个参数传递给函数。同样也只有四行代码。
from functools import partial
class FF(partial):
def __ror__(self, other):
return self(*other)
使用样例
看看管道的魅力！
以下代码均基于 Python3，你可以自行在 Python2 中编写代码进行实践。
求十以下所有奇数的合：
range(10) | F(filter, lambda x: x % 2) | F(sum)
筛选字典中值为真的数据：
your_dict | F(lambda d: {k: v for k, v in d.items() if v})
为一段文本每行前面都加上四个空格。
text = text.splitlines() | F(map, lambda line: " " * 4 + line) | F("\n".join)
插入一个函数来打印所有值用于临时 DEBUG
... | F(map, lambda x: print(x) or x) | ...
把同一个参数丢给多个函数处理，并返回它们的处理结果
... | F(map, lambda x: (f0(x), f1(x), f2(x))) | ...
接下来是一点实际应用，以下是一段满足需求的代码：
result = {}
for method in view.__methods__:
if method == "OPTIONS":
continue
method = method.lower()
method_docs = self._generate_method(
getattr(view, method), path, definitions
)
if not method_docs:
continue
result[method] = method_docs
由于这段 for 循环里的代码过多，无法使用推导式进行优化结构。看看完成相同的功能但使用管道的代码：
generate_method_docs = lambda method: (
method,
self._generate_method(getattr(view, method), path, definitions),
)
result = dict(
view.__methods__
| F(map, lambda method: method.lower())
| F(filter, lambda method: method != "options")
| F(map, generate_method_docs)
| F(filter, lambda method_and_docs: bool(method_and_docs[1]))
)
管道的好处
管道中每个函数完成自己的功能，并且不会出现类似于 dict(filter(map(filter(map(...), ...), ...), ...)) 这种嵌套的不友好情况。
管道操作使得数据的流向和阅读、思考代码的顺序达成一致，虽然在性能上有微不足道的损耗。但，你都开始使用 Python 了，在自己的业务代码上扣那么 1μs 的性能（此数字经过 CPython3.7、3.8 的测试得到）似乎没有必要？可能你把 Python 版本从 3.6 升级到 3.9 就足够弥补这方面的损耗了（笑）。
更优雅的写法
实际上，在此文刚写出来的时候我曾经发送到某程序员居多的论坛。在大家的讨论里，我思考了更多的问题。比如有人提出 | F(...) 这种写法很麻烦，每次都需要写一个 F，能不能去掉，然后有人发出了一个不完全的解决方案。
但这个问题实际上是无法从运行时来解决的。因为在运行时解决这个问题的前提是需要代码自己知道它后面是不是还跟着一个 | 来进行管道运算，以便于它判断是该返回一个类似于 F 的对象还是真实的运算结果。这是一个不可能在运行时完成的任务。
在 Python 社区也曾经出现过类似的库，它的解决方案是 PIPE | range(100) | sum | END。通过 PIPE 来创建一个可以反复进行 | 运算的对象，直到这个对象与 END 这个对象进行了 | 运算，于是求值开始。这便失去了 F 的优点——你可以在任何一个管道运算之后停下来并放心的删掉后面所有的代码，你拿到的依旧是真实运算的结果。
最优雅的解决方案是把这件事放在编译期做，在 Python 代码编译到 Python 字节码的时候判断 | 并传参，既不需要额外的运行时负担，又可以做到管道的功能。