いいね！大事なポイントをおさらいしよう。

基本編では、SQLAlchemyの3つの柱（Engine、Base、Session）と、CRUD操作を学んだよね。

# エンジンとSessionの作成
engine = create_engine("sqlite:///users.db")
SessionLocal = sessionmaker(bind=engine)
session = SessionLocal()

# モデルの定義
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String(50))

# CRUD操作
user = User(name="田中太郎")
session.add(user)          # Create
users = session.query(User).all()  # Read
user.name = "田中次郎"      # Update
session.delete(user)       # Delete
session.commit()           # 確定

これが基本だったね。

そう！実際のアプリケーションでは、複数のテーブルを関連付けて使うことが多いんだ。例えば、ブログアプリなら「ユーザー」と「投稿」、ECサイトなら「商品」と「注文」みたいにね。

リレーションシップを使うんだ。まず、一番よく使う「1対多」の関係から見ていこう。

例えば、ブログアプリを考えてみて。1人のユーザーは複数の投稿を書けるよね？これが「1対多」の関係なんだ。

そう！コードで見てみよう。

from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime
from sqlalchemy.orm import relationship
from datetime import datetime

# ユーザーモデル
class User(Base):
    __tablename__ = "users"

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

    # リレーションシップ: このユーザーの投稿一覧
    posts = relationship("Post", back_populates="author")

# 投稿モデル
class Post(Base):
    __tablename__ = "posts"

    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(Text, nullable=False)
    created_at = Column(DateTime, default=datetime.now)

    # 外部キー: どのユーザーの投稿か
    user_id = Column(Integer, ForeignKey("users.id"), nullable=False)

    # リレーションシップ: この投稿の著者
    author = relationship("User", back_populates="posts")

順番に説明するね。まず ForeignKey（外部キー） から。

user_id = Column(Integer, ForeignKey("users.id"))

これは、「この投稿はどのユーザーのものか」を示すカラムなんだ。ForeignKey("users.id")は、「usersテーブルのidカラムを参照する」という意味。

そう！そして relationship は、Pythonのコードからもっと便利にアクセスするための仕組みなんだ。

# Userモデル側
posts = relationship("Post", back_populates="author")

# Postモデル側
author = relationship("User", back_populates="posts")

これによって、こんな風に使えるようになるんだよ。

# ユーザーから投稿を取得
user = session.query(User).first()
for post in user.posts:  # user.postsで投稿一覧にアクセス！
    print(post.title)

# 投稿から著者を取得
post = session.query(Post).first()
print(post.author.name)  # post.authorで著者にアクセス！

そう！これがrelationshipの便利なところ。SQL文を書かなくても、直感的にデータを取得できるんだ。

back_populatesは、双方向の関連を定義するためのものなんだ。

• Userのpostsは、Postのauthorと対応
• Postのauthorは、Userのpostsと対応

こうやって双方向に関連付けることで、どちら側からでもデータにアクセスできるようになるんだよ。

じゃあ、実際にデータを作って動かしてみよう。

# ユーザー作成
user1 = User(name="田中太郎", email="tanaka@example.com")
user2 = User(name="佐藤花子", email="sato@example.com")

# 投稿作成（方法1: user_idを直接指定）
post1 = Post(
    title="SQLAlchemyの使い方",
    content="SQLAlchemyはPythonの便利なORMです。",
    user_id=1
)

# 投稿作成（方法2: relationshipを使う）
post2 = Post(
    title="Pythonでデータベース操作",
    content="データベース操作が簡単になります。"
)
post2.author = user1  # 著者を設定

post3 = Post(
    title="Flaskで Webアプリ開発",
    content="FlaskとSQLAlchemyを組み合わせて使えます。"
)
post3.author = user2

# データベースに追加
session.add_all([user1, user2, post1, post2, post3])
session.commit()

どちらでもいいんだけど、post.author = user1の方が読みやすいし、間違いにくいかな。user_id=1だと、IDを間違えるリスクがあるからね。

データを取得してみよう。

# ユーザーから投稿を取得
user = session.query(User).filter_by(name="田中太郎").first()
print(f"{user.name}の投稿:")
for post in user.posts:
    print(f"  - {post.title}")

# 投稿から著者を取得
posts = session.query(Post).all()
for post in posts:
    print(f"『{post.title}』 by {post.author.name}")

実行結果:

田中太郎の投稿:
  - Pythonでデータベース操作
  - SQLAlchemyの使い方

『Pythonでデータベース操作』 by 田中太郎
『Flaskで Webアプリ開発』 by 佐藤花子
『SQLAlchemyの使い方』 by 田中太郎

その場合は、join()を使うんだ。例えば、「田中太郎が書いた投稿」を検索したい時。

