Feb 10, 2021

Notes for Effective Python

1. Pythonic Thinking

1. version

python -V
python --version
import sys
print(sys.version_info)
print(sys.version)
# try dir(sys) to find out more

2. PEP8 Python Enhancement Proposal #8

3. bytes, str and unicode

4. 从复杂表达式抽出帮助 (helper) 函数 - 不要写过长的单行逻辑 - if/else 优于 and/or

5. 序列分片 Slice Sequence - 从左向右（注1），[起始序号（含）:中止序号（不含）]，起始在中止右边则返回 [] - A[:end] == A[0:end]， A[start:] == A[start:len(A)] - 起始序号和中止序号越界也不会出错，会分别换成 0 和 len(A) - 给分片赋值相当于把原序列的内容先删除，再插入要赋值的东西：

a = [1, 2, 3, 4, 5, 6, 7]
a[0:2] = 'a', 'b' # ['a', 'b', 3, 4, 5, 6, 7]
a[0:3] = 'X'      # ['X',         4, 5, 6, 7]
a[:]   = 'A'      # ['A'] 整个序列 a 都被换掉了

6. 序列分片不要同时指定起始，中止和步长(stride) - 避免负数步长（遇到unicode时可能会出错）u'美国'.encode('utf-8')[::-1].decode('utf-8') - 一定要指定起始，中止和步长时，将截取和步长分两步走，但是会浪费内存，可以试试 itertools.islice

注1: 默认步长是 1，即从左向右。

7. List Comprehension 优于 map/filter

map 更啰嗦

a = [1, 2, 3, 4, 5, 6, 7]
asquare = [x**2 for x in a]
asquare2 = map(lambda x: x ** 2, a) # bad

含有过滤条件时比 filter 更清楚

a = [1, 2, 3, 4, 5, 6, 7]
even_squares = [x**2 for x in a if x % 2 == 0]
even_squares2 = map(lambda x: x ** 2, filter(lambda x: x % 2 == 0, a))

还有 Dict Comprehension 和 Set Comprehension

8. List Comprehension 里不要多过两个表达式 - 两个表达式的限制是指 for（loop） 和 if（condition）加起来不超过 2 个 - 超过两个，还不如用 for loop

matrix = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]
flat = [x for row in matrix for x in row]         # 注意 for 的顺序
squared = [[x**2 for x in row] for row in matrix] # 还能接受


big_matrix = [
  [[1, 2, 3], [4, 5, 6]],
  [[7, 8, 9], [10, 11, 12]],
]
flat = [m3 for m1 in big_matrix                   # TOO MUCH!!!
        for m2 in m1
        for m3 in m2
]
flat2 = []
for m1 in big_matrix:                             # Better
  for m2 in m1:
    flat2.extend(m2)
  • 含有多个 if 过滤条件的时候相当于 and
a = [1, 2, 3, 4, 5, 6, 7]
b = [x for x in a if x > 4 if x % 2 == 0]    # 连续两个 if 相当于 if a and b
c = [x for x in a if x > 4 and x % 2 == 0]

9. List Comprehension 太大不妨用生成器 (Generator)

参考资料

10. enumerate 优于 range（有时候）

for idx, fruit in enumerate(fruits, 1):  # idx 从 1 开始数，对现实世界计数时比较方便
  print(idx, fruit)

11. zip (python2 izip) 来并行处理迭代器 （参考 #7）

12. for … else 和 while … else 不可取 - 这里的 else 非常反直觉

for i in range(3):
  print(i)
else:
  print("ELSE!")      # else 可及：看上去 else 是「之前」没有运行才会运行的，实际上之前 for loop 完成之后还是运行了 else
  

for i in range(3):
  print(i)
  if i == 1:
    break
else:
  print("ELSE!")      # else 不可及：看上去 for loop 退出了，应该运行 else 了，实际上 else 会被跳过。
  
# for loop 是空的时候 直接执行 else
for i in []:
  print(i)
else:
  print("ELSE!")       # else 可及

# while 初始条件是 False 时直接执行 else
while False:
    print("ok")
else:
    print("ELSE")

a = iter([None, 1, 2, 3, None, 5, 6])
while next(a):
    print('hi')
else:
    print("ELSE")
  • 唯一有用的 case 是在 loop 中找东西的时候，即使这样仍然不推荐
for i in range(2, min(a, b) + 1):
    print('testing', i)
    if a % i == 0 and b % i == 0:
        print('Not coprime')
        break           # 只有 break 才不会运行 else
else:
    print("Coprime")

# 可以在发现满足条件的时候提前 return
def coprime(a, b):
    for i in range(2, min(a, b) + 1):
        if a % i == 0 and b % i == 0:
            return False
    return True

# 或者使用一个变量来表示搜索的结果
def coprime2(a, b):
    is_coprime = True
    for i in range(2, min(a, b) + 1):
        if a % i == 0 and b % i == 0:
            is_coprime = False
            break
    return is_coprime

