インストール方法

pip install sqlalchemy

特定のデータベースを使用する場合は追加パッケージが必要です：

# PostgreSQL
pip install psycopg2-binary

# MySQL
pip install mysql-connector-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://user:pass@localhost:5432/mydb')

# セッションファクトリの作成
SessionFactory = sessionmaker(bind=engine)
db_session = SessionFactory()

モデル定義

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

Base = declarative_base()

class Department(Base):
    __tablename__ = 'departments'
    id = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True)
    employees = relationship("Employee", back_populates="department")

class Employee(Base):
    __tablename__ = 'employees'
    id = Column(Integer, primary_key=True)
    name = Column(String(100))
    department_id = Column(Integer, ForeignKey('departments.id'))
    department = relationship("Department", back_populates="employees")
    projects = relationship("Project", secondary="employee_projects")

class Project(Base):
    __tablename__ = 'projects'
    id = Column(Integer, primary_key=True)
    title = Column(String(150))
    employees = relationship("Employee", secondary="employee_projects")

class EmployeeProject(Base):
    __tablename__ = 'employee_projects'
    employee_id = Column(Integer, ForeignKey('employees.id'), primary_key=True)
    project_id = Column(Integer, ForeignKey('projects.id'), primary_key=True)

テーブル操作

# テーブルの作成
Base.metadata.create_all(bind=engine)

# テーブルの削除
# Base.metadata.drop_all(bind=engine)

CRUD操作

データ作成

# 単一レコード作成
dept = Department(name="開発部")
db_session.add(dept)
db_session.commit()

# 複数レコード一括作成
db_session.bulk_save_objects([
    Employee(name="山田太郎", department_id=1),
    Employee(name="佐藤花子", department_id=1)
])
db_session.commit()

データ検索

# 条件付き検索
dev_team = db_session.query(Employee).filter(
    Employee.name.like("%山田%")
).all()

# 集計関数使用例
from sqlalchemy import func
emp_count = db_session.query(func.count(Employee.id)).scalar()

トランザクション管理

# 例外処理付きトランザクション
try:
    new_dept = Department(name="新規部署")
    db_session.add(new_dept)
    db_session.flush()  # 仮コミット
    
    new_emp = Employee(name="管理者", department_id=new_dept.id)
    db_session.add(new_emp)
    db_session.commit()
except Exception as e:
    db_session.rollback()
    print(f"エラー発生: {str(e)}")

ベストプラクティス

• セッションはリクエスト単位で管理
• 大量データ処理時はバッチ操作を活用
• N+1クエリ問題を回避するためのプリロード
• 接続プールの適切な設定
• アプリケーション層でのデータ検証