ライブラリ

Pydantic

概要

Pydanticは「Data validation using Python type hints」として開発されたPython向けの最高速データバリデーションライブラリです。Pythonの型ヒントを活用した直感的なAPI設計で、信頼できないデータを安全なPythonオブジェクトに変換します。Rustで書かれたコアロジックにより他のライブラリより最大10倍高速な処理を実現。FastAPI、LangChain、HuggingFaceなど8,000以上のPyPIパッケージで採用される、Pythonエコシステムの標準的なバリデーションライブラリです。

詳細

Pydantic v2は2025年現在の最新版で、Rustベースのpydantic-coreによる完全な書き直しにより大幅なパフォーマンス向上を実現しています。Python 3.9+の純粋で標準的な型ヒント記法でデータ構造を定義し、Pydanticが自動的にバリデーション・シリアライゼーション・デシリアライゼーションを処理します。StrictモードとLaxモードの選択、JSON Schema生成、SQLAlchemy統合など、データサイエンス・機械学習・Web API開発での豊富な機能を提供します。

主な特徴

• 最高速パフォーマンス: Rustコアによる他ライブラリ比10倍の処理速度
• Python型ヒント統合: 標準的な型ヒント記法による直感的なAPI
• 堅牢な型変換: 柔軟なデータ型強制変換と厳密な型チェック
• JSON Schema生成: OpenAPI/JSON Schemaの自動生成機能
• 豊富なエコシステム: FastAPI、SQLAlchemy、Django等との深い統合
• エラー詳細性: 詳細で分かりやすいバリデーションエラーメッセージ

メリット・デメリット

メリット

• Pythonエコシステムで最も広く使用される実績と信頼性
• Rustコアによる圧倒的なパフォーマンス（10倍高速）
• FastAPI連携による最強のWeb API開発環境
• データサイエンス・機械学習分野での豊富な活用事例
• 型ヒント記法による学習コストの低さ
• 充実した公式ドキュメントとコミュニティサポート

デメリット

• 大規模なプロジェクトでのメモリ使用量増加
• 複雑なネストしたモデルでの型推論パフォーマンス
• Python専用（他言語での相互運用性なし）
• カスタムバリデーターの学習コストがやや高い
• レガシーPythonバージョン（3.8以下）のサポート終了
• 一部の高度な機能で実行時オーバーヘッド

参考ページ

• Pydantic 公式ドキュメント
• Pydantic GitHub
• Pydantic PyPI

書き方の例

インストールと基本セットアップ

# Pydanticのインストール
pip install pydantic

# オプション：追加機能付きインストール
pip install "pydantic[email]"  # メールバリデーション
pip install "pydantic[dotenv]"  # .env設定読み込み
pip install "pydantic[all]"     # 全ての追加機能

# 開発環境用
pip install "pydantic[dev]"

基本的なスキーマ定義とバリデーション

from datetime import datetime
from typing import Optional, List
from pydantic import BaseModel, ValidationError

# 基本的なモデル定義
class User(BaseModel):
    id: int
    name: str
    email: str
    age: Optional[int] = None
    is_active: bool = True
    created_at: datetime = datetime.now()
    tags: List[str] = []

# データのバリデーション
user_data = {
    'id': '123',  # 文字列から自動的にintに変換
    'name': '田中太郎',
    'email': '[email protected]',
    'age': '30',  # 文字列から自動的にintに変換
    'tags': ['developer', 'python']
}

try:
    user = User(**user_data)
    print(user)
    print(f"ユーザーID: {user.id}, 型: {type(user.id)}")  # int型
    print(f"年齢: {user.age}, 型: {type(user.age)}")       # int型
except ValidationError as e:
    print(f"バリデーションエラー: {e}")

# モデルの辞書形式出力
print(user.model_dump())
# {'id': 123, 'name': '田中太郎', 'email': '[email protected]', ...}

# JSON形式出力
print(user.model_dump_json())

# 部分的なデータ更新
updated_user = user.model_copy(update={'age': 31, 'is_active': False})
print(updated_user)

# データクラス風の使用方法
class Product(BaseModel):
    name: str
    price: float
    in_stock: bool

# バリデーションのテスト
valid_product = Product(name="ノートPC", price=99999.99, in_stock=True)
print(valid_product.price)  # 99999.99

高度なバリデーションルールとカスタムバリデーター

