python中一些很好用的工具类模块

contextlib

此模块为 with 语句提供了一些工具，简化了 with的使用。

1. @contextlib.contextmanager

   它将一个生成器函数转换为一个上下文管理器。而无需创建一个类或单独的 __enter__() 和 __exit__() 方法。

   被装饰的生成器函数需要遵循以下规则：

   • 在生成器中，yield 之前的代码会在进入上下文时执行（相当于 __enter__ 方法）。

   • yield 之后的代码会在退出上下文时执行（相当于 __exit__ 方法）。

   • 如果在上下文代码块中发生异常，异常信息会传递给生成器函数，可以通过捕获异常来处理。

   from contextlib import contextmanager
   ​
   @contextmanager
   def managed_resource(*args, **kwds):
       # 获取资源的代码，例如：
       resource = acquire_resource(*args, **kwds)
       try:
           yield resource
       finally:
           # 释放资源的代码，例如：
           release_resource(resource)
           
   # 此时可以直接使用with调用        
   with managed_resource(timeout=3600) as resource:
       # 资源将在此代码块的末尾被释放，
       # 即使代码块中的代码引发了异常

2. contextlib.closing(thing)

   # 接收一个函数，在执行完后调用其 close方法
   from urllib.request import urlopen
   ​
   with closing(urlopen('https://www.python.org')) as page:
       for line in page:
           print(line)  

   closing的功能类似于

   from contextlib import contextmanager
   ​
   @contextmanager
   def closing(thing):
       try:
           yield thing
       finally:
           thing.close()
   ​

contextvars

可以更简单的实现上下文管理及隔离，线程及异步安全。

普通线程中使用

from contextvars import ContextVar
import threading
​
var = ContextVar('var', default='default')
​
def worker():
    print(f"worker: {var.get()}")  # 输出 'be set'
​
def main():
    print(f"main-start: {var.get()}")  # 输出 'default'
    var.set('be set')
    worker()
    print(f"main-end: {var.get()}")  # 输出 'be set'
​
# 主线程
if __name__ == '__main__':
    main()
    

异步使用

import asyncio
from contextvars import ContextVar
​
var = ContextVar('var', default='default')
​
async def task(name, value):
    print(f"{name}: before set -> {var.get()}")  # 默认值
    var.set(value)
    print(f"{name}: after set -> {var.get()}")  # 当前协程的值
​
async def main():
    # 创建两个协程，分别设置不同的值
    t1 = asyncio.create_task(task("Task-A", "value-A"))
    t2 = asyncio.create_task(task("Task-B", "value-B"))
    await t1
    await t2
    print("Main after tasks:", var.get())  # 主协程的上下文不受影响
​
asyncio.run(main())
​
​
# 输出
Task-A: before set -> default
Task-A: after set -> value-A
Task-B: before set -> default
Task-B: after set -> value-B
Main after tasks: default