zip 特点

  1. 结果长度和短的 sequence 一致

    1
2

    In [60]: list(zip([1,2,3], [2,3]))
Out[60]: [(1, 2), (2, 3)]

  2. 如何和最长的 sequence 一致， itertools.zip_longest

    1
2

    In [72]: list(itertools.zip_longest([1,2,3], [2,3]))
Out[72]: [(1, 2), (2, 3), (3, None)]

itertools

参考：

简介

tee
复制 iterator
islice
切边工具 slice tool, 类似 list 一样来 slice 切片 iterator
pairwise
两两配对工具
count
类似 start::step, 给的 start 和 step，无限增长的 iterator,

takewile 和 dropwhile 辨析

  1. 它们针对的是从头开始的逐个元素

  2. takewhile, 如果 true, 保留； 如果 false, 停止保留

    • 所以结果是开头的部分

  3. dropwhile, 如果 true, 丢弃； 如果 false, 停止丢弃

    • 所以结果是结尾的部分
1
2
3
4
5
6

# * dropwhile
dropwhile(lambda x: x<5, [1,4,6,4,1]) --> 6 4 1

# * takewile
In [71]: list(itertools.takewhile(lambda x: x<5, [1,4,6,4,1]))
Out[71]: [1, 4]

pipe 串联调用函数

工具：

用法：

1
2
3
4
5
6
7
8
9

# 使用 pipe.Pipe
import pipe
In [147]:  range(5)| fp.pairwise | fp.to_list | pipe.Pipe(lambda x: itertools.starmap(X**X, x)) | fp.to_list
Out[147]: [0, 1, 8, 81]

# 使用 funcy_pipe.PipeFirst
import funcy_pipe as fp
In [148]:  range(5)| fp.pairwise | fp.to_list | fp.PipeFirst(lambda x: itertools.starmap(X**X, x)) | fp.to_list
Out[148]: [0, 1, 8, 81]

funcy

注意

  1. sequence Vs. collection

    • funcy 中面向 sequence 的函数可以针对 generator 使用
    • 面向 collection 的对象，不能对 generator 使用

    • 举例：

      • 过滤工具： funcy.filter 和 funcy.select 不同， funcy.select 只能用于 list, set 等 collection

        1
2
3
4
5

        In [158]: range(10) | fp.to_list| fp.select(X > 3) | fp.to_list
Out[158]: [4, 5, 6, 7, 8, 9]

In [159]: range(10) | fp.filter(X > 3) | fp.to_list
Out[159]: [4, 5, 6, 7, 8, 9]

函数工具（高阶函数）

参考：

以下函数都是接受函数作为参数，返回一个新的函数

identity
返回参数本身
constantly
永远返回 constantly 存储的参数
caller
接收函数 f 作为参数，f (*args, **kwargs), 这里 args 和 kwargs 是 caller 存储的参数， 可以用于同样的参数，应用到不同的函数的使用案例
juxt
juxtapositon 把提供的函数 并列执行
autocurry
自动 curry
iffy
不确定函数，判断合格（predicate），执行 action, 否则执行 default

逻辑函数

complement
not predicate, 返回补集判断函数
all_fn(*fs)

满足所有 fs 的 predicat 判断

  • 类似函数 any_fn, none_fn, one_fn
  • 注意：这些函数是 短路执行 的，例如 all_fn, 有一个是 False 停止执行
some_fn(*fs)

并列执行，第一个 True 的返回

  • 注意：它不是短路执行的，而是 并列执行

decorator 工具

可以通过一个普通的函数来创建一个高阶装饰器

参考：

例子：

1
2
3
4

@decorator
def log(call):
    print(call._func.__name__, call._args, call._kwargs)
    return call()

解释：

  • 这里的 call 是一个 funcy.decorators.Call 对象
  • Call._func 字段 :: 被装饰的原始函数
  • Call._args 字段 ::
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

import funcy


