月: 2026年8月

設定やシークレット管理で困っていませんか。ローカルでは動くのに本番で失敗した、誰かのAPIキーが誤って公開リポジトリに混入した、などの経験は多くの実務担当者にとって身近な問題です。本記事は「実務で確実に使える」手順とチェックリストを、具体例（.env／pydantic、クラウドシークレット、CIルール、ローテーション）を交えて提供します。

なぜ設定とシークレット管理が重要か（現実的リスク）

AIワークフローではAPIキーやモデル用トークンが多数発生します。これらが漏れると、予期せぬコスト発生、不正利用、顧客データ流出などの重大インシデントに直結します。よくある現実的リスク：

• 誤ってコミットされた.envファイル（公開リポジトリに流出）
• テスト環境のキーが本番で使用されていたため権限過剰になっていた
• ローテーションが未実施で長期間同一キーが使われていた

基本原則と設計方針

• 環境の分離（local / staging / production）を明示する
• 最小権限：キーには必要最小限のスコープだけ付与する
• 構成（config）とシークレット（secret）を明確に分ける
• 運用チェックリストを作り、CIで自動検査する

要件化の簡単な例

リスク	要件
公開リポジトリにキーが入る	git-precommitで秘密値検査、CIで再スキャン
テスト用と本番用の混同	環境ごとに設定を分離し、環境変数で明示
漏洩検出後の対処が遅い	インシデント対応フローとローテーション手順の整備

ローカル開発の典型パターン（.env + python-dotenv + pydantic）

ローカルではシークレットを直接ファイルで扱いがちですが、.envは開発専用に限定し、チームでの共有は避けます。安全な基本パターン：.env（ローカル・例外的）＋環境変数（CI/本番）＋型検証（pydantic）です。

サンプル：.env と python-dotenv の読み込み（最小例）

開発時のみ .env を使い、常に環境変数で上書きできるようにします。

# .env (例、絶対にコミットしない)
OPENAI_API_KEY=sk-xxxxx
ENV=local

pydantic の BaseSettings を使った型・バリデーションの例：

from pydantic import BaseSettings, Field

class Settings(BaseSettings):
    openai_api_key: str = Field(..., env='OPENAI_API_KEY')
    env: str = Field('production', env='ENV')

    class Config:
        env_file = '.env'
        env_file_encoding = 'utf-8'

settings = Settings()

ポイント：

• env_file を指定してローカルでの利便性を保ちつつ、本番では環境変数で上書きする運用にする
• 必須値に対しては Field(… ) で明示し、起動時に不足があれば例外を出す
• 型・制約を入れることで誤った値の混入を防ぐ

本番での安全な配置パターン（比較表）

本番環境ではファイルベースよりも環境変数やマネージドシークレットが望ましいです。以下は現場でよく選ばれる選択肢の比較です。

方式	長所	短所	推奨規模
環境変数（直接）	導入容易、ランタイムで読み取り簡単	プロセス内でアクセス可能な全員が見える	小〜中規模、単一ホスト
コンテナのシークレットマウント	ファイルとしてマウントし管理しやすい	シークレットをファイルとして扱うためアクセス管理が必要	中規模、コンテナ運用
Kubernetes Secrets	クラスタ管理、RBACや監査と連携可能	Base64エンコードで平文化のまま管理されるケースがあり運用が必要	中〜大規模、K8s利用
クラウド Secret Manager	自動ローテーション・監査・強力なアクセス制御	コストがかかる、導入運用の学習コスト	推奨（中〜大）

推奨パターン：小規模チームはまず「環境変数 + CIでの検査」から始め、中〜大規模は「クラウドSecret Manager + IAM/RBAC + ローテーション自動化」を目指すとよいでしょう。

クラウドシークレット管理の実務例（概要と最小実装）

ここでは AWS Secrets Manager の呼び出しの最小フロー例を示します。設計では「ローカルフォールバック」を用意しておくと便利です（例：本番はシークレットマネージャ、ローカルは環境変数）。

# boto3 を使った最小取得例（実際は IAM ロールでアクセス）
import boto3
import base64
from botocore.exceptions import ClientError

def get_secret(secret_name, region_name='us-east-1'):
    client = boto3.client('secretsmanager', region_name=region_name)
    try:
        resp = client.get_secret_value(SecretId=secret_name)
        if 'SecretString' in resp:
            return resp['SecretString']
        else:
            return base64.b64decode(resp['SecretBinary'])
    except ClientError:
        raise

実務ポイント：

• 呼び出しは最小限にし、アプリ起動時にキャッシュする（頻繁な API 呼び出しは避ける）
• IAM ポリシーは最小権限で Secret の読み取りのみを許可する
• ローテーションは可能なら自動化（Secrets Manager は Lambda 連携で自動ローテーション可）

CI とコード側での漏洩対策・テスト

コード側・CI パイプラインで自動検査を入れることで漏洩リスクを大幅に下げられます。代表的な対策：

• pre-commit フック（git-secrets など）によるコミット時スキャン
• CI ビルドでの再スキャン（push 時や PR 時）
• テスト環境は固定のモックキーを用意し、本物のキーは一切配備しない
• 証跡（audit）を残す：誰がいつキーを作成・参照したかをログに残す

pre-commit の簡単な設定例（概念）：

# .pre-commit-config.yaml の一部（例）
-   repo: https://github.com/awslabs/git-secrets
    rev: v1.3.0
    hooks:
    -   id: git-secrets

インシデント準備とキーのローテーション手順

鍵が漏れた場合の手順を事前に決めておくと対応が早くなります。基本フロー：

• 検出：自動検出（CI・監査ログ）または手動報告
• 無効化：該当キーを即時無効化（可能なら読み取り専用から取り消す）
• ローテーション：新しいキーを発行し、アプリへ段階的に展開
• 確認：動作確認とアクセスログの確認
• 対策：なぜ漏れたかを特定しルールを改訂

ロールプレイ用の簡易チェックリスト（自動化可能）

ステップ	実施内容
検出	CI スキャンアラート、監査ログ、セキュリティチーム報告
無効化	管理コンソール或いは API で即時無効化
新キー発行	Secret Manager などで新キーを発行し、必要権限を付与
展開	段階的に環境変数/デプロイで差し替え、ステージングで検証
レビュー	原因究明と再発防止策の実施

現場での導入手順（段階的チェックリスト）

小規模チームが今日から始められる短期〜長期ステップを示します。

期間	作業
短期（今週〜1ヶ月）	1) .gitignore に .env を追加 2) pre-commit で git-secrets 導入 3) pydantic で起動時バリデーション
中期（1〜3ヶ月）	1) CI に漏洩スキャンを追加 2) 本番は環境変数を利用 3) 簡易ローテーション手順を文書化
長期（3〜12ヶ月）	1) Secret Manager の導入 2) 自動ローテーションと監査ログの整備 3) RBAC の見直し

関連記事との接続（シリーズ案内）

本記事は「AIとPythonの実務」シリーズの一部です。関連回：

• 第126回：外部API連携（設定の読み替えポイント）
• 第137回：トークン管理（設計の一貫性）
• 第139回：推論API運用（運用面の補完）

まとめ

設定とシークレット管理は、AIワークフローの信頼性と安全性の基礎です。まずは小さな改善（.env の扱い、pydantic によるバリデーション、pre-commit/CI の導入）から始め、組織の成長に合わせて Secret Manager や自動ローテーションを導入するのが現実的な道筋です。最後に、必ず「検出→無効化→ローテーション→再発防止」のフローを文書化して運用に落とし込んでください。

次回は、実際に AWS Secret Manager を使って Lambda と連携する具体的なハンズオンを予定しています。シリーズを通じて、現場で使える手順を積み上げていきましょう。

はじめに — うまく説明できないエラーで時間を取られていませんか？

AIを業務に組み込むと、予期せぬエラーが発生しがちです。外部APIの一時障害、入力データの不整合、モデルの推論失敗、ファイルI/Oの問題……どれも現場では頻出です。しかし重要なのは“何が起きたかを説明できて、対応できること”です。本記事では、実務で使える例外設計とハンドリング手順を、コード例・ログ設計・テスト・運用チェックリストとともに示します。読了後には実装に移れるレベルを目指します。

問題提起：現場でよくある失敗パターン

カテゴリ	症状	原因（例）
外部API	タイムアウト、502/503	ネットワーク/レート制限/一時的障害
データ不整合	パース失敗、スキーマ違反	入力CSVの欠損・型違い
モデル推論	推論例外、メモリ不足	モデルの入力前提違反、リソース不足
ファイルI/O	読み書き失敗	権限、ディスク容量、同時アクセス

例外設計の基本

実務向けには、標準例外を乱用せず、業務ドメインに応じたカスタム例外を追加することを推奨します。命名規約は「場所＋原因＋Action」で分かりやすくします（例：CsvParseError、ApiTimeoutError、ModelInferenceError）。
下記は基本の階層例です。

基底	目的	例
AppError(Exception)	アプリケーション全体で共通の基底	ログやアラートの一括ハンドリング用
TransientError(AppError)	再試行で回復が期待できる一時的エラー	ApiTimeoutError、ServiceUnavailableError
PermanentError(AppError)	再試行しても意味がない恒久的エラー	InvalidInputError、DataSchemaError
ResourceError(AppError)	外部リソースに関わる問題	FileIOError、DatabaseError

