概要

SQLAlchemyはPythonのORMフレームワークです。

SQLデータベースとのやりとりを抽象化し、Pythonで簡単にデータベースの操作ができるようになります。

インストール

まず、SQLAlchemyをインストールします。

インストールしたバージョンは2.0.20でした。

pip install SQLAlchemy

接続

データベースに接続します。

echo=True はログを標準出力に書き出すための設定です。

from sqlalchemy import create_engine

engine = create_engine("sqlite:///database.sqlite3", echo=True)

モデル定義

モデルベースクラスを作り、それを継承したモデルを作ります。

from typing import List
from typing import Optional
from sqlalchemy import ForeignKey, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    pass

class Team(Base):
    __tablename__ = "teams"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(16))

    players: Mapped[List["Player"]] = relationship(
        back_populates="team", cascade="all, delete-orphan"
    )

    def __repr__(self) -> str:
        return f"Team(id={self.id!r}, name={self.name!r})"

class Player(Base):
    __tablename__ = "players"

    id: Mapped[int] = mapped_column(primary_key=True)
    fullname: Mapped[str]
    age: Mapped[Optional[int]]
    team_id: Mapped[int] = mapped_column(ForeignKey("teams.id"))

    team: Mapped["Team"] = relationship(back_populates="players")

    def __repr__(self) -> str:
        return f"Player(id={self.id!r}, fullname={self.fullname!r}, age={self.age!r}, )"

テーブル作成

定義したモデルでテーブルを作成します。

実行するとteamsテーブルとplayersテーブルが作成されます。

Base.metadata.create_all(engine)

テーブル削除

テーブルを削除したい場合はdrop_all()を実行します。

Base.metadata.drop_all(engine)

個別にテーブルを削除することもできます。

Team.__table__.drop(engine)

セッション操作

作成したテーブルを操作するためにデータベースへ接続します。

接続にはSessionオブジェクトを使用します。

from sqlalchemy.orm import Session

session = Session(engine)

データの追加

データを追加します。

team = Team(name="giants", players=[Player(fullname="Foster Griffin", age=28)])

session.add(team)
session.commit()

または、withステートメントを使って書くこともできます。

with Session(engine) as session:
		team = Team(name="giants", players=[Player(fullname="Foster Griffin", age=28)])

		session.add(team)
		session.commit()

データの検索

selectを使ってデータを検索します。

from sqlalchemy import select

stmt = select(Team).where(Team.name == "giants")
team = session.scalars(stmt).one()

PrimaryKeyを使った検索であればSession.get()を使って簡潔に書くことができます。

player = session.get(Player, 1)

データの更新

データの更新にはupdateを使います。

from sqlalchemy import update

stmt = update(Player).where(Player.id == 1).values(age=20)
session.execute(stmt)

または、オブジェクトに値を直接代入してデータの更新を行うこともできます。

player = session.get(Player, 1)
player.age = 30

session.commit()

データの削除

データの削除にはSession.delete()を使います。

player = session.get(Player, 1)

session.delete(player)
session.commit()