SQLModel で簡単なテーブルを作成する

プログラミング

はじめに

SQLModelとは、PythonのモダンなORM（Object Relational Mapping）です。

SQLModelは、SQLAlchemyとPydanticの利点を組み合わせたライブラリで、Web APIの開発に、特にFastAPIでの開発に適しています。

この記事では、SQLModel をつかってテーブルを作成する手順を簡単に述べます。この内容は SQLModelのチュートリアル を非常に参考にしていますので、ぜひそちらもご覧ください。

この記事に記載したソースコードは以下の GitHub で確認できます。

GitHub - Joichiro433/Blog-sqlmodel
Contribute to Joichiro433/Blog-sqlmodel development by creating an account on GitHub.
github.com

SQLModel でテーブルを構成する

今回作成するテーブルは以下のER図の通りです。2つのテーブル team と hero を作成します。team と hero は 『1対多』の関係にあります。

このテーブル構成を SQLModel で定義していきましょう。使用するソースコードの構成は以下となります。

.
├── app.py
├── db.py
└── models/
    ├── hero.py
    └── team.py

今回のテーブル構造はシンプルですが、 hero と team の定義を練習のために、あえて別々のファイルで管理しました。

テーブルの定義: models/

hero.py

from typing import Optional, TYPE_CHECKING

from sqlmodel import Field, Relationship, SQLModel

if TYPE_CHECKING:
    from models.team import Team


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

    team_id: Optional[int] = Field(default=None, foreign_key='team.id')
    team: Optional['Team'] = Relationship(back_populates='heroes')

team.py

from typing import Optional, TYPE_CHECKING

from sqlmodel import Field, Relationship, SQLModel

if TYPE_CHECKING:
    from models.hero import Hero


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

    heroes: list['Hero'] = Relationship(back_populates='team')

models/ ディレクトリ配下には、hero テーブルを定義する hero.py と、team テーブルを定義する team.py を配置しました。

ここでは、コード 5, 6 行目の if TYPE_CHECKING: について説明します。

今回は 2つのテーブル定義をそれぞれ別のファイルに分割しました。それぞれのコードでは型ヒントとして hero.pyでは Team型 を、 team.py では Hero型 を使用したいです。

しかし型ヒントを使用するために、普通にそれぞれをインポートしようとすると、『hero.py から team.py をインポート、team.py から hero.py をインポート』と両方がインポートを行い、循環インポートが生じてしまいエラーが発生します。

循環インポートが生じている原因の型ヒントは、あくまで実行時には不要で、実装時のみ必要です。このような、実装時のみ「インポート」として機能するものが

if TYPE_CHECKING:
    from models.team import Team

になります。これにより実装時では型ヒントの恩恵を享受することができ、実行時では TYPE_CHECKING の値が False となり循環インポートを回避することができます。

参考:

typing --- 型ヒントのサポート
ソースコード: Lib/typing.py This module provides runtime support for type hints. Consider the function below: surface_area_of_cube 関数は、 edge_length: float という type hi...
docs.python.org

if TYPE_CHECKING: のブロックは実行時には評価されないので、型ヒントは string で記載しましょう。
❌ team: Optional[Team]
✅ team: Optional['Team']

設定ファイル: settings.py

settings.py

SQLITE_FILE_NAME = 'database.sqlite3'
DB_URL = f'sqlite:///{SQLITE_FILE_NAME}'

settings.py にはデータベースの接続情報を記載します。今回は簡単な SQLite をDBとして使用します。

テーブルの作成: db.py

db.py

from sqlmodel import Session, SQLModel, create_engine

from models.hero import Hero
from models.team import Team
import settings


engine = create_engine(settings.DB_URL, echo=True)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


def create_heroes():
    with Session(engine) as session:
        team_preventers = Team(name='Preventers', headquarters='Sharp Tower')
        team_z_force = Team(name='Z-Force', headquarters="Sister Margaret's Bar")

        hero_deadpond = Hero(name='Deadpond', secret_name='Dive Wilson', team=team_z_force)
        hero_rusty_man = Hero(name='Rusty-Man', secret_name='Tommy Sharp', age=48, team=team_preventers)
        session.add(hero_deadpond)
        session.add(hero_rusty_man)
        session.commit()

        session.refresh(team_preventers)
        session.refresh(team_z_force)
        session.refresh(hero_deadpond)
        session.refresh(hero_rusty_man)

        print(f'Created team: {team_preventers}')
        print(f'Created team: {team_z_force}')
        print(f'Created hero: {hero_deadpond}')
        print(f'Created hero: {hero_rusty_man}')


