【Python】SQLAlchemy入門と使い方｜最速で覚えるCRUD＆モデル定義

# 仮想環境の作成(任意)
python -m venv venv
source venv/bin/activate  # Windowsなら: venv\Scripts\activate

# SQLAlchemy本体のインストール
pip install sqlalchemy

# ORM+SQLiteで試すなら、ドライバは標準のsqlite3でOK
# PostgreSQLなどを使う場合は、例えば:
# pip install psycopg2-binary

python -c "import sqlalchemy; print(sqlalchemy.__version__)"

# engine_example.py
from sqlalchemy import create_engine

# SQLiteファイルに接続するEngineを作成
# echo=True にすると、実行されるSQLがコンソールに表示される
engine = create_engine("sqlite:///example.db", echo=True, future=True)

# Engineオブジェクトを確認するだけ
print(engine)

# session_example.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Engineを先に作成
engine = create_engine("sqlite:///example.db", echo=True, future=True)

# Sessionを作るためのクラス(Session factory)を定義
SessionLocal = sessionmaker(
    bind=engine,
    autoflush=False,  # 自動フラッシュを無効化(明示的に扱う方が初心者向け)
    autocommit=False, # 明示的なcommitを前提にする
    future=True,
)

# 実際にSessionインスタンスを1つ作ってみる
def main():
    db = SessionLocal()
    try:
        print(db)
        # ここでクエリなどを実行していく
    finally:
        db.close()

if __name__ == "__main__":
    main()

# models_base.py
from sqlalchemy.orm import declarative_base

# Declarative Base を作成
Base = declarative_base()

# models_user.py
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = "users"  # 対応するテーブル名

    # 主キー。primary_key=Trueを必ずどれか1つ以上のカラムに指定する
    id = Column(Integer, primary_key=True, index=True)

    # ユーザー名。長さ制限付きの文字列
    name = Column(String(50), nullable=False)

    # メールアドレス。一意制約を付けるケースが多い
    email = Column(String(255), unique=True, nullable=False)

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

# models_article.py
from datetime import datetime

from sqlalchemy import Column, Integer, String, Text, Boolean, DateTime
from sqlalchemy.orm import declarative_base

Base = declarative_base()

class Article(Base):
    __tablename__ = "articles"

    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False, index=True)
    body = Column(Text, nullable=False)
    is_published = Column(Boolean, nullable=False, default=False)
    created_at = Column(
        DateTime,
        nullable=False,
        default=datetime.utcnow,  # 関数自体を渡す(呼び出さない)
    )

# models_relationship.py
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import declarative_base, relationship

Base = declarative_base()

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)

    # Addressとの一対多。back_populatesで双方向リンクを張る
    addresses = relationship("Address", back_populates="user")

class Address(Base):
    __tablename__ = "addresses"

    id = Column(Integer, primary_key=True)
    email = Column(String(255), nullable=False)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=False)

    # User側との対応を定義
    user = relationship("User", back_populates="addresses")

# crud_create.py
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import declarative_base, sessionmaker

# Base定義
Base = declarative_base()

# モデル定義
class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50), nullable=False)
    email = Column(String(255), unique=True, nullable=False)

    def __repr__(self):
        return f"User(id={self.id}, name={self.name!r}, email={self.email!r})"

# Engine & Session factory
engine = create_engine("sqlite:///example.db", echo=True, future=True)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)

# テーブルを作成(存在しない場合のみ)
Base.metadata.create_all(bind=engine)

def create_user(name: str, email: str) -> None:
    db = SessionLocal()
    try:
        # Userオブジェクトを作成
        user = User(name=name, email=email)

        # セッションに追加
        db.add(user)

        # トランザクションを確定
        db.commit()

        # commit後にidが自動的に反映される
        db.refresh(user)  # DBから最新状態を再取得
        print("Created:", user)
    except Exception as e:
        db.rollback()
        raise
    finally:
        db.close()

if __name__ == "__main__":
    create_user("Alice", "alice@example.com")
    create_user("Bob", "bob@example.com")

# crud_read.py
from sqlalchemy import create_engine, select
from sqlalchemy.orm import sessionmaker

from crud_create import Base, User  # 先ほどの定義を再利用

engine = create_engine("sqlite:///example.db", echo=True, future=True)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)

def get_user_by_id(user_id: int) -> User | None:
    db = SessionLocal()
    try:
        stmt = select(User).where(User.id == user_id)
        result = db.execute(stmt)
        # scalars()でUserオブジェクトのストリームを取得し、first()で1件取り出す
        user = result.scalars().first()
        return user
    finally:
        db.close()

def get_users_by_name(name: str) -> list[User]:
    db = SessionLocal()
    try:
        stmt = select(User).where(User.name == name)
        result = db.execute(stmt)
        users = result.scalars().all()
        return users
    finally:
        db.close()

def get_all_users() -> list[User]:
    db = SessionLocal()
    try:
        stmt = select(User)
        result = db.execute(stmt)
        return result.scalars().all()
    finally:
        db.close()

if __name__ == "__main__":
    print("User id=1:", get_user_by_id(1))
    print("Users name='Alice':", get_users_by_name("Alice"))
    print("All users:", get_all_users())

# crud_update.py
from sqlalchemy import create_engine, select
from sqlalchemy.orm import sessionmaker

from crud_create import Base, User

engine = create_engine("sqlite:///example.db", echo=True, future=True)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)

