Python 中 with 语句的学习

hresh 565 0

Python 中 with 语句的学习

with 语句适用于对资源进行访问的场合,确保不管使用过程中是否发生异常都会执行必要的“清理”操作,释放资源,比如文件使用后自动关闭、线程中锁的自动获取和释放等。

一个很好的例子是文件处理,你需要获取一个文件句柄,从文件中读取数据,然后关闭文件句柄。
如果不用 with 语句,代码如下:

fr = open('anc.txt', 'r')
data = fr.read()
fr.close()

这里有两个问题。一是可能忘记关闭文件句柄;二是文件读取数据发生异常,没有进行任何处理。下面是处理异常的加强版本:

fr = open('anc.txt', 'r')
try:
    data = fr.read()
finally:
    fr.close()

虽然这段代码运行良好,但是太冗长了。除了有更优雅的语法,with 还可以很好的处理上下文环境产生的异常。下面是 with 版本的代码:

with open('anc.txt','r') as fr:
    data = fr.read()

with 如何工作

一段基本的 with 表达式,其结构是这样的:

with EXPR as VAR:
    BLOCK

其中:EXPR 可以是任意表达式;as VAR 是可选的。其一般的执行过程是这样的:

  1. 计算 EXPR,并获取一个上下文管理器。
  2. 上下文管理器的 __exit()__ 方法被保存起来用于之后的调用。
  3. 调用上下文管理器的 __enter()__ 方法。
  4. 如果 with 表达式包含 as VAR,那么 EXPR 的返回值被赋值给 VAR。
  5. 执行 BLOCK 中的表达式。
  6. 调用上下文管理器的 __exit()__ 方法。如果 BLOCK 的执行过程中发生了一个异常导致程序退出,那么异常的 type、value 和 traceback (即 sys.exc_info() 的返回值)将作为参数传递给 __exit()__ 方法。否则,将传递三个 None。

将这个过程用代码表示,是这样的:

mgr = (EXPR)
exit = type(mgr).__exit__ # 这里没有执行
value = type(mgr).__enter__(mgr)
exc = True

try:
    try:
        VAR = value # 如果有 as VAR
        BLOCK
    except:
        exc = False
        if not exit(mgr, *sys.exc_info()):
            raise
finally:
    if exc:
        exit(mgr, None, None, None)

这个过程有几个细节:

  • 如果上下文管理器中没有 __enter()__ 或者 __exit()__ 中的任意一个方法,那么解释器会抛出一个 AttributeError。
  • 在 BLOCK 中发生异常后,如果 __exit()__ 方法返回一个可被看成是True的值,那么这个异常就不会被抛出,后面的代码会继续执行。

实例分析

自定义对象,构造 __enter()____exit()__方法。

class Test():
    def __enter__(self):
        print('__enter__() is call!')
        return self

    def dosomething(self):
        x = 1/0#异常
        print('dosomething!')

    def __exit__(self, exc_type, exc_val, exc_tb):
        print('__exit__ call!')
        print(f'type:{exc_type}')
        print(f'value:{exc_val}')
        print(f'trace:{exc_tb}')
        print('__exit__ end!')
        # return True

with Test() as sample:
    sample.dosomething()

此时输出结果为:

__enter__() is call!
Traceback (most recent call last):
__exit__ call!
  File "E:/PycharmWspace/pythonTip_challenge/interview_ques_study/with_use_scenarios.py", line 45, in <module>
type:<class 'ZeroDivisionError'>
    sample.dosomething()
value:division by zero
  File "E:/PycharmWspace/pythonTip_challenge/interview_ques_study/with_use_scenarios.py", line 29, in dosomething
trace:<traceback object at 0x0000024307F20508>
    x = 1/0
__exit__ end!
ZeroDivisionError: division by zero

如果将 __exit()__ 方法中返回值置为 True,后续代码会正常执行。此时结果为:

__enter__() is call!
__exit__ call!
type:<class 'ZeroDivisionError'>
value:division by zero
trace:<traceback object at 0x0000027C9BD71288>
__exit__ end!

另一个实例就是线程中锁的自动获取和释放。

import threading

num = 0 #全局变量多个线程可以读写,传递数据
mutex = threading.Lock()    #创建一个锁

class MyThread(threading.Thread):
    def run(self):
        global num

        # with mutex:#with Lock的作用相当于自动获取和释放锁(资源)
        #     for i in range(100000):#全局变量作为共享资源,一个线程拥有锁,便可以对资源进行操作;而其他线程无法获取操作共享资源
        #         num += 1
        # print(num)
        #上面和下面的是等价的
        if mutex.acquire(1):  # 获取锁
            for i in range(100000):  # 全局变量作为共享资源,一个线程拥有锁,便可以对资源进行操作;而其他线程无法获取操作共享资源
                num += 1
            mutex.release()  # 释放锁
        print(num)

