前言

年后要参与 Kafka Python SDK 开发,遂读《流畅的 Python》,笔记仅供个人 Review 使用


ch01. 数据模型

Python 对象只需实现约定的特殊方法,就能复用丰富的库函数和语法规则;如二维向量类Vector实现__add__特殊方法重载了+运算符,来定制加法逻辑:

1
2
3
4
5
6
7
8
9
10
class Vector:
def __init__(self, x=0, y=0):
self.x, self.y = x, y
def __repr__(self):
return 'Vector(%r, %r)' % (self.x, self.y) # 对象的字符串表示
def __add__(self, other):
return Vector(self.x + other.x, self.y + other.y) # 重载 + 操作返回新值对象
v1 = Vector(1,2)
v2 = Vector(10,20)
print(v1+v2) # 'Vector(11,22)'

再如实现__getitem__, __len__来使用切片语法、复用类库:

1
2
3
4
5
6
7
8
9
10
11
12
User = collections.namedtuple('User', ['name', 'score'])
class Group:
users = [User('userA', 93), User('userB', 89), User('userC', 91)]
def __getitem__(self, idx):
return self.users[idx]
def __len__(self):
return len(self.users)

> group = Group()
> group[1], group[2:] # 解释器会将切片操作转为 __getitem__
> reversed(group) # 使用 reversed 翻转遍历结果
> random.choice(group) # 复用类库取随机值

如上可总结 Python 数据模型的概念:解释器开放特殊方法给类去实现,用来定制化当语法结构、库函数应用在其对象时的行为;特殊方法有 50+ 个,可分为运算符重载、数据结构操作模拟和类属性读写三类,详细参考 special-method-names


ch02. 序列结构

2.1 list

列表推导[]:拍扁 for 结构,一次性生成多个元素组成 list,比等效的 filter, map 更简洁

1
2
l1 = [ord(c) for c in 'abc' if ord(c) > 10]
l2 = list(filter(lambda c: c>10, map(ord, 'abc'))) # l1 == l2 # True

生成器表达式():迭代时惰性产出元素

1
for c in ((ord(c) for c in 'abc')): print(c)

2.2 tuple

不可变列表,位置固定的元素集;拆包:将 tuple 展开为可迭代对象

1
2
3
4
5
def sum_div_mod(x, y):
return x + y, x / y, x % y # 用 tuple 返回多值
xy = (13, 10)
*vs, mod = sum_div_mod(*xy) # 拆包传参
print(vs, mod) # [23, 1.3] 3 # _ 占一位,* 不定

具名元组:构建只有字段名的类,底层用元组存储。

1
2
3
4
City = namedtuple('City', 'name age scores')  # 空格分割的字符串,或其他字符串可迭代对象
pek = City('Peking', 200, (98.1, 99.5))
print(pek._fields, pek.name, pek[1]) # 获取字段名元组;引用指定字段名;本质是元组,可索引读
print(pek._asdict() == City._make(pek)._asdict()) # 将具名元组展开为 OrderedDict;从可迭代对象中生成实例

2.3 slice

对序列类型进行切片操作,语法 [start:end:step],若 step 为负则反向切片,底层实际调用__getitem__;切片可对序列进行嫁接、切除和就地修改

1
2
3
4
5
6
s = 'abcdef'
print(s[slice(1, 4, 2)]) # bd # slice 对象描述切片操作
vs = list(range(6)) # [0,1,2,3,4,5]
vs[2:5] = (20, 30) # [0,1,20,30,5]
del vs[:2:] # [20,30,5]
vs[1:2] = [10] # [20,10,5]

增量赋值

  • +, *:+ 连接同类序列,* 倍增序列,都会生成新的结果序列;注意,若序列元素是引用,则新序列复制的也是引用

    1
    2
    3
    4
    5
    6
    7
    board = [['_'] * 2 for i in range(2)] 
    board = [['_'] * 2] * 2 # board 的 2 个元素都引用同一个 list
    board[0][1] = 'X' # [['_', 'X'], ['_', 'X']]

    row, board = ['_'] * 2, []
    for i in range(2):
    board.append(row) # 同样是重复引用的错误
  • +=, *=:对于+=增量赋值,解释器会先调用__iadd__就地修改,若未实现此方法则调用__add__生成新对象;同理*=也是先调__imul__或再调__mul__;可变序列如 list 会就地修改,不可变序列如 tuple 则会生成新 tuple 覆盖

    1
    2
    3
    4
    5
    6
    7
    8
    l = [1, 2]
    old_id = id(l)
    l += [30]
    print(id(l) == old_id) # True
    t = (1, 2)
    old_id = id(t)
    t += (3,) # type((3)) 为 int, type((3,)) 为 tuple
    print(id(t) == old_id) # False # t 已指向新对象

元组的增量赋值不是原子操作,尽量别把可变对象放元组中:直接修改抛出异常,但可变对象已修改成功

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
t = (1, 2, [3, 4])
try:
t[2] += [5] # t[2].extend([5]) # 正确用法
except TypeError as e:
print(e) # 'tuple' object does not support item assignment
print(t) # (1, 2, [3, 4, 5])
print(dis.dis('t[2] += [5]')) # list 对象被修改成功后,存入元组时才抛的异常

1 0 LOAD_NAME 0 (t) # 载入 tuple t
2 LOAD_CONST 0 (2)
4 DUP_TOP_TWO # 读取 t,2
6 BINARY_SUBSCR # 执行 t[2] 切片操作
8 LOAD_CONST 1 (5)
10 BUILD_LIST 1 # 创建 [5] list
12 INPLACE_ADD # 就地修改 [3,4] + [5]
14 ROT_THREE # 移到栈顶
16 STORE_SUBSCR # 执行 t[2] = [3,4,5] 的赋值操作,失败

排序搜索

  • list.sort(self, key, reserve):list 内部就地排序,可选 key 函数生成元素的权重进行排序,可选 reserve 布尔值控制是否倒排
  • sorted(iterable, key, reserve):对任意可迭代对象进行排序,统一返回 list

bitsect 模块:维护新值插入后序列有序性

1
2
3
4
5
6
l = [1, 7, 5, 3]
l.sort() # [1,3,5,7]
print(bisect.bisect(l, 5.0, lo=0, hi=len(l))) # 3 # 在有序序列 l 的 [lo,hi] 索引区间内,搜索值为 5.0 的插入位置
bisect.insort(l, 5.0) # [1, 3, 5, 5.0, 7] # 默认右侧执行插入
print(bisect.bisect_left(l, 5.0)) # 2 # 值相等则返回左侧位置,bisect 是 bisect_right 别名
bisect.insort_left(l, 5.0) # [1, 3, 5.0, 5, 5.0, 7]

其他序列类型

  • array:高效存取大量相同基本类型的元素,由指定的 typecode 解释元素类型,同时序列化到文件非常省空间
  • memoryview:高效共享同一片内存数据,修改时拷贝
  • deque:线程安全的双端队列

ch03. 字典与集合

3.1 dict

hashable:只有可散列类型才能作为 dict 键值;不可变数据类型如 str 和 bytes、不可变结构类型如 frozenset 和只包含不可变类型的元组等;用户类若实现__hash__则必须也实现__eq__来保证哈希值相同的两个对象相等

字典的五种初始化方式:

1
2
3
4
5
a = dict(a=1, b=2)                # 关键字参数
b = {'a': 1, 'b': 2} # 字面量 # 无函数调用,速度最快
c = dict([('a', 1), ('b', 2)]) # kv 元组列表
d = dict(zip(['a', 'b'], [1, 2])) # 可迭代对象打包生成 kv 元组列表
e = dict({'a': 1, 'b': 2}) # 字面量参数

