合理利用 try/catch/else/finally

在 Python 代码中处理异常需要合理利用 try/except/else/finally 这个结构中的四个代码块。

try = 正常的代码逻辑
except = 报错后的处理
else = try 运行完没有报错的处理
finally = 最后一定要执行的逻辑

考虑用 contextlib 和 with 语句来改写可复用的 try/finally 代码

使用 with 来简化代码

Python 里的 with 语句可以用来强调某段代码需要在特殊情境之中执行。例如,如果必须先持有互斥锁,然后才能运行某段代码。

使用 try-finally 实现该逻辑:

from threading import Lock
 
lock = Lock()
lock.require()
try:
	# do something
finally:
	lock.release()

而使用 with 语句的实现方式要简洁的多:

from threading import Lock
 
lock = Lock()
with lock:
	# do something
	pass

将自定义对象和函数放在 with 语句内

如果想让其他的对象跟函数,也能像 Lock 这样用在 with 语句里面,那么可以通过内置的 contextlib 模块来实现。这个模块提供了 contextmanager 修饰器,它可以使没有经过特别处理的普通函数也能受到 with 语句支持。这要比标准做法简单得多,因为那种做法必须定义新类并实现名为 __enter____exit__ 的特殊方法。

假如,当前有一个日志打印程序如下:

import logging
 
def my_function():
	logging.info("info")
	logging.debug("debug")
	logging.error("error")

现在想临时改变日志级别,可以用 @contextmanager 来修饰一个辅助函数,来实现这种需求。将该辅助函数用在 with 语句时,with 语句内的代码的日志输出级别就被临时的修改了。待 with 语句块执行完毕后,再恢复原有日志级别。

系统开始执行 with 语句时,会先把 @contextmanager 所修饰的辅助函数推进到 yield 表达式所在的地方,然后开始执行 with 结构的主体部分。如果执行 with 语句块的过程中发生异常,那么这个异常会重新抛出到 yield 表达式所在的那一行里,从而为辅助函数中的 try 结构所捕获。

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:')
    my_function()
	# INFO:root:info
	# DEBUG:root:debug
	# ERROR:root:error
 
print('* After:')
my_function()
# ERROR:root:error

带目标的 with 语句

with 语句还有一种写法,叫作 with…as…,它可以把情境管理器所返回的对象赋给 as 右侧的局部变量,这样的话,with 结构的主体部分代码就可以通过这个局部变量与情境管理器所针对的那套情境交互了。

如果要使得自定义函数也支持这种功能,只需要在 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, 'my-log') as logger:
    logger.debug(f'This is a message for {logger.name}!')
    logging.debug('This will not print')

用 datetime 模块优雅的处理时区

把当前的 UTC 时间转换成本地时间。

from datetime import datetime, timezone
 
now = datetime(2019, 3, 16, 22, 14, 35)
now_utc = now.replace(tzinfo=timezone.utc)
now_local = now_utc.astimezone()
print(now_local)     # 2019-03-17 06:14:35+08:00

本地时间转换成 UTC 格式的 UNIX 时间戳。

from datetime import datetime, timezone
 
time_str = '2019-03-16 15:14:35'
time_format = '%Y-%m-%d %H:%M:%S'
now = datetime.strptime(time_str, time_format)
time_tuple = now.timetuple()
utc_now = time.mktime(time_tuple)
print(utc_now)      # 1552720475.0

datetime 模块里面有相应的机制,可以把一个时区的本地时间可靠地转化成另一个时区的本地时间。但问题是,datetime 的这套时区操纵机制必须通过 tzinfo 类与相关的方法来运作,而系统在安装 Python 的时候,并不会默认安装 UTC 之外的时区定义信息。好在其他 Python 开发者提供了 pytz 模块,能够把这些缺失的时区定义信息给补上。

例如下面的代码将纽约时间转换为 UTC 时间,然后再转换为太平洋时间。

from datetime import datetime
import pytz
 
# 创建一个时区对象代表纽约时区
new_york_tz = pytz.timezone('America/New_York')
# 给定的纽约时间
ny_time_str = '2019-03-16 23:33:24'
# 格式化时间字符串并设置正确的时区
ny_time = datetime.strptime(ny_time_str, '%Y-%m-%d %H:%M:%S')
ny_time = new_york_tz.localize(ny_time)
# 转换为UTC时间
utc_time = ny_time.astimezone(pytz.utc)
print(utc_time)
 
 
# 再将 UTC 时间转换为其他时间
pacific_tz = pytz.timezone('US/Pacific')
sf_dt = pacific.normalize(utc_dt.astimezone(pacific_tz))

当 pickle 反序列化遇到原对象变化

pickle 可以进行将对象序列化和反序列化,但是如果被序列化的对象类型发生了变化,可能会遇到意想不到的问题。

Python 类属性增减

如下是最初的 GameState 类,以及对该类的实例对象进行序列化。

class GameState:
    def __init__(self):
        self.level = 0
        self.lives = 4
 
state = GameState()
state.level += 1  # Player beat a level
state.lives -= 1  # Player had to try again
print(state.__dict__)  # {'level': 1, 'lives: 3}
 
import pickle
state_path = 'game_state.bin'
with open(state_path, 'wb') as f:
    pickle.dump(state, f)

倘若反序列时,GameState 类与原来的不同,那么就会出现意外情况。例如下面在反序列化后,state_after 对象就缺失了 points 成员变量。

class GameState:
    def __init__(self):
        self.level = 0
        self.lives = 4
        self.points = 0  # New field
 