# 田中太郎が書いた投稿を検索
posts = session.query(Post).join(User).filter(User.name == "田中太郎").all()

for post in posts:
    print(f"{post.title} by {post.author.name}")

join(User)で投稿とユーザーを結合して、filter(User.name == "田中太郎")で条件を指定してるんだ。

joinは、2つのテーブルを「結合」する操作なんだ。例えば、postsテーブルとusersテーブルを、user_idをキーにして結合することで、投稿と著者の情報を一緒に取得できるんだよ。

SQLで書くと、こんな感じ。

SELECT posts.*
FROM posts
JOIN users ON posts.user_id = users.id
WHERE users.name = '田中太郎';

SQLAlchemyなら、これをPythonで書けるってわけ。

いい質問だね！実は、アクセスするたびにデータベースにクエリが飛んでるんだ。そして、これが N+1問題 という有名なパフォーマンス問題の原因になるんだよ。

例えば、こんなコードを見てみよう。

# 全投稿を取得
posts = session.query(Post).all()  # 1回のクエリ

# 各投稿の著者名を表示
for post in posts:
    print(f"{post.title} by {post.author.name}")  # 投稿ごとにクエリ！

投稿が50件あったら、どのくらいクエリが発行されると思う？

実は、51回なんだ！

1. 投稿を全件取得: 1回のクエリ
2. 各投稿の著者を取得: 50回のクエリ

合計で1 + N（投稿数）= 51回のクエリが発行される。これがN+1問題なんだ。

これが、パフォーマンスの大きな問題になるんだ。投稿が1000件あったら、1001回もクエリが飛ぶことになる。データベースへの問い合わせは時間がかかるから、これは避けたいんだよ。

安心して！SQLAlchemyには、これを解決する方法があるんだ。joinedload を使うんだよ。

from sqlalchemy.orm import joinedload

# 投稿と著者を一緒に取得
posts = session.query(Post).options(joinedload(Post.author)).all()

# これで何回アクセスしても、追加のクエリは発行されない
for post in posts:
    print(f"{post.title} by {post.author.name}")  # 追加クエリなし！

joinedload(Post.author)を指定することで、投稿と著者を 1回のクエリ でまとめて取得してくれるんだ。

実際に比較してみよう。

# 悪い例: N+1問題あり
posts = session.query(Post).all()
for post in posts:
    _ = post.author.name  # 各投稿ごとにクエリが飛ぶ

# 良い例: joinedload使用
posts = session.query(Post).options(joinedload(Post.author)).all()
for post in posts:
    _ = post.author.name  # 追加クエリなし

投稿が50件ある場合のクエリ数:

• 悪い例: 51回（1 + 50）
• 良い例: 1回

パフォーマンスが50倍以上改善されるんだ！

そう！実務では、このN+1問題に気をつけることがとても重要なんだ。特に、リスト表示のような場面では、必ずjoinedloadを使うようにしよう。

確かにね。実は、モデル定義の時にlazy="joined"を指定することもできるんだ。

class Post(Base):
    __tablename__ = "posts"
    # ...
    author = relationship("User", back_populates="posts", lazy="joined")

ただ、これだとすべてのクエリで自動的にjoinされちゃうから、必要な時だけjoinedloadを使う方が柔軟性があるんだよ。

あるよ！多対多 の関係っていうのもあるんだ。例えば、ブログの「投稿」と「タグ」の関係。

• 1つの投稿に、複数のタグがつけられる
• 1つのタグは、複数の投稿で使われる

これが多対多の関係なんだ。

多対多の関係は、中間テーブルを使って実現するんだ。

from sqlalchemy import Table

# 中間テーブル（投稿とタグの関連を保存）
post_tags = Table(
    "post_tags",
    Base.metadata,
    Column("post_id", Integer, ForeignKey("posts.id"), primary_key=True),
    Column("tag_id", Integer, ForeignKey("tags.id"), primary_key=True)
)

# 投稿モデル
class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    title = Column(String(200))
    content = Column(Text)

    # 多対多のリレーションシップ
    tags = relationship("Tag", secondary=post_tags, back_populates="posts")

# タグモデル
class Tag(Base):
    __tablename__ = "tags"
    id = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True)

    # 多対多のリレーションシップ
    posts = relationship("Post", secondary=post_tags, back_populates="tags")

secondaryは、中間テーブルを指定するパラメータなんだ。これによって、SQLAlchemyが自動的に中間テーブルを通じて関連付けをしてくれるんだよ。

使い方は1対多と同じような感じ。

# タグ作成
tag_python = Tag(name="Python")
tag_database = Tag(name="Database")

# 投稿作成とタグ付け
post = Post(title="SQLAlchemyの使い方", content="...")
post.tags.append(tag_python)
post.tags.append(tag_database)