13. 充分利用 try/except/else/finally （参考 #51 的例子） - finally 用来做收尾/善后/清理工作，比如及时关闭 file handle （当然可以使用 context，参见 #43）

f = open('/tmp/abc.txt')       # 如果文件不存在，这里会 raise IOError，这时不需要 finally 中的清理工作（都没 open，何必 close 呢）
try:
    data = handle.read()
finally:
    handle.close()              # 不管 try 的结果如何，finally 一定会执行。raise 异常的时候会控制转移，后面的代码不会执行，所以清理工作只能放在 finally 中。
  • else 可以减少 try block 中的代码
  • else 放一些 try 成功之后做的事。当然通用的清理工作还是在 finally

2. Functions

14. 抛异常优于返回 None - 如果用 None 表示异常可能会有意外。比如可能正常返回的 False，0 和异常的 None 对于 if 判断来说都是 False。 - 写好文档，放心地抛异常，调用方会（也应该能）处理好。 （文档 参见 #49）

15. 理解闭包（Closure）中的变量作用域

使用场景：有一个列表，其中优先排列一些元素。

方法1: 使用 sort 方法，在优先 group 的元素排在前面

def sort_priority(values, group):
    def helper(x):
        if x in group:           # helper 可以访问到外层定义的 group
            return (0, x)        # sort 排序的对象是这里生成的 tuple (0, x) 或 (1, x)。先比较第一个元素（0 或 1），再比较第二个元素（x）
        return (1, x)
    values.sort(key=helper)      # helper 是一个closure function，可以作为 key 参数传入 sort。函数是 first-class object

numbers = [1, 2, 3, 4, 5, 6, 7]
group = [6, 7]
sort_priority(numbers, group)
print(numbers)
# [6, 7, 1, 2, 3, 4, 5]

如果想知道 numbers 中有没有 group 中的元素，

def sort_priority2(numbers, group):
    found = False                # scope: 'sort_priority2'
    def helper(x):
        if x in group:
            found = True         # scope: 'helper' 因为不在同一 scope，这里实际上是定义了新变量，而不是对旧变量赋值
            return (0, x)
        return (1, x)
    numbers.sort(key=helper)
    return found

那如何让 helper 中的 found 指向外面的 found 呢？

python3 引入了 nonlocal 表达式(statement) （慎重使用）

def sort_priority2(numbers, group):
    found = False
    def helper(x):
        nonlocal found           # found 不再是 helper 这个 closure function 内的，而是指向外层的 found
        if x in group:
            found = True
            return (0, x)
        return (1, x)
    numbers.sort(key=helper)
    return found

python2 没有 nonlocal 只能使用黑科技，利用 list 来传递数据（常见 Python 技巧）：

def sort_priority(numbers, group):
    found = [False]
    def helper(x):
        if x in group:
            found[0] = True     # 这里分两步：1. 找到 found 定义（在外层），2. 修改。变相实现了 nonlocal
            return (0, x)
        return (1, x)
    numbers.sort(key=helper)
    return found[0]

使用变量的时候，会依据如下顺序从内到外去寻找，修改变量的时候则只用自己的 scope 中的变量，没有找到就新建一个：

  1. The current function’s scope
  2. Any enclosing scopes (like other containing functions)
  3. The scope of the module that contains the code (also called the global scope)
  4. The built-in scope (that contains functions like len and str)

16. 返回生成器(generator)优于返回列表

  • 注意 generator 的结果拿一次就耗尽了。（参见 #17）

例子，找到一句话中每个单词的位置： 比较一下，返回 list 的实现：

def index_words(text):
    result = []                         # 先定义一个 list 来存放结果。这里可能会占用过多内存。
    if text:
        result.append(0)
    for index, letter in enumerate(text):
        if letter == ' ':
            result.append(index + 1)    # 每找到一个词都要调用一次 append；`index + 1` 这个信息更重要，但是被 append 淡化了
    return result

返回 generator 的实现：

def index_words_iter(text):             # 不需要一个list来存储结果，省内存
    if text:
        yield 0
    for index, letter in enumerate(text):
        if letter == ' ':
            yield index + 1              # 突出重点

17. 应对生成器只能用一次的问题

  • 自己实现一个 iterable 容器，支持多次使用

18. Variable Positional Arguments 可变（位置）参数 （* 操作符）

def log(msg, values):
    print(msg)
    for v in values:
        print(v)

# 调用时要传一个 list，很麻烦，尤其是第二个参数用不到的时候，传入空列表很无语
log("it's my msg", [1, 2])
log("msg again", [])

# 这时可以使用 * 操作符
def log(msg, *values):                    # 只需要加一个 * 符号就搞定了
    print(msg)
    for v in values:
        print(v)
log("it's my msg", 1, 2)
log("msg again")
# 还可以接着用 * 操作符
values = [1, 2]
log("it's my msg", *values)
log("it's my msg", *[1, 2])

使用可变参数有两个问题：

  1. variable argument 在传到函数内部之前就会转成 tuple，如果传入一个 generator，就会完全遍历一遍，浪费内存
  2. 如果代码更新引入了新位置参数，旧调用方必须修改代码，兼容性差
