上下文管理器

定义

# 大多数情况下,与装饰器是等价的
# 进入时,保证代码执行的一致性,退出时,保证相关资源会被正确回收
# 一定会有退出, 即使是发生了异常
# 类似于 try...except...finally

语法

# try finally
try:
    file = open('Cp1.ipynb', 'r', encoding='utf8')
    content = file.read()
finally:
    file.close()

# with
with open('Cp1.ipynb', 'r', encoding='utf8') as f:
    content = f.read()

enter 与 exit

# 返回一个遵循特定协议的对象
# 该对象必须定义一个__enter__与__exit__方法
# __enter__只接受self
# 	返回值赋值给as后面的变量
# 且__exit__必须接受特定的参数
# 	self
# 	异常类型
# 	异常实例
# 	回溯

class ContextManager:
    def __init__(self):
        self._entered = False
        
    def __enter__(self):
        self._entered = True
        return self
        
    def __exit__(self, exc_type, exc_instance, traceback):
        self._entered = False
c1 = ContextManager()
print(c1._entered)

with ContextManager() as c2:
    print(c2._entered)

False
True

异常处理

# 传播异常
# 中止异常
# 抛出不同的异常

何时写上下文处理器

资源清理

# 数据库连接
# 例如PyMysql

避免重复

传播异常

# __exit__的内容实现了
# 发给它的异常被重新抛出
class BubbleExc:
    def __enter__(self):
        return self
    def __exit__(self, exc_type, exc_instance, traceback):
        print('in __exit__')
        if exc_instance:
            print(exc_instance)
        return False
with BubbleExc():
    5+5
    
with BubbleExc():
    5 / 0

in __exit__
in __exit__
division by zero

终止异常

# __exit__的内容实现了
# 发给它的异常被终止
class SuppressExc:
    def __enter__(self):
        return self
    def __exit__(self, exc_type, exc_instance, traceback):
        print('in __exit__')
        if exc_instance:
            print(exc_instance)
        return True
with SuppressExc():
    5+5
    
with SuppressExc():
    5 / 0
    print(1)

in __exit__
in __exit__
division by zero



---------------------------------------------------------------------------

ZeroDivisionError                         Traceback (most recent call last)

~\AppData\Local\Temp/ipykernel_13988/2733532279.py in <module>
     13 
     14 with SuppressExc():
---> 15     5 / 0
     16     print(1)


ZeroDivisionError: division by zero

# 在with中try了异常
# 就不回传给上一层__exit__
with SuppressExc():
    try:
        5 / 0
    except ZeroDivisionError:
        print('catch exception by context')
    print(1)

catch exception by context
1
in __exit__

处理特定异常类

# 能处理好就return True
class HandleError:
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_instance, traceback):
        if exc_type is None:
            return True
        if issubclass(exc_type, ValueError):
            print('__exit__ handle a Value Error')
            return True
        return False
    
with HandleError():
    raise ValueError('test')
    
with HandleError():
    raise RuntimeError()

__exit__ handle a Value Error



---------------------------------------------------------------------------

RuntimeError                              Traceback (most recent call last)

~\AppData\Local\Temp/ipykernel_13988/2716100519.py in <module>
     15 
     16 with HandleError():
---> 17     raise RuntimeError()


RuntimeError:

不包括的子类

# 只捕获异常而不捕获它的子类
class ValueErrorSubclass(ValueError):
    pass

class HandleError:

    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_instance, traceback):
        if exc_type is None:
            return True
        if exc_type == ValueError:
            print('__exit__ handle a Value Error')
            return True
        return False
    
with HandleError():
    raise ValueError()
    
with HandleError():
    raise ValueErrorSubclass()

__exit__ handle a Value Error



---------------------------------------------------------------------------

ValueErrorSubclass                        Traceback (most recent call last)

~\AppData\Local\Temp/ipykernel_13988/2102615219.py in <module>
     20 
     21 with HandleError():
---> 22     raise ValueErrorSubclass()


ValueErrorSubclass:

基于属性的异常处理

import subprocess

class ShellException(Exception):
    def __init__(self, code, stdout='', stderr=''):
        self.code = code
        self.stdout = stdout
        self.stderr = stderr
        
    def __str__(self):
        return f'exit code {self.code}- {self.stderr}'
    
    
def run_command(command):
    proc = subprocess.Popen(command.split(), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    proc.wait()
    stdout, stderr = proc.communicate()
    
    if proc.returncode > 0:
        raise ShellException(proc.returncode, stdout, stderr)
    return stdout

run_command('rm 123.txt')

---------------------------------------------------------------------------

ShellException                            Traceback (most recent call last)

~\AppData\Local\Temp/ipykernel_13988/2086617344.py in <module>
     20     return stdout
     21 
---> 22 run_command('rm 123.txt')


~\AppData\Local\Temp/ipykernel_13988/2086617344.py in run_command(command)
     17 
     18     if proc.returncode > 0:
---> 19         raise ShellException(proc.returncode, stdout, stderr)
     20     return stdout
     21 


ShellException: exit code 1- b"rm: cannot remove '123.txt': No such file or directory\n"

更简单的语法

import os
import contextlib

@contextlib.contextmanager
def accept_error_codes(*codes):
    try:
        yield
    except ShellException as exc_instance:
        if exc_instance.code not in codes:
            raise
        print('get code', exc_instance.code)
        pass
    
with accept_error_codes(1):
    run_command('rm 123.txt')