session.add_all([tag_python, tag_database, post])
session.commit()

# 投稿のタグ一覧
for tag in post.tags:
    print(tag.name)  # Python, Database

# タグがついた投稿一覧
for post in tag_python.posts:
    print(post.title)

トランザクションは、データベース操作をまとめて扱う仕組みなんだ。「全部成功」か「全部失敗」の2択にすることで、データの整合性を保つんだよ。

例えば、銀行の送金を考えてみて。

1. A さんの口座から1000円引く
2. B さんの口座に1000円足す

この2つの処理が、片方だけ成功したらまずいよね？

そう！だから、トランザクションで 両方成功するか、両方失敗するか を保証するんだ。

try:
    # トランザクション開始
    user_a = session.query(User).filter_by(name="Aさん").first()
    user_b = session.query(User).filter_by(name="Bさん").first()

    # 送金処理
    user_a.balance -= 1000
    user_b.balance += 1000

    # 問題がなければコミット
    session.commit()
    print("送金成功！")

except Exception as e:
    # エラーが起きたらロールバック（全部取り消し）
    session.rollback()
    print(f"送金失敗: {e}")

エラーが起きたら、rollback()で変更を全部なかったことにできるんだ。

実務では、コンテキストマネージャーを使うともっと安全だよ。

from contextlib import contextmanager

@contextmanager
def get_session():
    session = SessionLocal()
    try:
        yield session
        session.commit()  # 成功したら自動コミット
    except:
        session.rollback()  # 失敗したら自動ロールバック
        raise
    finally:
        session.close()  # 必ず閉じる

# 使い方
with get_session() as session:
    user = User(name="山田太郎", email="yamada@example.com")
    session.add(user)
    # ブロックを抜けると自動でcommit/rollback/close

これなら、commit()やclose()を忘れる心配がないんだ。

いくつか紹介するね。

1. カウント

# ユーザー数を取得
user_count = session.query(User).count()
print(f"登録ユーザー数: {user_count}人")

# 条件付きカウント
adult_count = session.query(User).filter(User.age >= 20).count()

2. 存在チェック

# メールアドレスが既に登録されているか
exists = session.query(User).filter_by(email="tanaka@example.com").first() is not None

# もっとシンプルな書き方
from sqlalchemy import exists as sql_exists
exists = session.query(sql_exists().where(User.email == "tanaka@example.com")).scalar()

3. 一括更新

# 全ユーザーの年齢に1を足す
session.query(User).update({User.age: User.age + 1})
session.commit()

# 条件付き更新
session.query(User).filter(User.age < 20).update({User.status: "minor"})
session.commit()

一件ずつ更新するより、ずっと速いんだ。

4. 一括削除

# 条件に合うレコードを全削除
session.query(User).filter(User.age < 18).delete()
session.commit()

ただし、一括削除は取り消せないから、慎重に使う必要があるよ。

そうなんだ。だから、本番環境では削除の前に必ず確認するようにしよう。

# 削除対象を確認
users_to_delete = session.query(User).filter(User.age < 18).all()
print(f"{len(users_to_delete)}人のユーザーを削除します")
for user in users_to_delete:
    print(f"  - {user.name}")

# 確認してから削除
if input("本当に削除しますか？ (yes/no): ") == "yes":
    session.query(User).filter(User.age < 18).delete()
    session.commit()

まず、echo=True を設定するのが一番簡単だよ。

engine = create_engine("sqlite:///users.db", echo=True)

これで、実行されるSQL文が全部表示されるから、何が起きてるか分かりやすいんだ。

あと、クエリを実行する前に、どんなSQL文が生成されるか確認することもできるよ。

query = session.query(User).filter(User.age >= 25)
print(query)  # SELECT文が表示される

これで、期待通りのクエリが生成されてるか確認できるんだ。

いくつか紹介しよう。

1. DetachedInstanceError

session = SessionLocal()
user = session.query(User).first()
session.close()

print(user.posts)  # エラー！Sessionが閉じられている

これは、Sessionを閉じた後にリレーションシップにアクセスしようとした時のエラー。必要なデータは、Sessionを閉じる前に読み込んでおこう。

joinedloadで事前に読み込むか、Sessionを閉じる前にアクセスするんだ。

# 方法1: joinedloadで事前読み込み
user = session.query(User).options(joinedload(User.posts)).first()
session.close()
print(user.posts)  # OK

# 方法2: Sessionを閉じる前にアクセス
user = session.query(User).first()
_ = user.posts  # ここでアクセスして読み込む
session.close()
print(user.posts)  # OK

2. IntegrityError

user = User(name="田中太郎", email="existing@example.com")
session.add(user)
session.commit()  # エラー！メールアドレスが重複