def log(version, msg, *values):
    print(version)
    print(msg)
    for v in values:
        print(v)
log("it's my msg", 1, 2)     # 出错了 但是不会抛异常！！！所以要加参数，最好用关键字参数（参见 #21）
log(1, "it's my msg", 1, 2)  # 必须修改代码做兼容

19. 关键字参数

20. 动态默认参数应该用 None 和 Docstrings

21. 限制只能使用关键字参数会更清楚

  • Python3：def safe_division_c(number, divisior, *, ingore_overflow=False, ignore_zero_division=False):
  • Python2 没有天然支持，需要特殊技巧：
def save_division_d(number, divisior, **kwargs):
    ignore_overflow = kwargs.pop('ignore_overflow', False)
    ignore_zero_div = kwargs.pop('ignore_zero_division', False)
    if kwargs:
        raise TypeError('Unexpected **kwargs: %r' % kwargs)

3. Classes and Inheritance

22. 使用辅助类(Helper Classes)来维护程序的状态优于字典和元组 - 避免嵌套的字典和过长的元组 - namedtuple 可以当作轻型的类。数据是不可变的（immutable data container) - 当程序状态无法使用简单的字典维护时，考虑重构成类

namedtuple 的限制：

  1. 无法指定默认值
  2. 除了当作类用 . 属性的方式来访问外，namedtuple 还可以用索引下标以及迭代器来访问，如果滥用会导致将来转换成类的时候出现问题

23. 简单接口应该接受函数而不是类做参数

Many of Python’s built-in APIs allow you to customize behavior by passing in a function. These hooks are used by APIs to call back your code while they execute.

比如 list 类的 sort 方法，接受一个 key 参数来确定顺序。

names = ['peiqi', 'qiaozhi', 'suxi', 'anqi']
names.sort(key=lambda x: len(x))

再比如 defaultdict，接受函数作为参数用来定义发现缺失的 key 的时候的行为，这个函数必须返回缺少的 key 应该有的默认值：

def log_missing():
  print('Key added')
  return 0
  
result = defaultdict(log_missing)
result
# 初始值
# defaultdict(<function __main__.log_missing>, {})
result[0]
#Key added
#0

Supplying functions like log_missing makes APIs easy to build and test because it separates side effects from deterministic behavior.

比如要定义一个新钩子来统计遇到了多少次缺失 key 的情况，

第一个方法是使用有状态的闭包（参见 #15），这个方法可读性差些。

def increment_with_report(current, increments):
    added_count = 0
    def missing():
        nonlocal added_count   # 有状态的闭包
        added_count += 1
        return 0
    result = defaultdict(missing, current)
    for key, amount in increments:
        result[key] += amount
    return result, added_count
    
current={'blue': 3, 'green': 12}
increments=[('red', 5), ('blue', 17), ('orange', 9)]

result, count = increment_with_report(current, increments)

第二个方法是定义一个类来封装想要的状态，可读性更好。

class CountMissing(object):
    def __init__(self):
        self.added = 0
    def missing(self):
        self.added += 1
        return 0

在 Python 中可以把 CountMissing.missing 方法当作 hook 传入 defaultdict：

counter = CountMissing()
result = defaultdict(counter.missing, current)

for key, amount in increments:
    result[key] += amount
assert counter.added == 2

但是单独看 CountMissing 还是会比较困惑它是干嘛用的，只有结合 defaultdict 的用法才能明白。可以直接把这个类的实例当作函数来调用吗？（而不是调用其方法）方法就是使用 __call__。

class BetterCountMissing(object):
    def __init__(self):
        self.added = 0
    def __call__(self):
        self.added += 1
        return 0
counter = BetterCountMissing()
counter()
assert callable(counter)

current={'blue': 3, 'green': 12}
increments=[('red', 5), ('blue', 17), ('orange', 9)]
result = defaultdict(counter, current) # 直接把实例当作函数来用了，借助 __call__ 这个方法
for key, amount in increments:
    result[key] += amount
assert counter.added == 2

24. @classmethod 实现的多态

Python 中对象和类都支持多态。（参见 #28）

  • Python 每个类只支持一个构造器，就是 __init__
  • @classmethod 可以用来定义额外的构造器
  • Use class method polymorphism to provide generic ways to build and connect concrete subclasses.

25. 用 super 来初始化父类 - method resolution order(MRO) 解决父类继承以及菱形继承时的顺序问题 - 初始化父类一定要用 super（注意 py2 和 py3 的语法差异）

26. 只有 Mix-in 工具类的时候才使用多重继承

  • 能用 Mix-in 的地方就不要多重继承
  • 在实例的层面使用可插入的「行为」 
