Python 中异常处理库 merry 的用法

扫码或搜索： 进击的Coder

发送

即可 立即永久 解锁本站全部文章

写程序时，异常处理是在所难免的，但你有没有考虑过怎样让异常处理机制变得扩展性更好，让写法更优雅呢？

实例引入

比如写 Python 的时候，举个最简单的算术运算和文件写入的例子，代码如下：

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')

这些用例跑起来一定是会报错的。如果面试写出来这个代码，肯定就挂了。

当然最好的方式是通过一些判断条件把一些该处理的问题和判定都加上。

但这里我们为了说异常处理，如果把这几类的错误都进行异常处理的话，会写成什么样子呢？

由于 Python 的语法是有缩进的，所以我们可能会首先将这些代码缩进四个空格，然后外面包上一个 try 和 except，可能写成这个样子：

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

Merry，这个库是 Python 的一个第三方库，非常小巧，它实现了几个装饰器。通过使用 Merry 我们可以把异常检查和异常处理的代码分开，并可以通过装饰器来定义异常检查和处理的逻辑。

GitHub 地址： https://github.com/miguelgrinberg/merry，这个库的安装非常简单，使用 pip3 安装即可：

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)

但我个人觉得这个写法很鸡肋，我个人觉得应该能获取主逻辑方法里面的 context 对象，context 里面包含了主逻辑方法里面的变量状态，然后异常处理的方法的某个参数可以接收到这个 context 对象，就能直接获取变量值了。

不过这只是个个人想法，还没有实现，有兴趣的朋友可以试试。

源码剖析

最后，我们来看看它的源码吧，其实源码就这么多：

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 方法来找寻最小能处理的子类，最终找到最佳匹配方法，实现有点妙。

好了，本节整体介绍了 Merry 这个库的基本使用，利用它我们可以使得 Python 的异常处理变得更优雅可扩展，来试试吧。

更多的用法可以看官方的说明 https://github.com/miguelgrinberg/merry 或者源码，谢谢。

希望对大家有帮助。