In [282]: @funcy.decorator
     ...: def second(call, *dargs, **dkwargs):
     ...:     print(f'{call = }, {call._func = }, {call._args = }')
     ...:     print(f'{dargs = }, {dkwargs = }')
     ...:     print(f'{call._args = }, {call._kwargs = }')
     ...:     return call()
     ...:

In [283]: @second('d1', 'd2', dkw_1=3)
     ...: def orignal(x, y):
     ...:     print(f'orignal args, {x}, {y}')
     ...:

In [284]: orignal(1,2)
call = <Call orignal(1, 2)>, call._func = <function orignal at 0x7f694f4a8ea0>, call._args = (1, 2)
dargs = ('d1', 'd2'), dkwargs = {'dkw_1': 3}
call._args = (1, 2), call._kwargs = {}
orignal args, 1, 2

flow, error 流程和异常处理

错误处理

工具：

  1. 异常捕获

    slient
    错误返回 None, 忽略所有异常
    ignore
    捕获指定异常，可以指定异常发生时的 default 值
    suppress
    压制给定的 errors 异常
    nullcontext
    啥也不干的占位符 content manager

  2. 重试

    retry
    重试给定次数，捕获异常，设置 timeout

  3. 异常抛出

    raiser
    抛出给定的异常
    reraiser
    捕获一种异常，以指定的新异常抛出

  4. 异常发生，换一种处理方法

    fallback(*approaches)

    依次尝试给定的 approaches, 如果发生异常

    • approach 可以是 callable, 或者 (callable, Exception 子类)
    • 全部失败，返回 None

    • 一种实现方法：最后一个函数，抛出异常，即使用 raiser

      1
2
3
4
5

      fallback(
    (partial(send_mail, ADMIN_EMAIL, message), SMTPException),
    partial(log.error, message),          # Handle any Exception
    (raiser(FeedbackError, "Failed"), ()) # Handle nothing, 抛出异常
)

  5. 异常速率限定

    limit_error_rate(fails: int, timeout: int | timedelta)

    重复调用次数和超额 timeout 限制 decorator

    • 如果重复调用次数超额 (fails 限制), 触发计时，只有计时达到 timeout, 才可以再次执行
    • 计时达到 timeout 之前执行，抛出 ErrorRateExceeded 异常

执行次数

根据设定条件只执行函数一次，适合初始化的使用案例。

once
无论函数参数如何变化，都只执行一次
once_per_args
每种参数组合，只执行一次
once_per('arg_names')
给的的参数名组合，只执行一次
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

# * once_per_args
In [294]: saved = []
     ...: @funcy.once_per_args
     ...: def unique_args(x, y):
     ...:     global saved
     ...:     saved.append((x,y))
     ...:

In [295]: unique_args(1,2)

In [296]: unique_args(1,2)

In [297]: saved
Out[297]: [(1, 2)]

In [298]: saved
Out[298]: [(1, 2)]

In [299]: unique_args(2,2)

In [300]: saved
Out[300]: [(1, 2), (2, 2)]

# * once_per
In [301]: saved = []
     ...: @funcy.once_per('x')
     ...: def unique_args(x, y):
     ...:     global saved
     ...:     saved.append((x,y))
     ...:

In [302]: unique_args(1,2)

In [303]: unique_args(1,1)

In [304]: saved
Out[304]: [(1, 2)]

重试 retry

1
2
3
4

@retry(3, errors=HttpError)
def download_image(url):
    # ... make http request
    return image

执行时间限制

throttle(period: int | timedelta)

只有给定时间段可以执行

  • 第一次执行 now, 到 now + period 之间可以执行
  • 超时（ > now + period）禁止执行

genator 结果收集

collecting

装饰器，收集 generator 的结果到一个 list

1
2
3
4
5
6
7

In [2]: @funcy.collecting
   ...: def gen():
   ...:     for i in range(3):
   ...:         yield i

In [3]: gen()
Out[3]: [0, 1, 2]
joining

str.join, bytes.join, 把 generator 结果 join 起来

1
2
3
4
5
6
7

