• 上下文管理器与with关键字为业务增加辅助功能


    上下文管理器与with关键字为业务增加辅助功能

    1.概述

    上下文管理器与with关键字需要配合起来使用,他们的使用场景是为业务执行前后增加额外功能
    比如执行业务前添加辅助功能,执行业务后添加额外功能。下面通过实际的例子展示它在业务中的用法。

    2.上下文管理器与with关键字

    with是一个神奇的关键字,他可以在代码中开辟一段由他管理的上下文,并控制程序在进入和退出时的行为。并非所有的对象都能和with配合使用,只有满足上下文管理器协议的对象才行。
    上下文管理器:是一种定义了“进入”和“退出”动作的特殊对象,要创建一个上下文管理器,只要实现__enter__ 和 __exit__两个魔法方法即可。

    2.1.创建一个上下文管理器

    class DummyContext:
        def __init__(self, name):
            self.name = name
    
        def __enter__(self):
            # __enter__会在进入管理器时自动被调用,同时返回结果
            return f'{self.name} - {random.random()}'
    
        def __exit__(self, exc_type, exc_val, exc_tb):
            # __exit__会在退出管理器时被调用
            print('Exiting DummyContext')
            return False
    
    with DummyContext('foo') as name:
        print(f'name: {name}')
    
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16

    2.2.上下文管理器关闭资源

    下面操作文件就是普通的写法,每次都需要在finally里关闭资源。

    conn = create_conn(host, port, timeout=None)
    try:
    	conn.send_text('Hello, world')
    except Exception as e:
    	print(f'Unable to use connection:{e}')
    finally:
    	conn.close()
    
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8

    使用上下文管理器后,就不需要每次都写一个关闭资源的句子,而是交给上下文来处理。

    # 创建上下文管理器
    class create_conn_obj:
        def __init__(self, host, port, timeout=None):
            self.conn = create_conn(host, port, timeout=timeout)
    
        def __enter__(self):
            return self.conn
        
        def __exit__(self, exc_type, exc_val, exc_tb):
            self.conn.close()
            return False
    
    # 使用上下文管理器
    with create_conn_obj(host, port, timeout=None) as conn:
        try:
            conn.send_text('Hellod world')
        except Exception as e:
            print(f'Unable to use connection:{e}')
    
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19

    2.3.with语句中as的作用

    as后面的名称是enter魔法方法返回的对象,下面通过一个例子说明它的用法。
    在例子中创建了两个类Context和WithinContext,在Context类中的enter魔法方法中调用了WithinContext,因此返回的对象就是WithinContext类对象,不再是Context对象。
    当使用with打开的是Context类对象,as接收的是WithinContext对象,通过该对象调用WithinContext类中的方法。

    class WithinContext:
    
        def __init__(self, context):
            print('WithinContext.__init__({})'.format(context))
    
        def do_something(self):
            print('WithinContext.do_something()')
    
        def __del__(self):
            print('WithinContext.__del__')
    
    
    class Context:
    
        def __init__(self):
            print('Context.__init__()')
    
        def __enter__(self):
            print('Context.__enter__()')
            return WithinContext(self)
    
        def __exit__(self, exc_type, exc_val, exc_tb):
            print('Context.__exit__()')
    
    
    with Context() as c:
        c.do_something()
    
    • 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

    运行上面的代码,c是enter返回的WithinContext对象,所以可以调用类中的方法。

    Context.__init__()
    Context.__enter__()
    WithinContext.__init__(<__main__.Context object at 0x109a2ffd0>)
    WithinContext.do_something()
    Context.__exit__()
    WithinContext.__del__
    
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7

    2.4.exit方法处理异常信息

    exit方法接收一些参数,其中包含with块中产生的所有异常信息。如果上下文可以处理这个异常,那么exit应当返回一个true指示这个异常不需要传播。如果返回false,则会在exit返回后再次抛出这个异常。
    exit方法三个参数介绍

    • exc_type:异常的类型
    • exc_val:异常对象
    • exc_tb:错误的堆栈对象
    class Context:
    
        def __init__(self, handle_error):
            print('__init__({})'.format(handle_error))
            self.handle_error = handle_error
    
        def __enter__(self):
            print('__enter__()')
            return self
    
        def __exit__(self, exc_type, exc_val, exc_tb):
            print('__exit__()')
            print('  exc_type =', exc_type)
            print('  exc_val  =', exc_val)
            print('  exc_tb   =', exc_tb)
            return self.handle_error
    
    # true
    with Context(True):
        raise RuntimeError('error message handled')
    
    print()
    
    # false
    with Context(False):
        raise RuntimeError('error message propagated')
    
    • 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

    运行上面代码,输出异常详细信息

    __init__(True)
    __enter__()
    __exit__()
      exc_type = <class 'RuntimeError'>
      exc_val  = error message handled
      exc_tb   = <traceback object at 0x10a8769c0>
    
    __init__(False)
    __enter__()
    __exit__()
      exc_type = <class 'RuntimeError'>
      exc_val  = error message propagated
      exc_tb   = <traceback object at 0x10a876940>
    Traceback (most recent call last):
      File 
        raise RuntimeError('error message propagated')
    RuntimeError: error message propagated
    
    Process finished with exit code 1
    
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20

    2.5.ContextDecorator上下文管理器装饰函数

    ContextDecorator类增加了对常规上下文管理器的支持,因此它既可以当做上下文管理器也可以装饰函数。
    下面的例子创建了子类继承ContextDecorator,装饰函数。并且调用了被装饰的函数,实现了为函数增加辅助功能。

    import contextlib
    
    # 创建上下文管理器继承ContextDecorator
    class Context(contextlib.ContextDecorator):
    
        def __init__(self, how_used):
            self.how_used = how_used
            print('__init__({})'.format(how_used))
    
        def __enter__(self):
            print('__enter__({})'.format(self.how_used))
            return self
    
        def __exit__(self, exc_type, exc_val, exc_tb):
            print('__exit__({})'.format(self.how_used))
    
    # 使用ContextDecorator子类装饰函数
    @Context('as decorator')
    def func(message):
        print(message)
    
    
    print('ContextDecorator当做上下文管理器使用')
    with Context('as context manager'):
        print('Doing work in the context')
    
    print('调用被ContextDecorator装饰的函数')
    func('Doing work in the wrapped function')
    
    • 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

    运行上面的代码,输出装饰函数结果

    __init__(as decorator)
    ContextDecorator当做装饰器使用
    __init__(as context manager)
    __enter__(as context manager)
    Doing work in the context
    __exit__(as context manager)
    调用被ContextDecorator装饰的函数
    __enter__(as decorator)
    Doing work in the wrapped function
    __exit__(as decorator)
    
    Process finished with exit code 0
    
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13

    2.6.使用python内置的上下文管理器

    上面的上下文管理器是我们手动创建的,python还提供了内建的模块contextlib,在该模块中通过contextmanager装饰器可以快速创建一个上下文管理器,下面来介绍下他的使用方法。
    contextmanager返回的上下文管理器派生自ContextDecoretor,所以也可以用作装饰函数。

    假如我们在调试程序,项目设置的日志输出级别是error,需要在执行特定的函数时设置日志级别为waring,出现错误时抛出详细的日志方便调试。
    通常的做法是在这个函数里添加代码设置日志级别为waring,等调试完毕后在删除该代码。如果调试的函数很多就需要做重复的动作,那么我们用上下文管理器来解决这个问题就变得非常的简单。只需要在调试的函数上使用with即可。

    # 导入contextmanager 上下文管理器
    from contextlib import contextmanager
    
    @contextmanager
    def debug_logging(level):
        logger = logging.getLogger()
        old_level = logger.getEffectiveLevel()
        logger.setLevel(level)
        # try里面的代码是执行业务前运行的代码,等效于__enter__
        try:
            yield
        # finally里面的代码是执行业务后运行的代码,等效于__exit__
        finally:
            logger.setLevel(old_level)
    
    
    # 通过with运行上下文管理器,首先运行上下文管理器里try中的代码,
    #然后运行with语句快里面的代码也就是prit和my_function(),
    #最后运行上下文管理器里finally代码。
    with debug_logging(logging.DEBUG):
        print('* Inside:')
        my_function()
    
    print('* After:')
    my_function()
    
    • 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

    2.7.关闭打开的句柄closing

    file类支持上下文管理器API ,因此可以通过上下文管理器关闭file的句柄。但并不是所有的对象都像file类支持上下文管理器。对于不支持的对象,为了能够确保关闭句柄使用contextlib.closing方法为他们创建一个上下文管理器。
    closing方法创建的上下文管理器会在离开时自动调用类中的close方法关闭对象的句柄。

    下面的示例中创建了一个普通的类Door,在类中定义close方法关闭对象的句柄。使用contextlib.closing为Door创建上下文管理器,并在离开时自动调用Door中close方法关闭对象句柄。

    import contextlib
    
    
    class Door:
    
        def __init__(self):
            print('  __init__()')
            self.status = 'open'
        # contextlib.closing会自动调用close方法,因此这个方法名称不能随意修改
        def close(self):
            print('  close()')
            self.status = 'closed'
    
    
    print('Normal Example:')
    with contextlib.closing(Door()) as door:
        print('  进入door类: {}'.format(door.status))
    print('  退出door对象自动调用close方法,关闭句柄: {}'.format(door.status))
    
    print('\n模拟出现异常时也会调用clse关闭句柄:')
    try:
        with contextlib.closing(Door()) as door:
            print('  raising from inside with statement')
            raise RuntimeError('error message')
    except Exception as err:
        print('  Had an error:', err)
    
    • 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

    运行上面的代码,正常离开上下文和遇到异常时都能自动关闭Door对象资源

    Normal Example:
      __init__()
      进入door类: open
      close()
      退出door对象自动调用close方法,关闭句柄: closed
    
    模拟出现异常时也会调用clse关闭句柄:
      __init__()
      raising from inside with statement
      close()
      Had an error: error message
    
    Process finished with exit code 0
    
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14

    2.8.忽略异常

    在开发过程中有些错误可能不需要被处理或抛出,忽略这个异常通常很有用。最常用的方法是利用try:except语句,在except块中只包含一个pass语句。
    这种方法在每个模块中都重复使用既不便于管理,又是冗余的代码。使用上下文管理器忽略异常可以做到统一管理,创建一次多次使用。

    import contextlib
    
    # 忽略异常
    class NonFatalError(Exception):
        pass
    
    # 业务逻辑
    def non_idempotent_operation():
        raise NonFatalError(
            'The operation failed because of existing state'
        )
    
    # 调用non_idempotent_operation函数,并忽略它的异常
    with contextlib.suppress(NonFatalError):
        print('trying non-idempotent operation')
        non_idempotent_operation()
        print('succeeded!')
    
    print('done')
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19

    运行上面的代码在上下文管理器中运行non_idempotent_operation函数,并没有抛出异常。实现了忽略异常功能。

    trying non-idempotent operation
    done
    
    
    • 1
    • 2
    • 3

    2.9.重定向输出流

    当有些库代码输出写死了sys.stdout或sys.stderr,没有提供参数配置不同的输出目标。可以使用redirect_stdout()和redirect_err()上下文管理器从这些函数捕获输出,重定向输出到其他位置。

    注意:redirect_stdout()和redirect_err()会修改全局状态,替换sys模块中的对象,要小心使用这两个函数。

    下面的例子misbehaving_function同时写到sys.stdout和sys.stderr,通过两个上下文管理器将这个输出发送到io.StringIO实例中,以备后续使用。

    from contextlib import redirect_stdout, redirect_stderr
    import io
    import sys
    
    
    def misbehaving_function(a):
        sys.stdout.write('(stdout) A: {!r}\n'.format(a))
        sys.stderr.write('(stderr) A: {!r}\n'.format(a))
    
    
    capture = io.StringIO()
    with redirect_stdout(capture), redirect_stderr(capture):
        misbehaving_function(5)
    
    print(capture.getvalue())
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15

    2.10.动态上下文管理器

    大多数上下文管理器一次处理一个对象,比如单个文件或数据库句柄。在这种情况下,对象都是提前已知的,并且上下文管理器的代码建立在一个对象上。
    另外一些情况下,可能需要在一个上下文中创建未知数目的对象,控制流退出这个上下文管理器所有对象都要清理。ExitStack就是用来处理这些动态的对象。

    ExitStack实例会维护清理回调的一个栈数据结构,这些回调在控制流退出上下文时会以逆序调用所有注册的回调。

    2.10.1.上下文管理器入栈

    下面通过一个例子使用enter_context来为栈增加一个新的上下文管理器,在退出时自动关闭所有对象的句柄。

    import contextlib
    
    
    @contextlib.contextmanager
    def make_context(i):
        print('{} entering'.format(i))
        yield {}
        print('{} exiting'.format(i))
    
    
    def variable_stack(n, msg):
        with contextlib.ExitStack() as stack:
            for i in range(n):
                stack.enter_context(make_context(i))
            print(msg)
    
    
    variable_stack(2, 'inside context')
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18

    运行上面的代码,为多个make_context对象创建入栈,在退出上下文时自动关闭所有make_context对象句柄。

    0 entering
    1 entering
    inside context
    1 exiting
    0 exiting
    
    • 1
    • 2
    • 3
    • 4
    • 5
  • 相关阅读:
    六、04【Java 多线程】之并发编程
    一生一芯13——linux设置环境变量
    django项目实战基于Python实现的飞机票销售系统
    SHA256 安全散列算法加速器实验
    PyG-GAT-Cora(在Cora数据集上应用GAT做节点分类)
    .Net Core_1_
    vue3 全局方法挂载
    Axure原型创建折线、柱状等图形,引用echarts
    元老职员离职申请书怎么写模板,共计10篇
    Mongoose web服务器的编译
  • 原文地址:https://blog.csdn.net/m0_38039437/article/details/128085871