MENU
あなたの考える力をトレーニング
  1. ホーム
  2. プログラミング力
  3. sqlmodelの使い方

sqlmodelの使い方

プログラミング力
2022年2月13日

この記事ではPythonライブラリのsqlmodelの使い方について説明します。sqlmodelは、クラスを効率的に定義できるpydanticと、SQLのORMライブラリであるSQLAlchemyを組み合わせたライブラリです。

sqlmodelは、pydanticと類似の方法でクラスを定義するだけで、そのクラスのORM機能も合わせて提供してくれます。pydanticとSQLAlchemyはそれぞれ便利なライブラリですが、一方にあって他方にない機能を利用したい場合や、それを解決するために二つのライブラリを自分で組み合わせて使用する場合に、機能を再実装する手間やそれぞれのクラスを相互変換する処理を書く手間などが生じます。sqlmodelはそのような手間を削減するためにfastapiの作者が開発しているライブラリです。

目次

基本的な使い方

まずは基本的な使い方を以下のコードにまとめました。大きな流れは次の通りです。

  1. クラス・テーブルを定義
  2. インスタンスの作成・属性アクセス
  3. インスタンスをレコードとしてテーブルに保存
  4. テーブルからレコードを取得
from typing import Optional

from sqlmodel import Field, SQLModel
from sqlmodel import Session, create_engine, select
from sqlmodel.sql.expression import SelectOfScalar


##############################################
# 1. クラス・テーブルを定義。
##############################################

# `SQLModel`を継承したクラスで
# 型アノテーション付きのクラス属性を定義する。
# テーブル化するには`table=True`が必要。
class User(SQLModel, table=True):
  # ユーザID。auto incrementのために以下のように定義。
  id: Optional[int] = Field(default=None, primary_key=True)

  # その他属性。そのままテーブルのフィールドとなる。
  name: str  # ユーザ名
  height: Optional[float]  # 身長[cm]
  weight: Optional[float]  # 体重[kg]


##############################################
# 2. インスタンスの作成・属性アクセス
##############################################

# 通常のクラスと同様にインスタンスを作成できる。
user1 = User(name="田中", height=142.0)

# pydanticの`BaseModel`の機能は一通り使える。
# `SQLModel`が`BaseClass`のサブクラスであるため。
print(SQLModel.__bases__)
#> (<class 'pydantic.main.BaseModel'>,)

print(user1)
#> id=None name='田中' height=142.0 weight=None

print(user1.json())
#> {"id": null, "name": "田中", "height": 142.0, "weight": null}

print(user1.copy())
#> id=None name='田中' height=142.0 weight=None

# parse_rawなどのメソッドも使用可能。
user2 = User.parse_raw('{"id": null, "name": "佐藤"}')
user3 = User.parse_raw('{"id": null, "name": "鈴木"}')


##############################################
# 3. インスタンスをレコードとしてテーブルに保存。
##############################################

# エンジンの作成（この段階では.dbファイルは作られない）
engine = create_engine(
  "sqlite:///database.db",
  echo=False,  # エンジンの内部メッセージをechoするか
)

# SQLModelのサブクラスを一括でCREATE TABLEし
# database.dbが作成される。
SQLModel.metadata.create_all(engine)

# セッションを作成して保存
with Session(engine) as session:
  # 各インスタンスをレコードとしてINSERT
  session.add(user1)
  session.add(user2)
  session.add(user3)

  # .dbファイルにレコードが追加される。
  # また、idがauto incrementされる。
  session.commit()

  print(user1.id)
  #> 1

  # なお`commit`直後のインスタンスは空になる。
  print(user2.json())
  #> {}

  # 属性アクセスなどで内部的に`refresh`が呼び出される。
  print(user2.id)
  #> 2
  print(user2.json())
  #> {"id": 2, "height": null, "name": "佐藤", "weight": null}

  # `user3`も同様に空になる。
  print(user3.json())
  #> {}

  # 自動的に`refresh`される操作を介さずに、明示的に`refresh`もできる。
  session.refresh(user3)
  print(user3.json())
  #> {"weight": null, "name": "鈴木", "id": 3, "height": null}


##############################################
# 4. テーブルからレコードを取得
##############################################

# （設定しないと`select`の使用時に警告が出る）
SelectOfScalar.inherit_cache = True

# userテーブルの値を全てSELECTして表示。
with Session(engine) as session:
  statement = select(User)
  users = session.exec(statement).all()
  for user in users:
    print(user)
  #> weight=None name='田中' id=1 height=142.0
  #> weight=None name='佐藤' id=2 height=None
  #> weight=None name='鈴木' id=3 height=None

SQLクエリの書き方

この節ではクエリの書き方を紹介します。

WHERE、AND、 OR

selectの戻り値にwhereメソッドを連ねることでWHERE句を追加できます。ANDとORはand_関数とor_関数で実現できます。論理積はwhereを複数つなげたりwhereの引数を増やしたりすることでも実現できます。

