异常处理库merry的用法
Contents
实例
算术运算和文件写入的例子
def process(num1, num2, file):
result = num1 / num2
with open(file, 'w', encoding='utf-8') as f:
f.write(str(result))
定义了一个 process 方法，接收三个参数，前两个参数是 num1 和 num2，第三个参数是 file。方法会首先将 num1 除以 num2，然后把除法的结果写入到 file 文件里面。
这个实现的漏洞：
- 没有判断 num1、num2 的类型，如果不是数字类型，那会抛出 TypeError。
- 没有判断 num2 是不是 0，如果是 0，那么会抛出 ZeroDivisionError。
- 没有判断文件路径是否存在，如果是子文件夹下的路径，文件夹不存在的话，会抛出 FileNotFoundError。 一些异常测试用例如下：
process(1, 2, 'result/result.txt')
process(1, 0, 'result.txt')
process(1, [2], 'result.txt')
用例跑起来一定会报错
最好的方式是通过一些判断条件把一些该处理的问题和判定都加上
def process(num1, num2, file):
try:
result = num1 / num2
with open(file, 'w', encoding='utf-8') as f:
f.write(str(result))
except ZeroDivisionError:
print(f'{num2} can not be zero')
except FileNotFoundError:
print(f'file {file} not found')
except Exception as e:
print(f'exception, {e.args}')
问题:
- 代码一下子臃肿了起来，这里的异常处理都没有实现，仅仅是 print 了一些错误信息，但现在可以看到异常处理代码可能比主逻辑还要多
- 主逻辑代码整块被硬生生地缩进进去了，如果主逻辑代码比较多的话，那么会有大片大片的缩进
- 如果再有相同的逻辑的代码，要再写一次 try except
几种改善的方案：
- 本身这个场景不需要这么多异常处理，使用判断条件把一些意外情况处理掉就好
- 异常处理本身就不应该这么写，每个功能区域应该和异常处理单独分开，另外各个逻辑模块建议再分方法解耦
- 使用 retrying 模块检测异常并进行重试。
- 使用上下文管理器 raise_api_error 来声明异常处理。
- 但上面的一些解决方案其实还不能彻底解决代码复用和美观上的问题，比如某一类的异常处理就统一在一个地方处理，另外任何代码都不想因为异常处理产生缩进。
Merry库
Merry库是 Python 的一个第三方库，非常小巧，实现了几个装饰器。通过使用 Merry 我们可以把异常检查和异常处理的代码分开，并可以通过装饰器来定义异常检查和处理的逻辑。
安装
pip3 install merry
使用
from merry import Merry
merry = Merry()
merry.logger.disabled = True
@merry._try
def process(num1, num2, file):
result = num1 / num2
with open(file, 'w', encoding='utf-8') as f:
f.write(str(result))
@merry._except(ZeroDivisionError)
def process_zero_division_error(e):
print('zero_division_error', e)
@merry._except(FileNotFoundError)
def process_file_not_found_error(e):
print('file_not_found_error', e)
@merry._except(Exception)
def process_exception(e):
print('exception', type(e), e)
if __name__ == '__main__':
process(1, 2, 'result/result.txt')
process(1, 0, 'result.txt')
process(1, 2, 'result.txt')
process(1, [1], 'result.txt')
可以看到，我们首先声明了一个 Merry 对象，然后 process 方法加上 merry 对象的 _try 方法的装饰器，这样就实现了异常的监听。
有了异常监听之后，怎么来进行异常处理呢？还是通过同一个 merry 对象，使用其 _except 方法作为装饰器即可。比如这里我们将几个异常处理方法分开了，如处理 ZeroDivisionError、FileNotFoundError、Exception 等异常分别都有一个方法的声明，分别加上对应的装饰器即可。
这样的话，就轻松实现了异常监听和处理
和上面不同之处：
- 主逻辑里面不用额外加异常处理代码了，显得简洁。
- 主逻辑不用因为 try except 而缩进了。
- 每个异常处理方法单独分开了，可以实现解耦和重用。
- 整个实现就舒服多了。
运行一下上面的代码，输出结果如下：
file_not_found_error [Errno 2] No such file or directory: 'result/result.txt'
zero_division_error division by zero
exception <class 'TypeError'> unsupported operand type(s) for /: 'int' and 'list'
类实现
上面的实现整个看起来结构比较松散，代码不好复用
如果封装好了类之后，可以直接把类拿过来使用，实现的时候继承这个类，就不用再关心各个异常处理是怎么实现的了。
在这里一个简单的实现如下：
import requests
from merry import Merry
from requests import ConnectTimeout
merry = Merry()
merry.logger.disabled = True
catch = merry._try
class BaseClass(object):
@staticmethod
@merry._except(ZeroDivisionError)
def process_zero_division_error(e):
print('zero_division_error', e)
@staticmethod
@merry._except(FileNotFoundError)
def process_file_not_found_error(e):
print('file_not_found_error', e)
@staticmethod
@merry._except(Exception)
def process_exception(e):
print('exception', type(e), e)
@staticmethod
@merry._except(ConnectTimeout)
def process_connect_timeout(e):
print('connect_timeout', e)
class Calculator(BaseClass):
@catch
def process(self, num1, num2, file):
result = num1 / num2
with open(file, 'w', encoding='utf-8') as f:
f.write(str(result))
class Fetcher(BaseClass):
@catch
def process(self, url):
response = requests.get(url, timeout=1)
if response.status_code == 200:
print(response.text)
if __name__ == '__main__':
c = Calculator()
c.process(1, 0, 'result.txt')
f = Fetcher()
f.process('http://notfound.com')
先实现了一个 BaseClass，里面通过 merry 定义了好多个异常处理方法和处理流程，异常处理方法定义成 staticmethod。
接着定义了两个类，一个是 Calculator，一个是 Fetcher，分别完成不同的功能，一个是计算，一个是抓取网页。另外 Merry 的 _try 方法我们给它取了个别名，比如叫做 catch，显得更简洁。
在 process 方法中，我们我们想要进行异常处理，那么就加上 @catch 这装饰器就好了，其他的不需要管。
这样，在实现子类的时候，只需要集成 BaseClass 然后实现对应的方法就好了，如果想加上异常处理就加个装饰器，也无需再关心 Merry 的具体实现，父类都实现好了.
运行结果如下：
zero_division_error division by zero
connect_timeout HTTPConnectionPool(host='notfound.com', port=80): Max retries exceeded with url: / (Caused by ConnectTimeoutError(<urllib3.connection.HTTPConnection object at 0x10d5b9310>, 'Connection to notfound.com timed out. (connect timeout=1)'))
数据传递
利用 try except 可以直接获取异常代码的变量信息，分了方法之后，一些上下文的变量不就拿不到了
Merry 是用了一个全局的变量来解决的，它使用了 merry.g 这个对象来存储上下文的变量，在 主逻辑方法里面要把想要传递的参数赋值进去，在异常处理的方法里面再用 merry.g 取出来，官方示例写法如下：
@merry._try
def app_logic():
db = open_database()
merry.g.database = db # save it in the error context just in case
# do database stuff here
@merry._except(Exception)
def catch_all():
db = getattr(merry.g, 'database', None)
if db is not None and is_database_open(db):
close_database(db)
print('Unexpected error, quitting')
sys.exit(1)
源码剖析
from functools import wraps
import inspect
import logging
getargspec = None
if getattr(inspect, 'getfullargspec', None):
getargspec = inspect.getfullargspec
else:
# this one is deprecated in Python 3, but available in Python 2
getargspec = inspect.getargspec
class _Namespace:
pass
class Merry(object):
def __init__(self, logger_name='merry', debug=False):
self.logger = logging.getLogger(logger_name)
self.g = _Namespace()
self.debug = debug
self.except_ = {}
self.force_debug = []
self.force_handle = []
self.else_ = None
self.finally_ = None
def _try(self, f):
@wraps(f)
def wrapper(*args, **kwargs):
ret = None
try:
ret = f(*args, **kwargs)
# note that if the function returned something, the else clause
# will be skipped. This is a similar behavior to a normal
# try/except/else block.
if ret is not None:
return ret
except Exception as e:
# find the best handler for this exception
handler = None
for c in self.except_.keys():
if isinstance(e, c):
if handler is None or issubclass(c, handler):
handler = c
# if we don't have any handler, we let the exception bubble up
if handler is None:
raise e
# log exception
self.logger.exception('[merry] Exception caught')
# if in debug mode, then bubble up to let a debugger handle
debug = self.debug
if handler in self.force_debug:
debug = True
elif handler in self.force_handle:
debug = False
if debug:
raise e
# invoke handler
if len(getargspec(self.except_[handler])[0]) == 0:
return self.except_[handler]()
else:
return self.except_[handler](e)
else:
# if we have an else handler, call it now
if self.else_ is not None:
return self.else_()
finally:
# if we have a finally handler, call it now
if self.finally_ is not None:
alt_ret = self.finally_()
if alt_ret is not None:
ret = alt_ret
return ret
return wrapper
def _except(self, *args, **kwargs):
def decorator(f):
for e in args:
self.except_[e] = f
d = kwargs.get('debug', None)
if d:
self.force_debug.append(e)
elif d is not None:
self.force_handle.append(e)
return f
return decorator
def _else(self, f):
self.else_ = f
return f
def _finally(self, f):
self.finally_ = f
return f
主要的逻辑都在 _try 方法里面了，就是仿照这标准的 try、except、else、finally 方法把流程实现下来了，只不过在对应的逻辑区块里面调用了装饰器修饰的方法。
其中有一个地方比较有意思：
handler = None
for c in self.except_.keys():
if isinstance(e, c):
if handler is None or issubclass(c, handler):
handler = c
这里是为异常处理找寻一个最佳的异常处理方法，可以看到这里通过各个 _except 修饰的方法，然后通过 issubclass 方法来找寻最小能处理的子类，最终找到最佳匹配方法.