class ToDictMixin(object):
    def to_dict(self):
        return self._traverse_dict(self.__dict__)
    def _traverse_dict(self, instance_dict):
        output = {}
        for key, value in instance_dict.items():
            output[key] = self._traverse(key, value)
        return output
    def _traverse(self, key, value):
        if isinstance(value, ToDictMixin):
            return value.to_dict()
        elif isinstance(value, dict):
            return self._traverse_dict(value)
        elif isinstance(value, list):
            return [self._traverse(key, i) for i in value]
        elif hasattr(value, '__dict__'):
            return self._traverse_dict(value.__dict__)
        else:
            return value

27. 公开属性优于私有属性

28. collections.abc

4. Metaclasses and Attributes

29. 直接用属性，不要自己实现 get 和 set 方法 - 如果要在 set 属性的时候实现一些特殊操作，可以用 @property （不要在 get 属性的时候做！非常奇怪！因为 get 就只是查询一下）

30. @property 优于重构属性

  • 利用 @property 为现有代码的实例属性添加新功能
  • 与其过度使用 @property 不如重构整个类

31. 利用描述符创建可复用的 @property 方法

@property 最大的问题是无法复用。比如：

class Exam(object):
    def __init__(self):
        self._writing_grade = 0
        self._math_grade = 0

    @staticmethod
    def _check_grade(value):
        if not (0 <= value <= 100):
            raise ValueError("Grade must be between 0 and 100")
    @property
    def writing_grade(self):
        return self._writing_grade

    @writing_grade.setter
    def writing_grade(self, value):
        self._check_grade(value)
        self._writing_grade = value

    @property
    def math_grade(self):
        return self._math_grade

    @math_grade.setter
    def math_grade(self, value):
        self._check_grade(value)
        self._math_grade = value

里面有大量的重复，每个「科目」都要来一遍，怎样才能重用呢？

这时需要使用*描述符 descriptor*，描述符用来定义Python如何访问（access）对象的属性。通过用描述符实现__get__和__set__方法就可以复用代码。在这种情况下，描述符也比 mix-in 要好。

技术细节：如何访问描述符定义的属性

以下面代码为例：

class Grade(object):
    def __get__(*args, **kwargs):
        pass
    def __set__(*args, **kwargs):
        pass


class Exam(object):
    math_grade = Grade()
    writing_grade = Grade()
    science_grade = Grade()

exam = Exam()
# 设置属性时
exam.writing_grade = 40
# 实际上 Python 会解析为：
Exam.__dict__['wriging_grade'].__set__(exam, 40)

# 读取属性时
print(exam.writing_grade)
# 实际上 Python 会解析为：
print(Exam.__dict__['writing_grade'].__get__(exam, Exam))

上述行为的本质是 object 的 __getattribute__ 方法（见 #32）。

简而言之，

如果 Exam 的实例没有找到 writing_grade 属性，Python 会 fall back 去找 Exam 的类属性。如果这个类属性是个有 __get__ 和 __set__ 方法的 object 对象，Python 就知道你要使用描述符。

现在真正实现一下 Grade

class Grade(object):
    def __init__(self):
        self._value = 0
    def __get__(self, instance, instance_type):
        return self._value
    def __set__(self, instance, value):
        if not (0 <= value <= 100):
            raise ValueError("Grade must be between 0 and 100")
        self._value = value

这里有个bug，

first_exam = Exam()
second_exam = Exam()
first_exam.math_grade = 23
second_exam.math_grade = 99

print(first_exam.math_grade)
99

说明 second_exam 的 Grade实例 和 first_exam 的 Grade实例 指向了同一个对象，原因是Grade实例只在 定义 Exam 类的时候初始化一次。

要 fix 这个 bug，需要让 Grade 实例记住每个 Exam 实例：

class Grade(object):
    def __init__(self):
        self._values = {} # 这里改成，每个 Exam 实例对应一个值
    def __get__(self, instance, instance_type):
        if instance is None:
            return self
        return self._values.get(instance, 0)
    def __set__(self, instance, value):
        if not (0 <= value <= 100):
            raise ValueError("Grade must be between 0 and 100")
        self._values[instance] = value

不过这样写仍然有问题，内存泄漏。_values 保存了每个 Exam 实例的应用，也就是说 Exam 的实例的应用计数无法减到 0，也就导致 Exam 的实例无法被内存回收。

这时需要使用 WeakKeyDictionary：

from weakref import WeakKeyDictionary
class Grade(object):
    def __init__(self):
        self._values = WeakKeyDictionary()
    def __get__(self, instance, instance_type):
        if instance is None:
            return self
        return self._values.get(instance, 0)
    def __set__(self, instance, value):
        if not (0 <= value <= 100):
            raise ValueError("Grade must be between 0 and 100")
        self._values[instance] = value

take away:

  • 要复用 @property 就自己写描述符 descriptor
  • 用 WeakKeyDictionary 避免内存泄漏
  • 用描述符的时候不要纠结 __getattribute__ 是怎么实现的。（实现细节看下一条 😂）

32. 用 __getattr__, __getattribute__ 和 __setattr__ 实现懒属性（按需生成的属性）

下面的代码示例围绕着数据库关系映射 ORM 展开

问题：如何用Python对象来表示数据库的行？如何在不知道数据库 schema 的前提下，动态的表示数据库呢？