振る舞いマップ（例外ごとの実務アクション）

次の表は例外種別と現場で望ましい振る舞いのマッピングです。実運用ではこれをルール化してワークフローに落とし込みます。

例外種別	現場アクション	ログ/メトリクス
TransientError（外部APIタイムアウト等）	自動再試行（指数バックオフ）、失敗後はアラート	retry_count、last_error_type
PermanentError（入力データ不備等）	処理をスキップしてユーザー通知/エラーレポート	bad_record_count、validation_errors
ResourceError（ディスク/DB権限）	即時停止＋オンコールへ通知、想定外ならロールバック	resource_status、failure_rate
ModelError（推論失敗）	再試行＋軽量なフォールバック（簡易モデル）または手動介入	model_failures、latency

実装ガイド

カスタム例外（dataclassベース）

from dataclasses import dataclass

class AppError(Exception):
    """アプリ全体の基底例外"""
    pass

@dataclass
class TransientError(AppError):
    message: str
    retry_after: float = 0.0

@dataclass
class PermanentError(AppError):
    message: str
    field: str = ""

例外にコンテキストを持たせる（traceback保持）

raise from を使い、元の例外を保持しておくと原因追跡が容易になります。

try:
    result = external_api.call(payload)
except TimeoutError as e:
    raise TransientError("API timeout", retry_after=2.0) from e

例外をラップしてドメインに翻訳するパターン

外部ライブラリの例外はそのまま流すのではなく、ドメイン例外へ翻訳します。

def fetch_with_wrap():
    try:
        return third_party.fetch()
    except third_party.HttpError as e:
        # 外部のHTTPエラーを自ドメインのTransientErrorに変換
        raise TransientError("third_party http error", retry_after=5.0) from e

ロギング＆可観測性連携

標準loggingで構造化ログを出し、SentryやPrometheusに必要な情報を渡します。ポイントは「何が」「どこで」「どの程度」の情報を一貫して出すことです。

構造化ログの例

import logging
logger = logging.getLogger(__name__)

def handle_error(exc, context):
    logger.error("processing_error", extra={
        "error_type": type(exc).__name__,
        "message": getattr(exc, 'message', str(exc)),
        "context": context,
    })

メトリクスの置き場所

Prometheus等には次のようなメトリクスを送ります：

• error_count{type=…}
• retry_count
• processing_latency_seconds

ログ/メトリクス	用途	閾値の目安
error_rate (5m)	全体の健全性	>1%で要調査
transient_error_rate	外部依存の劣化	急増は外部障害の兆候

テストとCI

例外処理はユニットテストとCIでの回帰チェックが必須です。外部依存はモックで一貫して疑似障害を作るとよいです。

ユニットテストの例（pytest）

def test_api_timeout(monkeypatch):
    def fake_call(_):
        raise TimeoutError("timeout")
    monkeypatch.setattr('external_api.call', fake_call)

    with pytest.raises(TransientError) as excinfo:
        fetch_with_wrap()
    assert excinfo.value.retry_after == 5.0

CIでの検証項目（例）

• 例外階層のドキュメントが最新か
• 主要な例外でのユニットテスト通過
• ログに必要なフィールドが出力されているかのスナップショット

デプロイ／運用チェックリスト

項目	確認内容
例外リスト	主要例外と期待される振る舞い（retry/skip/alert）がドキュメント化されている
アラートテスト	オンコールに通知されることをテスト済み
ログ/メトリクス	構造化ログと主要メトリクスが収集されている
ロールバック手順	障害時のロールバック手順が用意されている

小さなハンズオン：CSV一括推論パイプライン

以下は簡易サンプルです。CSVを読み、1行ずつ推論して結果を書き出す。APIタイムアウトは再試行、データ不備はスキップしてレポートします。

import csv
import time
import logging

logger = logging.getLogger(__name__)

class CsvParseError(PermanentError):
    pass

class ApiTimeoutError(TransientError):
    pass

def predict_with_retry(payload, max_retry=3):
    for i in range(max_retry):
        try:
            return external_api.predict(payload)
        except TimeoutError as e:
            if i == max_retry - 1:
                raise ApiTimeoutError("api timeout", retry_after=2.0) from e
            time.sleep(2 ** i)

def process_csv(in_path, out_path):
    bad_rows = []
    with open(in_path) as inf, open(out_path, 'w', newline='') as outf:
        reader = csv.DictReader(inf)
        writer = csv.DictWriter(outf, fieldnames=reader.fieldnames + ['prediction'])
        writer.writeheader()
        for row in reader:
            try:
                if not row.get('text'):
                    raise CsvParseError('missing text', field='text')
                pred = predict_with_retry(row['text'])
                row['prediction'] = pred
                writer.writerow(row)
            except PermanentError as e:
                logger.warning('skip_row', extra={'reason': e.message, 'row': row})
                bad_rows.append(row)
            except TransientError as e:
                logger.error('transient_failure', extra={'reason': e.message})
                raise
    return bad_rows

読後アクション（まず取り組む3つ）

• 自分のワークフローで発生しうる主要例外を一覧化する。
• 各例外に対して期待される振る舞い（retry/skip/alert）を書き出す。
• 1種類の例外（例：APIタイムアウト）を実装してユニットテスト＆デプロイする。

まとめ

実務で使えるエラーハンドリングは、例外を分類し「期待される振る舞い」を明文化することが出発点です。カスタム例外の設計、例外のラップ、構造化ログとメトリクス連携、そしてユニットテストとCIでの回帰検証をセットにすることで、何が起きたか説明でき、対応できる運用が作れます。本記事の表やチェックリストを自分のワークフローに当てはめ、まず一つの例外を実装・検証してみてください。

関連：第124回（再試行）、第126回（API連携）、第123回（可観測性）も合わせて参照すると、今回の設計を運用に落とし込みやすくなります。

はじめに — つまずきに寄り添って

標準ライブラリだけで仕事に使えるツールを作ろうとすると、「依存を減らしたい」「でも本当に標準で足りるのか？」と迷うことが多いはずです。本記事ではその迷いに寄り添い、まずは標準ライブラリだけで安全に、保守しやすく、実用的な小さなツールを作る手順とテンプレートを示します。

導入：なぜまず標準ライブラリを使うか

標準ライブラリを優先する理由は主に次の3点です。

• 依存管理が簡単（追加のパッケージ管理が不要）
• 保守性が高い（長期的互換性やドキュメントが豊富）
• セキュリティと信頼性（広くレビューされたコード群）

標準ライブラリで十分かを判断するチェックリスト

項目	標準で十分か？	コメント
HTTPクライアント	限定的	簡単なGET/POSTはurllibで可能。認証やセッション管理が複雑ならrequestsの検討を。
CSV/JSON処理	十分	標準のcsv/jsonでほとんどの業務処理はこなせる。
データ解析（大規模）	不十分	大量データや高度な集計はpandasの方が現実的。
非同期IOや高スループット	場合による	asyncioは強力だが、使いこなしコストを考慮。

まずは小さなPoCを標準ライブラリで作り、性能・可読性・開発コストを見てから外部依存を検討するのが実務的です。

ハンズオン：主要モジュールと短い実例コード

ここでは実務でよく使うモジュールを例とともに紹介します。各コードはそのまま貼って試せる最小例です。

モジュール	用途
pathlib	ファイル/パス操作（プラットフォーム非依存）
logging + RotatingFileHandler	ログ設計とローテーション
configparser / os.environ	設定管理（ファイル／環境変数）
csv / json	軽量データ入出力
itertools	ストリーム処理（メモリ節約の変換）
collections	defaultdict, Counter, deque などの便利構造
functools.lru_cache	簡易キャッシュ
subprocess	外部コマンド実行
tempfile / shutil	一時ファイルとアーカイブ

pathlib（パス操作）

from pathlib import Path
p = Path('data')
for f in p.glob('*.csv'):
    print(f.resolve())

logging + RotatingFileHandler（ログ設計）

import logging
from logging.handlers import RotatingFileHandler

logger = logging.getLogger('myapp')
handler = RotatingFileHandler('logs/app.log', maxBytes=10_000_00, backupCount=5)
formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.setLevel(logging.INFO)

logger.info('start')

configparser / os.environ（設定管理）

import configparser
import os

config = configparser.ConfigParser()
config.read('config.ini')
api_key = os.environ.get('API_KEY') or config.get('default', 'api_key', fallback=None)

csv / json（軽量データ入出力）

import csv
import json

# CSV読み取り
with open('input.csv', newline='', encoding='utf-8') as f:
    reader = csv.DictReader(f)
    for row in reader:
        print(row['id'], row['value'])

# JSON書き出し
with open('out.json', 'w', encoding='utf-8') as f:
    json.dump({'ok': True}, f, ensure_ascii=False, indent=2)

itertools（ストリーミング変換）

import itertools

def chunks(iterable, n):
    it = iter(iterable)
    while True:
        chunk = list(itertools.islice(it, n))
        if not chunk:
            break
        yield chunk