字典推导:类比列表推导

1
2
codes = [(86, 'China'), (1, 'USA'), (81, 'Japan')]
code2country = {code: country.upper() for code, country in codes if code > 80}

方法列表

dict 方法 描述
setdefault(k, default) k 存在则返回值,否则设置值为 default 后返回 default
update(m, [**kwargs]) 从 map 或 iterable kv 中批量更新 dict
fromkeys(iterable, [initval]) 将 iterable 中元素设为 key,initval 为初值,默认 None
pop(k, [default]) 删除 k 并返回其值,不存在返回 default 或 None
popitem() 删除随机 key 并返回
get(k, [default]) 读取 k 的值,返回 default 或 value
copy() 浅复制
keys(), values(), items() 返回所有键,值,键值对
__setitem__(k, v) d[k] = v
__delitem__(k, v) del d[k]
__getitem__(k) v = d[k]
__len__() len(d)
__contains__(k) k in d

defultdict:带默认值构造函数的字典

1
2
w2l = collections.defaultdict(list)
w2l['word1'].append((0,1)) # 击穿时调用 list() 生成其值

__missing__:当d[k]读取不存在的 key 时,__getitem__将委托__missing__处理;示例:

1
2
3
4
5
6
7
8
9
10
11
12
# 增强 d[k] 读取功能,不存在时转为字符串重试
class StrKeyDict(dict):
def __missing__(self, key):
if isinstance(key, str): # 递归退出条件
raise KeyError(key)
return self[str(key)]

def __contains__(self, key):
return key in self.keys() or str(key) in self.keys() # 同理 key in self 会导致递归

sd = StrKeyDict({1: 'one', '4': 'four'})
print(sd[4], 4 in sd) # four, True

字典变种

  • collections.OrderedDict:键按添加顺序排列的 dict,迭代顺序保持一致
  • collections.ChainMap:连接多个 dict,逐个查找 key
  • collections.Counter:键值计数 dict
  • types.MappingProxyType:不可变 dict,本质是源 dict 的实时只读视图
  • collections.UserDict:纯 python 版的 dict,数据记录在 data 字段,常用作基类继承
1
2
3
4
5
6
7
8
9
10
11
12
13
14
od = collections.OrderedDict({'b': 20, 'a': 10, 'e': 999})
od.popitem(last=False) # OrderedDict([('a', 10), ('e', 999)])

cd = collections.ChainMap({'A': 1}, od)
print(cd['a']) # 10

counter = collections.Counter('aabbccc')
counter.update('aaa')
print(counter.most_common(2)) # [('a', 5), ('c', 3)]

d = {1: 'a'}
mp = types.MappingProxyType(d)
d[2] = 'b' # mp[2] = 'B' # TypeError: 'mappingproxy' object does not support item assignment
print(mp[2]) # 'b'

3.2 set

方法列表:集合类型实现了大量运算符的特殊方法,实现了运算符重载

运算符 方法 描述
s&z;s&=z s.__add__(z),s.intersection(z);s.intersection_update(z) 取交集
s|z;s|=z s.__or__(z),s.union(z);s.update(z) 取并集
s-z;s-=z s.__sub__(z),s.difference(z);s.difference_update(z) 取差集
s^z;s^=z s.__xor__(z);s.__ixor__(z) 取对称差集
e in s s.__contains__(e) 是否属于
s<z;s<=z s.__lt__(z);s.__le__(z),s.issubset(z) 真子集;子集
s>z;s>=z s.__gt__(z);s.__ge__(z),s.issupperset(z) 真父集;父集

ch05. 函数

函数有 4 个数据特征:运行时创建、可赋值给变量、可作为实参、可作为函数返回值,与整型变量、字符串变量等一样都是 first class object,函数是类<class 'function'>的对象

高阶函数:有函数形参或返回函数的函数,如内置的归约函数、映射过滤函数等

1
2
3
> functools.reduce(operator.add, range(100)) == sum(range(100)) # True
> all([1, 3, 4, 0]) # False
> any([0, 0, 0, 1]) # True

匿名函数:即 lambda 函数,函数体只能是纯表达式,不能包含语句;只是语法糖,和def一样用来创建函数对象

1
2
> fruits = ['apple', 'banana', 'cherry', 'strawberry'] # 按谐音排序
> sorted(fruits, key=lambda s: s[::-1]) # ['banana', 'apple', 'strawberry', 'cherry']

可调用对象:callable()检测,分为 7 种:

  • 用户自定义函数、方法:def 或 lambda 表达式创建、用户类中的方法
  • 内置函数、方法:CPython 实现的函数如 len、各种结构配套的方法如dict.get
  • 类、对象:调用类会调用__new__创建对象再用__init__初始化,若定义了__call__则对象也可被调用
  • 生成器函数:使用yield的函数或方法,返回生成器对象

函数参数:位置参数(位置索引的普通参数)、变长参数(数量不定的参数元组*args)、关键字参数(以字典形式指定形参名和实参值**kwargs),注意位置参数和关键字参数可混用

1
2
3
4
5
6
7
8
9
10
11
12
def pr_args(name, *content, cls=None, **attrs):
print('name:', name)
if content:
print('content:', content)
if cls is not None:
print('cls:', cls)
if attrs:
print('attrs:', attrs)
pr_args('img', 'v1', 'v2') # name: img; content: ('v1', 'v2')
pr_args(name='img') # positional arg -> keyword arg
args = {'name': 'img', 'cls': 'dark', 'k1': 'v1'} # 展开后 key 绑定到同名 named arg,剩余参数被 **args 捕捉
pr_args(**args) # name: img; cls: dark; attrs: {'k1': 'v1'}

函数内省:获取函数对象的特征信息,如形参名字和类型等静态特征,实参值等动态特征,类比反射机制

1
2
3
4
5
> class Cls: pass
> obj = Cls()
> def f(): pass
> sorted(set(dir(f)) - set(dir(obj)))
['__annotations__', '__call__', '__closure__', '__code__', '__defaults__', '__get__', '__globals__', '__kwdefaults__', '__name__', '__qualname__']

函数对象相比普通类对象独有的属性:

属性 类型 说明
__name__ str 函数名
__annotations__ dict 参数和返回值的注解
__call__ method 函数体
__code__ code 函数元数据
__defaults__ tuple 形参默认值
__kwdefaults__ dict 关键字形参默认值
__globals__ dict 函数所在模块的全局变量

函数注解:声明函数时为形参和返回值附加的注解表达式,对解释器无意义,用作类型检查、实参值合法性检查

示例:使用函数内省获取函数参数信息

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
def passed(name:str, id:int=-1, score:'int > 0'=10) -> int:
edge = 60
if score >= edge:
return 1
return 0
> passed.__defaults__ # (-1,10,)
> passed.__code__.co_varnames # ('name', 'id', 'score', 'edge') # 形参名、函数体内局部变量名
> passed.__code__.co_argcount # 3 # 形参个数
# 原生版:带默认值的形参必须在参数表尾部,形参表 (name,id,score),需从后向前遍历可得 score 默认值为 10,id 为 -1
# 简化版:inspect 模块
> from inspect import signature
> sig = signature(passed)
> sig.return_annotation # <class 'int'>
> for name, p in sig.parameters.items(): # _empty 即无默认值
> name, p.default, p.annotation, p.kind # name, <class 'inspect._empty'>, str, POSITIONAL_OR_KEYWORD