直接使用的属性， @property 方法 和描述符 descriptor 都需要预先定义好，所以没办法动态表示数据库。

这时要用 __getattr__ 这个特殊方法！

如果类中定义了 __getattr__，每当对象的实例字典 instance dictionary 中找不到某个属性的时候，都会调用 __getattr__ 动态加载。

class LazyDB(object):
    def __init__(self):
        self.exists = 5
    def __getattr__(self, name):
        value = 'Value for %s' % name
        setattr(self, name, value)
        return value

data = LazyDB()
data.__dict__
{'exists': 5}
data.foo            # 访问一个没有预先定义的属性
data.__dict__       # 这个属性已经被加入 instance dictionary 了！
{'exists': 5, 'foo': 'Value for foo'}

为了进一步观察 __getattr__ 的工作原理，加上一些print：

class LoggingLazyDB(LazyDB):
    def __getattr__(self, name):
        print('Called __getattr__(%s)' % name)
        return super().__getattr__(name)     # 这种写法只适用于 Py3。为了避免无限调用。

data = LoggingLazyDB()
data.exists
>>> 5
data.foo       # 第一次访问未定义的属性时，调用了 __getattr__
>>> Called __getattr__(foo)
>>> 'Value for foo'

data.foo       # 第二次访问 foo 的时候直接返回了值。是因为已经通过 setattr 把 foo 放到 instance dictionary 了
>>> 'Value for foo'

对未定义的属性，__getattr__ 只获取一次，之后的属性访问直接拿结果。

如果要实现数据库的事务/交易（transaction），就要知道之前访问过的数据现在是否仍然有效（可能被你修改了），这时 __getattr__ 永远返回第一次调用时求的值，无法满足需求，需要有一个办法可以实现每次都重新取一次值，它就是 __getattribute__!

class ValidatingDB(object):
    def __init__(self):
        self.exists = 5
    def __getattribute__(self, name):
        print('Called __getattribute__(%s)' % name)
        try:
            return super().__getattribute__(name)
        except AttributeError:
            value = 'Value for %s' % name
            setattr(self, name, value)
            return value

data = ValidatingDB()
data.exists
>>> Called __getattribute__(exists) # 调用内置的属性都会触发 `__getattribute__`
>>> 5
data.foo
>>> Called __getattribute__(foo)
>>> Value for foo
data.foo
>>> Called __getattribute__(foo)
>>> Value for foo

可以利用 __getattribute__ 阻止访问一些属性：

class NoAccessToValue(object):
    def __getattribute__(self, name):
        if name == 'value':
            raise AttributeError('%s is forbidded' % name)

三个内置函数： hasattr，getattr 和 setattr。

如何惰性地将值写入函数属性？介绍最后一个特殊方法： __setattr__：

class SavingDB(object):
    def __setattr__(self, name, value):
        super().__setattr__(name, value)


class LoggingSavingDB(SavingDB):
    def __setattr__(self, name, value):
        print('Called __setattr__(%s, %s)' % (name, value))
        super().__setattr__(name, value)

注意要避免无限递归:

class BrokenDictionaryDB(object):
    def __init__(self, data):
        self._data = data
    def __getattribute__(self, name):
        print('Called __getattribute__(%s)' % name)
        return self._data[name]      # __getattribute__ -> self._data -> __getattribute__ -> self._data ... 无穷调用

data = BrokenDictionary({'foo': 3})
data.foo
>>> RecursionError: maximum recursion depth exceeded while calling a Python object

修改为

class DictionaryDB(object):
    def __init__(self, data):
        self._data = data
    def __getattribute__(self, name):
        data_dict = super().__getattribute__('_data')
        return data_dict[name]

data = DictionaryDB({'foo': 3})         # 如果执行过👆的 BrokenDictionaryDB，这里不要用 data 这个名字
data.foo
>>> 3

take away

  • __getattr__ 和 __setattr__ 可以实现懒加载，懒保存
  • __getattr__ 只在第一次访问时调用一次， __getattribute__ 每次访问都调用
  • 用 __getattribute__ 和 __setattr__的时候要小心循环调用， 记得用 super()（这时其实访问的就是 object 类自己）来直接访问 instance attribute

33. 用元类 Metaclass 来验证子类

Metaclass 工作原理

元类是继承自 type 的类：

# py3
class Meta(type):       # 继承自 type
    def __new__(meta, name, bases, class_dict):
        print((meta, name, bases, class_dict))
        return type.__new__(meta, name, bases, class_dict)

class MyClass(object, metaclass=Meta):
    stuff = 123
    def foo(self):
        pass

>>> (<class '__main__.Meta'>, 'MyClass', (<class 'object'>,), {'__module__': '__main__', '__qualname__': 'MyClass', 'stuff': 123, 'foo': <function MyClass.foo at 0x10e6cb950>})

# py2 
class MyClassInPython2(object):
    __metaclass__ = Meta  # Python 2 的写法...
    # ...