from pydantic import BaseModel, Field, validator, EmailStr, ValidationError
from typing import Annotated
from annotated_types import Gt, Lt, Len

# 詳細なフィールドバリデーション
class UserProfile(BaseModel):
    username: Annotated[str, Len(min_length=3, max_length=20)] = Field(
        ..., 
        description="ユーザー名（3-20文字）"
    )
    email: EmailStr = Field(..., description="有効なメールアドレス")
    age: Annotated[int, Gt(0), Lt(150)] = Field(
        ..., 
        description="年齢（1-149歳）"
    )
    password: str = Field(..., min_length=8, description="パスワード（8文字以上）")
    bio: Optional[str] = Field(None, max_length=500, description="自己紹介（500文字以内）")

    @validator('username')
    def validate_username(cls, v):
        if not v.isalnum():
            raise ValueError('ユーザー名は英数字のみ使用可能です')
        return v.lower()

    @validator('password')
    def validate_password(cls, v):
        if not any(c.isupper() for c in v):
            raise ValueError('パスワードには大文字を含める必要があります')
        if not any(c.islower() for c in v):
            raise ValueError('パスワードには小文字を含める必要があります')
        if not any(c.isdigit() for c in v):
            raise ValueError('パスワードには数字を含める必要があります')
        return v

    class Config:
        # JSON Schema生成時の例
        schema_extra = {
            "example": {
                "username": "johndoe",
                "email": "[email protected]",
                "age": 30,
                "password": "SecurePass123",
                "bio": "Python開発者です"
            }
        }

# 複雑なネストしたモデル
class Address(BaseModel):
    street: str
    city: str
    postal_code: str = Field(..., regex=r'^\d{3}-\d{4}$')
    country: str = "日本"

class Company(BaseModel):
    name: str
    address: Address
    employees: List[UserProfile]
    founded_year: int = Field(..., ge=1800, le=2025)

# カスタムバリデーション例
company_data = {
    "name": "Tech Corp",
    "address": {
        "street": "1-1-1 丸の内",
        "city": "東京都千代田区",
        "postal_code": "100-0005"
    },
    "employees": [
        {
            "username": "Alice123",
            "email": "[email protected]",
            "age": 28,
            "password": "SecurePass123"
        }
    ],
    "founded_year": 2020
}

company = Company(**company_data)
print(f"会社名: {company.name}")
print(f"従業員数: {len(company.employees)}")

フレームワーク統合（FastAPI、Django、Flask等）

# FastAPIとの統合例
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, ValidationError

app = FastAPI()

class UserCreate(BaseModel):
    name: str
    email: EmailStr
    age: int = Field(..., ge=18, le=120)

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    created_at: datetime

@app.post("/users/", response_model=UserResponse)
async def create_user(user: UserCreate):
    # Pydanticモデルを使用した自動バリデーション
    # user.nameやuser.emailは型安全にアクセス可能
    new_user = UserResponse(
        id=123,
        name=user.name,
        email=user.email,
        created_at=datetime.now()
    )
    return new_user

# SQLAlchemyとの統合例
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()

class UserDB(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)

# PydanticとSQLAlchemyモデルの変換
class UserPydantic(BaseModel):
    id: int
    name: str
    email: str

    class Config:
        from_attributes = True  # SQLAlchemyオブジェクトから作成可能

# Django Rest Frameworkとの統合例
class ArticleSerializer(BaseModel):
    title: str = Field(..., max_length=200)
    content: str
    author_id: int
    published_at: Optional[datetime] = None

    class Config:
        # Djangoモデルとの互換性
        orm_mode = True

# Flaskとの統合例
from flask import Flask, request, jsonify

flask_app = Flask(__name__)

@flask_app.route('/api/users', methods=['POST'])
def create_user_flask():
    try:
        user_data = UserCreate(**request.json)
        # バリデーション済みデータの使用
        return jsonify({"status": "success", "user": user_data.dict()})
    except ValidationError as e:
        return jsonify({"status": "error", "errors": e.errors()}), 400

エラーハンドリングとカスタムエラーメッセージ

from pydantic import BaseModel, ValidationError, validator
from typing import List

