Python核心工具 contextmanager 装饰器详解

import requests# 传统类实现上下文管理器class SessionManager:    def __init__(self, base_url, token):        self.base_url = base_url        self.token = token        self.session = None    # 进入with块时执行（初始化资源）    def __enter__(self):        self.session = requests.Session()        self.session.headers.update({"Authorization": f"Bearer {self.token}"})        response = self.session.post(f"{self.base_url}/login")        response.raise_for_status()        return self.session    # 退出with块时执行（清理资源）    def __exit__(self, exc_type, exc_val, exc_tb):        self.session.post(f"{self.base_url}/logout")# 使用with SessionManager("https://my-base-url.com", "my-fake-token") as session:    refresh_reports(session)    download_reports(session)

from contextlib import contextmanagerimport requests# 用生成器函数 + 装饰器实现上下文管理器@contextmanagerdef session_manager(base_url, token):    try:        # 1. 进入with块时执行（对应__enter__）        session = requests.Session()        session.headers.update({"Authorization": f"Bearer {token}"})        response = session.post(f"{base_url}/login")        response.raise_for_status()        yield session  # 返回给with...as的变量（相当于__enter__的return）    finally:        # 2. 退出with块时执行（对应__exit__），无论是否报错都会执行        session.post(f"{base_url}/logout")# 使用方式完全一致！with session_manager("https://my-base-url.com", "my-fake-token") as session:    refresh_reports(session)    download_reports(session)    email_reports(session)

from contextlib import contextmanagerimport time@contextmanagerdef timer(label):    start = time.time()  # 进入时记录开始时间    try:        yield  # 不返回值，仅用于控制上下文    finally:        end = time.time()  # 退出时计算耗时        print(f"{label}: {end - start:.4f} seconds")# 使用with timer("数据处理耗时"):    # 模拟耗时操作    time.sleep(0.5)    do_something()# 输出：数据处理耗时: 0.5001 seconds

from contextlib import contextmanagerimport sqlite3@contextmanagerdef db_transaction(db_path, commit=False):    # 1. 初始化数据库连接和游标    conn = sqlite3.connect(db_path)    cursor = conn.cursor()    try:        yield cursor  # 返回游标，供with块内执行SQL        conn.commit()  # 无异常则提交事务    except Exception as e:        conn.rollback()  # 有异常则回滚        raise  # 重新抛出异常，让外部捕获（不隐藏错误）    finally:        conn.close()  # 无论是否报错，都关闭连接# 使用：执行两条插入语句，要么都成功，要么都失败with db_transaction("mydb.db") as cursor:    cursor.execute("INSERT INTO users (name) VALUES (?)", ("Alice",))    cursor.execute("INSERT INTO users (name) VALUES (?)", ("Bob",))

from contextlib import contextmanager# 错误写法：清理代码不在finally，可能不执行@contextmanagerdef bad_context():    resource = acquire_resource()    yield resource    release_resource(resource)  # 若yield后报错，这里不会执行！# 正确写法：清理代码放在finally，必执行@contextmanagerdef good_context():    resource = acquire_resource()    try:        yield resource    finally:        release_resource(resource)  # 无论是否报错，都执行

from contextlib import contextmanager@contextmanagerdef bad_multiple_yields():    yield 1    yield 2  # 这个永远不会执行with bad_multiple_yields() as value:    print(value)  # 输出：1

from contextlib import contextmanager# 不会屏蔽异常（异常会传播出去）@contextmanagerdef does_not_suppress():    try:        yield    finally:        pass# 手动捕获，实现异常屏蔽@contextmanagerdef suppress_errors():    try:        yield    except Exception as e:        print(f"屏蔽异常：{e}")  # 不重新抛出，就是屏蔽# 测试with suppress_errors():    raise ValueError("这个异常会被屏蔽")print("程序继续执行")  # 会正常执行，不会被异常中断