Pythonのコンテキストマネージャーを業務で活かす実装パターン集

Python

Pythonのコンテキストマネージャーを業務で活かす実装パターン集

Pythonを使った開発をしていると、リソースの確保と解放を安全に管理する必要が頻繁に発生します。ファイルを開いたら必ず閉じる、データベース接続を確立したら必ず切断する、といった処理はプログラミングの基本ですが、実装が煩雑になりやすいものです。そこで活躍するのがコンテキストマネージャーです。

本記事では、Pythonのコンテキストマネージャーの基本から、実務で実際に使われるパターンまで、具体的なコード例を交えて解説します。

1. コンテキストマネージャーとは何か

コンテキストマネージャーは、with文を使ってリソースの確保と解放を自動的に管理するPythonの仕組みです。

基本的な構文は以下の通りです：

with resource as variable:
    # リソースを使用する処理
    pass
# リソースは自動的に解放される

内部的には、以下の2つのメソッドが呼ばれます：

  • __enter__：withブロックに入るときに呼ばれる（リソース確保）
  • __exit__：withブロックを抜けるときに呼ばれる（リソース解放）

これにより、例外が発生した場合でも確実にリソースが解放されるため、バグの少ない安全なコードが書けます。

2. 業務でのユースケース

実務では、以下のようなシーンでコンテキストマネージャーが活躍します：

2.1 データベース接続の管理

Webアプリケーションやバッチ処理では、データベースへの接続と切断を確実に行う必要があります。コンテキストマネージャーを使えば、処理の終了時に自動的に接続を閉じられます。

2.2 ファイル操作の安全性確保

ファイルの読み込み・書き込みを行う際、処理中に例外が発生してもファイルが閉じられることを保証できます。

2.3 ロック機構による競合制御

マルチスレッド処理では、臨界領域を保護するためにロック（mutex）が使われますが、コンテキストマネージャーを使うと自動的にロック・アンロックできます。

2.4 一時的な設定の変更

ログレベルやデータベーストランザクションなど、一時的に設定を変更して、ブロック終了時に復元したい場合に便利です。

3. 実装コード：実務パターン

3.1 カスタムコンテキストマネージャー：データベース接続管理

まず最も一般的なパターンとして、データベース接続を管理するコンテキストマネージャーを実装してみます。

import sqlite3
from typing import Optional