from typing import Optional

from sqlmodel import Field, SQLModel
from sqlmodel import Session, create_engine, select, or_, and_
from sqlmodel.sql.expression import SelectOfScalar


##############################################
# 1. クラス・テーブルを定義。
##############################################

class User(SQLModel, table=True):
  id: Optional[int] = Field(default=None, primary_key=True)  # ユーザID
  name: str  # ユーザ名
  height: Optional[float]  # 身長[cm]
  weight: Optional[float]  # 体重[kg]


##############################################
# 2. デモ用データの保存
##############################################

users = [
  User(name="田中", height=142.0, weight=39.1),
  User(name="佐藤", height=128.3, weight=32.2),
  User(name="鈴木", height=133.5, weight=34.4),
  User(name="小林", height=141.8, weight=39.9),
  User(name="高橋", height=145.7, weight=37.4),
]

engine = create_engine("sqlite:///database.db")
SQLModel.metadata.create_all(engine)

with Session(engine) as session:
  for user in users:
    session.add(user)
  session.commit()


##############################################
# 3. WHERE句のデモ
##############################################

SelectOfScalar.inherit_cache = True

with Session(engine) as session:
  # クラス属性と比較演算子で条件を記述する。
  statement = select(User).where(User.name == "田中")
  users = session.exec(statement).all()
  for user in users:
      print(user)
  #> weight=39.1 name='田中' id=1 height=142.0

  # 複数の`where`を繋げるとAND条件になる。
  statement = (select(User)
    .where(User.height > 130)
    .where(User.height < 142))
  users = session.exec(statement).all()
  for user in users:
      print(user)
  #> weight=34.4 name='鈴木' id=3 height=133.5
  #> weight=39.9 name='小林' id=4 height=141.8

  # `where`の中に引数を追加してもAND条件になる。
  statement = select(User).where(User.height > 130, User.height < 142)
  users = session.exec(statement).all()
  for user in users:
      print(user)
  #> weight=34.4 name='鈴木' id=3 height=133.5
  #> weight=39.9 name='小林' id=4 height=141.8

  # `or_`や`and_`を組み合わせて複雑な論理式を構成できる。
  statement = select(User).where(
      or_(and_(User.height > 130, User.height < 140),
          and_(User.weight > 30, User.weight < 35)))
  users = session.exec(statement).all()
  for user in users:
      print(user)
  #> weight=32.2 name='佐藤' id=2 height=128.3
  #> weight=34.4 name='鈴木' id=3 height=133.5

LIMIT、OFFSET

ステートメントにlimitメソッドやoffsetメソッドを繋げるだけで実現できます。

# （WHERE句の説明コードと同様のレコードを用意）

with Session(engine) as session:
  # `limit`でレコード数を制限できる。
  statement = select(User).where(
      or_(and_(User.height > 130, User.height < 140),
          and_(User.weight > 30, User.weight < 35)))
  statement = statement.limit(1)
  users = session.exec(statement).all()
  for user in users:
      print(user)
  #> weight=32.2 name='佐藤' id=2 height=128.3

  # `offset`でレコードのオフセットを指定できる
  statement = select(User).where(
      or_(and_(User.height > 130, User.height < 140),
          and_(User.weight > 30, User.weight < 35)))
  statement = statement.offset(1)
  users = session.exec(statement).all()
  for user in users:
      print(user)
  #> weight=34.4 name='鈴木' id=3 height=133.5

UPDATE

selectで取得したインスタンスまたはadd（INSERT）されたインスタンスの属性を更新してaddすると自動でUPDATE扱いになります。

from typing import Optional

from sqlmodel import Field, SQLModel
from sqlmodel import Session, create_engine, select
from sqlmodel.sql.expression import SelectOfScalar


##############################################
# 1. クラス・テーブルを定義。
##############################################

class User(SQLModel, table=True):
  id: Optional[int] = Field(default=None, primary_key=True)  # ユーザID
  name: str  # ユーザ名
  height: Optional[float]  # 身長[cm]
  weight: Optional[float]  # 体重[kg]


##############################################
# 2. デモ用データの保存
##############################################

user1 = User(name="田中", height=142.0, weight=39.1)

engine = create_engine("sqlite:///database.db")
SQLModel.metadata.create_all(engine)

with Session(engine) as session:
  session.add(user1)
  session.commit()


##############################################
# 3. オブジェクトの値を更新してUPDATE
##############################################

user1.height = 142.5
user1.weight = 39.2

with Session(engine) as session:
  session.add(user1)
  session.commit()


##############################################
# 4. レコードの確認
##############################################

SelectOfScalar.inherit_cache = True

with Session(engine) as session:
  # クラス属性と比較演算子で条件を記述する。
  statement = select(User)
  users = session.exec(statement).all()
  for user in users:
      print(user)
  #> name='田中' id=1 height=142.5 weight=39.2

DELETE

セッションのdeleteメソッドでレコードをDELETEできます。

from typing import Optional