for c in chunks(range(10), 3):
    print(c)

collections（defaultdict / Counter / deque）

from collections import defaultdict, Counter, deque

cnt = Counter(['a','b','a'])
print(cnt.most_common())

d = defaultdict(list)
d['k'].append(1)

q = deque(maxlen=100)
q.append(1)

functools.lru_cache（キャッシュ）

from functools import lru_cache

@lru_cache(maxsize=128)
def expensive(x):
    # 重い計算や外部呼び出しの結果をキャッシュ
    return x * x

print(expensive(3))

subprocess（外部コマンド実行）

import subprocess

# 引数をリストで渡してシェルインジェクションを避ける
res = subprocess.run(['ls', '-la'], capture_output=True, text=True)
print(res.stdout)

tempfile / shutil（一時ファイルとアーカイブ）

import tempfile
import shutil
from pathlib import Path

with tempfile.TemporaryDirectory() as td:
    tmp = Path(td) / 'data.txt'
    tmp.write_text('hello')
    shutil.make_archive('archive', 'gztar', root_dir=td)

実務パターン：すぐ使えるテンプレート3本

ここではテンプレートごとに必要モジュール、設定例、ログ設定、最小テスト例を示します。

1) CLIバッチ用ユーティリティ（ファイル変換等）

• 主なモジュール：argparse / pathlib / csv / logging / configparser
• 設定ファイル例（config.ini）：

[default]
input_dir = ./data
output_dir = ./out
log_file = ./logs/app.log

• 簡易ログ設定：前述のRotatingFileHandlerを使う
• 最小テスト（pytest）例：

def test_cli_process(tmp_path):
    # 入出力のダミーファイルを作って処理が期待通り動くかを確認
    in_file = tmp_path / 'input.csv'
    in_file.write_text('id,value\n1,100\n')
    # process_cliは実装関数
    process_cli(str(in_file), str(tmp_path / 'out'))
    assert (tmp_path / 'out' / 'input.csv').exists()

2) ファイル監視→処理の小型ワーカー

• 主なモジュール：pathlib / time / logging / collections.deque / tempfile
• 構成：監視ループで新着ファイルをキュー（deque）に入れ、ワーカーが順次処理

from collections import deque
from pathlib import Path
import time

queue = deque()
watch = Path('inbox')
while True:
    for f in watch.glob('*.csv'):
        queue.append(f)
        f.rename(f.with_suffix('.processing'))
    if queue:
        process(queue.popleft())
    time.sleep(5)

• 設定は環境変数優先（os.environ）にして、config.iniはフォールバックとする
• テスト：ファイルの到着と処理開始をモックまたは一時ディレクトリで確認

3) 日次アーカイブ＋アップロードスクリプト

• 主なモジュール：shutil / tempfile / subprocess / logging / datetime
• 処理の流れ：古いファイルを集めて一時ディレクトリで圧縮→外部コマンド（curl/s3cli等）でアップロード→クリーンアップ

import shutil
from datetime import datetime
from pathlib import Path

src = Path('logs')
archive_name = f'logs-{datetime.today().date().isoformat()}'
shutil.make_archive(archive_name, 'gztar', root_dir=src)
# subprocessでアップロードコマンドを実行

既存ワークフローとの接続例

過去記事との連携例をテーブルで示します。

ステップ	該当記事	標準ライブラリでの実装例
CSV入出力	第134回	csv.DictReader → itertoolsでストリーム変換
CSV前処理	第135回	collections.Counterで簡易集計後、jsonで保存
推論API登録	第139回	subprocessやurllibでバッチ登録リクエスト

例：CSVクリーニング→itertoolsでストリーム処理→subprocessで外部ETL呼び出し→FastAPIにバッチ登録

import csv
import itertools
import subprocess
from urllib import request, parse

# 1. ストリームで読み、変換
with open('in.csv', newline='') as f:
    reader = csv.DictReader(f)
    cleaned = ({'id': r['id'], 'v': float(r['value'])} for r in reader if r['value'])
    # 2. バッチ処理（外部ETLを呼ぶ）
    for chunk in itertools.islice(cleaned, 1000):
        subprocess.run(['etl_tool', '--input', '-'], input=str(chunk).encode())
# 3. FastAPI に登録
data = parse.urlencode({'job': 'daily'}).encode()
req = request.Request('https://example/api/register', data=data)
request.urlopen(req)

運用チェックリスト

項目	確認ポイント
ログローテーション	適切なmaxBytes/backupCountが設定されているか
ログレベル運用	本番はINFO以上、デバッグ時にDEBUGを有効化する仕組み
例外ハンドリング	リトライポリシーとアラート（最低限stdout/exit codeで監視可能に）
一時ファイルのクリーンアップ	TemporaryDirectoryやtry/finallyで確実に削除
メトリクス出力	標準出力へ簡易メトリクス（処理数、エラー数）を出す

実装上の注意と落とし穴

• subprocessの呼び出しは引数をリストで渡し、ユーザー入力を直接渡さない（シェルインジェクション対策）。
• itertoolsでメモリ節約をしても、中間でlist化すると効果が消える。isliceやジェネレータを意識する。
• loggingはモジュール間でグローバルに影響するため、テストではログ設定のリセットを行う。
• 標準ライブラリで性能限界に達したら、代替ライブラリ導入の判断基準（処理時間、メモリ、開発コスト）を明確にする。

成果物と次の一歩

本記事で紹介したテンプレートはGitHub向けに最小リポジトリを用意すると実務導入が早まります。例：

ファイル／ディレクトリ	説明
README.md	導入手順と設定例
config.ini	サンプル設定ファイル（ENV優先）
src/	スクリプト本体（__main__.py とモジュール）
tests/	pytest用の最小テスト
.github/workflows/ci.yml	pytestを回すCI設定（例）

WordPressに貼りやすいコードブロック例（すでに本文中で複数提示）をそのままコピーして使ってください。

次回候補：itertools＋collectionsの深掘り、標準ライブラリだけで作る小さなCLIアプリ公開フロー。

まとめ

標準ライブラリは、依存管理や保守性、セキュリティ面で強い利点があります。まず標準でプロトタイプを作り、実際の負荷や運用を見てから外部ライブラリ導入を判断するのが実務的です。本稿のテンプレートやチェックリストを使えば、すぐに試せるツールが用意できます。まずは小さく始めて、運用で得た知見を元に改善していきましょう。

はじめに — つまずきに寄り添う一言

「ローカルでモデルは動くが、実際の業務にAPIとして出すと想定外の問題が出る」——こうした悩みはよく聞きます。本稿では、短時間で動く軽量な推論APIをFastAPIで設計・実装し、オンデマンドとバッチの両方を同一サービスで扱うパターンと、運用で必要なチェック項目を手順で示します。実務でそのまま使える観点を優先し、落とし穴と対処法も整理します。

1) 要件定義と設計（オンデマンド vs バッチ）

まず設計方針を明確にしましょう。用途によってSLA・レイテンシ・コスト感が変わります。

用途	要求	実装上の着眼点
オンデマンド推論（同期）	低レイテンシ（数百ms〜数秒）	非同期I/O、リクエストタイムアウト、スケール方針
バッチ推論（大量一括）	スループット重視、コスト効率	バッチング、バックグラウンドキュー、バッチサイズ制御
混在ケース	同一サービスで両立	優先度制御、リソース隔離、レート制限

設計時に決めるべき最小項目：

• SLA（最大許容遅延、成功率）
• 一回の推論当たりコスト想定
• 同時実行数とスケール戦略（垂直 vs 水平）
• 冪等性と再試行ポリシー

2) FastAPIの最小セットアップとエンドポイント設計

入力検証にはpydanticを使い、落ちる可能性を早期に検出します。ここではエンドポイント設計をテーブルで示します（コード例はポイントのみ記述）。

エンドポイント	メソッド	目的	備考（入力検証）
/predict	POST	オンデマンド推論（同期応答）	pydanticモデルで必須フィールド・型チェック
/predict_async	POST	非同期キュー登録（バッチング対象）	リクエストIDとコールバックURLを受け取る設計が実務向け
/batch/status/{job_id}	GET	バッチジョブの状態取得	ステータスとエラー情報を返す
/health, /ready	GET	ヘルスとレディネスチェック	監視用に簡潔な応答を返す

pydantic例（説明のみ）：入力モデルはサイズ制限、文字列長、必須キーを明示する。大きなペイロードは事前チェックで弾く。

3) バッチング戦略 — バックグラウンドキュー＋集約タイムウィンドウ

バッチ処理は「一定時間で集めて一度に処理する」か「サイズでトリガする」方式が基本です。ここでは簡易ワーカーと非同期キューの考え方を示します。

方式	特徴	向き不向き
時間ウィンドウ（例：100msごと）	レイテンシばらつきが小さくバッチ効率が良い	短めの応答要件を満たしつつスループットを稼ぎたい場合
サイズトリガ（例：batch_size=32）	バッチ効率が最大化されるが待ちが発生しうる	バッチ性能を最優先にする業務
ハイブリッド	どちらのしきい値でもトリガ可能	実運用でよく使われる

実装ヒント（要点のみ、簡潔に）：

