SQLModel チートシート

Pythonでデータベースをシンプルかつ型安全に扱う

SQLModelとは？

SQLModelは、PydanticとSQLAlchemyを組み合わせた最新のPython ORMライブラリです。 型ヒントを活用したシンプルなコードで、データベース操作を安全かつ効率的に実装できます。

型安全 シンプル FastAPI連携 自動マイグレーション

基本操作

インストール 

pip install sqlmodel

SQLAlchemy、Pydanticも自動でインストールされます

モデル定義 

from sqlmodel import SQLModel, Field
from typing import Optional

class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    email: str
    age: Optional[int] = None

Pythonのクラスでテーブル構造を定義

データベース/テーブル作成 

from sqlmodel import create_engine

# エンジン作成
engine = create_engine("sqlite:///database.db")

# すべてのテーブルを作成
SQLModel.metadata.create_all(engine)

モデル定義から自動でテーブルを生成

セッション管理 

from sqlmodel import Session

# セッション作成
with Session(engine) as session:
    # データベース操作をここに記述
    session.commit()  # 変更を保存

withブロックで自動的にクローズ

CRUD操作

Create（作成） 

user = User(name="田中太郎", email="tanaka@example.com", age=25)

with Session(engine) as session:
    session.add(user)
    session.commit()
    session.refresh(user)  # DBから最新データを取得
    print(f"作成されたユーザーID: {user.id}")

Read（読み取り） 

from sqlmodel import select

with Session(engine) as session:
    # 全件取得
    statement = select(User)
    users = session.exec(statement).all()

    # ID指定で1件取得
    user = session.get(User, 1)

Update（更新） 

with Session(engine) as session:
    user = session.get(User, 1)
    user.age = 26
    session.add(user)
    session.commit()
    session.refresh(user)

Delete（削除） 

with Session(engine) as session:
    user = session.get(User, 1)
    session.delete(user)
    session.commit()

データ検索・絞り込み

条件検索（WHERE） 

from sqlmodel import select

with Session(engine) as session:
    # 年齢が25歳以上のユーザー
    statement = select(User).where(User.age >= 25)
    users = session.exec(statement).all()

    # 複数条件（AND）
    statement = select(User).where(
        User.age >= 20,
        User.age <= 30
    )
    users = session.exec(statement).all()

並び替え・件数制限 

from sqlmodel import select, col

with Session(engine) as session:
    # 年齢の降順で並び替え
    statement = select(User).order_by(col(User.age).desc())

    # 最新10件のみ取得
    statement = select(User).limit(10)

    # 組み合わせ
    statement = select(User).where(User.age >= 20).order_by(User.name).limit(5)

部分一致検索（LIKE） 

from sqlmodel import select, col

with Session(engine) as session:
    # 名前に「田中」を含むユーザー
    statement = select(User).where(col(User.name).contains("田中"))

    # 名前が「山」で始まるユーザー
    statement = select(User).where(col(User.name).startswith("山"))

OR条件 

from sqlmodel import select, or_

with Session(engine) as session:
    # 年齢が20歳未満または30歳以上
    statement = select(User).where(
        or_(User.age < 20, User.age >= 30)
    )

リレーション（1対多）

モデル定義

from sqlmodel import SQLModel, Field, Relationship
from typing import Optional, List

class Team(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str

    # リレーション: 1つのチームは複数のユーザーを持つ
    members: List["User"] = Relationship(back_populates="team")

class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    team_id: Optional[int] = Field(default=None, foreign_key="team.id")

    # リレーション: 1人のユーザーは1つのチームに所属
    team: Optional[Team] = Relationship(back_populates="members")

リレーションの作成 

with Session(engine) as session:
    # チーム作成
    team = Team(name="開発チーム")
    session.add(team)
    session.commit()
    session.refresh(team)

    # ユーザー作成とチーム紐付け
    user = User(name="田中太郎", team_id=team.id)
    session.add(user)
    session.commit()

リレーションの取得 

with Session(engine) as session:
    # チームからメンバーを取得
    team = session.get(Team, 1)
    for member in team.members:
        print(member.name)

    # ユーザーからチームを取得
    user = session.get(User, 1)
    print(user.team.name)

Fieldオプション 

from sqlmodel import SQLModel, Field
from typing import Optional

class Product(SQLModel, table=True):
    # 主キー（自動採番）
    id: Optional[int] = Field(default=None, primary_key=True)

    # 必須項目
    name: str = Field(index=True)  # インデックス作成

    # デフォルト値
    price: int = Field(default=0)

    # 最小値・最大値
    stock: int = Field(ge=0, le=10000)  # 0以上10000以下

    # 文字列長制限
    code: str = Field(max_length=20)

    # ユニーク制約
    email: str = Field(unique=True)

    # NULL許可
    description: Optional[str] = None

実務Tips

トランザクション管理

複数の操作を1つのトランザクションにまとめることで、データの整合性を保つ。 エラー時は自動的にロールバックされる。

N+1問題対策

リレーション取得時はselectinload()やjoinedload()を使って クエリ数を削減する。

バリデーション

Pydanticの機能を活用して、データ保存前に自動バリデーションを実施。 型安全性を確保できる。

環境変数でDB接続

本番環境では接続文字列を環境変数で管理。 os.getenv("DATABASE_URL")で取得する。

FastAPI連携

基本的なAPI実装

from fastapi import FastAPI, Depends
from sqlmodel import Session, select

app = FastAPI()

# データベース接続の依存関係
def get_session():
    with Session(engine) as session:
        yield session

# ユーザー一覧取得API
@app.get("/users")
def get_users(session: Session = Depends(get_session)):
    statement = select(User)
    users = session.exec(statement).all()
    return users

# ユーザー作成API
@app.post("/users")
def create_user(user: User, session: Session = Depends(get_session)):
    session.add(user)
    session.commit()
    session.refresh(user)
    return user

よくあるエラーと対処法

IntegrityError: UNIQUE constraint failed

原因: ユニーク制約違反（同じメールアドレスなど）

対処: 既存データの確認、または制約の見直し

DetachedInstanceError

原因: セッションが閉じた後にリレーションにアクセス

対処: セッション内でリレーションを先に読み込む、またはselectinload()を使用

テーブルが作成されない

原因: モデル定義時にtable=Trueを忘れている

対処: class User(SQLModel, table=True)と明示的に指定

データベース接続文字列

SQLite（開発用）

sqlite:///database.db

PostgreSQL

postgresql://user:password@localhost/dbname

MySQL

mysql://user:password@localhost/dbname

参考リンク