回到我们最初的目的，如何验证子类？ 在 MetaClass 的 __new__ 中就可以检查子类的参数，因为Metaclass 可以拿到子类的名字，子类的父类是什么，子类的所有属性信息等。

例子，写一个类表示多边形，多边形有一些限制条件，可以把这些限制条件写成一个*验证用的元类*，然后让多边形的父类使用这个元类，再写集成了多边形父类的具体的子类。注意，元类中的限制条件只是给具体的多边形子类的，不要用在多边形父类上。

class ValidatePolygon(type):
    def __new__(meta, name, bases, class_dict):
        # Do not validate the abstract Polygon class
        if bases != (object,):      # 不要将限制用在父类上！
            if class_dict['sides'] < 3:
                raise ValueError('Polygons have more than three sides!')
        return type.__new__(meta, name, bases, class_dict)

class Polygon(object, metaclass=ValidatePolygon):
    sides = None
    @classmethod
    def interior_angles(cls):
        return (cls.sides - 2) * 180

class Triangle(Polygon):
    sides = 3

如果只指定2个边会报错：

print('1. Before Class')
class Line(Polygon):
    print("2. Before sides")
    sides = 1
    print("3. After sides")
print('4. After Class')
>>>
1. Before Class
2. Before sides
3. After sides
ValueError: Polygons have more than three sides!
# 没有 4... # 说明 class 要在跑完整个类定义之后，再跑元类的 __new__ 方法

take away

  • 用元类强制保证子类的定义符合规范
  • Python2 和 Python3 指定元类的方法有差别
  • 元类的 __new__ 方法只有在 class 的类定义跑完之后才会运行

34. 用元类注册子类

take away

  • 写模块化 Python 程序时，可以考虑类的注册
  • 每次从基类集成子类时，基类的元类可以自动执行注册代码
  • 用上面这种元类来自动注册可以防止漏掉注册某个子类

35. 用元类注释类的属性

5. 并发和平行计算

The key difference between parallelism and concurrency is speedup.

并发不会提高执行速度

36. 用 subprocess 管理子进程

父进程是 Python 解释器。

import subprocess

proc = subprocess.Popen(       # Popen 开启子进程
    ['echo', 'what a world'],
    stdout=subprocess.PIPE
)
out, err = proc.communicate() # communicate 读取子进程的输出，等待子进程结束，也就说这里阻塞了
out.decode('utf-8')
>>> 'what a world\n'

如果想在等待子进程完成前做点别的事情：

proc = subprocess.Popen(['sleep', '0.3'])
while proc.poll() is None:       # poll 轮询，可以在子进程结束之前干点别的事情
    print('Working ...')
print('Finished!', proc.poll())
>>>
Working ...
Finished! 0

更进一步的，把子进程和父进程解耦，就可以并行/平行运行多个子进程：

def run_sleep(period):
    proc = subprocess.Popen(['sleep', str(period)])
    return proc

start = time()
procs = []
for _ in range(10):
    proc = run_sleep(0.1)
    procs.append(proc)
for proc in procs:
    proc.communicate()
end = time()
print("Used %.3f seconds" % (end-start))

>>> Used 0.134 seconds  # 如果是顺序执行，需要 0.1 * 10 大概 1秒钟

还可以模拟 bash 的管道，让多个子进程的输入输出串在一起：

def run_openssl(data):
    env = os.environ.copy()
    env['password'] = b'\xe24U\n\xd0Ql3S\x11'
    proc = subprocess.Popen(
        ['openssl', 'enc', '-des3', '-pass', 'env:password'],
        env=env,
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE
    )
    proc.stdin.write(data)
    proc.stdin.flush()
    return proc

def run_md5(input_stdin):
    proc = subprocess.Popen(
        ['md5'],
        stdin=input_stdin,
        stdout=subprocess.PIPE)
    return proc


input_procs = []
hash_procs = []
for _ in range(3):
    data = os.urandom(10)
    proc = run_openssl(data)
    input_procs.append(proc)
    hash_proc = run_md5(proc.stdout)  # run_md5 的输入来自 run_openssl 的输出
    hash_procs.append(hash_proc)

for proc in input_procs:
    proc.communicate()
for proc in hash_procs:
    out, err = proc.communicate()
    print(out.strip())
>>>
b'adbbfc45ed870a3bc514eb144bd45b2e'
b'0808820193de0c5d9c4f2c6081c2e6bd'
b'6cac34413703849782144a6ecf20e4f4'

如果担心子进程无法在预计时间内完成，可以加上timeout参数（Python3.3 之后版本才可以！）

proc = run_sleep(10)
try:
    proc.communicate(timeout=0.1)
except subprocess.TimeoutExpired:
    proc.terminate()
    proc.wait()

print('Exit status', proc.poll())
>>> Exit status 0     # 并不是书里的 -15  ???

take away

  • 用 subprocess 模块处理子进程及其输入输出流
  • 子进程和父进程（Python 解释器）平行运行，可以提高 CPU 使用率
  • communicate(timeout=0.1) 加上超时限制避免死锁和挂住的子进程

