レビュー処理を作るとき、リソース漏れや例外時の不整合で運用に支障が出る――そんな経験はありませんか。この記事では第132回のレビューキュー設計の続きとして、コンテキストマネージャとデコレータを使い、資源管理と横断処理（ロギング・メトリクス・エラー変換など）を整理し、実務で運用できるハンドラを作る手順を示します。読み終えると、小さなプロダクション用ハンドラを組み込める水準を目指します。
導入：なぜコンテキストマネージャ／デコレータが有効か
実務では「確実に資源を解放すること」と「共通処理を中央集権化すること」が重要です。コンテキストマネージャはファイル・DB接続・ロック・一時ファイルなどの確実なクリーンアップを保証し、デコレータは横断的な関心事（ログ、認可、メトリクス、入力検証）を関数周辺に集中できます。第132回で設計したレビューキューにこれらを当てはめると、各ハンドラは最小限のビジネスロジックに集中し、運用やテストが容易になります。
コンテキストマネージャ実践編
基本の役割
- __enter__/__exit__：リソース確保と解放の場所。例外時も必ず実行される。
- contextlib.contextmanager：シンプルなジェネレータベースの実装に便利。
- トランザクション境界：明示的にコミット/ロールバックを扱う。
代表的パターンと使い分け
| 目的 | 同期実装（例） | チェックリスト |
|---|---|---|
| ファイル操作・一時ファイル | with open(…): / tempfile.TemporaryDirectory() | 必ずclose、例外での中断を想定、予期せぬ大容量を警告 |
| ファイルロック/DB行ロック | with FileLock(path): / with db.transaction(): | ロックのタイムアウト、デッドロック検出、ログ出力 |
| 外部セッション（HTTP/DB） | with requests.Session(): / with db.connect(): | セッション再利用、接続プール、接続の明示的閉鎖 |
同期コードテンプレ（簡易）
from contextlib import contextmanager
@contextmanager
def db_transaction(conn):
try:
conn.begin()
yield conn
conn.commit()
except Exception:
conn.rollback()
raise実務チェックリスト（同期）
- 例外時に必ずロールバック/解放されるか
- ロックのタイムアウトと再試行はどこで管理するか
- コンテキスト内で長時間処理がある場合の監視（タイムアウト/ハートビート）
- サイドエフェクト（外部API呼び出し等）は明示的に扱う
非同期版と互換性
async/awaitを使う場合はasync context manager（__aenter__/__aexit__やcontextlib.asynccontextmanager）を利用します。既存の同期コンポーネントを使いたいときはスレッドプールでラップするパターンが実務ではよく使われます。
簡単なasync例
from contextlib import asynccontextmanager
@asynccontextmanager
async def async_db_tx(async_conn):
await async_conn.begin()
try:
yield async_conn
await async_conn.commit()
except Exception:
await async_conn.rollback()
raise同期コンポーネントとの橋渡し
- run_in_executorでファイルI/Oや同期DBクライアントを非同期から呼ぶ
- 重要：ブロッキング処理はイベントループを塞がないよう明示的に分離
- テストでasync/sync双方のモックを用意する
デコレータ実践編
デコレータは関数の周辺に横断的処理を付与する最も分かりやすい方法です。実装ではfunctools.wrapsを必ず使い、メタ情報（__name__や__doc__）を保ちます。
代表的なデコレータと用途（表）
| 目的 | 説明 | 短い例 |
|---|---|---|
| timing | 処理時間を計測してメトリクス送信/ログ | @timing |
| auth（認可） | 実行前に権限チェックを行い、失敗時に早期リターン | @requires_role('reviewer') |
| idempotencyラッパ | 重複実行の検知と短絡的な成功返却 | @idempotent(key_fn) |
タイミングデコレータ（同期例）
import time
from functools import wraps
def timing(fn):
@wraps(fn)
def wrapper(*args, **kwargs):
start = time.time()
try:
return fn(*args, **kwargs)
finally:
elapsed = time.time() - start
print(f"{fn.__name__} took {elapsed:.3f}s")
return wrapper補助的なRetryの使い方
Retryは単独で「失敗を隠す」危険があるため、トランザクション境界やデータ整合性が担保できる場面で補助的に使います（124回の実装と重ならないよう、ここでは簡潔に言及）。
レビュー処理への統合例
ここでは、レビューキューからアイテムを取得して処理する際の全体テンプレを示します。コンテキストマネージャでロックとトランザクションを管理し、デコレータでメトリクス計測とエラーラッピングを付与します。
処理フローチャート（テーブルで概要）
| ステップ | 目的 | 失敗時の処置 |
|---|---|---|
| 1. キューからアイテム取得 | 処理対象の同定 | 取得失敗 → 再試行/ログ |
| 2. ロック取得（コンテキスト） | 二重処理防止 | ロック失敗 → 再キュー/アラート |
| 3. トランザクション開始 | DB整合性保護 | 例外 → ロールバック、再キュー判断 |
| 4. ビジネス処理（ハンドラ） | 実際のレビュー処理 | 例外 → エラーハンドラ、DLQへ送る判定 |
| 5. コミット/ロック解放 | 処理完了の確定 | コミット失敗 → ロールバック/再試行 |
ハンドラ全体テンプレ（同期）
from functools import wraps
def capture_exceptions(fn):
@wraps(fn)
def wrapper(*args, **kwargs):
try:
return fn(*args, **kwargs)
except Exception as e:
# エラー変換・ログ・メトリクス
print("handler error:", e)
raise
return wrapper
@capture_exceptions
def process_review(item, db_conn, lock):
with lock(item.id), db_transaction(db_conn) as conn:
# ビジネスロジック
result = do_review(item, conn)
return result失敗時の再キューと死活判定
- 短期エラー（ネットワーク等）：再試行カウントを増やしてキューに戻す
- 恒常的エラー（データ不整合等）：DLQ（死トピック）へ移動し運用で分析
- ロック取得失敗はすぐに再キュー、もしくはバックオフで再試行
テスト・デプロイの注意点
- ユニットテスト：モックでenter/exit、commit/rollback、ロックの取得/解放が呼ばれることを検証する。
- CI：破壊的な変更が入ったらDBマイグレーションや接続設定の動作を自動検証する。
- ステージングでの破壊試験：長時間ロック・高負荷キュー・部分的な外部API障害を再現。
- 観測性：処理時間、再試行回数、ロールバック回数をメトリクス化（第123回参照）。
運用上のベストプラクティスと落とし穴
| 項目 | 対策 |
|---|---|
| デコレータのネスト | 副作用の順序資産化（ログ→認可→トランザクションなど）とドキュメント化 |
| 隠れた遅延 | コンテキスト内での長時間処理を避け、タイムアウトを設定 |
| ログとメトリクスの分離 | 重要イベント（ロールバック/再試行）は構造化ログで残す |
| ロールバック戦略 | 部分コミットを避け、補償処理を用意する |
社内で簡単に使えるテンプレコードはライブラリ化し、ドキュメントとチェックリストをセットで配布すると実運用が楽になります。
まとめ
- コンテキストマネージャはリソース確実解放とトランザクション境界の明確化に有効。
- デコレータはロギング・認可・メトリクスなど横断的処理の集中化に向く。
- 非同期処理との連携はasync context managerやexecutorラップで対応し、ブロッキングを避ける。
- テストではenter/exitやcommit/rollbackの呼び出しを必ず検証し、ステージングで破壊試験を行う。
- 運用面ではログ・メトリクス・DLQ・再試行ポリシーを整備し、テンプレ化して共有する。
付録・実例コード一覧（貼りやすい簡易版）
同期コンテキストマネージャ（ファイルロック簡易）
import fcntl
from contextlib import contextmanager
@contextmanager
def file_lock(path):
with open(path, 'w') as f:
try:
fcntl.flock(f, fcntl.LOCK_EX)
yield
finally:
fcntl.flock(f, fcntl.LOCK_UN)asyncコンテキストマネージャ（簡易）
from contextlib import asynccontextmanager
@asynccontextmanager
async def async_temp_session(client):
await client.open()
try:
yield client
finally:
await client.close()代表的デコレータ（idempotencyの簡易版）
from functools import wraps
_seen = set()
def idempotent(key_fn):
def deco(fn):
@wraps(fn)
def wrapper(*args, **kwargs):
key = key_fn(*args, **kwargs)
if key in _seen:
return 'duplicate'
_seen.add(key)
return fn(*args, **kwargs)
return wrapper
return decoレビュー処理ハンドラの全体テンプレ（同期・簡易）
def handle_queue_item(item, db_conn, lock):
try:
with lock(item.id), db_transaction(db_conn) as conn:
result = process_business(item, conn)
return result
except TransientError:
requeue(item)
except PermanentError:
send_to_dlq(item)
except Exception:
alert_ops()
raise関連回／参考
- 第131回（async） — 非同期の基礎と注意点
- 第132回（HITLレビュー） — レビューキュー設計
- 第123回（観測性） — メトリクス設計の実務
- 第117回（ユニットテスト） — テスト設計の基本
次の学習ステップ：typing/dataclassesの復習、HITL運用の拡張、社内ライブラリ化の推進。公開予定日: 2026-08-19