def update_user_email(user_id: int, new_email: str) -> None:
    db = SessionLocal()
    try:
        stmt = select(User).where(User.id == user_id)
        user = db.execute(stmt).scalars().first()
        if user is None:
            print("User not found")
            return

        # 属性を書き換える
        user.email = new_email

        # commitでUPDATEが発行される
        db.commit()
        print("Updated:", user)
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()

if __name__ == "__main__":
    update_user_email(1, "alice.updated@example.com")

# crud_delete.py
from sqlalchemy import create_engine, select
from sqlalchemy.orm import sessionmaker

from crud_create import Base, User

engine = create_engine("sqlite:///example.db", echo=True, future=True)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)

def delete_user(user_id: int) -> None:
    db = SessionLocal()
    try:
        stmt = select(User).where(User.id == user_id)
        user = db.execute(stmt).scalars().first()
        if user is None:
            print("User not found")
            return

        db.delete(user)
        db.commit()
        print(f"Deleted user id={user_id}")
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()

if __name__ == "__main__":
    delete_user(2)

# transaction_basic.py
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import declarative_base, sessionmaker

Base = declarative_base()

class Item(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)

engine = create_engine("sqlite:///tx_example.db", echo=True, future=True)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)
Base.metadata.create_all(bind=engine)

def demo_commit_and_rollback():
    db = SessionLocal()
    try:
        item1 = Item(name="Committed item")
        db.add(item1)

        # ここでcommitするとitem1はDBに保存される
        db.commit()

        item2 = Item(name="Rolled back item")
        db.add(item2)

        # 例外が起きたと仮定してrollback
        raise RuntimeError("Something went wrong!")

    except Exception as e:
        print("Error:", e)
        db.rollback()
    finally:
        db.close()

if __name__ == "__main__":
    demo_commit_and_rollback()

# transaction_with.py
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import declarative_base, sessionmaker

Base = declarative_base()

class Log(Base):
    __tablename__ = "logs"
    id = Column(Integer, primary_key=True)
    message = Column(String(100), nullable=False)

engine = create_engine("sqlite:///tx_with.db", echo=True, future=True)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)
Base.metadata.create_all(bind=engine)

def write_log_safely(message: str):
    # with構文でSessionを管理
    with SessionLocal() as db:
        log = Log(message=message)
        db.add(log)
        # withブロックを正常に抜けると、自動的にcommitされる
        # 例外が発生した場合はrollbackされる

if __name__ == "__main__":
    write_log_safely("Hello transaction with 'with'!")

# models_relations_practice.py
from sqlalchemy import (
    Column, Integer, String, Text, ForeignKey, Table, create_engine
)
from sqlalchemy.orm import declarative_base, relationship, sessionmaker

Base = declarative_base()

# 中間テーブル(多対多用)の定義
article_tag = Table(
    "article_tag",
    Base.metadata,
    Column("article_id", ForeignKey("articles.id"), primary_key=True),
    Column("tag_id", ForeignKey("tags.id"), primary_key=True),
)

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)

    articles = relationship("Article", back_populates="author")

class Article(Base):
    __tablename__ = "articles"

    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    body = Column(Text, nullable=False)
    author_id = Column(Integer, ForeignKey("users.id"), nullable=False)

    author = relationship("User", back_populates="articles")
    tags = relationship("Tag", secondary=article_tag, back_populates="articles")

class Tag(Base):
    __tablename__ = "tags"

    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False, unique=True)

    articles = relationship("Article", secondary=article_tag, back_populates="tags")

# relations_demo.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from models_relations_practice import Base, User, Article, Tag

engine = create_engine("sqlite:///relation_example.db", echo=True, future=True)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)
Base.metadata.create_all(bind=engine)

def create_sample_data():
    with SessionLocal() as db:
        user = User(name="Author A")
        tag_python = Tag(name="Python")
        tag_orm = Tag(name="ORM")

        article = Article(
            title="SQLAlchemy入門",
            body="SQLAlchemyの基本を解説します。",
            author=user,           # Userオブジェクトを直接渡す
            tags=[tag_python, tag_orm],  # 多対多の関連付け
        )

        db.add(article)
        db.commit()

def query_relations():
    with SessionLocal() as db:
        author = db.query(User).filter_by(name="Author A").first()
        print("Author:", author.name)
        for article in author.articles:
            print("Article:", article.title)
            print(" Tags:", [t.name for t in article.tags])

if __name__ == "__main__":
    create_sample_data()
    query_relations()

# db/base.py
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, sessionmaker

DATABASE_URL = "sqlite:///app.db"

engine = create_engine(DATABASE_URL, echo=False, future=True)

SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)

Base = declarative_base()

# Alembicのインストール
pip install alembic

# 初期化(現在のディレクトリにalembic/が作成される)
alembic init alembic

# モデル変更に応じて新しいマイグレーションファイルを作る
alembic revision --autogenerate -m "create user and article tables"

# マイグレーションをDBに適用
alembic upgrade head

# raw_sql_example.py
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker

engine = create_engine("sqlite:///example.db", echo=True, future=True)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)

def get_user_count():
    with SessionLocal() as db:
        # :name のようなプレースホルダを使い、パラメータを別渡しにする
        sql = text("SELECT COUNT(*) FROM users WHERE name = :name")
        result = db.execute(sql, {"name": "Alice"})
        count = result.scalar_one()
        return count

if __name__ == "__main__":
    print("Alice count:", get_user_count())