UNIQUE制約違反や、NOT NULL制約違反の時に起きるエラーだよ。

事前にチェックするか、try-exceptで捕捉するんだ。

from sqlalchemy.exc import IntegrityError

try:
    user = User(name="田中太郎", email="existing@example.com")
    session.add(user)
    session.commit()
except IntegrityError:
    session.rollback()
    print("メールアドレスが既に使われています")

いいね！簡単なブログシステムを作ってみよう。

from sqlalchemy import create_engine, Column, Integer, String, Text, ForeignKey, DateTime
from sqlalchemy.orm import declarative_base, sessionmaker, relationship, joinedload
from datetime import datetime

engine = create_engine("sqlite:///blog.db")
Base = declarative_base()

# ユーザーモデル
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    email = Column(String(100), unique=True, nullable=False)
    posts = relationship("Post", back_populates="author")

# 投稿モデル
class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(Text, nullable=False)
    created_at = Column(DateTime, default=datetime.now)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
    author = relationship("User", back_populates="posts")
    comments = relationship("Comment", back_populates="post")

# コメントモデル
class Comment(Base):
    __tablename__ = "comments"
    id = Column(Integer, primary_key=True)
    content = Column(Text, nullable=False)
    created_at = Column(DateTime, default=datetime.now)
    post_id = Column(Integer, ForeignKey("posts.id"), nullable=False)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
    post = relationship("Post", back_populates="comments")
    author = relationship("User")

Base.metadata.create_all(engine)
SessionLocal = sessionmaker(bind=engine)

# 使用例
def create_post(user_email, title, content):
    """投稿を作成"""
    session = SessionLocal()
    try:
        user = session.query(User).filter_by(email=user_email).first()
        if not user:
            print("ユーザーが見つかりません")
            return None

        post = Post(title=title, content=content, author=user)
        session.add(post)
        session.commit()
        print(f"✓ 投稿『{title}』を作成しました")
        return post.id
    except Exception as e:
        session.rollback()
        print(f"✗ エラー: {e}")
        return None
    finally:
        session.close()

def get_post_with_comments(post_id):
    """投稿とコメントを取得（N+1問題を回避）"""
    session = SessionLocal()
    try:
        post = (
            session.query(Post)
            .options(
                joinedload(Post.author),
                joinedload(Post.comments)
            )
            .filter_by(id=post_id)
            .first()
        )

        if post:
            print(f"\n【{post.title}】")
            print(f"著者: {post.author.name}")
            print(f"投稿日: {post.created_at}")
            print(f"\n{post.content}\n")
            print(f"コメント ({len(post.comments)}件):")
            for comment in post.comments:
                print(f"  - {comment.author.name}: {comment.content}")

        return post
    finally:
        session.close()

# テストデータ作成
session = SessionLocal()
user1 = User(name="田中太郎", email="tanaka@example.com")
user2 = User(name="佐藤花子", email="sato@example.com")
session.add_all([user1, user2])
session.commit()
session.close()

# 投稿作成
create_post("tanaka@example.com", "SQLAlchemyの紹介", "SQLAlchemyは便利なORMです。")

# 投稿とコメントを表示
get_post_with_comments(1)

そう！これで、ユーザー、投稿、コメントの3つのテーブルを関連付けて使えるようになったよ。実際のアプリケーションも、基本的にはこの延長なんだ。

応用編で学んだことをまとめてみよう。

• リレーションシップ: ForeignKeyとrelationshipでテーブルを関連付ける
• 1対多: ユーザーと投稿のような関係
• 多対多: 投稿とタグのような関係、中間テーブルが必要
• N+1問題: relationshipのアクセスで大量のクエリが発行される問題
• joinedload: N+1問題の解決方法、事前にデータをまとめて取得
• トランザクション: commit()で確定、rollback()で取り消し
• デバッグ: echo=Trueで実行されるSQLを確認

これで、実務レベルのSQLAlchemyが使えるようになったよ！

そうだね！でも、SQLAlchemyにはまだまだ機能があるんだ。例えば:

• マイグレーション: Alembicを使ったスキーマの変更管理
• 非同期対応: asyncioを使った非同期データベース操作
• カスタムデータ型: 独自のデータ型の定義
• イベントリスナー: 挿入・更新時の自動処理

興味が出てきたら、公式ドキュメントを読んでみるといいよ。

SQLAlchemyは、Pythonでデータベースを扱う上で最も人気のあるライブラリの一つなんだ。Django ORMやPeeweeなど他の選択肢もあるけど、SQLAlchemyは柔軟性と機能の豊富さで選ばれることが多いんだよ。

実際のプロジェクトで使いながら、少しずつ慣れていこう。困った時は、公式ドキュメントやStack Overflowが役に立つから、ぜひ活用してね！