• APIは受信時にリクエストを軽くバリデートし、非同期キューに格納する。
• バックグラウンドタスクがタイマーやサイズでバッチを切り出し、推論ワーカーに渡す。
• ワーカーは外部モデルAPI呼び出し／ローカルモデル推論を行い、結果を保存またはコールバック。
• 非同期実装はasyncio.QueueやRedis/RQを利用すると堅牢。単純構成ならuvicornのバックグラウンドタスクで十分。

4) エラー処理と再試行方針

実運用では予期せぬ例外、タイムアウト、外部API障害が起きます。方針を明確にしておきましょう。

• 例外ハンドラでHTTP 500を返す前に、詳細はログに残す。クライアントには説明的なエラーコードとメッセージを返す。
• 再試行は冪等性が担保できる場合のみ行う（重複結果の影響を考慮）。指数バックオフ＋最大試行回数を設定する。
• タイムアウトはAPIゲートウェイとアプリ双方で設定する（例：APIは短め、バッチジョブでは長めに）。
• 長時間処理のバッチはジョブ状態を保存し、失敗時に部分成功を扱えるようにする。

5) ロギング・メトリクス・ヘルスチェック

監視のための最低限のエンドポイントとログ設計を示します。

名称	用途	実装例（返す情報）
/health	プロセスが稼働しているか（簡易）	{“status”: “ok”}
/ready	依存（モデルロード、外部サービス接続）が整っているか	{“ready”: true, “model_loaded”: true}
/metrics	Prometheus互換のメトリクス公開	リクエスト数、エラーレート、レイテンシヒストグラム

ログ設計のポイント：

• 構造化ログ（JSON）でリクエストID、ジョブID、処理時間、エラー詳細を出力する。
• ログレベルは運用と開発で分ける。デバッグはファイル/外部でのみ残す。
• 例外はスタックトレースを追えるように保存しつつ、機密情報はマスクする。

6) Docker化と簡易CI

軽量APIはコンテナ化してデプロイするのが運用しやすいです。CIは次の流れを推奨します。

• pytestでFastAPI TestClientを使ったエンドポイントテストを用意する（/predictの成功系・異常系、バッチ登録の統合テスト）。
• Dockerfileは小さなベースイメージ（python:3.x-slim）を使い、依存は最小化する。
• CI（例：GitHub Actions）でテスト→イメージビルド→（必要なら）イメージのスキャン→registryへpushの流れを作る。

CIに含める最小ステップ（例）:

ステップ	目的
pytest実行	機能回帰チェック
lint（optional）	コード品質チェック
docker build & push	イメージ化とレジストリ反映

7) デプロイと運用チェックリスト

以下は現場ですぐチェックできる実用的な項目です。最後に表で整理します。

• リソース制限（CPU/メモリ）の設定とOOM対策
• レートリミットと優先度制御（オンデマンド優先／バッチはスロット確保）
• セキュリティ：TLS、認証、シークレット管理
• 監視：メトリクス収集、アラートルール（エラーレート、レイテンシ、スループット）

レベル	項目	チェック内容（そのまま使える）
必須	ヘルス/レディネス	/health と /ready を監視に登録し、復旧不能時は自動再起動
必須	リクエストタイムアウト	APIゲートウェイとアプリに最大タイムアウトを設定（例：30s/300s）
必須	ログレベル・フォーマット	JSONログでリクエストIDを必ず出力
推奨	メトリクス	Prometheus互換でリクエスト数/エラー/レイテンシを収集
推奨	リソース制限	コンテナごとにCPU/メモリ上限を設ける
発展	認証・シークレット管理	VaultやクラウドKMSでキー管理、APIはBearerトークンで保護

8) よくある落とし穴とトラブルシューティング

実務で遭遇しやすい問題と対処法を短くまとめます。

• 問題：バッチの遅延が増える。対処：ウィンドウ幅を短縮、優先度でオンデマンド優先にする。
• 問題：メモリ不足でOOM。対処：バッチサイズ制限、ワーカー数削減、モデルを軽量化。
• 問題：外部APIの不安定。対処：タイムアウト、回路遮断（circuit breaker）、キャッシュ導入。
• 問題：再試行で重複データ。対処：リクエストIDによる冪等性チェック、または重複排除ロジック。

まとめ

本稿では、FastAPIを使った軽量推論APIの設計と実装の考え方を、オンデマンドとバッチを一つのサービスで扱う観点から整理しました。重要なポイントは次の通りです：

• 設計段階でSLAとコスト感を明確にすること
• 入力検証（pydantic）と構造化ログで問題の早期把握を行うこと
• バッチは時間ウィンドウとサイズトリガのハイブリッドがおすすめで、非同期キューを使って安全に処理すること
• ヘルス/レディネス/メトリクスを整備し、CIでテスト・コンテナ化を自動化すること

最後に、運用チェックリストを実運用で回しながら改善を続けることが、安定稼働への近道です。次回は第138回で触れたテスト方針を踏まえ、具体的なpytestによるエンドツーエンドテストの例を紹介します。

AIワークフローを業務に組み込もうとすると、「モデルは動くけれど、変更すると突然動かなくなる」「外部APIの遅延で処理が止まる」「実運用データでの想定外ケースに気づけない」といった悩みに直面しがちです。本記事では、Pythonで実装した日常的なAIワークフロー（CSV前処理、API/LLM呼び出し、推論バッチ）を対象に、実務で使えるテスト戦略とCIの構成を手順とコード例で解説します。現場で再現しやすい具体例に沿って進めますので、まずは小さく始め、運用に耐える品質を徐々に高めていきましょう。

本記事の対象と目的

想定読者：AIを仕事で活用したい実務担当者、個人事業主、中小企業の担当者。目的は、既存のPythonワークフローにテストとCIを導入して運用リスクを下げ、変更時の回帰を防ぐことです。

章立て（概要）

• テスト戦略の決め方（ユニット／統合／エンドツーエンド）
• pytestを使った単体テストの書き方（fixtures, parametrize）
• 外部API/LLM呼び出しのモック手法
• テストデータの生成と管理（faker, factory_boy, CSVサンプル）
• 統合テストと小型データセットでの実行方針
• GitHub ActionsでのCIパイプライン例
• 運用時の注意点と失敗パターン
• 導入チェックリストとテンプレート／30分クイックスタート

1. テスト戦略（どのテストをどこまで書くか）

まずは役割を明確にします。小さなチームや一人運用では、全てのテストを同時に充実させるのは非現実的です。優先度を決めて段階的に増やすと良いです。

テスト種別	対象	速度	例	導入優先度（実務）
ユニットテスト	関数単位、前処理ロジック	速い	CSVの列変換、正規化関数	高
統合テスト	モジュール間連携、データパイプライン	中	CSV読み込み→前処理→推論ラッパー	中
エンドツーエンド（E2E）	本番に近いフロー（API/外部サービス含む）	遅い	小さなデータセットでのバッチ実行	低→定期実行推奨

実務的な優先度の目安

• まずユニットテスト：データ前処理、ビジネスロジック、スコア計算。
• 次に統合テスト：モジュールの連携、不整合を早期検出。
• E2Eは定期実行（夜間／週次）で本番寄せの検証を行う。

2. pytestを使った単体テストの書き方（基本とコツ）

pytestはfixtureやparametrizeで可読性の高いテストが書けます。ここではCSV前処理関数の例を示します。

CSV前処理（例）

def clean_row(row):
    # 入力: dict（CSVの1行）
    # 戻り値: 正規化された dict
    name = row.get("name", "").strip()
    age = row.get("age")
    try:
        age = int(age)
    except (TypeError, ValueError):
        age = None
    return {"name": name, "age": age}

pytestの例

import pytest
from myproject.csv_utils import clean_row

@pytest.mark.parametrize(
    "input_row,expected",
    [
        ({"name": " Alice ", "age": "30"}, {"name": "Alice", "age": 30}),
        ({"name": "", "age": "x"}, {"name": "", "age": None}),
    ],
)
def test_clean_row_param(input_row, expected):
    assert clean_row(input_row) == expected

@pytest.fixture
def sample_row():
    return {"name": " Bob ", "age": "45"}

def test_clean_row_fixture(sample_row):
    out = clean_row(sample_row)
    assert out["name"] == "Bob"
    assert isinstance(out["age"], int)

assertion のコツ

• 等価性は == を使い、浮動小数点は pytest.approx を使う。
• 複雑なオブジェクトは必要最小限に比較（キーのみや主要値のみ）。
• 失敗時に読みやすいメッセージが出るよう、pytestのassertをご活用ください。

3. 外部API/LLM呼び出しのモック

外部依存はテストを不安定にするため、ユニット/統合レベルで適切にモックします。代表的な手法と使い分けを下表に示します。

ライブラリ	用途	向く場面
unittest.mock	関数/メソッドの置換（軽量）	内部ラッパーの戻り値制御、エラー再現
responses	requestsによるHTTPレスポンスのモック	外部REST APIのユニットテスト
requests-mock	requestsのSession単位でのモック	細かいHTTP振る舞いを制御したい場合
VCR.py	実際のHTTP交流を録画して再生	既知のAPI応答をキャッシュして再現性を得たいとき

外部APIラッパーの例とテスト

