Pythonコンテキストマネージャを自作してwith文を極める！初心者でもわかる実装方法と活用術

with コンテキストマネージャ as 変数名:
    # ここで処理を実行
    pass
# ここで自動的に後片付けが実行される

def __enter__(self):
    # リソースの取得や初期化処理
    print("リソースを取得しました")
    return self  # 通常は自分自身を返す

def __exit__(self, exc_type, exc_value, traceback):
    # リソースの解放や後片付け処理
    print("リソースを解放しました")
    return False  # 例外を再発生させる場合はFalse

import time

class Timer:
    def __enter__(self):
        print("処理開始...")
        self.start_time = time.time()
        return self
    
    def __exit__(self, exc_type, exc_value, traceback):
        end_time = time.time()
        elapsed_time = end_time - self.start_time
        print(f"処理時間: {elapsed_time:.2f}秒")
        return False

# 使用例
with Timer():
    # 時間のかかる処理をシミュレート
    time.sleep(2)
    print("何かの処理を実行中...")

処理開始...
何かの処理を実行中...
処理時間: 2.00秒

class SafeFileManager:
    def __init__(self, filename, mode='r'):
        self.filename = filename
        self.mode = mode
        self.file = None
    
    def __enter__(self):
        try:
            self.file = open(self.filename, self.mode)
            print(f"ファイル '{self.filename}' を開きました")
            return self.file
        except IOError as e:
            print(f"ファイルの打開に失敗: {e}")
            raise
    
    def __exit__(self, exc_type, exc_value, traceback):
        if self.file:
            self.file.close()
            print(f"ファイル '{self.filename}' を閉じました")
        
        if exc_type:
            print(f"エラーが発生しました: {exc_value}")
        
        return False  # 例外を再発生させる

# 使用例
with SafeFileManager('test.txt', 'w') as f:
    f.write("Hello, World!")
    f.write("コンテキストマネージャのテストです")

class TemporarySettings:
    def __init__(self, **settings):
        self.new_settings = settings
        self.old_settings = {}
    
    def __enter__(self):
        # 現在の設定を保存
        for key, value in self.new_settings.items():
            if hasattr(self, key):
                self.old_settings[key] = getattr(self, key)
            setattr(self, key, value)
        
        print("設定を一時的に変更しました")
        return self
    
    def __exit__(self, exc_type, exc_value, traceback):
        # 元の設定に戻す
        for key, value in self.old_settings.items():
            setattr(self, key, value)
        
        print("設定を元に戻しました")
        return False

# グローバル設定クラス（例）
class AppConfig:
    debug_mode = False
    log_level = "INFO"

config = AppConfig()

# 使用例
print(f"変更前: debug_mode = {config.debug_mode}")

with TemporarySettings(debug_mode=True, log_level="DEBUG"):
    print(f"変更中: debug_mode = {config.debug_mode}")
    # ここでデバッグ処理を実行

print(f"変更後: debug_mode = {config.debug_mode}")

class DatabaseConnection:
    def __init__(self, connection_string):
        self.connection_string = connection_string
        self.connection = None
    
    def __enter__(self):
        print("データベースに接続中...")
        # 実際の接続処理（ここではシミュレート）
        self.connection = f"接続: {self.connection_string}"
        print("データベース接続完了")
        return self.connection
    
    def __exit__(self, exc_type, exc_value, traceback):
        if self.connection:
            print("データベース接続を切断中...")
            # 実際の切断処理
            self.connection = None
            print("データベース接続切断完了")
        
        if exc_type:
            print("エラーが発生したため、トランザクションをロールバックします")
        
        return False

# 使用例
with DatabaseConnection("localhost:5432/mydb") as db:
    print("データベース操作を実行中...")
    # ここでSQLクエリを実行

from contextlib import contextmanager
import time

@contextmanager
def timer():
    print("処理開始...")
    start_time = time.time()
    
    try:
        yield  # ここでwithブロックの処理が実行される
    finally:
        end_time = time.time()
        elapsed_time = end_time - start_time
        print(f"処理時間: {elapsed_time:.2f}秒")

# 使用例
with timer():
    time.sleep(1)
    print("何かの処理...")

@contextmanager
def file_manager(filename, mode='r'):
    print(f"ファイル '{filename}' を開きます")
    f = open(filename, mode)
    
    try:
        yield f  # ファイルオブジェクトを返す
    finally:
        f.close()
        print(f"ファイル '{filename}' を閉じました")

# 使用例
with file_manager('test.txt', 'w') as f:
    f.write("contextlibを使った例です")

import logging
from contextlib import contextmanager

@contextmanager
def temporary_log_level(level):
    logger = logging.getLogger()
    old_level = logger.level
    
    print(f"ログレベルを {old_level} から {level} に変更")
    logger.setLevel(level)
    
    try:
        yield logger
    finally:
        logger.setLevel(old_level)
        print(f"ログレベルを {level} から {old_level} に戻しました")

# 使用例
logging.basicConfig(level=logging.WARNING)

with temporary_log_level(logging.DEBUG) as logger:
    logger.debug("これはデバッグメッセージです")
    logger.info("これは情報メッセージです")

class ResourceManager:
    def __init__(self, resource_name):
        self.resource_name = resource_name
        self.resource = None
    
    def __enter__(self):
        print(f"リソース '{self.resource_name}' を取得")
        self.resource = f"取得済み_{self.resource_name}"
        return self.resource
    
    def __exit__(self, exc_type, exc_value, traceback):
        print(f"リソース '{self.resource_name}' を解放")
        
        if exc_type is not None:
            print(f"例外が発生しました:")
            print(f"  タイプ: {exc_type.__name__}")
            print(f"  値: {exc_value}")
            print("それでもリソースは正しく解放されました")
        
        # Falseを返すと例外が再発生する
        # Trueを返すと例外を抑制する
        return False

# 正常なケース
print("=== 正常なケース ===")
with ResourceManager("ファイル1") as resource:
    print(f"リソースを使用中: {resource}")

print("\n=== 例外が発生するケース ===")
try:
    with ResourceManager("ファイル2") as resource:
        print(f"リソースを使用中: {resource}")
        raise ValueError("何かエラーが発生!")
except ValueError as e:
    print(f"例外をキャッチ: {e}")

class IgnoreSpecificError:
    def __init__(self, ignored_exception):
        self.ignored_exception = ignored_exception
    
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_value, traceback):
        if exc_type is self.ignored_exception:
            print(f"{exc_type.__name__} を無視しました: {exc_value}")
            return True  # 例外を抑制
        return False  # その他の例外は再発生

# 使用例
print("=== FileNotFoundErrorを無視する例 ===")
with IgnoreSpecificError(FileNotFoundError):
    with open("存在しないファイル.txt") as f:
        content = f.read()

print("処理は継続されます")

from contextlib import contextmanager

@contextmanager
def safe_operation(operation_name):
    print(f"{operation_name} を開始")
    
    try:
        yield
        print(f"{operation_name} が正常に完了")
    except Exception as e:
        print(f"{operation_name} でエラー発生: {e}")
        print("クリーンアップ処理を実行")
        # ここでクリーンアップ処理
        raise  # 例外を再発生
    finally:
        print(f"{operation_name} の後処理完了")

# 使用例
try:
    with safe_operation("重要な処理"):
        print("何かの重要な処理...")
        # raise RuntimeError("予期しないエラー")  # コメントアウトを外すとエラー
except RuntimeError as e:
    print(f"最終的にキャッチされたエラー: {e}")