if __name__ == '__main__':
    create_db_and_tables()
    create_heroes()

db.py ではテーブルの作成と、レコードの追加をおこなっています。

テーブルの作成は、関数 create_db_and_tables() でおこないます。

from models.hero import Hero
from models.team import Team

のインポートを行うことにより 2 つのテーブル定義が読み込まれ、適切にテーブルが作成されます。逆に言うと、テーブル定義 ("SQLModel, table=True" を継承したクラス) をインポートしなければ、そのテーブルは作成されません。

レコードの追加は、関数 create_heroes() でおこないます。今回はサンプルとして それぞれ 2 レコード作成します。

テーブルの作成とデータの取得

db.py を実行してテーブルを作成しましょう。

$ python db.py
## 出力
...
CREATE TABLE team (
        id INTEGER NOT NULL, 
        name VARCHAR NOT NULL, 
        headquarters VARCHAR NOT NULL, 
        PRIMARY KEY (id)
)
...
CREATE TABLE hero (
        id INTEGER NOT NULL, 
        name VARCHAR NOT NULL, 
        secret_name VARCHAR NOT NULL, 
        age INTEGER, 
        team_id INTEGER, 
        PRIMARY KEY (id), 
        FOREIGN KEY(team_id) REFERENCES team (id)
)
...
Created team: headquarters='Sharp Tower' name='Preventers' id=2
Created team: headquarters="Sister Margaret's Bar" name='Z-Force' id=1
Created hero: secret_name='Dive Wilson' name='Deadpond' team_id=1 id=1 age=None
Created hero: secret_name='Tommy Sharp' name='Rusty-Man' team_id=2 id=2 age=48

実行が完了すると、database.sqlite3 が作成されます。

DBが作成できたら、次はデータの取得を確認しましょう。データ取得コードは app.py に記載します。

app.py

from sqlmodel import Session, create_engine, select

from models.hero import Hero
from models.team import Team
import settings


engine = create_engine(settings.DB_URL, echo=True)


def exec_query():
    with Session(engine) as session:
        print('Query all rows from the Hero table')
        query = select(Hero)
        result = session.exec(query)
        print(result.all())

        print('Query the Hero table for a specific row with id = 1')
        query = select(Hero).where(Hero.id == 1)
        result = session.exec(query)
        hero = result.one()
        print(hero)
        print(hero.team)

        print('Query all rows from the Team table')
        query = select(Team)
        result = session.exec(query)
        print(result.all())

        print('Query the Team table for a specific row with id = 2')
        query = select(Team).where(Team.id == 2)
        result = session.exec(query)
        team = result.one()
        print(team)
        print(team.heroes)


if __name__ == '__main__':
    exec_query()

app.py を実行します。

$ python app.py
## 出力
'Query all rows from the Hero table'
[
    Hero(id=1, age=None, name='Deadpond', secret_name='Dive Wilson', team_id=1),
    Hero(id=2, age=48, name='Rusty-Man', secret_name='Tommy Sharp', team_id=2)
]

'Query the Hero table for a specific row with id = 1'
Hero(id=1, age=None, secret_name='Dive Wilson', name='Deadpond', team_id=1)
Team(headquarters="Sister Margaret's Bar", name='Z-Force', id=1)

'Query all rows from the Team table'
[
    Team(headquarters="Sister Margaret's Bar", name='Z-Force', id=1), 
    Team(headquarters='Sharp Tower', name='Preventers', id=2)
]

'Query the Team table for a specific row with id = 2'
Team(headquarters='Sharp Tower', name='Preventers', id=2)
[Hero(id=2, age=48, secret_name='Tommy Sharp', name='Rusty-Man', team_id=2)]

データが取得できました！例えば、hero 取得では、hero.py で team: Optional['Team'] = Relationship(back_populates='heroes') を定義したおかげで、hero.team にアクセスすると関連した team のレコードが取得できます。team の取得でも同様です。

まとめ

  • SQLModel をつかって簡単なテーブルを作成しました
  • 『1対多』のテーブルは Relationship(...) を用いることで、データの取得が簡単になります
タイトルとURLをコピーしました