# myproject/api_client.py
import requests

class APIClient:
    def __init__(self, base_url):
        self.base_url = base_url

    def get_score(self, payload):
        r = requests.post(f"{self.base_url}/score", json=payload, timeout=5)
        r.raise_for_status()
        return r.json()

# tests/test_api_client.py
from unittest.mock import patch
from myproject.api_client import APIClient

@patch("myproject.api_client.requests.post")
def test_get_score_mock(mock_post):
    mock_resp = mock_post.return_value
    mock_resp.json.return_value = {"score": 0.9}
    mock_resp.raise_for_status.return_value = None

    client = APIClient("https://api.example")
    out = client.get_score({"x": 1})
    assert out["score"] == 0.9
    mock_post.assert_called_once()

よりHTTP層で試したい場合は responses を使います：

import responses
from myproject.api_client import APIClient

@responses.activate
def test_get_score_responses():
    responses.add(
        responses.POST,
        "https://api.example/score",
        json={"score": 0.8},
        status=200,
    )
    client = APIClient("https://api.example")
    assert client.get_score({"x": 1})["score"] == 0.8

4. テストデータの生成と管理

良いテストは良いデータから生まれます。現実の代表ケースと境界値、エッジケースを意図的に含めましょう。個人情報（PII）を含む実データは必ず合成化するか、マスクして使用してください。

ルール	説明
代表ケース抽出	現場データから頻度の高いパターンを抽出してサンプル化
異常値／境界値	欠損、極端値、異常文字列（例: 半角／全角混在）を含める
PIIの扱い	本番データを使う場合は必ず匿名化または合成データで代替
小さなCSVサンプル	数十行の代表的セットを用意して統合テストに利用

faker と factory_boy の簡単例

from faker import Faker
fake = Faker()

def make_user_record():
    return {"name": fake.first_name(), "email": fake.email(), "age": fake.random_int(18, 80)}

生成したレコードはCSVに書き出して統合テスト用の小データとして利用できます。

小さなCSVサンプル設計（例）

行種別	説明
正常	代表的な1〜2行（標準的な列と値）
欠損	必須列が欠けている行（ageが空など）
境界	年齢=0、最大値、長い文字列など
異常	数値が文字列、特殊文字を含む列

5. 統合テストの実行方針

統合テストは重くなりがちなので、次の方針を推奨します。

• ローカルではユニットテストを高速に回す（pre-commit / pre-pushで実行）。
• 統合テストは小さなデータセットで速く回るように設計し、CIでは長いE2Eは別ジョブ・夜間に実行。
• 外部APIはスタブやVCRで再現性を担保。定期的に実際のAPIで罹患テストを走らせる。

統合テスト（ワークフロー実行）の簡単な例

# myproject/workflow.py
import csv
from myproject.csv_utils import clean_row
from myproject.api_client import APIClient

def run_batch(csv_path, api_client):
    results = []
    with open(csv_path, newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        for r in reader:
            r2 = clean_row(r)
            res = api_client.get_score(r2)
            results.append(res)
    return results

# tests/test_workflow_integration.py
from myproject.workflow import run_batch
from types import SimpleNamespace

def test_run_batch(tmp_path):
    csv_file = tmp_path / "sample.csv"
    csv_file.write_text("name,age\nAlice,30\nBob,\n")

    class DummyClient:
        def get_score(self, payload):
            return {"score": 0.5 if payload.get("age") else 0.0}

    out = run_batch(str(csv_file), DummyClient())
    assert len(out) == 2
    assert out[0]["score"] == 0.5
    assert out[1]["score"] == 0.0

6. GitHub ActionsでのCIパイプライン（テンプレート）

ここではPRで自動的にpytestを回し、カバレッジを計測して報告する最小テンプレートを示します。ファイル名は .github/workflows/ci.yml を想定してください。

name: CI

on:
  pull_request:
  push:
    branches: [ main ]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.10'
      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements-dev.txt
      - name: Run tests with coverage
        run: |
          pytest --junitxml=reports/junit.xml --cov=myproject --cov-report=xml
      - name: Upload coverage report
        uses: actions/upload-artifact@v4
        with:
          name: coverage-report
          path: coverage.xml

実務では coverage の結果を Codecov や Coveralls と連携したり、PR上で失敗を通知する仕組み（Reviewdog や GitHub Checks）を追加します。

7. 運用時の注意点と失敗パターン

問題	原因	対策
フレークテスト（不安定なテスト）	外部依存、時間依存、並列性の問題	依存のモック化、retry回避、テストの分離
テストが遅い	大きなデータセット、E2Eを頻繁に実行	ユニットと統合を分離、夜間にE2Eを実行
カバレッジ偏重	量だけ増えて中身が薄いテスト	重要ロジックに対する深いアサーション、コードレビューでの品質担保

品質指標とゲート条件（目安）

指標	目安
全体テストカバレッジ	70〜85%（業務重要度に応じて高める）
重要モジュール	90%前後を目指す
PRゲート	ユニットテスト全通＋カバレッジの変化が大きい場合は要確認

CIで失敗した時の簡単なエスカレーションフロー

• 1) PRに修正コミットを追加して再実行（まず試す）
• 2) それで直らない場合は失敗ログを貼って担当者に@メンション
• 3) 夜間のE2Eで発生した場合は運用チームにチケット登録し、翌稼働日までブロッキング回避策を適用

8. 導入チェックリストとテンプレート

項目	完了	注記
requirements-dev.txt に pytest, pytest-cov を追加	[ ]	最低限のテスト実行に必要
少なくとも1つのユニットテストを追加	[ ]	CSV前処理などビジネスロジック
CIワークフローを追加（.github/workflows/ci.yml）	[ ]	PRでテストが回るように
テストデータ（小さなCSV）をリポジトリに保存	[ ]	例: tests/data/sample.csv（PIIに注意）
外部APIをモックするガイドを作成	[ ]	responses や unittest.mock の例を含める

9. 30分でできるクイックスタート（実践手順）

1. requirements-dev.txt に次を追加：pytest, pytest-cov, responses、保存（5分）。
2. リポジトリに tests/test_csv_utils.py を1つ作成し、上記の clean_row テストを貼る（10分）。
3. .github/workflows/ci.yml を追加し、上記テンプレートを貼る（10分）。
4. PRを作って動作を確認。失敗したらログを見て修正して再コミット（5分）。

補助リソースと推奨ライブラリ（導入順）

順	ライブラリ	目的
1	pytest, pytest-cov	テスト実行とカバレッジ測定
2	unittest.mock	軽量なモック・パッチ
3	responses / requests-mock	HTTP APIのモック
4	faker, factory_boy	テスト用データ生成
5	VCR.py	HTTP録画再生（必要時）
6	GitHub Actions	CIの実行環境

まとめ

テストとCIは、運用リスクを低減し、変更時の安心感を高めます。現場では小さく始めることが重要です。まずは主要な前処理やスコア計算に対するユニットテストを1つ書き、CIで自動実行する流れを作ってください。次に統合テストを小さなデータセットで追加し、外部依存はモックで切り離します。最終的には定期的に実運用に近いE2Eを走らせることで、運用上の齟齬を早期に発見できます。

Manage AI（https://manageai.online）では、今回のような実務寄りの手順を今後も扱っていきます。まずはこの記事のクイックスタートに従い、既存リポジトリに1つのユニットテストとCIを追加してみてください。

次の一歩（推奨）：ローカルで pytest が通ることを確認したら、PR を作成して GitHub Actions で自動化される流れを体験してください。困ったときはログ全文を保存してチームで共有すると原因特定が速くなります。

はじめに — 現場でよくあるつまずきに寄り添う

AI導入を進めると、次のような問題で手が止まりがちです：コストが予想以上に増える、応答のばらつきで業務フローが不安定になる、同じ処理が再現できない。この記事では「テンプレート設計」「トークン予算管理」「キャッシュ／並列化」「ログと検証」「運用チェックリスト」を中心に、Pythonで実装できる実務手順を落ち着いた手順で示します。すぐ試せる課題と関数概要も最後にまとめます。

この記事の範囲（短く）

• プロンプトテンプレートの作り方と単体テスト化
• トークン見積もりとコスト試算の実務的手順
• レスポンスキャッシュの設計と簡易実装案（sqlite3／shelve）
• 並列・バッチ呼び出しの安全策とリトライ設計
• ログスキーマと応答検証、運用チェックリスト

1. テンプレート設計手順

目的は「入力変動に強く、再現性のあるプロンプト」を作ること。手順は次の通りです。

ステップ	具体的な作業	失敗しやすい点
入力正規化	文字コード・空白・日付フォーマットを統一。不要文字を削る。	想定外の入力形式を見落とすとテンプレートが壊れる
プレースホルダ定義	必須／任意の区別を明示。{user_text} のように命名。	曖昧な命名で誤った差し込みが起きる
例示の組み込み	期待する出力例をテンプレート内に含める（短い例で十分）。	例が長すぎるとトークンコストが増える
安全なフォーマット	Pythonのstr.formatや独自テンプレート関数で差し込み。直接連結を避ける。	未エスケープの入力で構文破壊が起きる
単体テスト化	代表ケース（正常、境界、異常）でテンプレート出力を比較。	テストカバレッジ不足で運用時に問題が顕在化する

