Python から MySQL や PostgreSQL、SQLite を操作するなら、SQLAlchemy を使うと、接続の管理・SQL の組み立て・トランザクションを一貫した書き方で扱えます。この記事では、現在の推奨である 2.0 形式の書き方で、接続、テーブル定義、追加・検索・更新・削除、トランザクションまでを順に説明します。古い記事に多い 1.x 形式の session.query() との違いや、接続切れ・文字化け・N+1 問題など、実運用でつまずきやすい点もあわせて扱います。
準備：インストールとドライバ
SQLAlchemy 本体のほかに、使うデータベースに合わせたドライバが必要です。仮想環境を作ってからインストールします。
python -m venv .venv
source .venv/bin/activate # Windows は .venv\Scripts\activate
pip install sqlalchemy pymysql # MySQL / MariaDB の場合
# pip install sqlalchemy "psycopg[binary]" # PostgreSQL の場合SQLite は Python 標準の sqlite3 を使うため、追加のドライバは要りません。手元で試すなら SQLite から始め、接続先の URL だけを本番用に切り替える進め方が手軽です。
この記事のコードは SQLAlchemy 2.0 系の書き方です（2026年10月現在）。pip show sqlalchemy でバージョンを確認し、1.3 以前が入っている場合は、ここで紹介する DeclarativeBase や Mapped が使えないため更新してください。既存のアプリを更新する場合は、公式の移行ガイドに沿って、テスト環境で動作を確かめてから進めます。
接続する：Engine を作る
最初に、データベースへの接続を管理する Engine を作ります。Engine は接続プールを内部に持っており、アプリ全体で1つ作って使い回すものです。リクエストのたびに作り直すと、接続が増え続ける原因になります。
import os
from sqlalchemy import create_engine, text
from sqlalchemy.engine import URL
url = URL.create(
"mysql+pymysql",
username=os.environ["DB_USER"],
password=os.environ["DB_PASSWORD"],
host=os.environ.get("DB_HOST", "127.0.0.1"),
port=3306,
database="appdb",
query={"charset": "utf8mb4"},
)
engine = create_engine(
url,
pool_pre_ping=True, # 使う前に接続が生きているか確認する
pool_recycle=3600, # 1時間使った接続は作り直す
)
# 接続確認
with engine.connect() as conn:
print(conn.execute(text("SELECT 1")).scalar())ここでのポイントは次のとおりです。
- パスワードはコードに書かない：環境変数や、権限を絞った設定ファイルから読み込みます。リポジトリに含めないよう注意してください。
- URL.create を使う：
mysql+pymysql://user:pass@host/dbの文字列で書くと、パスワードに@や/が含まれているときに解釈を誤ります。URL.createなら記号を気にせず渡せます。 - charset=utf8mb4 を指定する：MySQL で絵文字などを扱うと、文字コードが
utf8（3バイト）のままでは保存に失敗します。テーブル側も utf8mb4 にしておきます。 - pool_pre_ping と pool_recycle：MySQL はしばらく使われない接続をサーバー側で切断します（
wait_timeout）。切れた接続を使うと「MySQL server has gone away」などのエラーになるため、この2つで防ぎます。
SQLite なら create_engine("sqlite:///app.db") だけで動きます。
テーブルをモデルとして定義する
2.0 形式では、DeclarativeBase を継承した基底クラスを作り、型ヒント付きで列を定義します。
from datetime import datetime
from sqlalchemy import ForeignKey, String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
class Base(DeclarativeBase):
pass
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
email: Mapped[str] = mapped_column(String(255), unique=True)
name: Mapped[str] = mapped_column(String(100))
created_at: Mapped[datetime] = mapped_column(server_default=func.now())
posts: Mapped[list["Post"]] = relationship(back_populates="author")
class Post(Base):
__tablename__ = "posts"
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str] = mapped_column(String(200))
body: Mapped[str | None]
user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
author: Mapped["User"] = relationship(back_populates="posts")
Base.metadata.create_all(engine) # 無いテーブルだけを作成するMapped[str | None] のように None を含めた列は NULL を許可し、含めない列は NOT NULL になります。str | None の書き方は Python 3.10 以降です。それより古い環境では Optional[str] を使います。
create_all は存在しないテーブルを作るだけで、既存テーブルへの列の追加や型の変更は行いません。運用中のテーブル構造を変えるなら、マイグレーションツールの Alembic を使って変更履歴を管理します。本番のテーブルを変更する前には、データベースのバックアップを取ってください。
Session で追加・検索・更新・削除する
ORM の操作は Session を通して行います。sessionmaker で作成用の設定を1つ用意しておき、処理のまとまりごとに Session を開いて閉じます。
from sqlalchemy import select
from sqlalchemy.orm import sessionmaker
SessionLocal = sessionmaker(engine)
# 追加
with SessionLocal.begin() as session: # ブロックを抜けると自動で commit、例外なら rollback
user = User(email="taro@example.com", name="Taro")
user.posts.append(Post(title="はじめての投稿", body="本文"))
session.add(user)
# 検索
with SessionLocal() as session:
stmt = select(User).where(User.email == "taro@example.com")
user = session.scalars(stmt).one_or_none()
stmt = select(Post).where(Post.title.like("%投稿%")).order_by(Post.id.desc()).limit(10)
for post in session.scalars(stmt):
print(post.id, post.title)
found = session.get(User, 1) # 主キーで1件取得
# 更新と削除
with SessionLocal.begin() as session:
user = session.scalars(select(User).where(User.name == "Taro")).first()
if user is not None:
user.name = "Taro Yamada" # 属性を変えるだけで UPDATE される
old = session.get(Post, 1)
if old is not None:
session.delete(old)取得結果の受け取り方は、one()（ちょうど1件でなければ例外）、one_or_none()（0件なら None、2件以上なら例外）、first()（先頭の1件）を目的に合わせて使い分けます。「メールアドレスで1件」のように1件であるはずの検索では、one_or_none() を使うと、重複データがあったときに気づけます。
1.x 形式の session.query(User).filter_by(...) も 2.0 で動きますが、旧来の書き方という位置づけです。新しく書くコードは select() と session.scalars() / session.execute() にそろえておくと、今後の更新で困りません。
SQL を直接書くときはパラメータで渡す
集計や既存の SQL を流用したい場合は、text() で SQL を直接実行できます。このとき、値を文字列の連結や f-string で埋め込んではいけません。SQL インジェクションの原因になります。
from sqlalchemy import text
keyword = input("検索語: ")
# 悪い例：入力がそのまま SQL になる
# conn.execute(text(f"SELECT * FROM posts WHERE title LIKE '%{keyword}%'"))
# 良い例：:name の形でパラメータとして渡す
with engine.connect() as conn:
rows = conn.execute(
text("SELECT id, title FROM posts WHERE title LIKE :kw"),
{"kw": f"%{keyword}%"},
)
for row in rows:
print(row.id, row.title)テーブル名や列名のような識別子はパラメータにできません。ユーザー入力で並び替える列を選ぶ場合は、許可する列名の一覧を用意して、その中から選ぶ形にします。
つまずきやすい点
N+1 問題：投稿一覧を取得してから、ループの中で post.author.name を参照すると、投稿の件数分だけユーザーの SELECT が発行されます。件数が多いと遅くなるため、まとめて読み込む指定をします。
from sqlalchemy.orm import selectinload
with SessionLocal() as session:
stmt = select(Post).options(selectinload(Post.author)).limit(100)
for post in session.scalars(stmt):
print(post.title, post.author.name)発行される SQL は、create_engine(..., echo=True) にすると標準出力に表示されます。開発中にこれで確認すると、想定外のクエリに気づけます。本番では値がログに出るため無効にしておきます。
Session を閉じた後の参照：with ブロックを抜けた後に、まだ読み込んでいない関連（user.posts など）を参照するとエラーになります。必要なデータは Session の中で読み込むか、上の selectinload で先に取得しておきます。
commit 後の属性：既定では commit 後にオブジェクトの属性が期限切れになり、次に参照したときに再読み込みされます。Session を閉じた後に値を使う設計なら、sessionmaker(engine, expire_on_commit=False) を検討します。
長いトランザクション：大量の行を1つのトランザクションで更新すると、ロックが長く続きます。数千件単位に分けて commit するなど、ほかの処理を止めない工夫が要ります。
まとめ
- Engine はアプリ全体で1つ作り、
URL.createと環境変数で接続情報を渡す - MySQL では
charset=utf8mb4、pool_pre_pingとpool_recycleで文字化けと接続切れを防ぐ - モデルは
DeclarativeBase・Mapped・mapped_columnで定義し、構造の変更は Alembic で管理する - 検索は
select()とsession.scalars()、書き込みはSessionLocal.begin()のブロックで行う - 生の SQL では値をパラメータで渡し（連結しない）、識別子は許可リストから選ぶ
- 関連の読み込みは
selectinloadで N+1 問題を避け、echo=Trueで発行される SQL を確かめる