実務でPythonを使っていると、「似た処理が何度も現れる」「一度作った処理のテストや運用が難しい」と感じることはありませんか。この記事では、クラス設計のパターンを、CSVクリーナー、推論ラッパー、Pipelineのコードで示します。標準ライブラリを中心に実装し、pytestによるテストまで確認します。
なぜクラスを使うか（現場メリット）
関数の集まりでも十分なケースは多いですが、状態やリソースの管理、初期化コストの節約、モックしやすいインターフェースを作る点でクラスは有用です。以下の表は代表的な利点と注意点です。
| 利点 | 現場で役立つ理由 | 注意点 |
|---|---|---|
| 状態管理 | 重いリソースを初回だけロードして再利用できる（モデル、セッション） | メモリやマルチプロセスを考慮する必要あり |
| リソース管理 | ファイルやセッションを確実に開放しやすい | 誤ったライフサイクル設計でリークする |
| テスト性 | モックしやすいインスタンスメソッドで外部依存を切れる | 複雑な状態はテストを書く負担を増やす |
状態を持つコンポーネント vs ステートレス関数
どちらを選ぶかは次の観点で判断します。
- 初期化コストが高い処理（モデルロード、外部接続）には、状態を持つクラスを検討する。
- 受け取ったデータだけを変換する処理には、ステートレス関数を使う。
- テストやモックのしやすさを優先する場合は、外部依存をコンストラクターから注入できるようにする。
リソース管理：context managerを使う
ファイルやセッションは明示的に開閉する必要があります。context manager（withステートメント）を使うと、処理中に例外が発生した場合もファイルを閉じられます。
例1：ファイルハンドルを安全に扱うCSVクリーナークラス
以下のコードをcomponents.pyとして保存します。入力CSVの各文字列から前後の空白を取り除き、別のCSVへ保存する例です。
# components.py
import csv
from threading import Lock
class CSVCleaner:
def __init__(self, input_path, output_path, encoding="utf-8"):
self.input_path = input_path
self.output_path = output_path
self.encoding = encoding
self._input_file = None
self._output_file = None
def __enter__(self):
self._input_file = open(
self.input_path,
"r",
encoding=self.encoding,
newline="",
)
try:
self._output_file = open(
self.output_path,
"w",
encoding=self.encoding,
newline="",
)
except Exception:
self._input_file.close()
self._input_file = None
raise
return self
def __exit__(self, exc_type, exc, traceback):
try:
if self._output_file is not None:
self._output_file.close()
finally:
if self._input_file is not None:
self._input_file.close()
return False
def clean(self):
if self._input_file is None or self._output_file is None:
raise RuntimeError("CSVCleaner must be used with a with statement")
reader = csv.DictReader(self._input_file)
if reader.fieldnames is None:
raise ValueError("CSV header is required")
writer = csv.DictWriter(
self._output_file,
fieldnames=reader.fieldnames,
)
writer.writeheader()
count = 0
for row in reader:
cleaned_row = {
key: value.strip() if isinstance(value, str) else value
for key, value in row.items()
}
writer.writerow(cleaned_row)
count += 1
return count
次のようにwith構文で使用します。clean()の途中で例外が発生した場合も、__exit__が入出力ファイルを閉じます。
from components import CSVCleaner
with CSVCleaner("raw.csv", "clean.csv") as cleaner:
processed_count = cleaner.clean()
print(processed_count)
合成（composition）と継承（inheritance）の使い分け
一般の実務では合成を優先します。合成は役割ごとにコンポーネントを分け、再利用とテストを容易にします。継承は明確に「is-a」の関係があるとき、またはフレームワークの拡張時に限定して使います。
| 選び方 | 合成（composition） | 継承（inheritance） |
|---|---|---|
| 使う場面 | 異なる振る舞いを組み合わせるとき | 共通の振る舞いを拡張するとき |
| メリット | 変更に強く、テストしやすい | コードの重複を減らせるが複雑になりがち |
実践例：推論ラッパーとPipeline
次は、モデルを必要になるまでロードしない推論ラッパーと、CSV読み込み、前処理、推論、保存を合成するPipelineクラスです。以下のコードは、先ほど作成したcomponents.pyの末尾へ追記します。
例2：推論ラッパークラス（モデルロード／predict）
class ModelWrapper:
def __init__(self, loader):
self.loader = loader
self._model = None
self._load_lock = Lock()
def _get_model(self):
if self._model is None:
with self._load_lock:
if self._model is None:
self._model = self.loader()
return self._model
def predict(self, records):
model = self._get_model()
return model.predict(records)
loaderには、引数なしでモデルを返す関数を渡します。モデルは最初のpredict()呼び出し時にロードされ、その後は同じインスタンスが再利用されます。ロックが保護するのはロード処理であり、各モデルのpredict()自体がスレッドセーフであることを保証するものではありません。
例3：Pipelineクラス（CSV読み込み→前処理→推論→保存）
class Pipeline:
def __init__(self, model, preprocessor):
self.model = model
self.preprocessor = preprocessor
def run(self, input_path, output_path, encoding="utf-8"):
with open(input_path, "r", encoding=encoding, newline="") as input_file:
reader = csv.DictReader(input_file)
if reader.fieldnames is None:
raise ValueError("CSV header is required")
fieldnames = list(reader.fieldnames)
if "prediction" in fieldnames:
raise ValueError("Input CSV already has a prediction column")
original_rows = list(reader)
processed_rows = [
self.preprocessor(row) for row in original_rows
]
predictions = list(self.model.predict(processed_rows))
if len(predictions) != len(original_rows):
raise ValueError(
"The number of predictions must match the number of input rows"
)
output_fieldnames = fieldnames + ["prediction"]
with open(output_path, "w", encoding=encoding, newline="") as output_file:
writer = csv.DictWriter(
output_file,
fieldnames=output_fieldnames,
)
writer.writeheader()
for row, prediction in zip(original_rows, predictions):
output_row = dict(row)
output_row["prediction"] = prediction
writer.writerow(output_row)
return len(original_rows)
Pipelineは、predict(records)を持つオブジェクトと、CSVの1行を前処理する関数を受け取ります。特定の機械学習ライブラリには依存していません。
Pipelineをローカルで実行する
動作確認用として、次の内容をinput.csvへ保存します。
id,amount
1,80
2,120
続いて、以下をrun_pipeline.pyとして保存します。
from components import ModelWrapper, Pipeline
class ThresholdModel:
def __init__(self, threshold):
self.threshold = threshold
def predict(self, records):
return [
"high" if record["amount"] >= self.threshold else "low"
for record in records
]
def load_model():
return ThresholdModel(threshold=100.0)
def preprocess(row):
return {"amount": float(row["amount"].strip())}
model = ModelWrapper(loader=load_model)
pipeline = Pipeline(model=model, preprocessor=preprocess)
processed_count = pipeline.run("input.csv", "output.csv")
print(f"processed: {processed_count}")
components.py、input.csv、run_pipeline.pyを同じディレクトリへ置き、次の順序で実行します。
python run_pipeline.py
実行後のoutput.csvは次の内容になります。
id,amount,prediction
1,80,low
2,120,high
CSVが存在しない場合のFileNotFoundError、数値へ変換できない場合のValueError、モデル内部の例外は呼び出し元へ伝わります。ここでは広い例外を握りつぶさず、バッチ処理やAPIなどの呼び出し側でログ記録、再試行、処理中断を選べるようにしています。
テストとモックの方針
外部依存（ファイル、HTTP、モデル）はモックで切り離し、入出力の最小契約をテストします。ここではpytestのtmp_pathと、標準ライブラリのunittest.mock.Mockを使います。
次のコードをtests/test_components.pyとして保存します。
import csv
from unittest.mock import Mock
from components import CSVCleaner, ModelWrapper, Pipeline
def test_csv_cleaner_closes_files_and_strips_values(tmp_path):
input_path = tmp_path / "raw.csv"
output_path = tmp_path / "clean.csv"
input_path.write_text(
"id,name\n1, Alice \n",
encoding="utf-8",
)
with CSVCleaner(input_path, output_path) as cleaner:
assert cleaner.clean() == 1
with output_path.open("r", encoding="utf-8", newline="") as file:
rows = list(csv.DictReader(file))
assert rows == [{"id": "1", "name": "Alice"}]
def test_model_wrapper_loads_model_only_once():
loaded_model = Mock()
loaded_model.predict.side_effect = [["first"], ["second"]]
loader = Mock(return_value=loaded_model)
wrapper = ModelWrapper(loader=loader)
assert wrapper.predict([{"amount": 1}]) == ["first"]
assert wrapper.predict([{"amount": 2}]) == ["second"]
loader.assert_called_once_with()
assert loaded_model.predict.call_count == 2
def test_pipeline_writes_predictions(tmp_path):
input_path = tmp_path / "input.csv"
output_path = tmp_path / "output.csv"
input_path.write_text(
"id,amount\n1,120\n",
encoding="utf-8",
)
model = Mock()
model.predict.return_value = ["high"]
pipeline = Pipeline(
model=model,
preprocessor=lambda row: {"amount": float(row["amount"])},
)
assert pipeline.run(input_path, output_path) == 1
model.predict.assert_called_once_with([{"amount": 120.0}])
with output_path.open("r", encoding="utf-8", newline="") as file:
rows = list(csv.DictReader(file))
assert rows == [
{"id": "1", "amount": "120", "prediction": "high"}
]
pytestが利用できる環境で、プロジェクトのルートディレクトリから次のコマンドを実行します。
python -m pytest
テストのポイントは次の通りです。
ModelWrapperのloaderはMockへ差し替えられ、ロード回数を検査できる。- Pipelineに渡すモデルもMockにできるため、実際のモデルをロードせず入出力を確認できる。
- ファイル入出力には
tmp_pathを使い、テスト終了後に残るファイルを減らす。
シリアライズ・設定・バージョン対応（dataclass併用例）
設定をdataclassで型付きにし、JSONで保存する例です。長期運用では、設定データのバージョンを明示し、読み込めないバージョンを検出できるようにします。
| 項目 | 実務での推奨 |
|---|---|
| 設定管理 | dataclassとJSONを使い、設定バージョンを明示する。環境変数で上書きする場合は優先順位も文書化する |
| シリアライズ | 長期保存するデータでは、Python固有の形式だけを前提にせず、JSONなどでスキーマを管理する |
| モデルファイル | ファイル名にバージョンを含め、manifestでメタ情報を管理する |
次のコードは、設定の保存と読み込みを行う最小例です。
# config.py
import json
from dataclasses import asdict, dataclass
from pathlib import Path
@dataclass(frozen=True)
class AppConfig:
model_version: str
threshold: float
schema_version: int = 1
def save_config(config, path):
Path(path).write_text(
json.dumps(asdict(config), ensure_ascii=False, indent=2),
encoding="utf-8",
)
def load_config(path):
data = json.loads(Path(path).read_text(encoding="utf-8"))
if data.get("schema_version") != 1:
raise ValueError("Unsupported configuration schema version")
return AppConfig(**data)
config = AppConfig(model_version="v1", threshold=100.0)
save_config(config, "config.json")
loaded_config = load_config("config.json")
print(loaded_config)
load_config()は未対応のschema_versionを検出するとValueErrorを送出します。実運用で環境変数による上書きを追加する場合は、JSONを読み込んだ後に適用し、その優先順位を明示してください。
運用チェックリストとデプロイ時の注意点
実務で詰まりやすい点をチェックリスト形式でまとめます。
| カテゴリ | チェック項目 |
|---|---|
| 初期化コスト | 重いモデルは遅延ロードやwarmupを行っているか |
| メモリ／プロセス | マルチプロセスやコンテナでモデルを使うときのメモリ増加を確認したか |
| シリアライズ | 特定バージョンのPythonやライブラリだけで読める形式を長期互換の前提にしていないか |
| 設定管理 | 環境変数とファイルの優先順位を文書化したか |
| ログ・メトリクス | 処理時間、エラー率、入力サイズを計測しているか |
| 障害時 | 再起動手順、再実行時の重複対策、ロールバック手順があるか |
まとめと次の一歩
この記事の要点は次の通りです。
- 状態を持つ処理（モデル、セッション）はクラスにして、初回ロードやライフサイクルを管理する。
- 合成を優先してコンポーネントを小さく保ち、テストとモックで外部依存を切り離す。
- context managerでファイルや接続を安全に閉じ、設定はdataclassとJSONで明示的に管理する。
- 例外を無条件に握りつぶさず、ログや再試行を担当する呼び出し側へ伝える。
読了後のアクション提案：
components.pyとrun_pipeline.pyを保存し、サンプルPipelineをローカルで実行する。tests/test_components.pyを追加し、python -m pytestでテストする。- 業務の小さな処理を1つ選び、外部依存を注入できるクラスへ分割する。
- デプロイ前に、ログ、設定の優先順位、再実行手順、バージョン互換性を確認する。
Manage AI（https://manageai.online）では、次回以降で運用・点検軸をさらに掘り下げます。まずは再利用可能な小さなコンポーネントを作り、実装からテストまでの流れを確認してみてください。