Python上下文管理器实现优雅处理资源释放的实战指南

# 危险的文件操作示例
def read_large_file(file_path):
    file = open(file_path, 'r')  # 获取文件句柄
    data = file.read()
    # 忘记调用 file.close()
    return data

# 错误的双重释放示例
class ResourceHolder:
    def __init__(self):
        self.resource = acquire_resource()
    
    def cleanup(self):
        if self.resource:
            release_resource(self.resource)
            self.resource = None

holder = ResourceHolder()
holder.cleanup()
holder.cleanup()  # 第二次调用导致异常

# 异常处理中的资源泄漏
def process_data():
    file = open('data.txt', 'r')
    db_conn = connect_to_db()
    try:
        data = file.read()
        db_conn.execute(f"INSERT INTO logs VALUES('{data}')")
    except Exception as e:
        print(f"Error occurred: {e}")
    # 无论是否发生异常，都需要关闭资源
    # 但实际代码中经常忘记处理

class ManagedResource:
    def __enter__(self):
        # 资源获取逻辑
        print("Acquiring resource...")
        self.resource = acquire_expensive_resource()
        return self.resource  # 可返回不同对象
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        # 资源释放逻辑
        print("Releasing resource...")
        if exc_type is not None:
            print(f"Handling exception: {exc_val}")
        release_resource(self.resource)

# 安全文件操作示例
def read_file_safely(file_path):
    with open(file_path, 'r', encoding='utf-8') as file:
        return file.read()

# 等价于手动实现：
def read_file_manual(file_path):
    file = None
    try:
        file = open(file_path, 'r', encoding='utf-8')
        return file.read()
    finally:
        if file is not None:
            file.close()

# 数据库连接池上下文管理器
class DatabaseConnection:
    def __init__(self, connection_string):
        self.connection_string = connection_string
    
    def __enter__(self):
        self.conn = psycopg2.connect(self.connection_string)
        return self.conn.cursor()
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is None:
            self.conn.commit()
        else:
            self.conn.rollback()
        self.conn.close()

# 使用示例
with DatabaseConnection("dbname=test user=postgres") as cursor:
    cursor.execute("SELECT * FROM users")
    print(cursor.fetchall())

# 线程锁上下文管理器
from threading import Lock

class ThreadLock:
    def __init__(self):
        self.lock = Lock()
    
    def __enter__(self):
        self.lock.acquire()
        print("Lock acquired")
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.lock.release()
        print("Lock released")

# 使用示例
shared_resource = 0
lock = ThreadLock()

def increment_resource():
    global shared_resource
    with lock:
        shared_resource += 1

# 临时文件上下文管理器
import tempfile
import os

class TemporaryDirectory:
    def __enter__(self):
        self.path = tempfile.mkdtemp()
        return self.path
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        import shutil
        shutil.rmtree(self.path)

# 使用示例
with TemporaryDirectory() as tmpdir:
    file_path = os.path.join(tmpdir, 'test.txt')
    with open(file_path, 'w') as f:
        f.write("Temporary content")
    # 退出with块后，临时目录自动删除

# 同时管理文件和数据库连接
with open('data.txt', 'r') as file, \
     DatabaseConnection("dbname=test") as cursor:
    data = file.read()
    cursor.execute("INSERT INTO logs VALUES(%s)", (data,))

# 上下文管理器装饰器
from functools import wraps

def contextmanager_decorator(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        # 模拟__enter__
        resource = func(*args, **kwargs)
        try:
            yield resource
        finally:
            # 模拟__exit__
            resource.cleanup()
    return wrapper

# 使用示例
class Resource:
    def cleanup(self):
        print("Cleaning up...")

@contextmanager_decorator
def get_resource():
    return Resource()

with get_resource() as res:
    print("Using resource...")

# 异步文件操作示例
import aiofiles

async def async_file_example():
    async with aiofiles.open('async.txt', mode='w') as f:
        await f.write("Async content")
    # 自动关闭文件

# 延迟获取资源的上下文管理器
class LazyResource:
    def __init__(self):
        self.resource = None
    
    def __enter__(self):
        if self.resource is None:
            print("Initializing resource...")
            self.resource = acquire_expensive_resource()
        return self.resource
    
    def __exit__(self, *args):
        pass  # 不释放资源，适合单例模式

# 区分异常类型的释放逻辑
class FineGrainedExit:
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is ValueError:
            print("Handling ValueError")
            return True  # 抑制异常传播
        elif exc_type is KeyboardInterrupt:
            print("Handling KeyboardInterrupt")
            return False  # 允许异常传播
        print("Cleaning up normally")

class WrongExit:
    def __exit__(self, *args):
        return True  # 抑制所有异常！

with WrongExit():
    1 / 0  # 异常被静默吞噬

class ValuableResource:
    def __enter__(self):
        return {"key": "value"}
    
    def __exit__(self, *args):
        pass

# 错误用法：忽略返回值
with ValuableResource() as res:
    print("Inside context")  # 未使用res

# 正确用法
with ValuableResource() as res:
    print(res["key"])  # 使用返回的资源

class DangerousExit:
    def __exit__(self, *args):
        if args[0] is not None:
            raise RuntimeError("Cleanup failed")  # 危险操作！

with DangerousExit():
    raise ValueError("Original error")
# 将同时存在 ValueError 和 RuntimeError

from typing import ContextManager

def process_with_resource() -> ContextManager[Resource]:
    ...

with DistributedLock("resource_key") as lock:
    # 跨多台机器的同步操作