Pydanticを使った型安全なデータ構造を持ったPythonコードのススメ

from datetime import datetime as dt
from pydantic import BaseModel, PositiveInt


class Order(BaseModel):
    id: int
    customer_name: str = 'ヤマダタロウ'
    ordered_at: dt | None = dt.now()
    price: PositiveInt = 1000


order_data = {'id': 111,
              'customer_name': 'ヤマダハナコ',
              'ordered_at': '2023-08-27 12:00:00',
              'price': 2000}

order = Order(**order_data)

print(order.model_dump())

$ python app.py
{'id': 111, 'customer_name': 'ヤマダハナコ', 'ordered_at': datetime.datetime(2023, 8, 27, 12, 0), 'price': 2000}

order_data = {'id': 'testID',
              'customer_name': 'ヤマダハナコ',
              'ordered_at': '2023-08-27 12:00:00',
              'price': 2000}

pydantic_core._pydantic_core.ValidationError: 1 validation error for Order

├── app.py # エントリーポイント
├── models # モデル
│   ├── base.py
│   └── product.py
├── schemas # リクエスト、レスポンスなどの定義
│   ├── __init__.py
│   └── product.py
└── services # ルーターと共にロジックを提供
    └── product.py

from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass

from sqlalchemy import Integer, String
from sqlalchemy.orm import Mapped, mapped_column

from pydantic import PositiveInt

from .base import Base


class Product(Base):
    __tablename__ = "products"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(255), nullable=False)
    price: Mapped[PositiveInt] = mapped_column(Integer, nullable=False)
    description: Mapped[str] = mapped_column(String(255), nullable=True)

from pydantic import BaseModel, ConfigDict


class ProductBase(BaseModel):
    name: str
    price: int
    description: str = None


class ProductCreate(ProductBase):
    pass


class ProductResponse(ProductBase):
    id: int

    # 今回は省略しているが、from_attributesはmodel_validateを使う設定
    model_config = ConfigDict(from_attributes=True)

from .product import ProductCreate, ProductResponse

from fastapi import APIRouter

import schemas
from models.product import Product

router = APIRouter()


@router.get("/{id}", operation_id="get_product")
async def get_product(id: int) -> schemas.ProductResponse:
    return Product(
        id=id,
        name=f"Product {id}",
        price=100,
        description=f"Description of product {id}",
    )


@router.post("", operation_id="create_product")
async def create_product(product: schemas.ProductCreate) -> schemas.ProductResponse:
    return Product(
        id=2,
        name=product.name,
        price=product.price,
        description=product.description,
    )

from fastapi import FastAPI

from services import product

app = FastAPI()

app.include_router(product.router, prefix="/product", tags=["product"])

{
  "name": "Product 1",
  "price": 100,
  "description": "Description of product 1",
  "id": 1
}

$ curl -X POST -H 'Content-type: application/json' --data '{"name":"Product2", "price": 200, "description": "Description of product 2"}' http://localhost:8080/product

{
  "name": "Product2",
  "price": 200,
  "description": "Description of product 2",
  "id": 2
}

$ curl -X POST -H 'Content-type: application/json' --data '{"price": 200, "description": "Description of product 2"}' http://localhost:8080/product

{
  "detail": [
    {
      "type": "missing",
      "loc": [
        "body",
        "name"
      ],
      "msg": "Field required",
      "input": {
        "price": 200,
        "description": "Description of product 2"
      },
      "url": "https://errors.pydantic.dev/2.3/v/missing"
    }
  ]
}

from pydantic import BaseModel, computed_field


class Product(BaseModel):
    unit_weight: int
    quantity: int

    @computed_field
    @property
    def total_weight(self) -> int:
        return self.unit_weight * self.quantity

print(Product(unit_weight=10, quantity=5).model_dump())
# => {'unit_weight': 10, 'quantity': 5, 'total_weight': 50}

from pydantic import ValidationError, validate_call


@validate_call
def say_hello(name: str, age: int = 17) -> str:
    return f"Hello, {name}! You are {age} years old."

print(say_hello("Tanabe", 34))
print(say_hello("Hanako"))

try:
    print(say_hello("Kei", "Suke"))
except ValidationError as e:
    print(e)

Hello, Tanabe! You are 34 years old.
Hello, Hanako! You are 17 years old.
1 validation error for say_hello
1
  Input should be a valid integer, unable to parse string as an integer [type=int_parsing, input_value='Suke', input_type=str]

from pydantic import ValidationError, BaseModel


class Product(BaseModel):
    price: int

print(Product.model_validate({"price": '100'}))

try:
    Product.model_validate({"price": '100'}, strict=True)
except ValidationError as e:
    print(e)

price=100
1 validation error for Product
price
  Input should be a valid integer [type=int_type, input_value='100', input_type=str]

from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict
import os


class MySQLSettings(BaseSettings):
    host: str
    port: int
    user: str
    password: str
    database: str


class Settings(BaseSettings, case_sensitive=True):
    api_key: str = Field('xxx', alias='app_api_key')
    mysql: MySQLSettings


os.environ['mysql'] = '{"host": "localhost", "port": 3306, "user": "testuser", "password": "p@ssw0rd", "database": "testdb"}'

print(Settings().model_dump())

{
  "api_key": "xxx",
  "mysql": {
    "host": "localhost",
    "port": 3306,
    "user": "testuser",
    "password": "p@ssw0rd",
    "database": "testdb"
  }
}

try:
    # Hoge
except ValidationError as e:
    print(e)

# e.errors = 見つかったエラー一覧
# e.error_count = エラーの数
# e.json = JSON形式