37. 对于阻塞 I/O 任务，要用线程，不要用并行 > 对于I/O密集的任务，比如频繁读取文件，可以利用读取文件的时间做一些别的事情，可以使用线程。

先串行试试：

# 分解因数比较慢，用来做示例
def factorize(number):
    for i in range(1, number + 1):
        if number % i == 0:
            yield i

numbers = [2139079, 1214759, 1516637, 1852285]
start = time()
for number in numbers:
    list(factorize(number))
print("Took %.3f seconds" % (time() - start))
>>> 
Took 0.474 seconds

再用线程试试：

from threading import Thread

class FactorizeThread(Thread):   # 定义一个继承自 Thread 的类，来实现多线程
    def __init__(self, number):
        super().__init__()
        self.number = number
    def run(self):
        self.factors = list(factorize(self.number))

start = time()
threads = []
for number in numbers:
    thread = FactorizeThread(number)
    thread.start()
    threads.append(thread)
    
for thread in threads:
    thread.join()
print("Took %.3f seconds" % (time() - start))
>>>
Took 0.481 seconds      # 比顺序执行还慢了！

Thread 类无法利用多核CPU的性能。

换一个 I/O密集的函数（select 系统调用）：

# 顺序
import select
def slow_systemcall():
    select.select([], [], [], 0.1)
    
start = time()
for _ in range(5):
    slow_systemcall()
print("Took %.3f seconds" % (time() - start))
>>>
Took 0.523 seconds

# 多线程

start = time()
threads = []
for _ in range(5):
    thread = Thread(target=slow_systemcall)
    thread.start()
    threads.append(thread)

def other_thing_to_do(index):
    print('Working on %s' % index)
    
for i in range(5):
    other_thing_to_do(i)
for thread in threads:
    thread.join()

print("Took %.3f seconds" % (time() - start))
>>>
Took 0.105 seconds          # 快了 5 倍！

System call 不受 GIL 影响，可以真正平行处理。

对于阻塞 I/O 型任务，还可以使用 asyncio 模块，但是这里的 Thread 方法对代码的改动最小。

take away

  • 由于 GIL 的限制，Python 线程不能平行运行字节码
  • 即使不能平行运行（不能充分利用多核 CPU），线程仍然可以让程序看起来在同时做很多事情
  • System call 不受 GIL 影响

38. 用锁来避免资源竞争

GIL 并不能保证多个线程访问同一数据结构时，数据是一致的。要保证数据一致，请使用 Thread.Lock

take away

  • GIL 不能保证多线程访问的同一数据是一致的
  • 如果没有锁，多个线程同时修改同一对象，可能发生混乱，造成错误数据
  • from threading import Lock

39. 用 Queue 在多线程直接协调，通信

故事：想要从相机里读取照片，调整大小，然后上传到网络相册，这个 pipeline 如何实现？

自己实现的Queue可能存在各种问题，直接 from queue import Queue

take away

  • 用 Queue 来管理有顺序的工作流
  • 自己构建 Queue 要注意许多问题
  • 直接使用 from queue import Queue: 阻塞操作，buffer 大小，join

40. 协程

线程的问题：

  • 额外的代码保证数据的一致性（#39）
  • 每个线程大约占 8Mb 内存
  • 线程启动开销较

协程是生成器的一种拓展用法。启用协程的开销更调用函数一样。启动之后每个协程只占用 1KB 内存。

协程和生成器的区别：

generators are data producers coroutines are data consumers

协程的工作原理：

让生成器的消费者，在每次 yield 表达式之后，给生成器发送（send）回一个值。生成器收到值，将其作为 yield 表达式的结果。

def minimize():
    current = yield
    while True:
        value = yield current
        current = min(value, current)

it = minimize()
# 开始， next 推进到 yield 这里
# 不使用 next(it) 会报 TypeError: can't send non-None value to a just-started generator
next(it)
it.send(10)
>>> 10
it.send(100)
>>> 10
it.send(-1)
>>> -1

# 结束
it.close()

关于协程的参考资料

take away

  • 利用协程可以同时高效运行上万个函数
  • 生成器内，yield 表达式的值就是通过 send 传入的值
  • 协程可以隔离 程序核心逻辑 和 程序的对外交互代码
  • Python 2 不支持 yield from

41. concurrent.futures

用 from concurrent.futures import ProcessPoolExecutor（低层是 multiprocessing）实现多进程，加速 CPU 密集型任务。

multiprocessing 开销较大，主进程和子进程之间要进行序列化-反序列化操作。所以，更适合较为*孤立*（待运行的函数不需要和程序的其他部分共享状态）而且*数据利用率高*（主进程和子进程之间传递的数据较少）的任务。

take away

  • 用 C 改写 Python 模块可以提高速度，但是容易引入 bug
  • multiprocessing 模块最好通过 from concurrent.futures import ProcessPoolExecutor 使用
  • multiprocessing 提供的其他高级工具最好不要直接使用（非常复杂）

6. 内置模块