函数式编程

  • operator 模块:提供实现算术运算符的函数如multi等效于lambda a,b: a*b;特别地,itemgetter构建读索引函数,attrgetter构建读字段函数

  • functools 模块:特别地,partitial可冻结函数参数生产新函数

    1
    2
    > double = functools.partial(operator.mul, 2)
    > list(map(double, [1, 2, 3])) # [2,4,6]

ch07. 装饰器和闭包

装饰器:在语法层面支持的装饰器模式,用于增强函数行为;装饰器会在模块加载时立即执行生效,类似全局变量的赋值操作

1
2
3
4
5
6
7
8
9
10
11
12
13
def deco(func):
print('invoke decorator deco')
def inner():
print('invoked deco.inner()')
return inner

@deco
def target():
print('invoked target()')

> import deco # invoke decorator deco
> target() # invoked deco.inner()
> target # <function deco.<locals>.inner at 0x10b859d30> # target 已更新为 deco.inner 的引用

作用域:省略变量声明的副作用是,函数体内被赋值的变量将被视为局部变量,而非同名全局变量的引用

1
2
3
4
5
6
7
8
9
10
11
12
b = 10
def f(a):
print(a)
# global b # 显式声明使用全局变量 b
print(b) # UnboundLocalError: local variable 'b' referenced before assignment
b = 3 # b 声明为局部变量
> dis.dis(f)
74 8 LOAD_GLOBAL 0 (print)
10 LOAD_FAST 1 (b) # load local scope var b
# 10 LOAD_GLOBAL 1 (b) # load global scope var b
12 CALL_FUNCTION 1
14 POP_TOP

闭包:在函数体中引用了函数体外定义的非全局变量,延伸了函数的作用域

1
2
3
4
5
6
7
8
9
10
11
12
13
def make_avg():
prices = [] # 自由变量:make_avg 的局部变量,在闭包内部被绑定,make_avg 调用结束后不会被释放
def avg(price):
prices.append(price)
total = sum(prices)
return total / len(prices)
return avg
> avg = make_avg()
> avg(10), avg(11) # 10, 10.5
> avg.__code__.co_varnames # ('price', 'total')
> avg.__code__.co_freevars # ('price')
> avg.__closure__ # (<cell at 0xxx list object at 0xyy>) # 闭包由 cell 对象描述
> avg.__closure__[0].cell_contents # [10, 11] # 自由变量的值暂存在 cell.cell_contents

nolocal 声明:强制将局部变量标记为自由变量,将绑定到闭包中的同名变量

1
2
3
4
5
6
7
8
9
10
def make_avg():
count, total = 0, 0
def avg(price):
# nonlocal count, total # 将 count, total 标记使用自由变量
total += price # UnboundLocalError # 不可变类型 += 后将引用新对象
count += 1
return total / count
return avg
> avg = make_avg()
> avg(10), avg(11) # 10, 10.5

functools 中的常用装饰器

  • wraps:拷贝保留被装饰函数的__name__, __annotation__等属性

  • lru_cache:缓存相同参数表的函数返回值,避免重复计算

    1
    2
    3
    4
    5
    6
    7
    8
    9
    @functools.lru_cache()
    def fib(n):
    if n < 2:
    return n
    return fib(n - 2) + fib(n - 1) # fib(n) 只会被调用 N+1 次,避免递归计算

    @functools.lru_cache() # 内部实现为 {arg: return_val},故不可散列类型的参数无法使用
    def f(vs: list): # TypeError: unhashable type: 'list'
    pass
  • singledispatch:将被装饰函数变为泛函数(generic function),根据第一个参数类型,执行对应的“重载”函数;类似类型断言后各种 case 执行不同的函数

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    @functools.singledispatch
    def pr(obj):
    print('base object', obj)

    @pr.register(str)
    def _(s):
    print('str', s)

    @pr.register(numbers.Integral)
    def _(n):
    print('{0}: 0x{0:x}'.format(n))

    @pr.register(tuple)
    @pr.register(collections.abc.MutableSequence)
    def _(seq):
    print('seq: ', ','.join(str(n) for n in seq))

    > pr(1) # 1: 0x1
    > pr('1') # str 1
    > pr([1, 2, [10], [20]]) # 1,2,[10, 20]

    参数化装饰器:装饰函数被限制只能有一个形参,即被装饰的函数;当装饰器本身需要额外参数时,需再套一层装饰器工厂函数,传递参数后返回装饰器

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    registry = set()

    def decorator_factory(active=True):
    def decorator(func): # 真正的装饰器
    print('running register_factory(active=%s) -> register(%s)' % (active, func))
    if active: # 使用了装饰器工厂的实参
    registry.add(func)
    else:
    registry.discard(func)
    return func
    return decorator

    @decorator_factory(active=True) # 调用工厂函数,生成装饰器函数
    def f():
    print('invoked f()') # 等效于 decorator_factory(active=True)(f)

ch08. 对象引用与拷贝

对象标识:对象创建后不可变,标识可认为是其内存地址;id()返回标识符的整数表示;is运算符检查标识是否相等,而==运算符检查是否相等,实际调用__eq__

元组的相对不变性:元组的不变性,是指内部数据的引用不变,与引用的对象可变性无关

1
2
3
4
5
> t1 = (1, 2, [3, 4])
> t2 = (1, 2, [3, 4])
> t1 == t2 # True
> t1[-1].append(5) # (1,2,[3,4,5])
> t1 == t2 # Flase

浅拷贝:列表等可变序列的构造方法、[:]操作等,只会复制最外层容器来生成副本,若内部有可变元素则共享其引用,易出错

1
2
3
4
5
6
7
8
> l1 = [1, [6, 5], (8, 9)]
> l2 = list(l1) # 浅拷贝
> l1.append(10) # [1, [6,5], (8,9), 10]
> l1[1].remove(5) # [1, [6], (8,9), 10] # l1,l2 引用同一个 [6,5] list 子对象
> l2 # [1, [6], (8,9)]
> l2[1] += [3, 2] # [1, [6,3,2], (8,9)]
> l2[2] += (10, 11) # [1, [6,3,2], (8,9,10,11)] # l2 引用了 += 生成的新 tuple
> l1 # [1, [6,3,2], (8,9), 10] # l1 还在引用旧 tuple

深拷贝:内部对象也会逐层拷贝,引用不会被共享;特别地,会记住已拷贝过的对象,避免循环引用

1
2
3
4
5
6
7
8
9
> v1 = [1, 2, [3, 4]]
> v2 = copy.deepcopy(v1)
> v2[2].append(5) # [1, 2, [3,4,5]]
> v1 # [1, 2, [3,4]]

> a = [1, 2]
> b = [a, 3]
> a.append(b)
> copy.deepcopy(a) # [1, 2, [[...], 3]]

引用实参:Python 只支持共享传参(call by sharing),函数内的“形参”实际是引用实参的副本,故可变对象可被函数所修改

1
2
3
4
5
6
7
def add(a, b):
a += b
return a
> a, b = [1,2], [3,4]
> add(a, b) # [1,2,3,4] # a,b: [1,2,3,4], [3, 4] # 函数内通过引用副本修改了外部对象
> t, u = (1,2), (3,4)
> add(t, u) # (1,2,3,4) # t,u: (1,2), (3,4) # 返回的是新 tuple