実務メモ：テンプレート関数は「入力を受けて正規化→差し込み→結果を返す」責務だけにし、外部呼び出しでトークン見積りやキャッシュを行うと分離が明確になります。

テンプレートの検査ポイント（チェック表）

項目	確認方法
必須プレースホルダが埋まるか	単体テストで空文字やNoneケースを入れて確認
出力の一貫性	同一入力で複数回の出力差異を確認（乱数要素は抑制）
コスト影響	例示が長すぎないかトークンで試算

2. トークン予算管理（実務的見積もりと運用）

実務では厳密なトークン数の算出より「見積もり→検証→調整」のサイクルが重要です。

手順	実務上の注意点
単純推定	テンプレートの平均文字数を測り、1トークン=約4文字で粗算する
ライブラリ採用判断	正確な集計が必要ならトークンカウントライブラリを導入する（コスト計測用）
バッチサイズ設計	小さなバッチで試算→最適ポイントを見つける（応答遅延とコストのトレードオフ）

コスト試算（例）

モデル	平均プロンプトtokens	平均応答tokens	単価(1k tokens)	1件当たり概算
gpt-4-x	300	400	0.03 USD	(700/1000)*0.03 = 0.021 USD
gpt-3.5	200	150	0.002 USD	(350/1000)*0.002 = 0.0007 USD

実務メモ：高頻度バッチはモデルを混在させる運用（簡易タスクは安価モデル、重要タスクは高品質モデル）でコストと品質を両立できます。

3. キャッシュ／メモ化戦略

応答コストと遅延を下げ、再現性を上げるためにキャッシュは強力です。ただしPIIと整合性に注意する必要があります。

キャッシュキー設計の考え方

要素	説明
不変部分	テンプレートID、モデル名、温度など再現性のために必須
可変部分	入力テキスト（正規化後）、ユーザー固有フラグ（必要ならハッシュ化）
キー生成ルール	長い入力はSHA256等でハッシュ化してキーにする

TTLと無効化ルール

シナリオ	推奨TTL／無効化
静的説明文（頻繁に変わらない）	長めのTTL（数日〜数週間）
日次更新データに依存	短めのTTL（数分〜数時間）＋データ更新時に無効化フラグ
ユーザー固有応答（PII含む）	保存しないか、暗号化＋短TTL

簡易実装（方針説明）

小規模ならsqlite3やshelveを使ったファイルベースのレスポンスキャッシュが手早いです。実装方針は次のとおりです：

• キー列（ハッシュ）とシリアライズした応答、タイムスタンプを保存
• 取得時にTTLをチェックし、期限切れなら再取得して上書き
• 容量が増えたらLRU削除や最大件数で制限

注意点：キャッシュにPIIを入れない、あるいは暗号化する。複数プロセスでの同時更新は排他制御を入れる。

4. 並列・バッチ呼び出しとレート制御

バッチ化はAPIコール回数を減らし効率化しますが、レートや並列数に制限があるため設計が重要です。

手段	利点	選び方の目安
concurrent.futures（スレッド・プロセス）	同期コードに向く、導入が簡単	既存の同期処理や簡易な並列で十分なとき
asyncio	多数のI/O待ちがある場合に高効率	非同期設計に慣れている、長時間の多数接続で有利

リトライと冪等性の実務ルール

• 短時間のトランジェントエラーは指数バックオフでリトライ（上限回数を設定）
• リクエストにIDを付与して冪等性を担保（重複処理を防ぐ）
• サーキットブレーカーで連続失敗時は処理を保護する

短いコードスニペット（方針説明）：同一入力は先にキャッシュを確認→無ければAPI呼び出し→応答を保存。並列は最大Nスレッドで制御。

5. ログ設計と応答検証

運用で役に立つログは「追えること」が重要です。ログは必要最小限を構造化して残します。

推奨ログスキーマ

フィールド	例／説明
timestamp	ISO8601形式の発生時刻
template_id	使用したテンプレートの識別子
input_hash	入力のハッシュ（PIIは保存しない）
model	呼び出したモデル名
tokens_prompt/response	消費トークン数
cost	その呼び出しの概算コスト
response_status	成功／エラー／低信頼など

応答検証とアラート設計

検証項目	閾値／対応
エラー率	5分間で5%を超えたらアラート
低信頼応答比	定義した信頼スコアで10%以上なら調査
急激なトークン増	日次消費が予算の80%を超えたら通知

6. デプロイ／運用チェックリスト

項目	確認ポイント
テストケース	正常系・境界・エラー系を自動化しているか
ステージング負荷試験	実運用に近いバッチで負荷を確認
コストしきい値	日次・月次アラートしきい値を設定
ロールバック手順	旧バージョン復帰とデータ整合の手順を文書化
監査ログ	誰がいつ何を実行したか追えるようにする

7. ハンズオン：読後すぐに試せる3つの実践課題

課題	目的	所要時間の目安
テンプレート作成	入力正規化とプレースホルダ設計を実践	30〜60分
キャッシュ追加	sqlite3/shelveで簡易キャッシュを実装して効果測定	30〜90分
トークンコスト比較	同一タスクで複数モデルのコストを比較する	20〜60分

記事内で使う主要関数（概要）

関数名	目的	入力	出力
render_template	テンプレートに入力を差し込む	template_id, input_dict	生成されたプロンプト文字列
estimate_tokens	簡易トークン見積り（平均文字数から）	text	推定トークン数
response_cache_get/set	レスポンスキャッシュの取得／保存（TTL管理）	key, response, ttl	キャッシュヒット/保存結果

実装例の方針（擬似的に示す）：render_templateは入力正規化→必須チェック→str.formatで差し込み、estimate_tokensはlen(text)/4で概算、response_cacheはキーにsha256(hash)を用いるなどが実務的です。

まとめ

実務でAIを安定運用するには、テンプレートの堅牢さ、トークンの見積りと予算管理、キャッシュの安全な導入、並列化とリトライの慎重な設計、そして運用を支えるログと検証の仕組みが必要です。本記事で示したチェックリストと3つのハンズオン課題を順に実行することで、コストと品質のバランスを取りながら現場に導入しやすくなります。

次回は、この記事で使った小さなPythonサンプル（テンプレート関数・キャッシュデコレータ・トークン見積り関数）を具体的なコードとして掲載し、ステップバイステップで環境に組み込む方法を紹介します。

（シリーズ：AIとPythonの実務 ／ サイト：Manage AI）

はじめに — つまずきに寄り添って

現場で「とりあえず動く」スクリプトを書いた経験は多いはずです。しかし時間が経つと、同じ処理が別の場所でコピペされ、テストがなく、変更がこわくなります。本記事では第135回（CSVクリーニング）から自然につながる実践的な手順で、そうした“一発スクリプト”を再利用可能でテストしやすいコンポーネントに変える方法を示します。

設計原則（短く実務視点で）

単一責務（Single Responsibility）

関数は一つの目的だけを持ちます。読み込み・変換・書き出しは別々にし、組み合わせは上位の関数で行います。

純粋関数と副作用の分離

データ変換は入力を受け取り出力を返す純粋関数にし、ファイルやログなどの副作用は別モジュールにまとめます。こうするとユニットテストが容易になります。

依存注入

外部リソース（ファイルパス、DB接続、設定）は引数で渡すか、IOアダプターを介して渡します。テスト時はモックやスタブに差し替えます。

パターン実例：CSVクリーナーのリファクタ（before / after）

まず典型的な一発スクリプト（before）です。

実装メモ: コード例は環境に合わせて調整してください。例: # before: csv_cleaner.py

問題点：読み込み・変換・書き出しが混在。テストが難しい。

リファクタ後は3つの責務に分けます：pure functions（transform）、io_adapter（読み書きラップ）、cli（エントリポイント）。

実装メモ: コード例は環境に合わせて調整してください。例: # package layout (例)

この構成の利点：transformは純粋関数なのでユニットテストが容易。io_adapterをモックすれば統合テストもしやすい。

ディレクトリとパッケージ構成（推奨）

小規模プロジェクトの最低限の構成例：

実装メモ: コード例は環境に合わせて調整してください。例: mycsv/

__init__.pyで外部に公開する関数を明記し、内部実装は隠すと保守性が高まります。

テスト設計：pytestでの例

pure functionは通常のユニットテスト、IOはtmp_pathやモックで扱います。例：

実装メモ: コード例は環境に合わせて調整してください。例: # tests/test_transform.py

実行コマンド例：

• pip install -e .[dev]
• pytest -q

CIと品質ゲート（最低ライン）

テスト・型チェック・lintを最低限組み込みます。簡単なGitHub Actionsジョブ例：

実装メモ: コード例は環境に合わせて調整してください。例: # .github/workflows/ci.yml

pyproject.toml の最低例：

実装メモ: コード例は環境に合わせて調整してください。例: [project]

実務的チェックリスト