with open(state_path, 'rb') as f:
    state_after = pickle.load(f)
 
print(state_after.__dict__)  # {'level': 1, 'lives: 3}

解决的办法就是使用 Python 内置的 copyreg 模块来辅助 pickle 进行序列化和反序列化。

方法一:对于上面的代码,最简单的办法是定义带有默认参数的构造函数。

class GameState:
    def __init__(self, level=0, lives=4, points=0):
        self.level = level
        self.lives = lives
        self.points = points

同时,还需要定义辅助函数来进行序列化和反序列化。

首先定义一个序列化函数 pickle_game_state,该函数接受要被序列化的对象作为参数,并返回一个元组。元组的第一个元素是负责反序列化用的函数,第二个元素本身也是一个元组,代表反序列化函数所接受的参数。由于此处 unpickle_game_state 函数只接受一个参数,因此第二个元素所代表的元组也只有一个元素。

def pickle_game_state(game_state):
    kwargs = game_state.__dict__
    return unpickle_game_state, (kwargs,)
 
def unpickle_game_state(kwargs):
    return GameState(**kwargs)

然后使用内置的 copyreg 模块来注册序列化函数。

import copyreg
 
copyreg.pickle(GameState, pickle_game_state)

假如注册后修改了 GameState 类,但是仅仅是增加了新属性并且存在参数默认值,那么仍然可以还原回去。

class GameState:
    def __init__(self, level=0, lives=4, points=0, magic=5):
        self.level = level
        self.lives = lives
        self.points = points
        self.magic = magic  # New field
 
state_after = pickle.loads(serialized)
print('After: ', state_after.__dict__)
# After:{'level':1, 1ives':3, points': 0, 'magic':5}

方法二:用版本号标注同一个类的不同定义。

有时候做的改动是无法向后兼容的,比如将 GameState 中的某个字段删除掉,那么方法一就失效了。

此时可以通过版本号机制,根据不同的版本号来对序列化和反序列化过程进行处理。

class GameState:
    def __init__(self, level=0, points=0, magic=5):
        self.level = level
        self.points = points
        self.magic = magic
 
def pickle_game_state(game_state):
    kwargs = game_state.__dict__
    kwargs['version'] = 2
    return unpickle_game_state, (kwargs,)
 
def unpickle_game_state(kwargs):
    version = kwargs.pop('version', 1)
    if version == 1:
        del kwargs['lives']
    return GameState(**kwargs)
 
copyreg.pickle(GameState, pickle_game_state)
state_after = pickle.loads(serialized)

正确处理类名的变化

使用 pickle 模块的时候,还会碰到一个问题,就是类名会发生变化。在开发程序的过程中,我们需要重构代码,这时我们可能会把类的名称改掉,或者把它放到别的模块里面。

例如,把 GameState 类的名称改为 BetterGameState,并把原来那个 GameState 的代码从程序里面彻底删掉。那么此时正确的做法仍然是借助于 copyreg,在调用 pickle 函数时,将新的类传入进去,由反序列化函数来决定应该欢迎成哪个类对象。

如果先用 copyreg 注册,然后做序列化,那么写到数据里面的引入路径就不是 BetterGameState 这样的类名了,而是 unpickle_game_state 这样的 unpickle 函数名。

class BetterGameState:
    def __init__(self, level=0, points=0, magic=5):
        self.level = level
        self.points = points
        self.magic = magic
 
copyreg.pickle(BetterGameState, pickle_game_state)
 
state = BetterGameState()
serialized = pickle.dumps(state)
print(serialized)

用 Decimal 甚至 Fraction 表示准确的小数

因为浮点数必须表示成 IEEE 754 格式,所以采用浮点数算出的结果可能跟实际结果稍有偏差。

此时若想要准确的小数,则应该用 Python 内置的 decimal 模块所提供的 Decimal 类来做。

rate = 1.45
seconds = 3*60 + 42
cost = rate * seconds / 60
print(cost) # 5.3649999999999
 
 
from decimal import Decimal
rate = Decimal('1.45')
seconds = Decimal(3*60 + 42)
cost = rate * seconds / Decimal(60)
print(cost)        # 5.365

Decimal 的初始值可以用两种办法来指定。第一种,是把含有数值的 str 字符串传给 Decimal 的构造函数,这样做不会让字符串里面的数值由于 Python 本身的浮点数机制而出现偏差。第二种,是直接把 float 或 int 实例传给构造函数。但这两种方式其实是存在结果差异:

print(Decimal('1.45'))    # 1.45
print(Decimal(1.45))    # 1.44999999

Decimal 类提供了 quantize 函数,可以根据指定的舍入方式把数值调整到某一位。首先用前面那个数值比较大的例子试试看,我们把计费结果向上调整(round up)到小数点后第二位。

from decimal import ROUND_UP
 
rounded = cost.quantize(Decimal('0.01'), rounding=ROUND_UP)
print(f'Rounded {cost} to {rounded}')  # Rounded 5.365 to 5.37
 
rounded = small_cost.quantize(Decimal('0.01'), rounding=ROUND_UP)
print(f'Rounded {small_cost} to {rounded}')  # Rounded 0.0041666667 to 0.01

Decimal 可以很好地应对小数点后面数位有限的值(也就是定点数,fixed point number)。但对于小数位无限的值(例如 1/3)来说,依然会有误差。如果想表示精度不受限的有理数,那么可以考虑使用内置的 fractions 模块里面的 Fraction 类。