SQLAlchemyはPythonで最も人気のあるORM（オブジェクトリレーショナルマッピング）フレームワークの一つで、効率的かつ柔軟なデータベース操作を提供します。この記事では、SQLAlchemy ORMを利用してデータベース操作を行う方法について説明します。
インストール
pip install sqlalchemy特定のデータベースに接続するには、それに応じたドライバもインストールする必要があります：
# PostgreSQL
pip install psycopg2-binary
# MySQL
pip install mysql-connector-python
# SQLite (Python標準ライブラリに含まれているため、追加インストール不要)
コアコンセプト
- Engine：データベースとの通信を行うエンジン
- Session：データベースセッションで、すべての永続化操作を管理する
- Model：データモデルクラスで、データベースのテーブルに対応する
- Query：データベースクエリを構築し実行するためのオブジェクト
データベースへの接続
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
# SQLiteの例
engine = create_engine('sqlite:///sample.db', echo=True)
# PostgreSQLの例
# engine = create_engine('postgresql://ユーザー名:パスワード@localhost:5432/データベース名')
# MySQLの例
# engine = create_engine('mysql+mysqlconnector://ユーザー名:パスワード@localhost:3306/データベース名')
# セッションファクトリーを作成
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# セッションインスタンスを作成
session = SessionLocal()
データモデルの定義
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, declarative_base
# 基底クラスを作成
Base = declarative_base()
class Person(Base):
__tablename__ = 'people'
id = Column(Integer, primary_key=True, index=True)
name = Column(String(50), nullable=False)
mail = Column(String(100), unique=True, index=True)
# 1対多の関係を定義
articles = relationship("Article", back_populates="writer")
class Article(Base):
__tablename__ = 'articles'
id = Column(Integer, primary_key=True, index=True)
headline = Column(String(100), nullable=False)
text = Column(String(500))
writer_id = Column(Integer, ForeignKey('people.id'))
# 多対1の関係を定義
writer = relationship("Person", back_populates="articles")
# 多対多の関係（関連テーブルを通じて）を定義
categories = relationship("Category", secondary="article_categories", back_populates="articles")
class Category(Base):
__tablename__ = 'categories'
id = Column(Integer, primary_key=True, index=True)
label = Column(String(30), unique=True, nullable=False)
articles = relationship("Article", secondary="article_categories", back_populates="categories")
# 関連テーブル（多対多の関係用）
class ArticleCategory(Base):
__tablename__ = 'article_categories'
article_id = Column(Integer, ForeignKey('articles.id'), primary_key=True)
category_id = Column(Integer, ForeignKey('categories.id'), primary_key=True)
データベーステーブルの作成
# すべてのテーブルを作成
Base.metadata.create_all(bind=engine)
# すべてのテーブルを削除
# Base.metadata.drop_all(bind=engine)
基本的なCRUD操作
データの作成
# 新しいユーザーを作成
new_person = Person(name="田中太郎", mail="tanaka@example.com")
session.add(new_person)
session.commit()
# 一括で作成
session.add_all([
Person(name="佐藤次郎", mail="sato@example.com"),
Person(name="鈴木三郎", mail="suzuki@example.com")
])
session.commit()
データの読み取り
# 全ユーザーを取得
people = session.query(Person).all()
# 最初のユーザーを取得
first_person = session.query(Person).first()
# IDでユーザーを取得
person = session.query(Person).get(1)
データの更新
# 取得して更新
person = session.query(Person).get(1)
person.name = "田中太郎二世"
session.commit()
# 一括で更新
session.query(Person).filter(Person.name.like("田中%")).update({"name": "田中氏"}, synchronize_session=False)
session.commit()
データの削除
# 取得して削除
person = session.query(Person).get(1)
session.delete(person)
session.commit()
# 一括で削除
session.query(Person).filter(Person.name == "佐藤次郎").delete(synchronize_session=False)
session.commit()
データの検索
基本的な検索
# すべてのレコードを取得
people = session.query(Person).all()
# 特定のフィールドを取得
names = session.query(Person.name).all()
# 並べ替え
people = session.query(Person).order_by(Person.name.desc()).all()
# 結果数の制限
people = session.query(Person).limit(10).all()
# オフセット
people = session.query(Person).offset(5).limit(10).all()
フィルタリング検索
from sqlalchemy import or_
# 等価性によるフィルタリング
person = session.query(Person).filter(Person.name == "田中太郎").first()
# 模糊検索
people = session.query(Person).filter(Person.name.like("田中%")).all()
# INによる検索
people = session.query(Person).filter(Person.name.in_(["田中太郎", "佐藤次郎"])).all()
# 複数条件での検索
people = session.query(Person).filter(
Person.name == "田中太郎",
Person.mail.like("%@example.com")
).all()
# OR条件での検索
people = session.query(Person).filter(
or_(Person.name == "田中太郎", Person.name == "佐藤次郎")
).all()
# 不等価性によるフィルタリング
people = session.query(Person).filter(Person.name != "田中太郎").all()
集計検索
from sqlalchemy import func
# カウント
count = session.query(Person).count()
# グループごとのカウント
person_article_count = session.query(
Person.name,
func.count(Article.id)
).join(Article).group_by(Person.name).all()
# 合計値や平均値などの計算
avg_id = session.query(func.avg(Person.id)).scalar()
結合検索
# 内部結合
results = session.query(Person, Article).join(Article).filter(Article.headline.like("%Python%")).all()
# 左外部結合
results = session.query(Person, Article).outerjoin(Article).all()
# 結合条件の指定
results = session.query(Person, Article).join(Article, Person.id == Article.writer_id).all()
関係操作
# 関係を持つオブジェクトを作成
person = Person(name="高橋四郎", mail="takahashi@example.com")
article = Article(headline="最初のブログ記事", text="こんにちは、世界！", writer=person)
session.add(article)
session.commit()
# 関係を通じてアクセス
print(f"記事 '{article.headline}' の著者は {article.writer.name}")
print(f"ユーザー {person.name} のすべての記事:")
for a in person.articles:
print(f" - {a.headline}")
# 多対多の関係操作
python_category = Category(label="Python")
sqlalchemy_category = Category(label="SQLAlchemy")
article.categories.append(python_category)
article.categories.append(sqlalchemy_category)
session.commit()
print(f"記事 '{article.headline}' のカテゴリ:")
for category in article.categories:
print(f" - {category.label}")
トランザクション管理
# 自動コミットトランザクション
try:
person = Person(name="テストユーザー", mail="test@example.com")
session.add(person)
session.commit()
except Exception as e:
session.rollback()
print(f"エラーが発生しました: {e}")
# トランザクションコンテキストマネージャーの使用
from sqlalchemy.orm import Session
def create_person(session: Session, name: str, mail: str):
try:
person = Person(name=name, mail=mail)
session.add(person)
session.commit()
return person
except:
session.rollback()
raise
# ネストされたトランザクション
with session.begin_nested():
person = Person(name="トランザクションユーザー", mail="transaction@example.com")
session.add(person)
# セーブポイント
savepoint = session.begin_nested()
try:
person = Person(name="セーブポイントユーザー", mail="savepoint@example.com")
session.add(person)
savepoint.commit()
except:
savepoint.rollback()
ベストプラクティス
- セッション管理：各リクエストごとに新しいセッションを作成し、リクエスト終了時に閉じる
- 例外処理：常に例外を処理し、適切にトランザクションをロールバックする
- 遅延ロード：N+1クエリ問題に注意し、eager loadingを使って最適化する
- コネクションプール：適切なコネクションプールサイズとタイムアウト設定を行う
- データ検証：モデル層やアプリケーション層でデータの整合性を検証する
# コンテキストマネージャーを使用してセッションを管理
from contextlib import contextmanager
@contextmanager
def get_db():
db = SessionLocal()
try:
yield db
db.commit()
except Exception:
db.rollback()
raise
finally:
db.close()
# 使用例
with get_db() as db:
person = Person(name="コンテキストユーザー", mail="context@example.com")
db.add(person)