threads = []
for i in range(5):
    t = MyThread()
    t.start()
    threads.append(t)

for t in threads:
    t.join()

print('game over')

mutex 即为 Lock() 对象,该对象同样包含 __enter()____exit()__方法。

拓展

1.实现上下文管理器类

第一种方法是实现一个类,其含有一个实例属性 db 和上下文管理器所需要的方法__enter()____exit()__

class Transaction(object):
    def __init__(self, db):
        self.db = db

    def __enter__(self):
        self.db.begin()

    def __exit__(self, type, value, traceback):
        if type is None:
            db.commit()
        else:
            db.rollback()

with Transaction(db):
    # do some actions

2.使用生成器装饰器

在 Python 的标准库中,有一个装饰器可以通过生成器获取上下文管理器。使用生成器装饰器的实现过程如下:

rom contextlib import contextmanager

@contextmanager
def transaction(db):
    db.begin()

    try:
        yield db
    except:
        db.rollback()
        raise
    else:
        db.commit()

第一眼上看去,这种实现方式更为简单,但是其机制更为复杂。看一下其执行过程吧:

  1. Python 解释器识别到 yield 关键字后,def 会创建一个生成器函数替代常规的函数(在类定义之外我喜欢用函数代替方法)
  2. 装饰器 contextmanager 被调用并返回一个帮助函数,这个帮助函数在被调用后会生成一个 GeneratorContextManager 实例。最终 with 表达式中的 EXPR 调用的是由 contentmanager 装饰器返回的帮助函数。
  3. with 表达式调用 transaction(db),实际上是调用帮助函数。帮助函数调用生成器函数,生成器函数创建一个生成器。
  4. 帮助函数将这个生成器传递给 GeneratorContextManager,并创建一个 GeneratorContextManager 的实例对象作为上下文管理器。
  5. with 表达式调用实例对象的上下文管理器的__enter()__方法。
  6. __enter()__方法中会调用这个生成器的 next()方法。这时候,生成器方法会执行到 yield db 处停止,并将 db 作为 next()的返回值。如果有 as VAR,那么它将会被赋值给 VAR。
  7. with 中的 BLOCK 被执行。
  8. BLOCK 执行结束后,调用上下文管理器的__exit()__方法。__exit()__方法会再次调用生成器的 next()方法。如果发生 StopIteration 异常,则 pass。
  9. 如果没有发生异常生成器方法将会执行 db.commit(),否则会执行 db.rollback()。

同 threading.Lock() 一样,装饰器 contextmanager 被调用后生成的实例中肯定存在 __enter()____exit()__ 方法。以下是详细代码:

def contextmanager(func):
    def helper(*args, **kwargs):
        return GeneratorContextManager(func(*args, **kwargs))
    return helper

class GeneratorContextManager(object):
    def __init__(self, gen):
        self.gen = gen

    def __enter__(self):
        try:
            return self.gen.next()
        except StopIteration:
            raise RuntimeError("generator didn't yield")

    def __exit__(self, type, value, traceback):
        if type is None:
            try:
                self.gen.next()
            except StopIteration:
                pass
            else:
                raise RuntimeError("generator didn't stop")
        else:
            try:
                self.gen.throw(type, value, traceback)
                raise RuntimeError("generator didn't stop after throw()")
            except StopIteration:
                return True
            except:
                if sys.exc_info()[1] is not value:
                    raise

标准输出重定向实例

import sys

savedStdout = sys.stdout
with open("anc.txt", "w") as f:
    sys.stdout = f
    print("afkajkfd")#写入

sys.stdout = savedStdout
print("This message is for screen!")#屏显

按照上述生成器装饰起方法进行修改。

from contextlib import contextmanager
import sys

@contextmanager
def stdout_redirect(new_stdout):
    old_stdout = sys.stdout
    sys.stdout = new_stdout
    try:
        yield   #因为sys.stdout值属于 Python 内置类变量,在执行到yield方法时已经做完修改操作,此时next()无返回值
    finally:
        sys.stdout = old_stdout

with open("anc.txt", "w") as f:
    print('first message...')#屏显
    with stdout_redirect(f):
        print("hello world")#写入
    print('second message....')#屏显

with open("anc.txt",'w') as f,stdout_redirect(f):
    print('useful message...')  #写入

print('hello kitty')#屏显

发表评论 取消回复
表情 图片 链接 代码

分享