class ProductOrder(BaseModel):
    product_name: str = Field(..., min_length=1, max_length=100)
    quantity: int = Field(..., gt=0, description="数量は1以上である必要があります")
    unit_price: float = Field(..., gt=0, description="単価は0より大きい必要があります")
    discount_rate: float = Field(0.0, ge=0.0, le=1.0, description="割引率は0-1の範囲")

    @validator('product_name')
    def validate_product_name(cls, v):
        if v.strip() != v:
            raise ValueError('商品名の前後に空白は使用できません')
        if any(char in v for char in ['<', '>', '&']):
            raise ValueError('商品名に特殊文字は使用できません')
        return v

    @validator('quantity')
    def validate_quantity(cls, v):
        if v > 1000:
            raise ValueError('一度に注文できる数量は1000個までです')
        return v

    def total_price(self) -> float:
        """合計金額の計算"""
        return self.quantity * self.unit_price * (1 - self.discount_rate)

# 詳細なエラーハンドリング例
def process_order(order_data: dict) -> ProductOrder:
    try:
        order = ProductOrder(**order_data)
        print(f"注文処理成功: {order.product_name}, 合計金額: {order.total_price():.2f}円")
        return order
    except ValidationError as e:
        print("バリデーションエラーが発生しました:")
        for error in e.errors():
            field = error['loc'][0] if error['loc'] else 'unknown'
            message = error['msg']
            value = error.get('input', 'N/A')
            print(f"  フィールド: {field}")
            print(f"  エラー: {message}")
            print(f"  入力値: {value}")
            print("---")
        raise

# エラーケースのテスト
invalid_orders = [
    {"product_name": "", "quantity": 1, "unit_price": 100},  # 商品名空
    {"product_name": "商品A", "quantity": 0, "unit_price": 100},  # 数量0
    {"product_name": "商品B", "quantity": 5, "unit_price": -100},  # 負の単価
    {"product_name": "商品C", "quantity": 5, "unit_price": 100, "discount_rate": 1.5},  # 無効な割引率
]

for i, order_data in enumerate(invalid_orders):
    print(f"\n=== エラーケース {i+1} ===")
    try:
        process_order(order_data)
    except ValidationError:
        pass  # エラーは既に出力済み

# カスタムエラーメッセージの設定
class CustomValidatedModel(BaseModel):
    name: str = Field(..., min_length=2, max_length=50)
    age: int = Field(..., ge=0, le=150)

    class Config:
        error_msg_templates = {
            'value_error.missing': '必須フィールドです',
            'value_error.str.max_length': '文字数が上限を超えています（最大{limit_value}文字）',
            'value_error.number.not_ge': '値が小さすぎます（{limit_value}以上である必要があります）',
        }

型安全性とTypeScript統合

from pydantic import BaseModel, Field
from typing import Dict, List, Union, Optional, Generic, TypeVar
from enum import Enum
import json

# Enum型の使用
class StatusEnum(str, Enum):
    PENDING = "pending"
    PROCESSING = "processing"
    COMPLETED = "completed"
    FAILED = "failed"

class Priority(int, Enum):
    LOW = 1
    MEDIUM = 2
    HIGH = 3
    URGENT = 4

# ジェネリクス型の使用
T = TypeVar('T')

class APIResponse(BaseModel, Generic[T]):
    success: bool
    data: T
    message: Optional[str] = None
    errors: Optional[List[str]] = None

class Task(BaseModel):
    id: int
    title: str
    description: Optional[str] = None
    status: StatusEnum = StatusEnum.PENDING
    priority: Priority = Priority.MEDIUM
    assignee_id: Optional[int] = None
    metadata: Dict[str, Union[str, int, float]] = Field(default_factory=dict)

# ジェネリクス型の使用例
task_data = {
    "id": 1,
    "title": "サンプルタスク",
    "status": "processing",
    "priority": 3,
    "metadata": {"estimated_hours": 5, "complexity": "medium"}
}

task = Task(**task_data)
response = APIResponse[Task](
    success=True,
    data=task,
    message="タスクが正常に作成されました"
)

print(f"タスク: {response.data.title}")
print(f"ステータス: {response.data.status}")
print(f"優先度: {response.data.priority}")

# JSON Schema生成（TypeScript型定義生成用）
schema = Task.schema()
print(json.dumps(schema, indent=2, ensure_ascii=False))

# TypeScript型定義の生成例（サードパーティツール使用）
# pip install pydantic-to-typescript が必要
"""
from pydantic2ts import generate_typescript_defs

generate_typescript_defs("path/to/models.py", "path/to/output.ts")
"""