问题:可变类型作为函数的默认参数,会导致多次无参调用同时引用一个可变对象

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
class HuntedBus:
def __init__(self, names=[]):
self._names = names
def pick(self, name):
self._names.append(name)
> b1 = HuntedBus()
> b1.pick('A')
> b2 = HuntedBus()
> b2._names # ['A']
> b2.pick('B')
> b1._names # ['A', 'B'] # 同步修改
> b1._names is b2._names # True # b1,b2 引用了同一个 names 对象
> HuntedBus.__init__.__defaults__ # 默认值存于函数对象 # (['A','B'])
> HuntedBus.__init__.__defaults__[0] is b1._names # True

class HuntedBus:
def __init__(self, names=None): # 实践:可变参数默认 None
if names is None:
self._names = [] # 正确的防御性可变参数,引用各自对象
else:
self._names = list(names) # 创建本地副本,不修改实参

垃圾回收del会解除变量到对象的引用,当解除最后一个引用时,对象变为 unreachable,调用其__del__析构函数,最后销毁对象完成回收

1
2
3
4
5
6
7
8
9
10
11
12
13
14
class User:
def __init__(self, name):
self._name = name

weak_users = weakref.WeakValueDictionary() # 弱引用不会增加对象的引用计数,不妨碍垃圾回收
users = [User('A'), User('B'), User('C')]
for user in users:
weak_users[user._name] = user

> sorted(weak_users.keys()) # ['A','B','C']
> del users
> sorted(weak_users.keys()) # ['C'] # 迭代变量还引用 User('C')
> del user
> sorted(weak_users.keys()) # [] # 三个 user 对象都被回收

字面量内存驻留:和 Java 的字符串共享池类似,Python 会将部分字符串字面量驻留内存,只读地共享引用

1
2
3
> s1 = 's'
> s2 = 's'
> id(s1) == id(s2), s1 is s2 # True, True # 比较字符串用 ==

ch09. 对象构造与属性

对象表示:自定义类可实现特殊方法,来定制对应的函数返回值

特殊方法 用户级方法 用途
__repr__ repr() 返回面向开发者的对象字符串表示
__str__ str() 返回面向用户的对象字符串表示
__bytes__ bytes() 返回对象的字节序列表示
__format__ format(), str.format() 返回对象的格式化字符串表示

格式化

  • %操作符:%[(name)][flags][width].[precision]typecode,其中(name)用于格式化指定字段
  • str.format()函数:使用:{}替换了%,支持位置映射、字典映射、对象属性映射、下表映射等
  • f-string模板字符串:直接将值嵌入格式化字符串中

classmethod, staticmethod:类方法可用@classmethod修饰,第一个参数是类自身;@staticmethod仅仅是定义在类内部的函数

1
2
3
4
5
6
7
8
9
class Demo:
@classmethod
def classmethod(*args):
return args
@staticmethod
def statmethod(*args):
return args
> Demo.classmethod('x') # (<class '__main__.Demo'>, 'x') # cls 对象隐式作为第一个参数,类比 self
> Demo.statmethod('x') # ('x') # 与普通函数调用无异

可散列性:用户类需实现__hash__计算散列值、实现__eq__保证散列值等的对象值也相等;此外,可引入@property实现属性只读,来保证对象的散列值始终不变;至此才能放入 set 或作为 dict 的键

示例

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
38
39
40
41
42
class Vector2d:
typecode = 'd'
def __init__(self, x, y):
self.__x = float(x) # 初始化时强转,符合 fail-fast 原则
self.__y = float(y) # 约定 __var 双下划线为私有属性

@property # 改写函数作为只读属性,来保证对象的哈希值始终不变
def x(self):
return self.__x
@property
def y(self):
return self.__y

def __hash__(self):
return hash(self.x) ^ hash(self.y) # 异或混合各属性的哈希值,尽量保证哈希值的雪崩特性
def __eq__(self, other):
return tuple(self) == tuple(other) # == 相等性比较

def __iter__(self):
return (i for i in (self.x, self.y)) # 构建可迭代对象,由 x,y 两个分量组成的 tuple

def __repr__(self):
cls = type(self).__name__
return '{}({!r}, {!r})'.format(cls, *self) # *self 展开 tuple
def __str__(self):
return str(tuple(self))

def __abs__(self):
return math.hypot(self.x, self.y)
def __bool__(self):
return bool(abs(self)) # 用作类型构造函数,如 bool(vec)

def __bytes__(self): # 字节序列表示:约定第一个字符为分量数据类型,再连接数组的字节序列
return (bytes(self.typecode)) + bytes(array(self.typecode, self))
@classmethod
def frombytes(cls, buf): # 类方法常用于定义额外的构造方法
memv = memoryview(buf[1:]).cast(chr(buf[0]))
return cls(*memv)

def __format__(self, fmt_spec): # 可让 fmt_spec 包含自定义标记,来扩展格式化
strs = (format(c, fmt_spec) for c in self) # 格式化各分量
return '({}, {})'.format(*strs)

私有属性:按约定__var双下划线的属性认为是 “private” 私有的,_var 单下划线的属性认为是 “protected” 受保护的;特别地,用户指定的__var会被解释器执行命名改写(name mangling),最终变为_Cls__var属性

1
2
3
> v = Vector2d(1, 2)
> v.__dict__ # {'_Vector2d__x': 1, '_Vector2d__y': 2} # 实际的属性名已被改写
> v._Vector2d__x = 10 # 可绕过只读机制,但违背约定

__slots__:对象默认用字典__dict__存储所有属性,当属性过多会导致 dict 占用内存多,可用__slots__进行优化,用元组来存放类的所有属性,不过__slots__并不会被继承


ch10. 鸭子类型

