この記事は Qiita Tech Festa 2026 の投稿記事です。
【連載】レガシーWebシステムを改修する技術：Feature Toggle, シャドーイング, リファクタリング
- 第1回：環境構築とFeature Toggle（フラグ管理）編
- 第2回：typing.Protocolとシャドーイング（ダークローンチ）編（本記事）
- 第3回：仕様化テスト・TDD実践と足場のクリーンアップ編
- 第4回：LLM Agentを活用した高速・安全なリファクタリング実践編
はじめに
前回の第1回では、Feature Toggle に対応したローカル開発環境の構築を行いました。
連載第2回となる今回は、web_server/app/services.py を中心に、既存コードのロジック部分をできるだけ変えずに新旧ロジックを切り替えるための実装 に進みます。
実際の修正フローとの相違
トピックごとの説明にまとめた都合上、コードの変更を一度に行っていますが、実際には変更の前に第3回で実施するテストを用意してから変更のサイクルを繰り返してRed-Green TDDの形式で変更を進めていきます。
本題・実装：Protocol化とWrapper、シャドーイング
今回の戦略の肝は「静的ダックタイピング（typing.Protocol）による抽象化」と、本番データを使って裏側で新コードを検証する「シャドーイング（ダークローンチ）」です。
既存のレガシーコードから新しい実装へ安全に移行するための段階的なステップを、順を追って解説します。
STEP 0: 既存コード (副作用があり、テストがない状態。パッチによる仕様テストでカバー)
まずはリファクタリングの対象となる既存のレガシーコードです。便宜上出力だけにしていますが、外部のDBや古いAPIへ直接依存している「副作用」があるコードを想定しています。
class OldDataProcessingService:
def execute_pipeline(self, user_id: int, payload: dict) -> dict:
# 外部のDBや古いサードパーティAPIへ直接依存している「副作用」があるコードを想定
print(f"[SideEffect] Writing to Legacy DB for user {user_id}")
result_score = len(payload.get("data", "")) * 10
return {"status": "success", "legacy_id": user_id + 1000, "score": result_score}
実際にはクラスになってない場合や、Type Hinting が付いてない場合もありますが、次のステップでの対応のためこの点は最低限修正することになります。
STEP 1: typing.Protocol による抽象化の定義
次に、既存の OldDataProcessingService を直接変更することなく、「この形状（メソッドとシグネチャ）を満たすものは仲間とみなす」という静的ダックタイピングを定義します。
from typing import Protocol
class DataProcessingService(Protocol):
def execute_pipeline(self, user_id: int, payload: dict) -> dict:
...
これで同一のメソッドとシグネチャを持つクラスは差し替え可能な状態になります。
STEP 2: 新コード (Red-Green TDD で実装を進める対象)
そして、新しく置き換えるロジックを作成します。これは DataProcessingService プロトコルを満たすように実装し、テストが容易な状態を目指します。
class NewDataProcessingService:
def execute_pipeline(self, user_id: int, payload: dict) -> dict:
# テストが容易な新しいロジックコードの想定
# ※サンプルなので副作用ある出力ですが、あくまで想定ということで
print(f"[ModernI/O] Writing to New Microservice for user {user_id}")
data_str = payload.get("data", "")
# 新ロジック: 空文字のハンドリングなどをしてみました
result_score = (len(data_str) * 10) if data_str else 0
return {"status": "success", "legacy_id": user_id + 1000, "score": result_score}
STEP 3: 呼び出し箇所を一層ラップする「抽象化による分岐」層
最後に、これらを束ねるラッパー層を作成します。ここでFeature Toggle（機能フラグ）を取得し、シャドーイング（ダークローンチ）の制御を行います。
ここが今回の改修のキモです。単純に enabled で新ロジックと既存ロジックを切り替えるというのもよいのですが、shadow_mode で新旧の結果を比較しつつ、ユーザに旧の結果を返すというパスを入れることで開発者テストでは検知しづらかった内容を発見することや第4回で取り扱う予定のLLM Agentによるエンドポイント実行のテスト時の評価材料に扱えることを目的にしています。
import httpx
import logging
logger = logging.getLogger("refactoring")
class FeatureToggledDataProcessingService:
def __init__(self, toggle_url: str):
self.toggle_url = toggle_url
self.old_service = OldDataProcessingService()
self.new_service = NewDataProcessingService()
def _fetch_toggle_status(self) -> tuple[bool, bool]:
try:
# タイムアウトを設定してトグルサーバーが落ちていた場合、既存コードへフォールバック
response = httpx.get(self.toggle_url, timeout=0.2)
if response.status_code == 200:
data = response.json()
return data.get("enabled", False), data.get("shadow_mode", False)
except Exception as e:
logger.error(f"Toggle server error: {e}. Falling back to default (Old).")
return False, False
def execute_pipeline(self, user_id: int, payload: dict) -> dict:
enabled, shadow_mode = self._fetch_toggle_status()
# パターンA: 完全切り替え完了モード
if enabled and not shadow_mode:
return self.new_service.execute_pipeline(user_id, payload)
# パターンB: シャドウモード (裏側で新旧両方を実行して挙動比較、ユーザーには旧の結果を返す)
if shadow_mode:
old_result = self.old_service.execute_pipeline(user_id, payload)
try:
# 新コードのバグでメイン処理を巻き添えに落とさないよう、完全に隔離して実行
new_result = self.new_service.execute_pipeline(user_id, payload)
# 同値性の検証 (Differential Testing)
if old_result != new_result:
logger.warning(
f"[Mismatch Detected] user_id: {user_id}. "
f"Old: {old_result} vs New: {new_result}"
)
else:
logger.info(f"[Match Success] Shadow run matched for user_id: {user_id}")
except Exception as shadow_err:
logger.error(f"[Shadow Failure] New code crashed: {shadow_err}")
return old_result
# デフォルト: 既存コードのみ実行
return self.old_service.execute_pipeline(user_id, payload)
全体コード（完成版）
最終的に、これらを1つのファイル（services.py）にまとめた全体像は以下のようになります。
実際の運用では、責務ごとに別ファイルへ分割して管理することが多いですが、ここでは全体像を把握しやすくするために1つのファイルにまとめています。
import httpx
import logging
from typing import Protocol
logger = logging.getLogger("refactoring")
# =====================================================================
# STEP 0: 既存コード (副作用があり、テストがない状態)
# =====================================================================
class OldDataProcessingService:
def execute_pipeline(self, user_id: int, payload: dict) -> dict:
# 外部のDBや古いサードパーティAPIへ直接依存している「副作用」の塊と仮定
print(f"[SideEffect] Writing to Legacy DB for user {user_id}")
result_score = len(payload.get("data", "")) * 10
return {"status": "success", "legacy_id": user_id + 1000, "score": result_score}
# =====================================================================
# STEP 1: typing.Protocol による抽象化の定義
# =====================================================================
# 既存の OldDataProcessingService の宣言を変更することなく、
# 「この形状を満たすものは仲間とみなす」という静的ダックタイピングを定義。
class DataProcessingService(Protocol):
def execute_pipeline(self, user_id: int, payload: dict) -> dict:
...
# =====================================================================
# STEP 2: 新コード (Red-Green TDD で実装を進める対象)
# =====================================================================
class NewDataProcessingService:
def execute_pipeline(self, user_id: int, payload: dict) -> dict:
# 洗練され、テストが容易で、高速な新しいロジック
print(f"[ModernI/O] Writing to New Microservice for user {user_id}")
data_str = payload.get("data", "")
# 新ロジック: 空文字のハンドリングを強化し、計算式をモダナイズしたとする
result_score = (len(data_str) * 10) if data_str else 0
return {"status": "success", "legacy_id": user_id + 1000, "score": result_score}
# =====================================================================
# STEP 3: 呼び出し箇所を一層ラップする「抽象化による分岐」層
# =====================================================================
class FeatureToggledDataProcessingService:
def __init__(self, toggle_url: str):
self.toggle_url = toggle_url
self.old_service = OldDataProcessingService()
self.new_service = NewDataProcessingService()
def _fetch_toggle_status(self) -> tuple[bool, bool]:
try:
# タイムアウトを設定してトグルサーバーが落ちていた場合、既存コードへフォールバック
response = httpx.get(self.toggle_url, timeout=0.2)
if response.status_code == 200:
data = response.json()
return data.get("enabled", False), data.get("shadow_mode", False)
except Exception as e:
logger.error(f"Toggle server error: {e}. Falling back to default (Old).")
return False, False
def execute_pipeline(self, user_id: int, payload: dict) -> dict:
enabled, shadow_mode = self._fetch_toggle_status()
# パターンA: 完全切り替え完了モード
if enabled and not shadow_mode:
return self.new_service.execute_pipeline(user_id, payload)
# パターンB: シャドウモード (裏側で新旧両方を実行して挙動比較、ユーザーには旧の結果を返す)
if shadow_mode:
old_result = self.old_service.execute_pipeline(user_id, payload)
try:
# 新コードのバグでメイン処理を巻き添えに落とさないよう、完全に隔離して実行
new_result = self.new_service.execute_pipeline(user_id, payload)
# 同値性の検証 (Differential Testing)
if old_result != new_result:
logger.warning(
f"[Mismatch Detected] user_id: {user_id}. "
f"Old: {old_result} vs New: {new_result}"
)
else:
logger.info(f"[Match Success] Shadow run matched for user_id: {user_id}")
except Exception as shadow_err:
logger.error(f"[Shadow Failure] New code crashed: {shadow_err}")
return old_result
# デフォルト: 既存コードのみ実行
return self.old_service.execute_pipeline(user_id, payload)
シャドーイング（ダークローンチ）のメリット
shadow_mode が有効な間は、実際のユーザーリクエストを使って新旧両方のロジックが走ります。しかし、ユーザーには旧コードの結果だけが返るため、万が一新しいコードにバグやパニックがあってもシステムは停止しません。
ログに出力される Mismatch Detected を監視し、ゼロになるまでバグ修正を繰り返すことで、本番デプロイ時の安心感が増しますし、万一新コードに内部テストでは発生しなかったエラーが発生しても Feature Toggle を切り替えることで旧コードに切り戻しが可能です。
本構成 ( Feature Toggling + Shadowing )の注意点
Feature Toggling のために Feature Toggle サーバ、 Shadowing のために新旧ロジック、Wrap 層の呼び出しと、呼び出しに追加のコストがかかる状態になっています。そのため、特にステージング環境以降のリリースで パフォーマンスの監視は必須になります 。
今回は大本の個所からの切り分けをしていますが、パフォーマンスやリクエスト数に応じてより局所的な切り分けをしたり、Shadowing は開発環境までにしたり、それ以降の環境では一部のサーバプロセスだけ設定を変えてリリースする(カナリアリリース)などの仕組みが必要になります。
おわりに
第2回では、typing.Protocol を活用したインターフェースの抽象化と、Feature Toggleサーバーと連動したシャドーイング・ラッパーの実装を行いました。
次回の第3回では、この抽象化を最大限に活かした「仕様化テスト」と「TDD」の実践、そして本番移行後の「クリーンアップ」について記載します。