チェック項目	説明	判定基準
関数の責務は明確か	一つの関数が複数のことをしていないかを確認	変換はpure、I/Oは別モジュール
グローバル状態はないか	モジュールレベルの可変変数が無いか	無ければOK
I/Oは分離されているか	ファイルやDBアクセスが専用アダプターにあるか	モック可能であればOK
テストカバレッジの最低ライン	重要な変換ロジックに対するユニットテストの有無	変換ロジックは100%を目指す（現実的最低は80%）
後方互換性の扱い	API変更時の互換性維持方針があるか	破壊的変更はバージョニングで管理

段階的リファクタ計画と落とし穴

段階的に置き換える手順：

• 1) transformをpure関数として切り出し、既存スクリプトから呼び出せるようにする
• 2) io_adapterを作成して既存I/Oを置換する（動作確認は並列運用で）
• 3) testsを追加、CIで確認してからマージ

注意点：

• 過度な抽象化は避け、複雑化してしまう場合はスコープを縮小する
• 既存運用中のスクリプトはブランチ戦略で段階的に切り替える（トグル可能にする）
• 大規模CSVはメモリに全ロードせずチャンク処理を使う。transformは行単位に保つと組み合わせやすい

まとめと次の実践課題

本稿のポイントは次の通りです。

• 関数は単一責務にし、変換ロジックは純粋関数にする
• 副作用（I/O）は別モジュールにまとめ、依存注入やモックでテストしやすくする
• パッケージ構成と最低限のテスト・CIを整えることで運用コストを下げる

手を動かす練習（3ステップ）

1. リポジトリを作り、上記構成で最小限のファイルを作る（cli.py, transform.py, io_adapter.py）。
2. transformのユニットテストを書いてpytestで実行する（pytest -q）。
3. 簡単なGitHub Actionsワークフローを追加してPushでテストが回ることを確認する。

次回はこの基盤を使って、AIを組み合わせた自動データ正規化パイプラインに進みます。小さく始めて、確実に保守できる形にすることを優先してください。

はじめに — つまずきに寄り添って

表データを扱うと、文字化け、区切り文字の混在、見出しがずれている、日付形式がバラバラ、Excel特有の余計なセル……といった問題に直面します。AIや機械学習に投入する前にこれらをそのままにしておくと、結果が不安定になったりエラーで処理が止まったりします。本稿では、現場で繰り返し発生する問題に対し、再現性のある手順と小さな関数群で構成する実務的なクリーニングパイプラインを示します。まずは落ち着いて、段階的に確認していきましょう。

全体の流れ（概要）

本記事で示す手順は次の5ステップです。各ステップはログ出力と中間成果物の保存を前提にしており、失敗時はロールバック方針に従って安全に戻せるようにします。

ステップ	主な作業	出力物（例）
1) サンプリングによる事前点検	ファイル種別・encoding・delimiter・シート構造の判定	検査レポート（JSON）
2) 安全な読み込み	チャンク／ジェネレータ読み込み、エラー回避設定	標準化されたDataFrameストリーム
3) 型推定と正規化	数値・日付・カテゴリの推定と一貫化	スキーマ（JSON）
4) 欠損・異常値の実務対応	置換、補完、除外の方針決定と適用	クリーニング済データ（中間保存）
5) 出力とメタ情報	JSONL／圧縮出力、チェックサム、品質レポート	最終ファイル + メタ（checksum, schema）

小さな関数群で設計するメリット

関数を細かく分けると、例外処理・ログ・ユニットテストが組み込みやすくなります。ここでは推奨する関数名と役割を表にまとめます。

関数名	役割	エラー処理・戻り値
detect_encoding()	少量サンプリングで文字コードを判定（chardet等）	見つからなければ既定のutf-8を返す。例外はログ化してデフォルトにフォールバック
detect_delimiter()	CSVの区切り文字を推定（カンマ/タブ/セミコロン等）	候補とスコアを返す。スコアが低ければユーザー確認フラグを立てる
infer_schema()	カラムごとの型推定（数値/日付/カテゴリ/混在）	型推定結果と不確実性メタを返す。閾値未満はstring指定
normalize_dates()	日付を標準形式（ISO 8601）に変換	変換失敗行は別ファイルに分離しログ化
clean_numeric()	数値列の小数点・カンマ・通貨記号の除去とnull化	変換失敗はNaNにしてカウントを返す
write_canonical_jsonl()	正規化された行をJSONLで出力＋gzip圧縮＋チェックサム生成	チェックサムを返し、失敗時は中間ファイルを残してエラーコードを返却

ツールとライブラリの実務的な使い分け

標準ライブラリと外部ライブラリの利点・注意点を比較します。現場では「目的に合わせた使い分け」が大切です。

ツール	利点	注意点
csv (標準)	軽量・低依存。ストリーム処理が容易	複雑な型変換やExcelは不得手
pandas	複雑な変換・集計、Excel読み込みに強い	メモリ消費に注意。大ファイルはチャンク化必須
chardet / charset-normalizer	文字コード推定に有効	100%ではないのでサンプリング＋ルールが必要
python-dateutil	柔軟な日付解析	あいまい解析は誤解析の可能性あり。明示変換を優先
openpyxl	Excelの細かいフォーマット読み書きが可能	Excel固有の余計なセルや数式の扱いに注意

実務的チェックリスト（出力前に必ず実行）

自動化するべき基本的な品質チェックを示します。閾値を超えた場合はアラートを出し、手動確認を促します。

チェック項目	目的	例：閾値と対応
欠損率	データ欠落の程度を把握	列欠損率 > 30% → 列除外または収集元確認
ユニーク数	カテゴリ安定性の確認	想定より多すぎる（例：ID列が重複）→ 対象列の再評価
分布差異	期待分布からの逸脱検出	大きな差異→ ログ詳細出力・差分確認
日付整合性	未来日や極端に古い日付の検出	不自然な日付が一定割合以上→ 分離して手動確認

実務上の注意点（運用を見据えて）

• メモリ対策：大ファイルはチャンク読み込みと生成器で処理。pandasはchunksizeを活用。
• 冪等性：同一入力から同一出力を得るため、変換ルール・タイムゾーン・乱数シードを明示的に保存。
• 中間成果物保存：各ステップでメタ（schema, checksum, ログ）を保存し、失敗時は最後の安全な状態にロールバック。
• エラーと戻りコード：関数は例外を投げるだけでなくエラーコードと説明を返す仕様にすると運用が楽。

CIと自動化のヒント

テストケースは代表的な不良データ（文字化け、複数区切り、混在型日付、Excelの余白行など）を用意し、パイプラインの各関数に対してユニットテストを用意します。CIでは小さなサンプルを使って差分チェックとスキーマ整合性テストを実行します。

現場でよくある失敗パターンと対策（簡潔）

失敗例	原因	対策
文字化けで読み込めない	未知のエンコード	detect_encoding()でサンプリング判定＋明示的エンコード指定
区切り文字が混在	手作業で編集されたCSV	detect_delimiter()で上位候補を提示し、最も一貫した解析結果を採用
日付が複数形式混在	ユーザー入力や外部システムの差異	normalize_dates()でISO化し、変換不能行は分離して報告

シリーズとの接続（次の一歩）

本稿で得た正規化データは、第122回／129回で扱ったバッチ推論やスケジューリングに直接つなげられます。さらに第117回のテスト／CIや第123回の可観測性の仕組みに組み込むことで、運用の信頼性を高められます。

まとめ

現場の表データは想定外の欠陥を多く含みます。重要なのは「一度きりの手直し」ではなく、再現性のある手順とログ／メタ情報を残すことです。本稿で示した5ステップと小さな関数群をベースにすれば、AIや機械学習に安全にデータを渡すための堅牢なパイプラインを作れます。まずはサンプリング→安全読み込み→型の明確化→欠損・異常の方針適用→出力とメタ保存、の順に実装して、CIと品質チェックを組み込むことをおすすめします。

次回は、ここで作った正規化データをバッチ推論に流す際の実例とスケジューリング、監視設計について紹介します。

現場でのデータ入出力は地味に厄介です。文字化け、途中落ち、巨大ファイル、フォーマットの不揃い――これらに遭遇すると、時間だけが無情に消耗します。本記事では「実務でよくある失敗」を避ける具体的な手順と、すぐ試せるサンプルスクリプト（Excel→JSONL（分割・gzip・原本退避））を示します。前回（第133回：contextmanager）や第119回のpathlibの知識を活かして進めてください。

1) ファイル入出力の基本と安全化の原則

まずは基本的な用語整理と、現場で守るべき原則を示します。

項目	要点
モード（open）	テキスト（’r’,’w’,’a’）とバイナリ（’rb’,’wb’）を用途に応じて使い分ける。圧縮やバイナリ形式は必ずバイナリモード。
改行	テキスト読み込み時はnewline=”（csv用）やnewline=Noneの違いに注意。CRLF/CRの混在に備える。
エンコーディング	明示指定を原則（UTF-8）。Shift-JISやBOM付きUTF-8は注意が必要。自動検出は補助手段。
BOM	CSVやTSVの先頭BOMは読み取り時に除去する。utf-8-sigが便利。

実務ルール（簡潔）

• 読み取りは明示的なエンコーディングか検出を使う。
• 書き込みは一時ファイル→atomic replaceで原子性を担保する。
• 巨大ファイルはストリーミング処理（chunked）で扱う。

