# 傳統(tǒng)資源管理方式
file = open('example.txt', 'r')
try:
    data = file.read()
    # 處理數(shù)據(jù)
finally:
    file.close()  # 必須確保文件被關(guān)閉

# 使用上下文管理器
with open('example.txt', 'r') as file:
    data = file.read()
    # 處理數(shù)據(jù)
# 文件會自動關(guān)閉，即使發(fā)生異常

class FileManager:
    """文件管理的上下文管理器"""
    
    def __init__(self, filename, mode):
        self.filename = filename
        self.mode = mode
        self.file = None
    
    def __enter__(self):
        """進(jìn)入上下文時調(diào)用，負(fù)責(zé)資源分配"""
        self.file = open(self.filename, self.mode)
        return self.file
    
    def __exit__(self, exc_type, exc_value, traceback):
        """退出上下文時調(diào)用，負(fù)責(zé)資源釋放"""
        if self.file:
            self.file.close()
        # 返回False表示不處理異常，異常會繼續(xù)傳播
        return False

# 使用示例
with FileManager('example.txt', 'w') as f:
    f.write('Hello, World!')

class ExceptionHandlingContext:
    """處理異常的上下文管理器"""
    
    def __enter__(self):
        print("進(jìn)入上下文")
        return self
    
    def __exit__(self, exc_type, exc_value, traceback):
        if exc_type is not None:
            print(f"捕獲異常: {exc_type.__name__}: {exc_value}")
            # 返回True抑制異常，返回False傳播異常
            return True  # 抑制異常
        print("正常退出")
        return False

# 測試異常處理
with ExceptionHandlingContext():
    raise ValueError("測試異常")
print("異常被抑制，繼續(xù)執(zhí)行")

class DatabaseTransaction:
    """數(shù)據(jù)庫事務(wù)管理的上下文管理器"""
    
    def __init__(self, db_connection):
        self.conn = db_connection
        self.success = False
    
    def __enter__(self):
        self.conn.begin_transaction()
        return self.conn
    
    def __exit__(self, exc_type, exc_value, traceback):
        if exc_type is None:
            # 沒有異常，提交事務(wù)
            self.conn.commit()
            self.success = True
            print("事務(wù)提交成功")
        else:
            # 有異常，回滾事務(wù)
            self.conn.rollback()
            print("事務(wù)回滾")
        # 不抑制異常，讓調(diào)用者知道發(fā)生了什么
        return False

# 使用示例
class MockDatabase:
    def begin_transaction(self): print("開始事務(wù)")
    def commit(self): print("提交事務(wù)")
    def rollback(self): print("回滾事務(wù)")

db = MockDatabase()
try:
    with DatabaseTransaction(db) as conn:
        # 模擬數(shù)據(jù)庫操作
        print("執(zhí)行數(shù)據(jù)庫操作...")
        # 模擬異常情況
        # raise Exception("操作失敗!")
        print("操作成功")
except Exception as e:
    print(f"捕獲到異常: {e}")
else:
    print("所有操作完成")

from contextlib import contextmanager

@contextmanager
def file_manager(filename, mode):
    """使用生成器實現(xiàn)的文件上下文管理器"""
    try:
        # yield之前的代碼相當(dāng)于__enter__
        f = open(filename, mode)
        yield f
    finally:
        # yield之后的代碼相當(dāng)于__exit__
        f.close()

# 使用方式與類實現(xiàn)完全相同
with file_manager('example.txt', 'w') as f:
    f.write('Hello, World!')

from contextlib import contextmanager

@contextmanager
def transaction_manager(db_connection):
    """數(shù)據(jù)庫事務(wù)管理的上下文管理器"""
    try:
        # 開始事務(wù)
        db_connection.begin()
        yield db_connection
        # 如果沒有異常，提交事務(wù)
        db_connection.commit()
    except Exception as e:
        # 發(fā)生異常，回滾事務(wù)
        db_connection.rollback()
        raise e  # 重新拋出異常
    finally:
        # 確保連接關(guān)閉
        db_connection.close()

# 使用示例
class MockDBConnection:
    def begin(self): print("開始事務(wù)")
    def commit(self): print("提交事務(wù)")
    def rollback(self): print("回滾事務(wù)")
    def close(self): print("關(guān)閉連接")
    def execute(self, query): print(f"執(zhí)行: {query}")

db = MockDBConnection()
try:
    with transaction_manager(db) as conn:
        conn.execute("INSERT INTO users VALUES (1, 'John')")
        # 模擬異常
        # raise Exception("模擬失敗!")
except Exception as e:
    print(f"捕獲異常: {e}")

from contextlib import suppress

# 抑制FileNotFoundError異常
with suppress(FileNotFoundError):
    os.remove('somefile.tmp')
    print("文件刪除成功")  # 如果文件不存在，這行不會執(zhí)行

from contextlib import redirect_stdout
import io

f = io.StringIO()
with redirect_stdout(f):
    print('Hello, World!')  # 輸出被重定向到f
print(f"捕獲的輸出: {f.getvalue()}")

from contextlib import ExitStack

# 動態(tài)管理多個上下文管理器
with ExitStack() as stack:
    files = [stack.enter_context(open(fname, 'r')) for fname in filenames]
    # 所有文件都會在退出時自動關(guān)閉
    contents = [f.read() for f in files]