# データクラスとの互換性
from dataclasses import dataclass
from pydantic.dataclasses import dataclass as pydantic_dataclass

@pydantic_dataclass
class DataClassUser:
    name: str
    age: int = Field(..., gt=0)
    email: Optional[str] = None

# 通常のdataclassと同様に使用可能、でもPydanticバリデーション付き
user = DataClassUser(name="田中", age=30, email="[email protected]")
print(user.name)  # 田中

# 設定とバリデーション
class DatabaseConfig(BaseModel):
    host: str = "localhost"
    port: int = Field(5432, ge=1, le=65535)
    database: str
    username: str
    password: str = Field(..., min_length=8)
    ssl_mode: bool = True
    pool_size: int = Field(10, ge=1, le=100)

    class Config:
        # 環境変数からの読み込み
        env_prefix = 'DB_'
        case_sensitive = False

# 環境変数 DB_HOST, DB_PORT などから自動読み込み可能
config = DatabaseConfig(
    database="myapp",
    username="admin",
    password="securepass123"
)

高度な機能とパフォーマンス最適化

from pydantic import BaseModel, Field, root_validator, validator
from typing import Any, Dict, List
import time
from decimal import Decimal

# カスタムデータ型とバリデーション
class Money(BaseModel):
    amount: Decimal = Field(..., decimal_places=2)
    currency: str = Field(..., regex=r'^[A-Z]{3}$')

    def __str__(self):
        return f"{self.amount} {self.currency}"

# 根本レベルバリデーション
class UserRegistration(BaseModel):
    username: str
    email: str
    password: str
    confirm_password: str
    terms_accepted: bool

    @root_validator
    def validate_passwords_match(cls, values):
        password = values.get('password')
        confirm_password = values.get('confirm_password')
        if password != confirm_password:
            raise ValueError('パスワードが一致しません')
        return values

    @root_validator
    def validate_terms_accepted(cls, values):
        if not values.get('terms_accepted'):
            raise ValueError('利用規約への同意が必要です')
        return values

# パフォーマンス測定
class PerformanceTest(BaseModel):
    name: str
    value: int
    metadata: Dict[str, Any] = Field(default_factory=dict)

def benchmark_validation():
    # 大量データのバリデーション性能テスト
    data_list = [
        {"name": f"item_{i}", "value": i, "metadata": {"index": i}}
        for i in range(10000)
    ]
    
    start_time = time.time()
    validated_objects = [PerformanceTest(**data) for data in data_list]
    end_time = time.time()
    
    print(f"10,000件のバリデーション時間: {end_time - start_time:.4f}秒")
    print(f"1件あたりの平均時間: {(end_time - start_time) / 10000 * 1000:.2f}ms")

# 実行
benchmark_validation()

# カスタムJSONエンコーダー
class CustomModel(BaseModel):
    timestamp: datetime
    price: Money
    
    class Config:
        json_encoders = {
            datetime: lambda v: v.isoformat(),
            Decimal: lambda v: float(v),
        }

# 部分的バリデーション（experimental）
class PartialUpdateModel(BaseModel):
    name: Optional[str] = None
    age: Optional[int] = None
    email: Optional[str] = None

    class Config:
        # 部分更新時のバリデーション設定
        validate_assignment = True
        allow_population_by_field_name = True

# StrictモードとLaxモードの切り替え
class StrictModel(BaseModel):
    number: int
    text: str

    class Config:
        # Strictモード: 型強制変換を行わない
        # Laxモード（デフォルト）: 可能な限り型変換を試行
        anystr_strip_whitespace = True
        validate_assignment = True

# Laxモード（デフォルト）
lax_model = StrictModel(number="123", text="  hello  ")  # 成功: "123" -> 123
print(lax_model.number, type(lax_model.number))  # 123 <class 'int'>
print(f"'{lax_model.text}'")  # 'hello' (空白が除去される)

# ValidationErrorの詳細解析
def analyze_validation_error():
    try:
        StrictModel(number="abc", text=123)
    except ValidationError as e:
        for error in e.errors():
            print(f"フィールド: {'.'.join(str(x) for x in error['loc'])}")
            print(f"エラータイプ: {error['type']}")
            print(f"メッセージ: {error['msg']}")
            print(f"入力値: {error['input']}")
            if 'ctx' in error:
                print(f"コンテキスト: {error['ctx']}")
            print("---")

analyze_validation_error()