2) 原子書き込み・一時ファイル・ロールバックパターン

途中で失敗して中途ファイルが残ると自動処理は停止します。安定運用には原子更新パターンが必須です。

パターン	説明
一時ファイル + os.replace	処理完了後にos.replaceで上書き。途中で落ちても元ファイルは残る。
.inprogress拡張子	処理中は拡張子を付ける。完成後にリネームして通知・転送。
チェックサム/タイムスタンプ	受け渡しの整合性確認に使う（受信側が完全性を検証できる）。

3) CSV実務：csvモジュール vs pandas

CSVをどう扱うかはデータサイズと処理内容で決めます。以下は選択の目安です。

用途	推奨	理由
大規模ストリーミング（メモリに乗らない）	csvモジュール	逐次読みでメモリ効率が良い。chunksizeは自作のジェネレータで実現。
集計・列操作・高速開発	pandas	便利で高速。ただしメモリ使用量に注意。
フォーマット自動判定	csv.Sniffer	区切り文字の推定に有用。ただし不完全なサンプルだと失敗する。

CSVの実務ヒント

• 読み取りはnewline=”、エンコーディングは可能ならutf-8-sigでBOMを無視。
• 不正行が混ざる可能性がある場合はtry/exceptでログに倒す（行をスキップして続行するポリシーを可視化）。
• パイプ区切りなど非標準区切りにも対応できるようSnifferを併用する。

4) Excel実務：openpyxl/xlrdの使い分け

Excelはバイナリ寄りで型推定が厄介です。現場ルールを決めておくとぶれにくくなります。

目的	ライブラリ	備考
.xlsxの読み書き	openpyxl	読み取り専用モード（read_only=True）でメモリを節約。
.xlsの読み取り	xlrd（古い）	最近はxlsが減少。互換性を確認。
大量行の読み取り	openpyxlのiter_rows	ストリーミングで低メモリ読み取り可能。

Excelで注意すべき点

• 空セルの型（数値→空→文字列）で列型が不安定になる。必要なら明示的にキャストする。
• 複数シートやヘッダが不規則なファイルは事前に簡易ルール（ヘッダ行を固定）を設ける。

5) JSONLとLLM向け行指向フォーマットの扱い

LLM用などで行指向（JSONL）を使う場合、各行が独立したJSONであることと文字列のエスケープに注意します。

観点	対策
1行が大きすぎる	行サイズをチェックして分割或いはパート化する。
不正なJSON	書き込み前にjson.dumpsで検証してから出力。
スキーマ検証	軽量にキー存在チェックや型チェックを行う。JSON Schemaは重めなので、最初は簡易チェックから。

6) 圧縮・アーカイブとストリーミング処理

gzip/zip/zstdなど圧縮はI/Oとストレージのトレードオフです。ストリーミング対応の書き方でメモリを保ちます。

• gzipはpython標準で簡単。gzip.openを使ってバイナリで書く。
• zipfileは複数ファイルのアーカイブに便利だがランダムアクセスに注意。
• zstdは高速・高圧縮だが追加ライブラリが必要。

7) Parquet/バイナリフォーマットの導入判断

Parquetは分析向けに優れるが、導入コストと運用面（ライブラリ互換、クラウドとの親和性）を考慮します。

用途	Parquet向き?	理由
繰り返し読み取り・列選択が多い	はい	列指向で読み取りが高速、サイズも小さくなる。
単一CSVの単発変換	いいえ	変換コストと運用負荷が割に合わない場合がある。

8) 実践レシピ：Excel→JSONL（分割＋gzip）を安全に出力する完全スクリプト

以下は現場で使える最小限のワークフロー例です。ポイントはストリーミング読み、分割（行数ごと）、gzip圧縮、一時ファイル＋atomic replace、原本退避です。openpyxlを想定しています。

事前準備: pip install openpyxl

説明: このスクリプトは入力.xlsxの指定シートをiter_rowsで逐次読みし、指定行数ごとにJSONL（各行は1つのJSON）ファイルを作成、gzip圧縮して出力します。出力は一時ファイルに書いてからfinalに置換します。

from pathlib import Path
import json
import gzip
import tempfile
import os
from openpyxl import load_workbook

INPUT = Path('inputs/data.xlsx')
SHEET_NAME = 'Sheet1'
ROWS_PER_FILE = 10000
OUT_DIR = Path('out')
OUT_DIR.mkdir(parents=True, exist_ok=True)
BACKUP_DIR = Path('backup')
BACKUP_DIR.mkdir(parents=True, exist_ok=True)

def iter_rows_from_excel(path, sheet_name):
    wb = load_workbook(path, read_only=True, data_only=True)
    ws = wb[sheet_name]
    it = ws.iter_rows(values_only=True)
    headers = next(it)
    for row in it:
        yield dict(zip(headers, row))
    wb.close()

def atomic_write_gzip(json_lines, out_path: Path):
    # 一時ファイルに書いてから原子置換
    with tempfile.NamedTemporaryFile(dir=out_path.parent, delete=False) as tf:
        tmp_path = Path(tf.name)
    try:
        with gzip.open(tmp_path, 'wt', encoding='utf-8') as gz:
            for obj in json_lines:
                gz.write(json.dumps(obj, ensure_ascii=False) + '\n')
        os.replace(tmp_path, out_path)
    finally:
        if tmp_path.exists():
            try:
                tmp_path.unlink()
            except Exception:
                pass

def excel_to_jsonl_gzip(input_path):
    base = OUT_DIR / input_path.stem
    part = 0
    buffer = []
    for i, obj in enumerate(iter_rows_from_excel(input_path, SHEET_NAME), start=1):
        buffer.append(obj)
        if i % ROWS_PER_FILE == 0:
            part += 1
            out_path = base.with_suffix(f'.part{part}.jsonl.gz')
            atomic_write_gzip(buffer, out_path)
            buffer = []
    if buffer:
        part += 1
        out_path = base.with_suffix(f'.part{part}.jsonl.gz')
        atomic_write_gzip(buffer, out_path)

if __name__ == '__main__':
    # 原本退避
    backup_path = BACKUP_DIR / INPUT.name
    if not backup_path.exists():
        INPUT.replace(backup_path)
        # 作業用に原本を戻す（運用では移動/コピーのルールを選択）
        backup_path.replace(INPUT)
    excel_to_jsonl_gzip(INPUT)

ポイント補足：

• openpyxlのread_only=Trueとvalues_only=Trueでメモリを節約。
• json.dumps(…, ensure_ascii=False)で日本語を維持。
• atomic_write_gzipで一時ファイル→os.replaceの原子性を確保。
• 分割サイズはROWS_PER_FILEで調整。クラウド転送の上限やLLMの入力制約を踏まえて決める。

9) デバッグ・運用チェックリストとトラブルシューティング

チェック項目	推奨対応
ファイルサイズ	メモリに収まらない場合はストリーミング。分割出力を規定する。
エンコーディングの不一致	utf-8-sigやchardetで検出、ログを残し変換ルールを定める。
処理途中で落ちる	.inprogress拡張子・ロギング・リトライ設計で再実行可能にする。
圧縮済みをテキストで開く	必ずバイナリモードで扱う。gzip.openやzipfileを使用。
Excelの型変化	明示的にキャスト、必要ならスキーマ変換レイヤを用意。

運用観測ポイント（ログ/メトリクス）

• 処理した行数、出力ファイル数、失敗行数
• 処理時間とスループット（行/秒）
• ストレージ使用量と圧縮比

よくある失敗と対応（簡易まとめ）

失敗例	原因	対応
文字化け	エンコーディング誤指定	utf-8-sigを試す・検出して変換
行分割のズレ	CRLF/改行の混在	newline=”でcsv処理、改行正規化
中途ファイル残存	原子性未担保の書き込み	一時ファイル→os.replaceパターンを導入
圧縮ファイルをテキストで開く	バイナリモードの無視	gzip.open等でバイナリ／テキスト適切に扱う

パフォーマンス／コスト判断ガイド

ざっくりした選び方：

• 少量データで素早く処理：pandasで楽に実装。
• 大量データで低メモリ：標準csv/openpyxlのストリーミング。
• 頻繁に分析・列選択するならParquetを検討。

まとめ（この記事の要点）

実務でのファイル入出力は「小さい工夫」の積み重ねで安定します。主なポイントを再掲します：

• エンコーディングと改行を明示的に扱う（utf-8-sig、newline=”など）。
• 書き込みは一時ファイル＋atomic replaceで原子性を担保する。
• 巨大ファイルはストリーミング／分割処理を採用する。
• JSONLは1行1レコードを守り、書き出し前の検証を行う。
• 圧縮とバイナリフォーマット導入は目的とコストを比較して判断する。

次回はこの記事の運用面をさらに深め、入出力アーティファクトのメタデータ管理（マニフェスト、チェックサム、保持ポリシー）やS3等外部ストレージとの安全連携を取り上げます。シリーズ「AIとPythonの実務」として、ここで提示したワークフローを基盤にさらに自動化・監視を進めてください。

レビュー処理を作るとき、リソース漏れや例外時の不整合で運用に支障が出る――そんな経験はありませんか。この記事では第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