◾️はじめに

https://dk521123.hatenablog.com/entry/2025/10/07/133808

の続き。

今回は、SQLAlchemyの基本的な知識を書き溜めておく。

【１】テーブル定義

* DeclarativeBase を継承して定義する
 => サンプル見た方が早い

https://docs.sqlalchemy.org/en/20/changelog/whatsnew_20.html#sql-expression-typing-examples

from sqlalchemy.orm import DeclarativeBase


# Define table
class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    name = Column(String)
    age = Column(Integer)

古い書き方

from sqlalchemy.orm import declarative_base

Base = declarative_base()

# Define table
class User(Base):
    __tablename__ = "users"

１）データ定義

* 基本は、以下のサンプルを見て、
　後は都度、ドキュメントで補うのがいいと思う

https://docs.sqlalchemy.org/en/20/changelog/whatsnew_20.html#migrating-an-existing-mapping
https://docs.sqlalchemy.org/en/20/core/type_basics.html

from sqlalchemy import Column, Integer, String, Text, Date, DateTime
from datetime import datetime, timezone 

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    name = Column(String(30), nullable=False)
    birthday = Column(Date)
    created_at = Column(DateTime,
                        default=lambda: datetime.now(timezone.utc))
    updated_at = Column(DateTime,
                        default=lambda: datetime.now(timezone.utc),
                        onupdate=lambda: datetime.now(timezone.utc))

使用上の注意

# lambda: なしだと固定した値が入ってしまう
# lambdaを使うことによって、関数自体を渡して逐一日付を取ってくるようにする
created_at = Column(DateTime,
                      default=datetime.now(timezone.utc))

２）初期化としてのCREATE TABLE

* Base.metadata.create_all() で行う

例

# Connect DB for SQLite
engine = create_engine("sqlite:///example.db", echo=True)

# Create table
Base.metadata.create_all(engine)

【２】SELECT

* query().all() で全データ取得

サンプル

users = session.query(User).all()
for user in users:
    print(user.name, user.age)

１）WHERE句

* filter() / filter_by() で指定

サンプル

user1 = session.query(User).filter(User.id == 1).first()
user2 = session.query(User).filter_by(name="Mike").first()

【３】INSERT

* session.add() + session.commit() で実現可能

サンプル

new_user = User(name="Mike", age=30)

session.add(new_user)
session.commit()

【４】UPDATE

* 以下の通り。
~~~~
[1] 対象ユーザー取得 ... 【２】SELECT参照
[2] その値を更新・保存 ... 【３】INSERT参照
~~~~

サンプル

# [1] 対象ユーザー取得
user = session.query(User).filter_by(name="Mike").first()

if user:
    # [2] その値を更新・保存
    user.age = 35
    session.commit()
    print(f"{user.name} の年齢を更新しました。")
else:
    print("対象のユーザーが見つかりませんでした。")

１）一括更新

session.query(User).update({User.age: User.age + 1})
session.commit()

【５】DELETE

* 以下の通り。
~~~~
[1] 対象ユーザー取得 ... 【２】SELECT参照
[2] その値をsession.delete()し、保存
~~~~

サンプル

user = session.query(User).filter_by(name="Mike").first()

if user:
    # ★UPDATEとの違いは、ここだけ
    session.delete(user)
    session.commit() 
    print(f"{user.name} を削除しました。")
else:
    print("削除対象のユーザーが見つかりませんでした。")

１）一括削除

session.query(User).filter(User.age < 20).delete()
session.commit()

【６】SQL直接実行

* session.execute() で実行

サンプル

# SELECT
result = session.execute(
    text("SELECT id, name, age FROM users WHERE age >= :age"),
    {"age": 30}
)

for row in result:
    print(row.id, row.name, row.age)

# DELETE
session.execute(
    text("DELETE FROM users WHERE name = :name"),
    {"name": "Mike"}
)
session.commit()