概要
複数のデータソース（DB、API、ファイル、SaaS）をLLMやRAG基盤に連携する際、Pythonでのデータ定義と受け渡しの設計は、保守性に関わります。型が統一されていないと、変換・検証・エラーハンドリングの責務が分散し、上流の仕様変更への対応が難しくなります。
DigitalBaseが社内AI基盤・データ連携の実装で標準としている設計指針は、入出力境界（API、DB、外部SaaS）にはpydanticモデルを置き、内部処理はdictで流す構成です。境界で型を確定させ、内部は柔軟に保つことで、データソースの増減や仕様変更に強い基盤を構築できます。
なぜ型定義が必要か
データ連携の運用では、次のような変更やデータの差異への対応が必要です。
- 上流のSaaSがフィールドをサイレントに変更する（
status→state） - DBのカラムが追加されるが、Python側は古い前提のまま動作する
- 文字列で来るはずの数値が、特定条件下でNullになる
- 日付フォーマットが拠点ごとに異なる
型定義の役割は、境界で必ず形を確定させ、内部では信頼してよい状態を作ることにあります。境界でバリデーションを一度通せば、下流の処理は値の妥当性を都度チェックする必要がなくなり、処理ロジックに集中できます。
型定義の3つの選択肢
データの取り回しには、用途に応じて3つの選択肢を使い分けます。
| 選択肢 | バリデーション | パフォーマンス | 主な用途 |
|---|---|---|---|
dict | なし | 最速 | 内部の中間データ、動的キー |
dataclass | なし（静的型のみ） | 速い | 内部のドメインモデル |
pydantic.BaseModel | あり（実行時） | やや重い | API/DB/外部I/Oの境界 |
実行時バリデーションは外部I/Oの境界で行い、内部の高頻度処理には持ち込まないことが基本方針です。
pydanticモデルの基本形
from datetime import datetime from pydantic import BaseModel, Field, field_validator class CustomerRecord(BaseModel): id: str name: str email: str company: str | None = None created_at: datetime revenue: float = Field(default=0.0, ge=0) @field_validator("email") @classmethod def normalize_email(cls, v: str) -> str: return v.strip().lower()
- 必須/任意を型で表現：
str | None = Noneで任意フィールドを明示する - Field でバリデーション：
ge=0、max_length=100などを宣言的に記述する - field_validator で正規化：trim・lowercase などの整形は型側で済ませる
正規化やバリデーションを型に寄せることで、レコードを受け取った下流のコードは整形済みであることを前提にできます。
dictでの取り回し
パイプライン内部でレコードを流す際は、dictが最も柔軟です。理由は次の3点です。
- オペレーターを汎用化できる：
filter・sort・aggregateは中身のスキーマを知らずに動作する - シリアライズコストが少ない：JSON/DBとの行き来がそのまま行える
- 動的フィールド追加が容易：上流で増えたカラムをそのまま下流に流せる
# 入口でpydantic検証 → 内部はdictで流す records: list[dict] = [ CustomerRecord.model_validate(raw).model_dump() for raw in raw_records ] # transform_filter filtered = [r for r in records if r["revenue"] >= 1_000_000] # transform_set for r in filtered: r["tier"] = "enterprise" if r["revenue"] >= 10_000_000 else "standard"
境界で一度 model_validate を通したあとは model_dump() でdict化し、以降のパイプラインは型に縛られず汎用的に処理します。
認証情報の扱い
外部APIの認証情報は、コードや設定JSONに直接記述しない設計が必要です。専用のVaultテーブル（またはシークレットマネージャー）を用意し、オペレーター側はキーのIDだけを参照する設計とします。
class HttpSourceConfig(BaseModel): url: str credential_id: str # vault のID参照のみ async def run_http_source(cfg: HttpSourceConfig) -> AsyncGenerator[dict, None]: cred = await vault.get(cfg.credential_id) # 復号して取得 headers = {"Authorization": f"Bearer {cred.token}"} ...
設定をそのままログやエラーメッセージに出力しても秘匿情報が漏れない構造になり、設定ファイルのバージョン管理やレビューも安全に行えます。
日付・タイムゾーンの正規化
拠点や外部システムごとに異なる日付表現は、境界で吸収します。原則は次の3点です。
- 境界では必ずタイムゾーン付きdatetimeへ変換し、UTCに揃える
- 内部はUTCで保持し、表示の直前にローカルタイムゾーンへ変換する
- DBもタイムゾーン付き型を用いる：PostgreSQLなら
TIMESTAMPTZ
from datetime import datetime, UTC def normalize_datetime(s: str) -> datetime: s = s.replace("/", "-") dt = datetime.fromisoformat(s) if dt.tzinfo is None: dt = dt.replace(tzinfo=UTC) return dt.astimezone(UTC)
タイムゾーンの扱いを境界に閉じ込めることで、内部ロジックは常にUTC前提で記述でき、拠点をまたぐ集計や比較のバグを防げます。
まとめ
社内AI基盤のデータ連携では、型検証と環境差への対応を入出力境界に集約します。設計の要点は次の2点です。
- 境界で型確定、内部はdict：pydanticは入口・出口に置き、内部はdictで柔軟に流す
- 認証・日付は型側で吸収：オペレーター本体に環境差や秘匿情報を持ち込まない
各オペレーターの責務を限定し、上流の変化を境界で吸収する構造にしておくことで、SaaSのAPI仕様変更や新フィールド追加に強い基盤を構築できます。