import asyncio

class AsyncDatabaseConnection:
    """異步數(shù)據(jù)庫連接管理器"""
    
    def __init__(self, connection_string):
        self.connection_string = connection_string
        self.connection = None
    
    async def __aenter__(self):
        """異步建立連接"""
        print(f"異步連接到 {self.connection_string}")
        # 模擬異步連接操作
        await asyncio.sleep(0.5)
        self.connection = f"connection_to_{self.connection_string}"
        return self
    
    async def __aexit__(self, exc_type, exc_value, exc_traceback):
        """異步關(guān)閉連接"""
        if self.connection:
            print(f"異步關(guān)閉連接 {self.connection}")
            await asyncio.sleep(0.2)
            self.connection = None
        return False
    
    async def execute_query(self, query):
        """異步執(zhí)行查詢"""
        if self.connection:
            print(f"執(zhí)行查詢: {query}")
            await asyncio.sleep(0.3)
            return f"結(jié)果: {query}"
        else:
            raise RuntimeError("數(shù)據(jù)庫未連接")

# 使用異步上下文管理器
async def main():
    async with AsyncDatabaseConnection("my_async_db") as db:
        result = await db.execute_query("SELECT * FROM users")
        print(result)

# 運行異步示例
# asyncio.run(main())

# 傳統(tǒng)嵌套方式
with open('input.txt', 'r') as fin:
    with open('output.txt', 'w') as fout:
        for line in fin:
            fout.write(line.upper())

# 簡化寫法（Python 3.10+）
with (open('input.txt', 'r') as fin,
      open('output.txt', 'w') as fout):
    for line in fin:
        fout.write(line.upper())

from contextlib import ExitStack

def process_files(sources, destination):
    """動態(tài)處理多個文件"""
    with ExitStack() as stack:
        # 動態(tài)進(jìn)入多個上下文
        input_files = [stack.enter_context(open(src, 'r')) for src in sources]
        output_file = stack.enter_context(open(destination, 'w'))
        
        for file in input_files:
            output_file.write(file.read().upper())

import time
from contextlib import contextmanager

@contextmanager
def timer(description="操作"):
    """計時上下文管理器"""
    start = time.time()
    try:
        yield
    finally:
        end = time.time()
        print(f"{description}執(zhí)行時間: {end - start:.4f}秒")

# 使用示例
with timer("數(shù)據(jù)處理"):
    data = [i**2 for i in range(1000000)]

import tempfile
import shutil
from contextlib import contextmanager

@contextmanager
def temporary_directory():
    """臨時目錄上下文管理器"""
    temp_dir = tempfile.mkdtemp()
    try:
        yield temp_dir
    finally:
        shutil.rmtree(temp_dir)  # 清理臨時目錄

# 使用示例
with temporary_directory() as temp_dir:
    print(f"使用臨時目錄: {temp_dir}")
    # 在臨時目錄中創(chuàng)建工作文件

class OptimizedConnectionManager:
    """帶連接池的優(yōu)化上下文管理器"""
    
    _pool = []  # 簡單的連接池
    
    def __init__(self, connection_string):
        self.connection_string = connection_string
        self.connection = None
    
    def __enter__(self):
        # 嘗試從連接池獲取
        if self._pool:
            self.connection = self._pool.pop()
        else:
            self.connection = self._create_connection()
        return self.connection
    
    def __exit__(self, exc_type, exc_value, traceback):
        # 將連接放回池中以便復(fù)用
        if self.connection:
            self._pool.append(self.connection)
        return False  # 傳播異常
    
    def _create_connection(self):
        """創(chuàng)建新連接"""
        print(f"創(chuàng)建新連接: {self.connection_string}")
        return f"connection_to_{self.connection_string}"

from contextlib import contextmanager

class LazyLoader:
    """惰性加載的上下文管理器"""
    
    def __init__(self, data_source):
        self.data_source = data_source
        self._data = None
    
    def load(self):
        if self._data is None:
            print("執(zhí)行惰性加載...")
            self._data = self.data_source()
    
    def __enter__(self):
        self.load()
        return self._data
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self._data = None  # 釋放資源

# 使用示例
with LazyLoader(lambda: [i for i in range(1000000)]) as data:
    print(f"數(shù)據(jù)長度: {len(data)}")

import unittest

class TestContextManagers(unittest.TestCase):
    
    def test_file_manager(self):
        """測試文件管理器上下文管理器"""
        with FileManager('test.txt', 'w') as f:
            f.write('test content')
        
        # 驗證文件已關(guān)閉
        with self.assertRaises(ValueError):
            f.write('should fail')
    
    def test_exception_handling(self):
        """測試異常處理機制"""
        with self.assertRaises(ValueError):
            with ExceptionHandlingContext() as ctx:
                raise ValueError("測試異常")
    
    def test_database_transaction(self):
        """測試數(shù)據(jù)庫事務(wù)回滾"""
        db = MockDatabase()
        try:
            with DatabaseTransaction(db) as conn:
                raise Exception("模擬失敗")
        except Exception:
            pass  # 期望異常被傳播
        
        # 驗證事務(wù)已回滾

if __name__ == '__main__':
    unittest.main()