Pipe — 中缀编程工具包

模块，启用类似于sh的中缀语法（使用管道）。

简介

例如，以下是对第二个欧拉项目问题的解决方案

找出所有不大于四百万的斐波那契数中偶数项的和。

给定 fib 一个斐波那契数生成器

sum(fib() | where(lambda x: x % 2 == 0) | take_while(lambda x: x < 4000000))

每个管道都是惰性评估的，可以重命名，并部分初始化，因此可以重写为

is_even = where(lambda x: x % 2 == 0)
sum(fib() | is_even | take_while(lambda x: x < 4000000)

安装

要安装库，只需运行以下命令

# Linux/macOS
python3 -m pip install pipe

# Windows
py -3 -m pip install pipe

使用

基本语法是在使用类似于shell中的 |

>>> from itertools import count
>>> from pipe import select, take
>>> sum(count() | select(lambda x: x ** 2) | take(10))
285
>>>

某些管道需要参数

>>> from pipe import where
>>> sum([1, 2, 3, 4] | where(lambda x: x % 2 == 0))
6
>>>

某些不需要

>>> from pipe import traverse
>>> for i in [1, [2, 3], 4] | traverse:
...     print(i)
1
2
3
4
>>>

在这种情况下，允许使用调用括号

>>> from pipe import traverse
>>> for i in [1, [2, 3], 4] | traverse():
...     print(i)
1
2
3
4
>>>

本模块中现有的管道

可用管道的字母顺序列表；对于给定管道的多个名称，这些是别名。

batched

类似于Python 3.12 itertool.batched

>>> from pipe import batched
>>> list("ABCDEFG" | batched(3))
[('A', 'B', 'C'), ('D', 'E', 'F'), ('G',)]
>>>

chain

链式连接一系列可迭代对象

>>> from pipe import chain
>>> list([[1, 2], [3, 4], [5]] | chain)
[1, 2, 3, 4, 5]
>>>

警告：chain仅展开只包含可迭代对象的序列

[1, 2, [3]] | chain

给出一个 TypeError: chain参数#1必须支持迭代 考虑使用traverse。

chain_with(other)

类似于itertools.chain，生成给定可迭代对象的元素，然后生成其参数的元素

>>> from pipe import chain_with
>>> list((1, 2, 3) | chain_with([4, 5], [6]))
[1, 2, 3, 4, 5, 6]
>>>

dedup(key=None)

使用提供的key函数（如果提供）去重值

>>> from pipe import dedup
>>> list([-1, 0, 0, 0, 1, 2, 3] | dedup)
[-1, 0, 1, 2, 3]
>>> list([-1, 0, 0, 0, 1, 2, 3] | dedup(key=abs))
[-1, 0, 2, 3]
>>>

enumerate(start=0)

内建的enumerate()作为一个管道

>>> from pipe import enumerate
>>> list(['apple', 'banana', 'citron'] | enumerate)
[(0, 'apple'), (1, 'banana'), (2, 'citron')]
>>> list(['car', 'truck', 'motorcycle', 'bus', 'train'] | enumerate(start=6))
[(6, 'car'), (7, 'truck'), (8, 'motorcycle'), (9, 'bus'), (10, 'train')]
>>>

filter(predicate)

是where(predicate)的别名，见where(predicate)。

groupby(key=None)

类似于itertools.groupby(sorted(iterable, key = keyfunc), keyfunc)

>>> from pipe import groupby, map
>>> items = range(10)
>>> ' / '.join(items | groupby(lambda x: "Odd" if x % 2 else "Even")
...                  | select(lambda x: "{}: {}".format(x[0], ', '.join(x[1] | map(str)))))
'Even: 0, 2, 4, 6, 8 / Odd: 1, 3, 5, 7, 9'
>>>

islice()

仅是itertools.islice函数作为一个管道

>>> from pipe import islice
>>> list((1, 2, 3, 4, 5, 6, 7, 8, 9) | islice(2, 8, 2))
[3, 5, 7]
>>>

izip()

仅作为管道使用的 itertools.izip 函数

>>> from pipe import izip
>>> list(range(0, 10) | izip(range(1, 11)))
[(0, 1), (1, 2), (2, 3), (3, 4), (4, 5), (5, 6), (6, 7), (7, 8), (8, 9), (9, 10)]
>>>

map()，select()

将给定的转换表达式应用于给定可迭代对象的每个元素

>>> list([1, 2, 3] | map(lambda x: x * x))
[1, 4, 9]

>>> list([1, 2, 3] | select(lambda x: x * x))
[1, 4, 9]
>>>

netcat

netcat 管道通过 TCP 发送和接收字节

data = [
    b"HEAD / HTTP/1.0\r\n",
    b"Host: python.org\r\n",
    b"\r\n",
]
for packet in data | netcat("python.org", 80):
    print(packet.decode("UTF-8"))

提供

HTTP/1.1 301 Moved Permanently
Content-length: 0
Location: https://pythonlang.cn/
Connection: close

permutations(r=None)

返回所有可能的排列

>>> from pipe import permutations
>>> for item in 'ABC' | permutations(2):
...     print(item)
('A', 'B')
('A', 'C')
('B', 'A')
('B', 'C')
('C', 'A')
('C', 'B')
>>>

>>> for item in range(3) | permutations:
...     print(item)
(0, 1, 2)
(0, 2, 1)
(1, 0, 2)
(1, 2, 0)
(2, 0, 1)
(2, 1, 0)
>>>

reverse

类似于 Python 内置的 reversed 函数。

>>> from pipe import reverse
>>> list([1, 2, 3] | reverse)
[3, 2, 1]
>>>

select(fct)

是 map(fct) 的别名，参见 map(fct)。

skip()

跳过给定数量的元素，然后产生

>>> from pipe import skip
>>> list((1, 2, 3, 4, 5) | skip(2))
[3, 4, 5]
>>>

skip_while(predicate)

类似于 itertools.dropwhile，在谓词为真时跳过给定可迭代对象的元素，然后产生其他元素

>>> from pipe import skip_while
>>> list([1, 2, 3, 4] | skip_while(lambda x: x < 3))
[3, 4]
>>>

sort(key=None, reverse=False)

类似于 Python 内置的 "sorted" 原语。

>>> from pipe import sort
>>> ''.join("python" | sort)
'hnopty'
>>> [5, -4, 3, -2, 1] | sort(key=abs)
[1, -2, 3, -4, 5]
>>>

t

类似于 Haskell 的运算符 ":"

>>> from pipe import t
>>> for i in 0 | t(1) | t(2):
...     print(i)
0
1
2
>>>

tail(n)

产生给定可迭代对象最后给定数量的元素。

>>> from pipe import tail
>>> for i in (1, 2, 3, 4, 5) | tail(3):
...     print(i)
3
4
5
>>>

take(n)

产生给定可迭代对象给定数量的元素，类似于 shell 脚本中的 head。

>>> from pipe import take
>>> for i in count() | take(5):
...     print(i)
0
1
2
3
4
>>>

take_while(predicate)

类似于 itertools.takewhile，在谓词为真时产生给定可迭代对象的元素

>>> from pipe import take_while
>>> for i in count() | take_while(lambda x: x ** 2 < 100):
...     print(i)
0
1
2
3
4
5
6
7
8
9
>>>

tee

tee 输出到标准输出并产生未更改的项目，对于逐步调试管道阶段非常有用

>>> from pipe import tee
>>> sum(["1", "2", "3", "4", "5"] | tee | map(int) | tee)
'1'
1
'2'
2
'3'
3
'4'
4
'5'
5
15
>>>

最后的 15 是返回 sum。

transpose()

转置矩阵的行和列。

>>> from pipe import transpose
>>> [[1, 2, 3], [4, 5, 6], [7, 8, 9]] | transpose
[(1, 4, 7), (2, 5, 8), (3, 6, 9)]
>>>

traverse

递归地展开可迭代对象

>>> list([[1, 2], [[[3], [[4]]], [5]]] | traverse)
[1, 2, 3, 4, 5]
>>> squares = (i * i for i in range(3))
>>> list([[0, 1, 2], squares] | traverse)
[0, 1, 2, 0, 1, 4]
>>>

uniq(key=None)

类似于 dedup()，但仅删除连续值，如果提供（否则使用身份）使用给定的 key 函数。

>>> from pipe import uniq
>>> list([1, 1, 2, 2, 3, 3, 1, 2, 3] | uniq)
[1, 2, 3, 1, 2, 3]
>>> list([1, -1, 1, 2, -2, 2, 3, 3, 1, 2, 3] | uniq(key=abs))
[1, 2, 3, 1, 2, 3]
>>>

where(predicate)，filter(predicate)

仅产生给定可迭代对象的匹配项

>>> list([1, 2, 3] | where(lambda x: x % 2 == 0))
[2]
>>>

别忘了它们可以被别名为

>>> positive = where(lambda x: x > 0)
>>> negative = where(lambda x: x < 0)
>>> sum([-10, -5, 0, 5, 10] | positive)
15
>>> sum([-10, -5, 0, 5, 10] | negative)
-15
>>>

自行构建

您可以使用 Pipe 类构建管道，如下所示

from pipe import Pipe
square = Pipe(lambda iterable: (x ** 2 for x in iterable))
map = Pipe(lambda iterable, fct: builtins.map(fct, iterable)
>>>

如您所见，编写通常非常短，并且如果您有点运气，您正在包装的函数已经接受可迭代对象作为第一个参数，这使得包装变得直接

>>> from collections import deque
>>> from pipe import Pipe
>>> end = Pipe(deque)
>>>

并且这就是 itrable | end(3) 是 deque(iterable, 3) 的原因

>>> list(range(100) | end(3))
[97, 98, 99]
>>>

如果情况更复杂，可以将 Pipe 作为装饰器用于接受可迭代对象作为第一个参数的函数，以及任何其他可选参数

>>> from statistics import mean

>>> @Pipe
... def running_average(iterable, width):
...     items = deque(maxlen=width)
...     for item in iterable:
...         items.append(item)
...         yield mean(items)

>>> list(range(20) | running_average(width=2))
[0, 0.5, 1.5, 2.5, 3.5, 4.5, 5.5, 6.5, 7.5, 8.5, 9.5, 10.5, 11.5, 12.5, 13.5, 14.5, 15.5, 16.5, 17.5, 18.5]
>>> list(range(20) | running_average(width=10))
[0, 0.5, 1, 1.5, 2, 2.5, 3, 3.5, 4, 4.5, 5.5, 6.5, 7.5, 8.5, 9.5, 10.5, 11.5, 12.5, 13.5, 14.5]
>>>

一次性管道

有时您只需要一行代码，创建管道时，您可以直接指定函数的位置参数和命名参数

>>> from itertools import combinations

>>> list(range(5) | Pipe(combinations, 2))
[(0, 1), (0, 2), (0, 3), (0, 4), (1, 2), (1, 3), (1, 4), (2, 3), (2, 4), (3, 4)]
>>>

具有初始起始值的一个简单的运行总和

>>> from itertools import accumulate

>>> list(range(10) | Pipe(accumulate, initial=1))
[1, 1, 2, 4, 7, 11, 16, 22, 29, 37, 46]
>>>

或者根据某些标准过滤数据

>>> from itertools import compress

list(range(20) | Pipe(compress, selectors=[1, 0] * 10))
[0, 2, 4, 6, 8, 10, 12, 14, 16, 18]
>>> list(range(20) | Pipe(compress, selectors=[0, 1] * 10))
[1, 3, 5, 7, 9, 11, 13, 15, 17, 19]
>>>

Euler 项目示例

找出所有小于 1000 的 3 或 5 的倍数的和。

>>> sum(count() | where(lambda x: x % 3 == 0 or x % 5 == 0) | take_while(lambda x: x < 1000))
233168
>>>

找出所有不大于四百万的斐波那契数中偶数项的和。

sum(fib() | where(lambda x: x % 2 == 0) | take_while(lambda x: x < 4000000))

找出前 100 个自然数的平方和与平方和之间的差。

>>> square = map(lambda x: x ** 2)
>>> sum(range(101)) ** 2 - sum(range(101) | square)
25164150
>>>

进一步探索

部分管道

管道可以参数化而不进行评估

>>> running_average_of_two = running_average(2)
>>> list(range(20) | running_average_of_two)
[0, 0.5, 1.5, 2.5, 3.5, 4.5, 5.5, 6.5, 7.5, 8.5, 9.5, 10.5, 11.5, 12.5, 13.5, 14.5, 15.5, 16.5, 17.5, 18.5]
>>>

对于多参数管道，可以部分初始化，您可以将它视为 currying

some_iterable | some_pipe(1, 2, 3)
some_iterable | Pipe(some_func, 1, 2, 3)

严格等同于

some_iterable | some_pipe(1)(2)(3)

因此，它可以用来专门化管道，首先是一个示例

>>> @Pipe
... def addmul(iterable, to_add, to_mul):
...     """Computes (x + to_add) * to_mul to every items of the input."""
...     for i in iterable:
...         yield (i + to_add) * to_mul

>>> mul = addmul(0)  # This partially initialize addmul with to_add=0
>>> list(range(10) | mul(10))
[0, 10, 20, 30, 40, 50, 60, 70, 80, 90]

它也适用于关键字参数

>>> add = addmul(to_mul=1)  # This partially initialize addmul with `to_mul=1`
>>> list(range(10) | add(10))
[10, 11, 12, 13, 14, 15, 16, 17, 18, 19]
>>>

但现在让我们做些有趣的事情

>>> import re
>>> @Pipe
... def grep(iterable, pattern, flags=0):
...     for line in iterable:
...         if re.match(pattern, line, flags=flags):
...             yield line
...
>>> lines = ["Hello", "hello", "World", "world"]
>>> for line in lines | grep("H"):
...     print(line)
Hello
>>>

现在让我们以两种方式重用它，首先是一个模式

>>> lowercase_only = grep("[a-z]+$")
>>> for line in lines | lowercase_only:
...     print(line)
hello
world
>>>

或者现在是一个标志

>>> igrep = grep(flags=re.IGNORECASE)
>>> for line in lines | igrep("hello"):
...    print(line)
...
Hello
hello
>>>

延迟评估

管道一路使用生成器，因此它自然是延迟的。

在以下示例中，我们将使用 itertools.count：一个无限的整数生成器。

我们还将使用 tee 管道，它打印通过它的每个值。

以下示例不执行任何操作，tee 没有打印任何内容，因此没有值通过它。这很棒，因为生成无限平方序列是“慢”的。

>>> result = count() | tee | select(lambda x: x ** 2)
>>>

链式更多管道仍然不会使之前的管道开始生成值，在以下示例中，没有从 count 中拉出任何值

>>> result = count() | tee | select(lambda x: x ** 2)
>>> first_results = result | take(10)
>>> only_odd_ones = first_results | where(lambda x: x % 2)
>>>

没有变量也是如此

>>> result = (count() | tee
...                   | select(lambda x: x ** 2)
...                   | take(10)
...                   | where(lambda x: x % 2))
>>>

只有在实际需要值时，生成器才开始工作。

以下示例中，只会从 count 提取两个值。

• 0 被平方（变为 0），轻松通过 take(10)，但被 where 丢弃。
• 1 被平方（变为 1），也轻松通过 take(10)，通过 where，并通过 take(1)。

此时，take(1) 已满足条件，因此无需进行其他计算。注意 tee 打印 0 和 1 通过它。

>>> result = (count() | tee
...                   | select(lambda x: x ** 2)
...                   | take(10)
...                   | where(lambda x: x % 2))
>>> print(list(result | take(1)))
0
1
[1]
>>>

弃用

在 pipe 1.x 中，许多函数返回可迭代对象，而许多其他函数返回非可迭代对象，这导致了混淆。返回非可迭代对象的函数只能用作管道表达式的最后一个函数，因此实际上它们是无用的。

range(100) | where(lambda x: x % 2 == 0) | add

可以重写为没有更低的可读性：

sum(range(100) | where(lambda x: x % 2 == 0))

因此，所有返回非可迭代对象的管道都被弃用（引发警告），并在 pipe 2.0 中最终被移除。

我该怎么做？

哦，你刚刚升级了 pipe，遇到了异常，来到了这里？你有三个解决方案

1. 停止使用关闭管道，将 ...|...|...|...|as_list 替换为 list(...|...|...|)，就是这样，甚至更短。

2. 如果你不介意“关闭管道”，并且真的喜欢它们，只需重新实现你真正需要的几个，这通常只需要几行代码，或者从这里复制。

3. 如果你仍然大量依赖它们，并且急于处理，只需运行 pip install pipe<2。

然后使用 Python 开发模式 开始测试你的项目，以便在它们咬你之前捕捉到这些警告。

但我喜欢它们，请重新引入它们！

这已经在 #74 中讨论过了。

@Pipe 常常可以很容易地在 1 到 3 行代码的函数中实现，并且 pipe 模块的目标不是提供所有可能性，而是提供 Pipe 装饰器。

因此，如果你需要更多管道、关闭管道、奇怪的管道、你自己命名的管道，请随意在你自己的项目中实现它们，并考虑已经实现的那些作为如何做的示例。

请参阅下面的 构建自己的 段落。