In [4]: @funcy.joining(' ')
   ...: def gen():
   ...:     for i in range(3):
   ...:         yield i

In [5]: gen()
Out[5]: '0 1 2'
post_processing(fun)

把 fun 应用到 generator 的所有结果

1
2
3
4
5
6
7
8

In [8]: @funcy.post_processing(list)
   ...: def gen():
   ...:     for i in range(3):
   ...:         yield i
   ...:

In [9]: gen()
Out[9]: [0, 1, 2]
  • 相当于 list(gen())

应用 context manager

wrap_with(context_manager)
装饰器，调用函数前，添加 context_manager 的 with 语句

funcy 使用案例

应用普通函数到 iterator 上的每一个元素

不同的解决方法：

  1. 装换成 list, 再使用 collection 系列工具，如 select select_value …

    1
2

    In [348]: range(3) | fp.to_list | fp.select(lambda x: (x**2, x))
Out[348]: [0, 1, 2]

  2. 使用 funcy.map

    1
2

    In [350]: range(3) | fp.map(lambda x: (x**2, x)) | fp.to_list
Out[350]: [(0, 0), (1, 1), (4, 2)]

应用多个普通函数到 iterator 上的每一个元素

  1. 使用 funcy.map + lambda x: [fun1(x), fun2(x), …]

    1
2

    In [104]: range(3) | fp.map(lambda x: (x, x**2, x**3)) | fp.to_list
Out[104]: [(0, 0, 0), (1, 1, 1), (2, 4, 8)]

  2. 注意： 避免使用 funcy.juxt, 太麻烦

    1
2
3
4

    from lambdas import _ as X

In [102]: range(3) | fp.map(funcy.juxt(X, X**2, X**3)) | fp.map(list) | fp.to_list
Out[102]: [[0, 0, 0], [1, 1, 1], [2, 4, 8]]

funcy_pipe

如何 map 使用一个 pipe_function

注意：

funcy_pipe.PipeFirst()(*new_args, ***new_kwargs)

接收的参数会被累加到 self.function 的 args 和 kwargs 中，

  • 这个 __call___, () 运算符，不是函数执行，而是修改 self.function 的参数
  • 因此 fp.map(a_pipe_fun) 实际上不会被执行，而是获取到新的 pipe_fun

  • 正确的用法： fp.pmap(a_pipe_function)

     1
 2
 3
 4
 5
 6
 7
 8
 9
10

    # 错误用法
In [140]: range(3) | fp.map(fp.inc) | fp.to_list
Out[140]:
[<funcy_pipe.pipe.PipeSecond at 0x7fa5f1029a10>,
 <funcy_pipe.pipe.PipeSecond at 0x7fa5f10281d0>,
 <funcy_pipe.pipe.PipeSecond at 0x7fa5f2238910>]

# 正确用法
In [141]: range(3) | fp.pmap(fp.inc) | fp.to_list
Out[141]: [1, 2, 3]

funcy_pipe Vs. funcy_chain

  1. funcy_chain

    • 导致 ipython 卡住（补全时）
    • not lazy, 会直接运行

  2. funcy_pipe

    • 支持原生的 layzy 运行

    • eg:

       1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

      # funcy_chain: 时间长， greedy
In [222]: %time Chain(range(1000000)).map(lambda x: x*2)
CPU times: user 115 ms, sys: 0 ns, total: 115 ms
Wall time: 115 ms
Out[222]: <funcy_chain.chain.Chain at 0x7fa5e3446c10>

# funcy_pipe: 时间短， lazy load
In [223]: %time range(1000000) | fp.map(lambda x: x*2)
CPU times: user 29 µs, sys: 0 ns, total: 29 µs
Wall time: 31.9 µs
Out[223]: <map at 0x7fa5e3935390>

      • 解释：

        • chain 直接把结果转成 list, 存储在 Chain._value 字段中
        • pipe 遵循的原生的 iterator 和 generator 机制，返回的是一个 map 对象，需要执行 list(result) 才能获取到最终结果