duck typing:是由一组方法表示的协议,实现协议的用户类就属于鸭子类型,能被用于任何处理鸭子类型的地方;故协议可理解为口头约定的非标准接口,既实现了接口特性,也避免了繁琐的显式实现声明(接口是 is-a,鸭子类型是 like-a

序列协议:实现__getitem__, __len__的用户类,可视为序列类型,可复用切片、连接等操作

切片原理:切片范围用内置类型slice描述,由start,end,step三部分描述,其方法indices(length)会处理负数索引(与 len 相加)、索引越界(重置到边界)和索引缺失(取默认值)等 case

1
2
3
4
5
6
7
8
class Seq:
def __getitem__(self, item):
return item
> s = Seq()
> s[1], s[1:2] # 1, slice(1,2,None) # None 即默认值,start 默认 0,end 默认 len(seq),step 默认 1
> s[1:4:3, 1:2] # (slice(1,4, 3), slice(1, 2, None)) # 同时支持多组切片,解释为 slice 列表
> sl = slice(-3, None, None)
> sl.indices(5), 'ABCDE'[sl] # (2,5,1), 'CDE'

动态读写属性:当读取对象不存在的属性时,会从obj.__class__父类开始沿继承树向上查找,若都没有则调用obj.__getattr__处理;写不存在的属性则由__setattr__处理

zip 函数:将多个可迭代对象,将相同索引的元素放入同一个元组,返回元组列表生成器,常用于惰性并行遍历

1
2
3
4
> list(zip(range(3), 'ABC', [10,20,30,40]))
[(0,'A',10), (1,'B',20), (2,'C',30)] # 当最短对象迭代结束时停止
> list(itertools.zip_longest(range(3), 'ABC', [10,20,30,40], fillvalue=0.1))
[(0,'A',10), (1,'B',20), (2,'C',30), (0.1,0.1,40)] # 可指定填充值,以最长对象为准

示例

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
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
class Vector:
typecode = 'd'
def __init__(self, vs):
self._vs = array(self.typecode, vs) # 内部存储为数组
def __iter__(self):
return iter(self._vs) # 迭代器委托数组实现

def __repr__(self):
vs = reprlib.repr(self._vs) # reprlib 限制字符串长度,超出则表示为...
return 'Vector({})'.format(vs[vs.find('['):-1])

def __len__(self): # 遵守序列协议
return len(self._vs)
def __getitem__(self, item):
cls = type(self)
if isinstance(item, slice):
return cls(self._vs[item]) # 切片结果还是 Vector 对象
elif isinstance(item, numbers.Integral):
return self._vs[item]
else: # 模仿内置对象的切片报错信息
raise TypeError('{cls.__name__} indices must be int'.format(cls=cls))

shortcut = 'xyzt' # 索引 0,1,2,3 的快捷访问方式
def __getattr__(self, item): # vec.x
cls = type(self)
if len(item) == 1:
pos = self.shortcut.find(item)
if 0 <= pos < len(self.shortcut):
return self._vs[pos]
raise AttributeError('{.__name__!r object has no attribute {!r}}'.format(cls, item))

def __setattr__(self, name: str, value): # vec.x 等为“只读”属性
cls = type(self)
if len(name) == 1:
if name in self.shortcut:
err_fmt = 'readonly attr {attr_name!r}'
elif name.islower():
err_fmt = 'can not set attr a-z in {cls_name!r}'
else:
err_fmt = ''
if err_fmt:
raise AttributeError(err_fmt.format(cls_name=cls.__name__, attr_name=name))
super().__setattr__(name, value) # 默认交由超类处理

import functools
import operator
def __hash__(self):
hashes = (hash(x) for x in self._vs) # 先逐个求散列值
return functools.reduce(operator.xor, hashes, 0) # 再归约取异或,生成对象的散列值

def __eq__(self, other): # 生成器函数更高效
return len(self) == len(other) and all(a == b for a, b in zip(self, other))
# if len(self) != len(other):
# return False
# for a, b in zip(self, other):
# if a != b:
# return False
# return True

ch11. 抽象基类

接口:Python 除了支持鸭子类型的动态协议,还支持通过抽象基类(Abstract Base Class)约束用户类必须遵守的静态协议(必须实现一组方法);可类比等效为杂糅了 Java 中接口和抽象类,来支持多重继承、默认方法等特性

序列抽象类collections.abc内的Sequence抽象类及其抽象父类;斜体即抽象方法

如上,用户类若实现__getitem__即为Sequence子类,可复用Sequence.__iter__()来迭代用户对象,同理复用__contains__的默认实现,迭代对象逐一比较来使用in操作符…

1
2
3
4
5
6
7
8
class seqx:
def __getitem__(self, item):
return range(0, 3)[item]
> vs = seqx()
> vs[1], 1 in vs # 1 # __getitem__
> 1 in vs # True # __contains__
> for v in vs:
> print(v) # 0,1,2 # __iter__

如上,seqx 虽未显式声明继承自Sequence,但实现了__getitem__,遵守了序列协议,如此解释器就知道可将 seqx 对象视为序列类型;总之,只要实现抽象类的抽象方法,就能复用其具体方法

抽象基类

  • collections.abc内的 16 个抽象基类

  • numbersNumber,Real,Integral,Complex,Rational

自定义抽象基类

  • 声明抽象类:继承父类abc.ABC,3.4 之前在 class 语句中指定metaclass=abc.ABCMeta,3.0 之前指定类属性 __metaclass__ = abc.ABCMeta
  • 抽象方法:@abc.abstractmethod即标记方法为抽象方法,可在其之上叠加@classmethod来标记该抽象方法属于类,叠加@staticmethod同理

示例:无重复随机挑选器

Tombola基类:定义 2 个抽象方法,2 个低效的后备具体方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
class Tombola(abc.ABC): # 继承自 abc.ABC
@abc.abstractmethod # 标记为抽象方法,子类必须实现
def load(self, iterable):
"""加载可迭代对象中所有元素"""
raise NotImplementedError('')

@abc.abstractmethod
def pick(self):
"""随机删除元素并返回,若为空则抛出 LookupError""" # 抽象方法的功能依赖文档说明

def inspect(self): # 具体方法的默认实现,子类可覆盖来优化逻辑
"""返回当前所有元素的有序元组"""
items = []
while True:
try:
items.append(self.pick())
except LookupError: # 序列为空抛出 IndexError,集合为空抛出 KeyError,此处使用二者父类
break
self.load(items) # 低效的后备机制:读取所有元素后重新写入
return tuple(sorted(items))

def loaded(self):
return bool(self.inspect())

Lottery子类:实现随机抽取,优化具体方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
class Lottery(Tombola):
def __init__(self, iterable):
self._balls = list(iterable) # 参考 ch08,不要直接引用实参

def load(self, iterable):
self._balls.extend(iterable)
def pick(self):
try:
pos = random.randrange(len(self._balls)) # 实参为 0 抛异常
except ValueError:
raise LookupError('pick from empty Lorry')
return self._balls.pop(pos)

def loaded(self):
return bool(self._balls) # 优化后的具体方法,不再依赖 inspect 做耗时的列表拷贝
def inspect(self):
return tuple(sorted(self._balls))

虚拟子类:将用户类显式注册为抽象类的子类,但不继承任何方法和属性,解释器不再检查用户是否实现了抽象方法;常用来说明继承关系,不直接使用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# @TomboList.register
class TomboList(list): # 扩展 list 内置类
def pick(self):
if self: # 复用了 list.__bool__
pos = random.randrange(len(self))
return self.pop(pos)
else:
raise LookupError('pop from empty TomboList')
load = list.extend # 属性赋值来复用 list 的函数对象

def loaded(self):
return bool(self)
def inspect(self):
return tuple(sorted(self))
Tombola.register(TomboList) # register 函数调用或装饰器

> issubclass(TomboList, Tombola) # True # 哪怕不实现 Tombola 的抽象方法,也是子类
> t = TomboList()
> isinstance(t, Tombola) # True
> TomboList.__mro__ # (<class 'TomboList'>, 'list', 'object') # 没有从 Tombola 继承任何方法
> Tombola.__subclasses__() # [<class 'BingoCage'>, 'Lottery'] # 虚拟子类不是直接子类

ch12. 继承

不要直接继承内置类型:由于 CPython 解释器的优化限制,用户类继承内置类型后,即使覆盖实现了其特殊方法,也不会被调用

1
2
3
4
5
6
class DictX(dict):
def __setitem__(self, key, value):
super().__setitem__(key, [value] * 2) # 故意存入重复值
> d = DictX(a=1) # {'a':1} # __init__ 忽略了被覆盖的 __setitem__
> d['b'] = 2 # {'a':1, 'b':[2,2]} # [] 生效
> d.update({'c': 3}) # {'a':1, 'b':[2,2], 'c':3}

解决:继承collections.UserList,UserDict,UserString来扩展内置类型list,dict,str,其对内置类型的包装、读写委托等逻辑对用户透明

1
2
3
4
5
6
class DictY(collections.UserDict):
def __getitem__(self, item):
return 10
> d1 = DictY(a=1) # {'a':1}
> d2 = {}
> d2['a'], d2.update(d1) # 10, {'a':10} # 操作接口与内置类型无异

多重继承:类的__mro__机制能确定多重继承中多个超类方法的调用顺序,不存在命名冲突的“菱形问题”

示例:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
class A:
def ping(self):
print('A ping', self)
class B1(A):
def pong(self):
print('B1 pong', self)
class B2(A):
def pong(self):
print('B2 PONG', self)
class C(B1, B2): # 继承顺序即 MRO 搜索顺序
def ping(self):
super().ping() # 委托父类 A 调用 ping()
print('C ping', self)
def ping_pong(self):
self.ping() # A ping, C ping
super().ping() # A ping
self.pong() # B1 pong
super().pong() # B1 pong
B2.pong(self) # B2 PONG # 显式调用某个超类方法,由于未绑定对象,需将传入对象

> C.__mro__ # [<class 'C'>,'B1','B2','A','object']

ch13. 运算符重载

重载分派机制:重载运算符时,若对应的特殊方法返回单例值NotImplemented,则解释器会对调操作数,调用该运算符的反向特殊方法重试;比如__add____radd__,流程如下:

示例:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
class Vector:
def __add__(self, other):
try:
pairs = itertools.zip_longest(self, other, fillvalue=0.0)
return Vector(x + y for x, y in pairs) # 特殊方法不修改操作数,生成新对象返回,保证逻辑正确
except TypeError: # 操作数类型无效
return NotImplemented # 通知解释器调用反向方法重试重试

def __radd__(self, other): # r: reserve
return self + other # 交换操作数,直接委托 __add__ 执行加法

> v = Vector([1, 2])
> v + [1] # Vector([2,2]) # Vector.__add__
> [1] + v # list.__add__ 无法处理 Vector 类型,返回 NotImplemented 分派给 Vector.__radd__
> 1 + v # TypeError: unsupported operand type(s) for +: 'Vector' and 'int' # 错误信息明确

如上的加法并未对另一操作数做类型预设,直接执行运算,出错才捕获异常,更灵活;此外,可用isinstance配合抽象基类对操作数做类型校验,更安全:

1
2
3
4
5
6
7
8
9
10
11
12
class Vector:
def __mul__(self, scalar):
if isinstance(scalar, numbers.Real): # 测试抽象类更通用
return Vector(scalar * v for v in self)
else:
return NotImplemented

def __rmul__(self, scalar):
return self * scalar
> v3 = Vector([1, 2])
> v3 * 10 # [10,20]
> v3 * True # [1,2]

比较运算符:特别地,==即使类型不匹配也不会抛出错误,是因为有比较id()的后备机制,同理!=的后备机制是==结果取反

1
2
3
4
5
6
7
class Vector:
def __eq__(self, other):
if isinstance(other, Vector): # 限制只能与同类对象比较
return len(self) == len(other) and all(a == b for a, b in zip(self, other))
else:
return NotImplemented
> Vector([1, 2]) == [x for x in (1, 2, 3)] # False

就地修改:额外返回self作为计算结果即可


ch14. 迭代器与生成器

本节迭代了 5 个版本的按单词分割文本的迭代器,分别通过序列协议、迭代器与生成器实现

v1. 实现序列协议

1
2
3
4
5
6
7
8
9
WORD_RE = re.compile('\w+')
class Sentence:
def __init__(self, text):
self._words = WORD_RE.findall(text) # 匹配文本中所有单词
def __getitem__(self, item): # 序列协议实现
return self._words[item]
> s = Sentence('"a" bc,def')
> for w in s:
> print(w) # 'a','bc','def' # 可迭代

iter() 函数:迭代对象时,若实现了__iter__则使用返回的迭代器,或实现了__getitem__则解释器创建隐式迭代器并从 0 开始顺序迭代,否则无法迭代

可迭代性检查:

1
2
3
4
5
6
7
8
9
try:
it = iter(Sentence(''))
except TypeError:
print('not iterable')
else:
print('Sentence is iterable') # bingo # iter() 会兼容 __getitem__

from collections import abc
> issubclass(Sentence, abc.Iterable) # False # 没有实现 __iter__

迭代器 Iterator 接口

  • __next__:返回下一个待迭代元素,若无可用元素则抛出StopIteration异常

  • __iter__:返回self实现Iterable抽象类

1
2
3
4
5
6
7
8
s = 'ABC'
it = iter(s) # 模拟 for 迭代操作
while True:
try:
print(next(it)) # 'A','B','C' # 会调用 it.__next__
except StopIteration: # 迭代结束
del it
break

v2. 分离可迭代序列及迭代器:手动实现SentenceIterator来实现迭代逻辑

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
class Sentence:
def __init__(self, text):
self._words = WORD_RE.findall(text)
def __iter__(self): # 取代 __getitem__,迭代逻辑委托给了 SentenceIterator
return SentenceIterator(self._words) # 返回迭代器,实现 abc.Iterable

class SentenceIterator:
def __init__(self, words):
self._words = words
self._idx = 0 # 维护迭代状态
def __next__(self): # 实现 abc.Iterator
try:
word = self._words[self._idx]
except IndexError:
raise StopIteration() # 迭代结束
self._idx += 1
return word
def __iter__(self): # 实现 abc.Iterable
return self

需要说明,为了能从一个Sentence对象获取多个独立的迭代器,Sentence内部不能维护迭代状态,所以该状态被委托给了sentenceIterator._idx,每次获取到的迭代器都是新的SentenceIterator对象;综上,可迭代对象一定不能是自身的迭代器,即必须实现__iter__,不能实现__next__

v3. 生成器 generator 函数

1
2
3
4
5
6
class Sentence:
def __init__(self, text):
self._words = WORD_RE.findall(text)
def __iter__(self): # 迭代逻辑委托给了下方的 generator 对象(已实现 __next__ 接口)
for word in self._words: # 可认为 generator 是 iterator 的工厂,每次迭代都生成新的 iterator
yield word # 迭代、产出元素、暂停、再迭代...

生成器函数:使用yield的函数,返回生成器对象

1
2
3
4
5
6
7
8
9
10
11
def gen_12():
yield 1
yield 2

> gen_12 # <function gen_12 at 0xxx>
> gen_12() # <generator object gen_12 at 0xxx>
> for i in gen_12():
> print(i) # 1,2
> g = gen_12()
> next(g), next(g) # 1,2
> next(g) # StopIteration

v4. 惰性求值

v1~v3 都是尽早求值(eager evaluation):初始化时就全量构建单次列表,若迭代只取前几个单词会导致内存浪费

v4 将实现惰性求值(lazy evaluation):使用正则匹配迭代器,真正逐步迭代时,才逐个匹配并产出单词

1
2
3
4
5
6
class Sentence:
def __init__(self, text):
self._text = text # 不再构建 _words 列表
def __iter__(self):
for match in WORD_RE.finditer(self._text): # 使用正则迭代器
yield match.group()

v5. 生成器表达式

使用生成器表达式简化yield语法

1
2
3
4
5
class Sentence:
def __iter__(self):
return (match.group() for match in WORD_RE.finditer(self._text))

> (v*10 for v in gen_12()) # <generator object <genexpr> at 0xx> # 生成器表达只是语法糖,可读性更高

标准库生成器函数:说明itertools,functools等标准库中的常用生成器函数

1. 过滤

1
2
3
4
5
6
7
8
9
10
11
12
def is_vowel(c: str):
return c.lower() in 'aeiou'

> s = 'Africa'
> list(filter(is_vowel, s)) # Aia # 过滤出 predicate 结果为真的元素,内置函数
> list(itertools.filterfalse(is_vowel, s)) # frc # 少用,放入 itertools 模块
> list(itertools.dropwhile(is_vowel, s)) # frica # 跳过 predicate 为真的第一个元素,立刻返回剩余元素
> list(itertools.takewhile(is_vowel, s)) # A # 获取 predicate 为真的第一个元素,立刻返回
> list(itertools.compress(s, [1,0,1,1,0])) # Ari # 将 selector_it 的元素与 it 一一对应,为真则产出
> list(itertools.islice(s, 1)) # A # 惰性切片产出元素 # s[1]
> list(itertools.islice(s, 1, 3)) # fr # s[1:3]
> list(itertools.islice(s, 1, 5, 2)) # fi # s[1:5:2]

2. 映射

1
2
3
4
5
6
> vs = [3, 1, 6, 0, 9]
> list(itertools.accumulate(vs, min)) # [3,1,1,0,0] # 累积规约结果
> list(itertools.accumulate(vs, operator.mul)) # [3,3,18,0,0] # 连乘结果列表
> list(enumerate('abc', start=1)) # [(1,a), (2,b), (3,c # 产出由 (idx,item) 组成的元组
> list(map(operator.add, range(1, 11), [10, 20])) # [11, 22] # 并行处理多个可迭代对象
> list(map(lambda a, b: (a, b), 'ab', 'ABCD')) # (a,A),(b,B) # 类似 zip

3. 合并

1
2
3
4
> list(itertools.chain('ABC', range(2)))                # A,B,C,0,1   # 连接多个可迭代对象
> list(itertools.chain.from_iterable(enumerate('ABC'))) # 0,A,1,B,2,C # 内部展开后再连接
> list(zip('ABC', range(5), [10, 20])) # (A,0,10),(B,1,20)
> list(itertools.zip_longest('ABC', range(2), [10, 20], fillvalue='?')) # (A,0,10),(B,1,20),(C,?,?)

4. 扩展

1
2
3
4
5
> list(itertools.combinations('ABC', 2)) # AB,AC,BC                   # 组合
> list(itertools.permutations('ABC', 2)) # AB,AC,BA,BC,CA,CB # 排列
> list(itertools.islice(itertools.count(1, 0.3), 3)) # 1,1.3,1.6 # 等差数列
> list(itertools.islice(itertools.cycle('ABC'), 7)) # ABCABCA # 循环产出
> list(map(operator.mul, range(11), itertools.repeat(5))) # 0,5,10,15 # 重复产出

5. 重排

1
2
3
4
5
6
7
8
9
> animals = ['rat', 'bat', 'duck', 'eagle', 'bear', 'lion', 'shark']
> list(reversed(animals)) # 翻转序列,倒序产出元素
> [list(it) for it in itertools.tee('AB', n=2)] # AB,AB # 产出 n 个相同生成器
> animals.sort(key=len) # 分组前排序是必须的
> for length, group in itertools.groupby(animals, key=len): # 按 key 结果分组
> print(length, list(group))
3 ['rat', 'bat']
4 ['duck', 'bear', 'lion']
5 ['eagle', 'shark']

ch15. 上下文管理

else 子句:与if/else的强互斥逻辑不同,for,while,try搭配的else子句,只有在完整迭代结束或无异常抛出时才会执行

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
d = dict(a=1, b=2, c=3)
try:
c = d['c']
except KeyError:
pass
else:
print('c in dict') # 无异常,执行

for k, v in d.items():
continue
else:
print('for done') # 完整迭代,不被中断,执行

while True:
break
else:
print('while done') # 非条件为假退出,不会执行

上下文管理器:简化版的try/except/finally,搭配with子句实例化,as子句绑定目标变量;管理协议:

  • __enter__:返回目标变量,供as子句绑定后使用
  • __exit__:类比finally,在退出前做清理资源、重置状态等工作

示例:实现管理协议,替换标准输出来翻转字符串

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
class LookingGlass:
def __enter__(self):
self.real_write = sys.stdout.write
sys.stdout.write = lambda a: self.real_write(a[::-1])
return 'ABCD'

def __exit__(self, exc_type, exc_val, exc_tb): # with 语句块若抛出异常,则入参为异常类型、异常对象、traceback 对象
sys.stdout.write = self.real_write # 重置 stdout 状态
if exc_type is ZeroDivisionError:
print('do not /0')
return True # 返回 True 告知解释器异常已被处理
# 返回 None 或者其他值,则异常向上冒泡

> with LookingGlass() as what:
> print(what == 'ABCD', what) # eurT DCBA # as 子句绑定的变量即 __enter__ 返回值
> print('with block end') # dne kcolb htiw #
> 1/0 # do not /0 # 抛出异常中断退出
> print('unreachable code')

contextlib 模块:包含大量现成的构建上下文管理器的工具集,以@contextmanager为例,将生成器函数装饰为上下文管理器,减少实现协议的样板代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
@contextlib.contextmanager
def look_glass():
real_write = sys.stdout.write

def reverse_write(txt: str):
real_write(txt[::-1]) # 闭包访问外部变量

sys.stdout.write = reverse_write
msg = ''
try: # 之前的代码即 __enter__
yield 'ABCD' # yield 产出一个值,绑定到 as 子句变量,暂停去执行 with 语句块
except ZeroDivisionError: # 之后的代码即 __exit__
msg = 'do not /0'
finally:
sys.stdout.write = real_write # finally 保证 stdout 一定重置
if msg:
print(msg)

> look_glass() # <contextlib._GeneratorContextManager object at 0xx> # 已被改写
> with look_glass() as what:
> print(what) # DCBA
> 1/0 # do not /0
> raise IndexError() # 不会被执行

原理:contextlib._GeneratorContextManager实现了上下文管理接口:

  • __enter__
    • 执行生成器函数,得到生成器对象 generator
    • 执行next(generator)产出一个值,绑定到as的变量
  • __exit__
    • 若有异常,则调用generator.throw(exception),将异常透传回yield处捕捉处理
    • 若无异常,则执行next(generator)恢复执行yield之后的代码

ch16. 协程

yield:面向数据传输的流程控制方式,控制数据来源、数据流向;x = yield y 会使生成器对象产出数据y,并暂停等待接收调用方发来数据放入x

示例:生成器作为协程的行为及其四种状态

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
def simple_cor2(a):
print('START a:', a) # GEN_RUNNING # 正在执行
b = yield a
print('RECV b:', b)
c = yield a + b
print('RECV c:', c)

> c = simple_cor2(14)
> inspect.getgeneratorstate(c) # GEN_CREATED # 已生成生成器对象,等待激活(prime)
> next(c) # START a: 14; 14
> inspect.getgeneratorstate(c) # GEN_SUSPENDED # 已在 yield 处暂停
> c.send(28) # RECV b: 28; 42
> try:
> print(c.send(99)) # RECV c: 99
> except StopIteration:
> pass
> inspect.getgeneratorstate(c) # GEN_CLOSED # 协程执行结束

三个执行阶段示意图:

示例:使用协程计算移动平均值

1
2
3
4
5
6
7
8
9
10
11
12
13
def averager():
total, count = 0.0, 0
avg = None
while True:
next_val = yield avg
total += next_val
count += 1
avg = total / count

> avger = averager()
> next(avger) # None # 使用前必须激活协程,使其在 yield 处暂停
> avger.send(10) # 10
> avger.send(11) # 10.5

激活协程的装饰器:将协程激活逻辑交由装饰器实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def coroutine(func):
@functools.wraps(func)
def primer(*args, **kwargs):
gen = func(*args, **kwargs)
next(gen)
return gen
return primer

@coroutine
def averager():
# ...

> avger = averager()
> inspect.getgeneratorstate(avger) # GEN_SUSPENDED # 已激活

协程的异常处理:协程内部未处理的异常会冒泡到 caller;此外 caller 也能显式地将异常发送给协程,将在yield暂停处抛出,两种方式:

  • generator.throw(exc_type[, exc_val[, traceback]]):发送指定类型的异常
  • generator.close():发送GeneratorExit异常,若协程未处理此异常或抛出StopIteration则说明关闭成功

示例:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
class DemoException(Exception):
pass

@coroutine
def demo_exc_handling():
while True:
try:
v = yield
except DemoException:
print('handled DemoException, continue yield...')
else:
print('RECV:', v)

> c = demo_exc_handling()
> c.send(1) # RECV: 1
> c.throw(DemoException) # handled...
> try:
> c.throw(ZeroDivisionError) # 向协程发送无法处理的异常,将导致协程退出
> except ZeroDivisionError: # 会冒泡回来
> pass
> inspect.getgeneratorstate(c) # GEN_CLOSED

协程返回值:返回值会放到StopIteration.value字段,caller 需在协程结束后自行读取

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
Result = collections.namedtuple('Result', 'count average')

@coroutine
def averager():
total, count = 0.0, 0
avg = None
while True:
next_val = yield # 不再产出平均值
if next_val is None: # 正常终止协程
break
total += next_val
count += 1
avg = total / count
return Result(count, avg) # 返回项数和平均值

> avger = averager()
> avger.send(10), avger.send(11) # None, None
> try:
> avger.send(None)
> except StopIteration as e:
> print(e.value) # Result(count=2, average=10.5)

yield fromsyntax for delagate to subgenerator,将生成器职责委托给子生成器;涉及到三个角色:

  • delegate generator:委派生成器,使用yield from实现委托逻辑的生成器
  • subgenerator:子生成器,被委托产出数据的生成器
  • caller:调用方,调用委派生成器获取数据

核心的委托逻辑:yield from subgen();此后 subgen 获得控制权,会将产出的值传回 caller,而 gen 将“阻塞”等待 subgen 协程终止

示例:现有一班级数据集,记录了男生、女生的身高、体重:

1
2
3
4
5
6
data = {
'girls;kg': [40.9, 38.5, 44.3, 42.2, 45.2, 41.7, 44.5, 38.0, 40.6, 44.5],
'girls;m': [1.6, 1.51, 1.4, 1.3, 1.41, 1.39, 1.33, 1.46, 1.45, 1.43],
'boys;kg': [39.0, 40.8, 43.2, 40.8, 43.1, 38.6, 41.4, 40.6, 36.3],
'boys;m': [1.38, 1.5, 1.32, 1.25, 1.37, 1.48, 1.25, 1.49, 1.46],
}

使用averager协程,分别统计如上 4 个维度的平均值

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
@coroutine
def averager(): ## subgen
# ...

@coroutine
def delegator(res, name): ## gen
while True: ### 可优化:while 死循环让 gen 永不结束,不抛出 StopIteration
res[name] = yield from averager() # 持续传递数据,暂停等待 averger 协程执行结束
# 解释器自动捕捉 StopIteration 并分离出 value,作为 yield from 表达式的值
def caller(data: dict): ## caller
res = {}
for name, vs in data.items():
gen = delegator(res, name)
for v in vs:
gen.send(v) # 持续将数据传给 averager 的 `next_val = yield`
gen.send(None) # 必需的,触发 averager 执行结束,才能让 group 从 yield from 返回
print(results)

> caller(data)
{'girls;kg': Result(count=10, average=42.04), ...}

类比装饰器工厂函数,Python 原生要实现类似 Go 的 channel 机制,也要用 yield from 套娃生成器,实在是费劲…


ch17. 并发

future:封装异步等待结果的操作,提供状态查询、结果获取、异常冒泡等机制

示例:使用concurrent.futures模块提供的线程池,并发执行网络 IO(下载资源),并用 future 等待结果

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
from concurrent import futures

def download_one(cc: str): # 线程执行的单个任务 task
img_buf = get_flag(cc)
save_flag(img_buf, cc.lower() + '.gif')
logging(cc)

def download_many_by_futures(cc_list):
workers = min(MAX_WORKERS, len(cc_list)) # 避免创建多余的线程
# executor 实现了上下文管理协议,退出时调用 executor.shutdown(wait=True) 阻塞等待线程池中所有任务执行完
with futures.ThreadPoolExecutor(max_workers=workers) as executor:
res = executor.map(download_one, sorted(cc_list)) # map 指定并发执行的函数及其参数,返回生成器获取各子任务结果
return len(list(res))

def download_many_as_completed(cc_list):
cc_list = cc_list[:5]
with futures.ThreadPoolExecutor(max_workers=5) as executor:
todo = [] # future 对象列表
for cc in cc_list:
future = executor.submit(download_one, cc) # 提交执行一个 task 排队,返回 future 对象
todo.append(future)

results = []
for future in futures.as_completed(todo): # 监测 future 列表,先结束的先被迭代返回
res = future.result() # 读取 future 对应的任务执行结果,或冒泡异常
results.append(res)

return len(results)

如上,concurrent 对用户屏蔽了任务调度和执行的逻辑,通过submit,map等方式提交任务,再通过返回的 future 对象自行等待结果,特别地,as_completed实现了类似 Go 的 select 机制,监控多个 future 对象先完成的先返回

GIL:历史原因,CPython 解释器并非线程安全,其使用 GIL(Global Interpreter Lock) 全局解释器锁,限制同一时间只有一个线程执行 Python 字节码,所谓多线程并发只是 Python 解释器单进程内的并发,无法利用多核;示意图(rohanvarma.me/GIL):

标准库中所有会阻塞 IO 的函数,在阻塞时都会释放 GIL,让其他线程运行,因此适合 IO 密集型应用

并行:CPU 密集型应用常用到多核实现并行计算,concurrent.ProcessPoolExecutor 提供的进程池绕过了 GIL 限制,实现了真正的并行执行;

底层的多线程由threading模块实现,多进程由multiprocessing模块实现,后者搭配Lock,Semaphore,Condition,Event 等资源来实现 IPC,更为灵活,后边写 Python 并发编程的文章梳理


总结

各章概要

  • ch01:Python 数据模型,动态实现模型协议,来复用面向模型的库函数、语法规则

  • ch02:list, tuple 及其推导表达式,增量赋值的本质,切片原理,序列的搜索和排序等

  • ch03:可散列性,dict 及其多个变种,set 及其数学操作

  • ch05:函数作为可调用对象,列举其内省属性,引出了高阶函数、lambda 表达式匿名函数

  • ch07:装饰器原理和闭包特性,列举常用装饰器,以及装饰器工厂函数的原理

  • ch08:深浅拷贝的区别,引用参数作为形参默认值存在的隐患,垃圾回收的概念

  • ch09:对象的格式化方式,私有属性等

  • ch10:实现序列协议说明鸭子类型

  • ch11:以序列为例说明抽象基类的实现关系,并自定义抽象基类

  • ch12:MRO 机制解决多重继承的菱形问题,以及内置类型的继承替代类

  • ch13:重载运算符存在重载分配机制

  • ch14:序列协议,迭代器的两个接口,yield 生成器,列举常用的标准库生成器函数

  • ch15:上下文管理器的两个接口,contextlib 模块用法

  • ch16:生成器作为协程的原理,说明预先激活、异常处理与返回值,最后说明实现协程的委派生成器的原理

  • ch17:使用 concurrent.futures 提供的线程池、进程池进行异步任务处理

Python 语法简洁,内置类型和函数丰富,既有 duck typing 的动态协议机制,也有抽象基类的静态约束机制,有独特的 decorator, iterator, generator 概念,也有 yield, yield from 等独创的协程机制