class DatabaseConnection:
    \"\"\"データベース接続を管理するコンテキストマネージャー\"\"\"
    
    def __init__(self, db_path: str):
        self.db_path = db_path
        self.connection: Optional[sqlite3.Connection] = None
    
    def __enter__(self) -> sqlite3.Connection:
        \"\"\"コンテキスト開始時に接続を確立\"\"\"
        self.connection = sqlite3.connect(self.db_path)
        print(f\"[DB] Connected to {self.db_path}\")
        return self.connection
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        \"\"\"コンテキスト終了時に接続を閉じる\"\"\"
        if self.connection:
            self.connection.close()
            print(f\"[DB] Connection closed\")\n        
        # 例外が発生した場合、Falseを返すと例外が伝播する
        if exc_type is not None:
            print(f\"[DB] Error occurred: {exc_type.__name__}: {exc_val}\")\n        return False

# 使用例
def fetch_user_data(user_id: int) -> dict:
    \"\"\"ユーザーデータを取得する業務ロジック\"\"\"
    with DatabaseConnection(\":memory:\") as conn:
        cursor = conn.cursor()
        cursor.execute(\n            \"SELECT id, name FROM users WHERE id = ?\", \n            (user_id,)\n        )\n        result = cursor.fetchone()\n        return {\"id\": result[0], \"name\": result[1]} if result else None

3.2 複数のリソースを同時に管理

実務では、複数のリソースを同時に管理する必要があることがよくあります。以下は、ログファイルを開きながらデータベースにアクセスする例です。

from datetime import datetime

def import_user_batch(csv_file_path: str, db_path: str, log_file_path: str):
    \"\"\"CSVからユーザーデータをインポートしながらログを記録\"\"\"
    with DatabaseConnection(db_path) as db_conn, \\\n         open(log_file_path, 'a', encoding='utf-8') as log_file:
        \n        log_file.write(f\"[{datetime.now()}] Starting batch import\\n\")\n        \n        try:\n            with open(csv_file_path, 'r', encoding='utf-8') as csv_file:\n                cursor = db_conn.cursor()\n                \n                for line_num, line in enumerate(csv_file, 1):\n                    try:\n                        user_id, name, email = line.strip().split(',')\n                        cursor.execute(\n                            \"INSERT INTO users (id, name, email) VALUES (?, ?, ?)\",\n                            (user_id, name, email)\n                        )\n                        db_conn.commit()\n                        log_file.write(f\"[{datetime.now()}] Row {line_num}: Imported user {name}\\n\")\n                    \n                    except ValueError:\n                        error_msg = f\"[{datetime.now()}] Row {line_num}: Invalid format\\n\"\n                        log_file.write(error_msg)\n                        print(error_msg)\n            \n            log_file.write(f\"[{datetime.now()}] Batch import completed successfully\\n\")\n        \n        except Exception as e:\n            log_file.write(f\"[{datetime.now()}] ERROR: {str(e)}\\n\")\n            raise

3.3 contextlib.contextmanagerデコレータを使ったシンプル実装

簡単なコンテキストマネージャーは、クラスではなくデコレータで実装する方が手軽です。

from contextlib import contextmanager
import time

@contextmanager
def timer(name: str = \"Processing\"):
    \"\"\"処理時間を計測するコンテキストマネージャー\"\"\"
    start_time = time.time()\n    print(f\"[TIMER] {name} started\")\n    \n    try:\n        yield  # コンテキストブロックの実行\n    finally:\n        elapsed_time = time.time() - start_time\n        print(f\"[TIMER] {name} completed in {elapsed_time:.2f}s\")\n\n# 使用例\ndef heavy_computation():\n    \"\"\"重い計算処理\"\"\"n    with timer(\"Heavy computation\"):\n        total = sum(i ** 2 for i in range(1000000))\n        return total

3.4 トランザクション管理の実装

データベースのトランザクション管理も、コンテキストマネージャーの典型的な用途です。

from contextlib import contextmanager\nimport sqlite3\n\n@contextmanager\ndef transaction(db_connection: sqlite3.Connection):\n    \"\"\"トランザクション管理コンテキストマネージャー\"\"\"n    try:\n        yield db_connection\n        db_connection.commit()\n        print(\"[TX] Transaction committed\")\n    except Exception as e:\n        db_connection.rollback()\n        print(f\"[TX] Transaction rolled back due to: {e}\")\n        raise\n\n# 使用例\ndef transfer_money(from_account: int, to_account: int, amount: float):\n    \"\"\"口座間送金処理\"\"\"n    with DatabaseConnection(\":memory:\") as conn:\n        with transaction(conn):\n            cursor = conn.cursor()\n            \n            # 送金元から減額\n            cursor.execute(\n                \"UPDATE accounts SET balance = balance - ? WHERE id = ?\",\n                (amount, from_account)\n            )\n            \n            # 受取人に加算\n            cursor.execute(\n                \"UPDATE accounts SET balance = balance + ? WHERE id = ?\",\n                (amount, to_account)\n            )\n            \n            print(f\"[TRANSFER] Sent {amount} from account {from_account} to {to_account}\")

4. よくある応用パターン

4.1 リトライロジック付きコンテキストマネージャー

ネットワーク通信やAPI呼び出しなど、失敗の可能性がある処理にはリトライが必須です。

from contextlib import contextmanager\nimport time\nfrom typing import Callable, TypeVar\n\nT = TypeVar('T')\n\n@contextmanager\ndef retry_on_failure(max_retries: int = 3, delay: float = 1.0):\n    \"\"\"リトライ機能付きコンテキストマネージャー\"\"\"n    attempt = 0\n    last_exception = None\n    \n    while attempt < max_retries:\n        try:\n            yield\n            return  # 成功したら終了\n        except Exception as e:\n            attempt += 1\n            last_exception = e\n            \n            if attempt < max_retries:\n                print(f\"[RETRY] Attempt {attempt} failed. Retrying in {delay}s...\")\n                time.sleep(delay)\n            else:\n                print(f\"[RETRY] All {max_retries} attempts failed\")\n    \n    if last_exception:\n        raise last_exception\n\n# 使用例\ndef fetch_from_api(url: str) -> dict:\n    \"\"\"不安定なAPIからデータを取得\"\"\"n    with retry_on_failure(max_retries=3, delay=2.0):\n        # APIコール処理（ここで例外が発生する可能性がある）\n        print(f\"[API] Fetching from {url}\")\n        # response = requests.get(url)\n        # return response.json()\n        return {\"status\": \"success\"}

4.2 リソースプーリング

データベース接続をプールして、複数の処理で効率的に再利用するパターンです。

from contextlib import contextmanager\nimport sqlite3\nfrom queue import Queue\nfrom typing import Optional\n\nclass ConnectionPool:\n    \"\"\"接続プール管理クラス\"\"\"n    def __init__(self, db_path: str, pool_size: int = 5):\n        self.db_path = db_path\n        self.pool_size = pool_size\n        self.pool: Queue = Queue(maxsize=pool_size)\n        \n        # 初期接続を作成\n        for _ in range(pool_size):\n            conn = sqlite3.connect(db_path)\n            self.pool.put(conn)\n    \n    @contextmanager\n    def get_connection(self):\n        \"\"\"接続をプールから取得\"\"\"n        conn = self.pool.get()  # プールから取得（満杯時はブロック）\n        try:\n            yield conn\n        finally:\n            self.pool.put(conn)  # プールに戻す\n    \n    def close_all(self):\n        \"\"\"すべての接続をクローズ\"\"\"n        while not self.pool.empty():\n            conn = self.pool.get_nowait()\n            conn.close()\n        print(\"[POOL] All connections closed\")\n\n# 使用例\npool = ConnectionPool(\":memory:\", pool_size=3)\n\ndef execute_query(query_id: int, sql: str):\n    \"\"\"クエリを実行\"\"\"n    with pool.get_connection() as conn:\n        cursor = conn.cursor()\n        cursor.execute(sql)\n        print(f\"[QUERY {query_id}] Executed: {sql}\")

4.3 コンテキスト変数の一時的な変更

ログレベルやタイムアウト値など、グローバル設定を一時的に変更する場合に便利です。

from contextlib import contextmanager\nimport logging\n\n@contextmanager\ndef temporary_log_level(logger: logging.Logger, level: int):\n    \"\"\"一時的にログレベルを変更\"\"\"n    original_level = logger.level\n    logger.setLevel(level)\n    print(f\"[LOG] Level changed from {original_level} to {level}\")\n    \n    try:\n        yield\n    finally:\n        logger.setLevel(original_level)\n        print(f\"[LOG] Level restored to {original_level}\")\n\n# 使用例\nlogger = logging.getLogger(__name__)\nlogger.setLevel(logging.WARNING)\n\ndef debug_heavy_function():\n    \"\"\"デバッグモードで関数を実行\"\"\"n    with temporary_log_level(logger, logging.DEBUG):\n        logger.debug(\"This is a debug message\")\n        logger.info(\"This is an info message\")\n        # 処理...\n    \n    logger.debug(\"This won't be printed (level is WARNING again)\")

5. コンテキストマネージャー使用時の注意点

5.1 例外処理と__exit__の戻り値

__exit__メソッドは、例外が発生した場合の挙動を制御する重要なポイントです。

class SmartContextManager:\n    \"\"\"例外処理の正しい実装例\"\"\"n    def __enter__(self):\n        print(\"[ENTER] Acquiring resource\")\n        return self\n    \n    def __exit__(self, exc_type, exc_val, exc_tb):\n        \"\"\"\n        戻り値の意味:\n        - True: 例外を抑止（コンテキスト外では例外が発生しない）\n        - False: 例外を伝播（コンテキスト外で例外が発生）\n        \"\"\"\n        print(f\"[EXIT] Releasing resource\")\n        \n        # 例外が発生した場合\n        if exc_type is not None:\n            print(f\"[EXIT] Exception detected: {exc_type.__name__}\")\n            \n            # 特定の例外だけ抑止したい場合\n            if issubclass(exc_type, ValueError):\n                print(\"[EXIT] ValueError is suppressed\")\n                return True  # 例外を抑止\n            \n            # その他の例外は伝播させる\n            return False\n        \n        return False\n\n# 使用例\ntry:\n    with SmartContextManager():\n        raise ValueError(\"This is a test error\")\n    print(\"After context (ValueError was suppressed)\")\nexcept ValueError:\n    print(\"This won't be printed\")

5.2 ネストされたコンテキストマネージャーの順序

複数のコンテキストマネージャーをネストする場合、破棄の順序に注意が必要です。

from contextlib import contextmanager\n\n@contextmanager\ndef resource_a():\n    print(\"[A] Acquired\")\n    try:\n        yield \"Resource A\"\n    finally:\n        print(\"[A] Released\")\n\n@contextmanager\ndef resource_b():\n    print(\"[B] Acquired\")\n    try:\n        yield \"Resource B\"\n    finally:\n        print(\"[B] Released\")\n\n# ネストの順序で破棄順序も決まる\nwith resource_a() as res_a, resource_b() as res_b:\n    print(f\"Using {res_a} and {res_b}\")\n\n# 出力:\n# [A] Acquired\n# [B] Acquired\n# Using Resource A and Resource B\n# [B] Released  ← 後に取得したものが先に破棄される\n# [A] Released

5.3 パフォーマンスへの配慮

コンテキストマネージャーは便利ですが、ループ内での使用はパフォーマンスに影響する可能性があります。

import time\n\n# ❌ 悪い例：毎回コンテキストマネージャーを作成\ndef slow_version(items: list):\n    for item in items:\n        with DatabaseConnection(\":memory:\") as conn:\n            cursor = conn.cursor()\n            cursor.execute(\"INSERT INTO data VALUES (?)\", (item,))\n\n# ✅ 良い例：外側でコンテキストマネージャーを作成\ndef fast_version(items: list):\n    with DatabaseConnection(\":memory:\") as conn:\n        cursor = conn.cursor()\n        for item in items:\n            cursor.execute(\"INSERT INTO data VALUES (?)\", (item,))\n        conn.commit()

5.4 コンテキストマネージャーとジェネレータのクリーンアップ

@contextmanagerを使う場合、yieldの前後で例外が発生する可能性を考慮する必要があります。

from contextlib import contextmanager\n\n@contextmanager\ndef safe_resource_manager():\n    \"\"\"例外安全なリソース管理\"\"\"n    resource = None\n    try:\n        # リソース獲得時に例外が発生する可能性\n        resource = acquire_expensive_resource()\n        print(\"[MGR] Resource acquired\")\n        yield resource\n    except Exception as e:\n        print(f\"[MGR] Error during acquisition or usage: {e}\")\n        raise\n    finally:\n        # リソースが取得できたかどうかに関わらず、必ずクリーンアップ\n        if resource is not None:\n            release_resource(resource)\n            print(\"[MGR] Resource released\")\n\ndef acquire_expensive_resource():\n    \"\"\"リソース取得（ここでは省略）\"\"\"n    return {\"connection\": \"dummy\"}\n\ndef release_resource(resource):\n    \"\"\"リソース破棄（ここでは省略）\"\"\"n    pass

6. 実務での導入のコツ

6.1 既存のコードをリファクタリング

既存の手動リソース管理コードをコンテキストマネージャーに置き換える際の例です。

## リファクタリング前
def old_style_processing():
    conn = sqlite3.connect(\":memory:\")\n    try:\n        cursor = conn.cursor()\n        cursor.execute(\"SELECT * FROM users\")\n        data = cursor.fetchall()\n        return data\n    finally:\n        conn.close()\n\n## リファクタリング後\ndef new_style_processing():\n    with DatabaseConnection(\":memory:\") as conn:\n        cursor = conn.cursor()\n        cursor.execute(\"SELECT * FROM users\")\n        data = cursor.fetchall()\n        return data

6.2 テスタビリティの向上

コンテキストマネージャーを使うと、テスト時にモック化しやすくなります。

from unittest.mock import MagicMock, patch\nimport pytest\n\ndef test_database_operation():\n    \"\"\"コンテキストマネージャーをモック化するテスト\"\"\"n    mock_conn = MagicMock()\n    mock_cursor = MagicMock()\n    mock_conn.cursor.return_value = mock_cursor\n    mock_cursor.fetchone.return_value = (1, \"Test User\")\n    \n    with patch('__main__.DatabaseConnection') as mock_db:\n        mock_db.return_value.__enter__.return_value = mock_conn\n        \n        result = fetch_user_data(1)\n        \n        assert result[\"name\"] == \"Test User\"\n        mock_cursor.execute.assert_called_once()

7. まとめ

Pythonのコンテキストマネージャーは、リソース管理を安全かつ簡潔に行うための強力な機能です。本記事で紹介したパターンは、実務で実際に使われているものばかりです。

重要なポイントを再度整理します：

  • 基本形：with文とクラスの__enter__/__exit__メソッドでリソース管理
  • シンプル実装：@contextmanagerデコレータとジェネレータで簡潔に書ける
  • 業務活用：DB接続、ファイル操作、ロック管理、トランザクション処理で活躍
  • 注意点：例外処理、ネスト時の順序、パフォーマンスへの配慮が必要

コンテキストマネージャーを適切に活用することで、バグが少なく、保守性の高いPythonコードが実現できます。ぜひ、プロジェクトに導入してみてください。

タイトルとURLをコピーしました