42. 用 functools.wraps 定义修饰器

不用 functools.wraps 的版本：

def trace(func):
    def wrapper(*args, **kwargs):
        result = func(*args, **kwargs)
        print('%s(%r, %r) => %r' % (func.__name__, args, kwargs, result))
        return result
    return wrapper

@trace
def fibonacci(n):
    if n in (0, 1):
        return n
    return (fibonacci(n - 2) + fibonacci(n - 1))

fibonacci
>>> 
<function __main__.trace.<locals>.wrapper(*args, **kwargs)> # 函数签名变成上面定义的 trace 里的 wrapper 了

使用 from functools import wraps 可以轻松的使用装饰器同时不破坏原有函数的内部变量。

def trace(func):
    @wraps(func)               # 只需要加这一行即可
    def wrapper(*args, **kwargs):
        result = func(*args, **kwargs)
        print('%s(%r, %r) => %r' % (func.__name__, args, kwargs, result))
        return result
    return wrapper

43. contextlib 和 with 表达式实现可复用的 try/finally (行为)

上面介绍 thread 资源竞争的时候用过 Lock：

from threading import Lock
lock = Lock()
with lock:
    print("lock is held")

# 上面的代码相当于：
lock.acquire()
try:
    print('lock is held')
finally:
    lock.release()

显然用 with 的方法更好，不用写 try/finally 结构。

要定义一个可以这样用的类，需要实现两个方法：__enter__ 和 __exit__，可以用 contextmanager 简化这个过程。

# 默认情况下 logging level 是 WARNING，所以下面的 debug 信息不会输出
def my_function():
    logging.debug('Some debug data')
    logging.error('Error log here')
    logging.debug('More debug data')


# 定义一个可以临时调整 logging level 的上下文
from contextlib import contextmanager
@contextmanager
def debug_logging(level):
    logger = logging.getLogger()
    old_level = logger.getEffectiveLevel()
    logger.setLevel(level)
    try:
        yield
    finally:
        logger.setLevel(old_level)

with debug_logging(logging.DEBUG):
    print("===Inside context===")
    my_function()
print("===After===")
my_function()

>>>
===Inside context===
DEBUG:root:Some debug data
ERROR:root:Error log
DEBUG:root:Extra debug data
===After===
ERROR:root:Error log

with 和 as

传给 with 的上下文管理可以返回一个对象，通过 as 可以把这个对象赋给只在 with 中生效的局部变量。 比如：

with open('xxx') as file:
    file.read()

只需要让上下文管理器中的 yield 返回一个值就行

@contextmanager
def log_level(level, name):
    logger = logging.getLogger(name)
    old_level = logger.getEffectiveLevel()
    logger.setLevel(level)
    try:
        yield logger
    finally:
        logger.setLevel(old_level)

with log_level(logging.DEBUG, 'awesome-log') as AWESOME_LOGGER:
    AWESOME_LOGGER.debug("my debug msg")   # 在上下文管理器环境内部，生成了一个 logger，级别是 debug
    logging.debug("this msg will not show") # 默认的 logger 级别是 warning，所以这个 msg 不会show

logger = logging.getLogger("awesome-log")
logger.debug("this msg will not show as well")
>>>

DEBUG:awesome-log:my debug msg
Outside

take away

  • with 表达式可以实现可复用的 try/finally 逻辑
  • contextlib.contextmanager 可以方便地实现上下文管理器
  • with 和 as

44. copyreg 让 pickle 更靠谱

只在可以信任的程序直接使用 pickle 传递数据。pickle 产生的二进制序列化数据是不安全的格式。pickle 产生的序列化数据实际上就是一个程序，描述了如何构建原始的 Python 对象。

使用 pickle 生成的序列化数据，如果原有的类的名称/路径改变之后，就无法进行反序列化了。因为 pickle 在序列化的时候直接把被序列化的对象所在的类的引入路径写在了输出结果里…

45. 用 datetime 来处理本地时间，不要用 time （时区问题）

46. 内置算法和数据结构

  • 双向队列/优先级队列
  • 有序字典
  • 默认值字典
  • 堆队列
  • 二分查找
  • itertools

47. 用 decimal 保证浮点数精度

take away

  • fractions.Fraction 可以用分数表示

48. 使用 pypi

7. 团队协作

49. docstring

50. 使用包 Package 来管理模块，提供稳定的 API 接口

51. 自定义root级别的异常类

52. 解决循环引用的问题

方法：

  • 调整 import 顺序（不推荐）
  • 先引入、再配置、最后运行
  • 动态引入（在 def，class 里 import，不推荐）

53. 虚拟环境 pyvenv

python2 用 virtualenv

8. 生产环境

54. 部署环境，比如开发环境和生产环境

55. repr 输出调试信息

56. unittest

57. pdb

58. cProfile

take away

  • 用 cProfile 不要用 profile
  • stats.print_callers() 可以打印出调用方的信息

59. tracemalloc 跟踪内存泄漏

take away

  • python2 需要使用第三方库， 比如 heapy