from sqlmodel import Field, SQLModel
from sqlmodel import Session, create_engine, select
from sqlmodel.sql.expression import SelectOfScalar


##############################################
# 1. クラス・テーブルを定義。
##############################################

class User(SQLModel, table=True):
  id: Optional[int] = Field(default=None, primary_key=True)  # ユーザID
  name: str  # ユーザ名
  height: Optional[float]  # 身長[cm]
  weight: Optional[float]  # 体重[kg]


##############################################
# 2. デモ用データの保存
##############################################

user1 = User(name="田中", height=142.0, weight=39.1)
user2 = User(name="佐藤", height=128.3, weight=32.2)

engine = create_engine("sqlite:///database.db")
SQLModel.metadata.create_all(engine)

SelectOfScalar.inherit_cache = True

with Session(engine) as session:
  session.add(user1)
  session.add(user2)
  session.commit()

  statement = select(User)
  users = session.exec(statement).all()
  for user in users:
      print(user)
  #> height=142.0 id=1 name='田中' weight=39.1
  #> height=128.3 id=2 name='佐藤' weight=32.2


##############################################
# 3. オブジェクトをDELEET
##############################################

with Session(engine) as session:
  session.delete(user1)
  session.commit()

  statement = select(User)
  users = session.exec(statement).all()
  for user in users:
      print(user)
  #> height=128.3 id=2 name='佐藤' weight=32.2

JOINについて

JOINは次の方法で実現できます。サンプルコードのうち、(1), (2), (3)に該当する行はコメントで示しています。

  • 子テーブルのFOREIGN KEY制約に相当する属性を定義する際に、Fieldのforeign_key引数に親テーブルの識別子・フィールドを指定。……(1)
  • 子テーブルのインスタンスにおいて、FOREIGN KEY制約の属性に親テーブルのIDを代入。……(2)
  • selectの引数にJOINするクラスを列挙し、whereメソッドで結合条件を指定。……(3)
from typing import List, Optional

from sqlmodel import Field, SQLModel
from sqlmodel import Session, create_engine, select
from sqlmodel.sql.expression import Select


##############################################
# 1. クラス・テーブルを定義。
##############################################

# 部活クラス
class Club(SQLModel, table=True):
  id: Optional[int] = Field(default=None, primary_key=True)  # 部活ID
  name: str  # 部活名


# 生徒クラス
class Student(SQLModel, table=True):
  id: Optional[int] = Field(default=None, primary_key=True)  # ユーザID
  name: str  # ユーザ名

  # (1) FOREIGN KEY制約に相当。親テーブルを指定。
  club_id: Optional[int] = Field(default=None, foreign_key="club.id")


engine = create_engine("sqlite:///database.db")

SQLModel.metadata.create_all(engine)


##############################################
# 2. 部活インスタンスを作成・保存
##############################################

# 部活インスタンスを生成
club1 = Club(name="野球部")
club2 = Club(name="サッカー部")

with Session(engine) as session:
  session.add(club1)
  session.add(club2)
  session.commit()
  session.refresh(club1)
  session.refresh(club2)


##############################################
# 3. 生徒インスタンスを作成・保存
##############################################

# (2) 生徒インスタンスを生成
student1 = Student(name="田中", club_id=club1.id)
student2 = Student(name="佐藤", club_id=club1.id)
student3 = Student(name="鈴木", club_id=club2.id)

with Session(engine) as session:
  session.add(student1)
  session.add(student2)
  session.add(student3)
  session.commit()


##############################################
# 4. JOIN・SELECT
##############################################

Select.inherit_cache = True

with Session(engine) as session:
  # (3) `select`にJOINする対象のテーブルを指定し、`where`にJOINの条件式を記載
  statement = select(Student, Club).where(Student.club_id == Club.id)
  # JOINされたレコードのタプルのリストを取得
  for student, club in session.exec(statement).all():
    print(student, club)
  #> id=1 name='田中' club_id=1 name='野球部' id=1
  #> id=2 name='佐藤' club_id=1 name='野球部' id=1
  #> id=3 name='鈴木' club_id=2 name='サッカー部' id=2

バージョン情報

この記事のスクリプトは以下のバージョンで確認しました。

  • Python3.9.1
  • sqlmodel==0.0.6
プログラミング力
記事：プログラミング
よかったらシェアしてね！
  • URLをコピーしました！

この記事を書いた人

いまいのアバター いまい

ITベンチャーでデータ分析、AI開発、システム設計、提案、営業、組織管理、公演、採用などなど多数の役割に従事してきました。

様々な職業や背景の方々と交流するうちに、幅広い分野で問題を解決したり価値を生み出したりするためには、個別の知識だけでなく、汎用的に物事を考える力を伸ばしていく必要があると考えるようになりました。

更に、自分自身の考える力だけでなく、より多くの人々の考える力のトレーニングを応援することで、社会全体を良くしていけるのではないかと考えて、このサイトを作りました。

関連記事

目次
閉じる