199_Python上下文管理器¶

# 基于类的上下文管理器
class FileManager:
    def __init__(self, filename, mode='r'):
        self.filename = filename
        self.mode = mode
        self.file = None

    def __enter__(self):
        self.file = open(self.filename, self.mode)
        return self.file

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.file:
            self.file.close()
        # 返回False（默认）：异常继续传播
        # 返回True：异常被抑制
        return False

with FileManager('data.txt', 'w') as f:
    f.write("Hello, Context Manager!")

# 数据库连接管理器
class DatabaseConnection:
    def __init__(self, connection_string):
        self.connection_string = connection_string
        self.conn = None

    def __enter__(self):
        import sqlite3
        self.conn = sqlite3.connect(self.connection_string)
        return self.conn

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is not None:
            self.conn.rollback()  # 异常时回滚
        else:
            self.conn.commit()    # 正常时提交
        self.conn.close()
        return False

with DatabaseConnection('mydb.sqlite') as conn:
    cursor = conn.cursor()
    cursor.execute("CREATE TABLE IF NOT EXISTS users (name TEXT)")
    cursor.execute("INSERT INTO users VALUES ('Alice')")

from contextlib import contextmanager
import time

@contextmanager
def timer(label=""):
    """计时上下文管理器"""
    start = time.perf_counter()
    try:
        yield  # 这里把控制权交给with块
    finally:
        elapsed = time.perf_counter() - start
        print(f"{label} 耗时: {elapsed:.4f}s")

with timer("数据处理"):
    time.sleep(1)
    # 数据处理 耗时: 1.0001s

@contextmanager
def temporary_directory():
    """创建临时目录，用完自动删除"""
    import tempfile, shutil
    tmpdir = tempfile.mkdtemp()
    try:
        yield tmpdir
    finally:
        shutil.rmtree(tmpdir)

with temporary_directory() as tmpdir:
    print(f"临时目录: {tmpdir}")
    # 使用临时目录...
# 退出后自动删除

@contextmanager
def change_directory(path):
    """临时切换工作目录"""
    import os
    old_dir = os.getcwd()
    os.chdir(path)
    try:
        yield
    finally:
        os.chdir(old_dir)

@contextmanager
def suppress_output():
    """抑制stdout输出"""
    import sys, os
    old_stdout = sys.stdout
    sys.stdout = open(os.devnull, 'w')
    try:
        yield
    finally:
        sys.stdout.close()
        sys.stdout = old_stdout

from contextlib import ExitStack, suppress, redirect_stdout

# 1. ExitStack：动态管理多个上下文
with ExitStack() as stack:
    files = [stack.enter_context(open(f)) for f in ['a.txt', 'b.txt', 'c.txt']]
    # 所有文件在退出时自动关闭

# 2. suppress：忽略特定异常
from contextlib import suppress
with suppress(FileNotFoundError):
    os.remove('nonexistent.txt')  # 不存在也不会报错

# 3. 可重入上下文管理器
@contextmanager
def managed_resource():
    print("获取资源")
    yield "resource"
    print("释放资源")

# 4. 上下文管理器作为装饰器
class log_execution:
    """同时作为上下文管理器和装饰器"""
    def __init__(self, func_or_label=None):
        if callable(func_or_label):
            self.label = func_or_label.__name__
            self.func = func_or_label
        else:
            self.label = func_or_label or "block"
            self.func = None

    def __enter__(self):
        print(f"[START] {self.label}")
        self.start = time.perf_counter()
        return self

    def __exit__(self, *exc):
        elapsed = time.perf_counter() - self.start
        print(f"[END] {self.label} ({elapsed:.4f}s)")
        return False

    def __call__(self, *args, **kwargs):
        if self.func:
            with self:
                return self.func(*args, **kwargs)

# 5. 异步上下文管理器
class AsyncDBConnection:
    async def __aenter__(self):
        self.conn = await asyncio.connect("db://localhost")
        return self.conn

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.conn.close()

# async with AsyncDBConnection() as conn:
#     await conn.execute("SELECT 1")

# 文件操作（最常见）
with open('file.txt', 'r') as f:
    data = f.read()

# 锁管理
import threading
lock = threading.Lock()
with lock:
    # 临界区代码
    pass

# 多上下文管理（Python 3.10+）
with (open('a.txt') as a, open('b.txt') as b):
    pass

# contextlib实用工具
from contextlib import closing, suppress, redirect_stdout, ExitStack
with closing(urllib.request.urlopen('http://example.com')) as page:
    pass