- Pythonでデータベースを使いたい！
- でもSQLクエリを覚えるのは面倒…
- SQLAlchemyのコードの意味を理解したい
SQLをほぼ触ったことがなかった僕が、SQLAlchemyを使ったらPythonコードだけでデータベース操作できるようになりました。
SQLAlchemyは本当に使いやすくて、「これまでにCSVやExcelファイル（.xlsx）などでデータ操作をしていたけれど、データベースにデータを蓄えてみたい」という方にはうってつけのライブラリだと思います。
とはいえ、Python初心者目線ではSQLAlchemyのコードはやや分かりづらいのも事実。
そこで本記事では僕が最初にSQLAlchemy使ってみたときに感じた疑問にも答える形でお伝えさせていただきます。
本記事を参考にしていただければ、コードの意味を理解しつつSQLAlchemyを使ったデータベース操作がスラスラとできるようになっているはずです。
SQLAlchemyについて簡単に解説
SQLAlchemyでSQLを操作するためには、データベースに繋ぐ部分などで準備のためのコードが数行必要です。
決まり文句的に書いてしまっても良いのですが、意味がわからず使うのも気持ち悪いと思うので（僕は結構気になるタイプ）ここで概略を説明します。
SQLAlchemyでデータベース操作するまでの流れ
SQLAlchemyを使ってデータベースを操作するためには、次のような手順が必要になってきます。
- データベースの準備
- 各列に何を格納するかを決定
- データベースに接続
- 定義したテーブルを作成
- データベースにアクセスする窓口を起動
- sessionが開始される
- データベースを操作
- トランザクションの開始・終了
最終的には「データベースを操作」をしたいのですが、そのための準備としてこのような手順が必要になってくるというわけですね。
データベースを扱うためには、まずデータを格納する列の構造を決めなければいけませんし、データベースの種類（sqliteやMySQLなど）を決めたり、データベースに接続する必要もあります。
ちょっと手順が多く思えるかもしれませんが、事前準備としてこれらが必要になることを知っておきましょう。
SQLAlchemyにはORMとCoreの二つの概念がある
最初はあまり気にしなくても良いのですが、SQLAlchemyの中にはORMとCoreの二つの概念があります。主な違いは次のとおり。
- ORM:
- 直感的に操作できる
- リレーションなどは自動的に管理される
- 複雑なSQLクエリは苦手
- Core
- SQLの概念をより直接的に扱える
- ORMに比べてより細かいコントロールが可能
最初のうちは分かりやすいORMの方を使えばOKだと思いますが、ORMで限界を感じた時には「Coreならできるかも！」と思い出していただければいいと思います。
SQLAlchemyの使い方
さて、ここから本題の具体的な使い方の説明に入っていきます。
インストール
まずはお使いのパッケージ管理ツールの方法にしたがって、SQLAlchemyのインストールしてください。
# pipの場合
pip install sqlalchemy
# poetryの場合
poetry add sqlalchemypoetryは別記事で導入方法などを解説しました。
簡単なデータベースを作成して保存する流れ
まずはsqliteを使って簡単なデータをカレントディレクトリのexample.dbに記録するだけのサンプルコードをご紹介します。
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# ベースクラスを定義
Base = declarative_base()
# テーブルの定義
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
name = Column(String)
email = Column(String)
# DBエンジンを作成し、DBに接続
engine = create_engine("sqlite:///example.db")
# 実際にテーブルを作成
Base.metadata.create_all(engine)
# セッションを作成
Session = sessionmaker(bind=engine)
session = Session()
# 新しいレコードを定義
new_user = User(name="xxx", email="xxx@example.com")
# ユーザーをデータベースに追加（実際にはまだ追加されない）
session.add(new_user)
# データベースに永続的に保存される
session.commit()example.dbはBase.metadata.create_all(engine)を実行するタイミングで作成されますので、ご自身で何か作成する必要はありません。
データの挿入（Insert）
次はORMでデータの挿入を行うコードです。
# ORMにより実行
new_user = User(name="xxx", email="xxx@example.com")
session.add(new_user)
session.commit()こちらは先ほどサンプルコードで示したものと同じですね。
普段からPythonを書き慣れている方からすると、かなり直感的で読みやすいのではないでしょうか？Userはモデルを定義したクラス名を指定します。
続いては、Coreでデータ挿入を行うコードです。
from sqlalchemy import insert
from sqlalchemy import create_engine, Table, MetaData
from sqlalchemy.orm import sessionmaker
# Coreでは必要
metadata = MetaData()
users = Table("users", metadata, autoload_with=engine)
statement = insert(users).values(name="xxx", email="xxx@example.com")
session.execute(statement)
session.commit()InsertというSQLステートメントをstatement変数に格納して、それを実行・コミットすることでデータベースに反映されます。
データの選択（Select）
ORMを使ってデータの選択をするコードは次のとおりです。
users = session.query(User).filter(User.name == "xxx").all()
for user in users:
print(user.email)少しDjangoのクエリセットを扱う場合に似ている気がします…該当のuserが取得できたら、ループで取り出していく形ですね。
続いてはCoreを使ったコード例です。
from sqlalchemy import select
from sqlalchemy import create_engine, Table, MetaData
from sqlalchemy.orm import sessionmaker
# Coreでは必要
metadata = MetaData()
users = Table("users", metadata, autoload_with=engine)
statement = select(users).where(users.c.name == 'xxx')
results = session.execute(statement).fetchall()
for result in results:
print(result)whereメソッドを使っているあたりが間接的にでもSQLを扱っていると実感。
データの更新（Update）
まずはORMの場合です。
session.query(User).filter(User.name == 'xxx').update({User.email: 'xxx2@example.com'})
session.commit()まずはfilterメソッドでレコードを取得した上で、該当の列を更新していきます。
データ更新はupdateメソッドを使い、更新後の値は辞書で渡しましょう。
続いてはCoreを使った場合です。
from sqlalchemy import update
from sqlalchemy import create_engine, Table, MetaData
from sqlalchemy.orm import sessionmaker
# Coreでは必要
metadata = MetaData()
users = Table("users", metadata, autoload_with=engine)
statement = update(users).where(users.c.name == 'xxx').values(email='xxx2@example.com')
session.execute(statement)
session.commit()user.c.nameのcはSQLAlchemyでColumnにアクセスするときに使われる書き方です。SQLではこのような表現がないので、あくまでSQLAlchemyだけに使う表現だということをご理解ください。
データの削除（Delete）
ORMを使った方法からご紹介します。書き方はかなり簡潔です。
session.query(User).filter(User.name == 'xxx').delete()
session.commit()続いてCoreを使った方法です。
from sqlalchemy import delete
from sqlalchemy import create_engine, Table, MetaData
from sqlalchemy.orm import sessionmaker
# Coreでは必要
metadata = MetaData()
users = Table("users", metadata, autoload_with=engine)
statement = delete(users).where(users.c.name == 'xxx')
session.execute(statement)
session.commit()こちらもややコードが長くなるものの、簡潔に表現できます。
データの取得
すべてのデータを取得する方法。
# 全取得
users = session.query(User).all()
for user in users:
print(user.name, user.email)特定のユーザーを取得する方法。以下の場合はid=1のユーザーを取得します。
user = session.query(User).get(1)
print(user.name, user.email)フィルタリング。
users = session.query(User).filter(User.name.like("%xxx%")).all()
for user in users:
print(user.name, user.email)%は0個以上の任意の文字列を表すワイルドカードです。Excelでいうところの*ですね。
任意の一文字を表したい場合には_を使います。
ソート
名前でソートしてすべてのユーザーを取得するにはorder_by()メソッドを使います。
users = session.query(User).order_by(User.name).all()
for user in users:
print(user.name, user.email)Pandasを使い慣れた方なら馴染みやすいかもしれません。
集計
ユーザーの総数を取得するコードはこちらです。
count = session.query(User).count()
print(count)ロールバックの実装方法
データベースへの登録中に何かエラーが発生した場合、データの整合性を保つためにロールバックということが行われることがあります。
try:
new_user = User(name="xxx", email="xxx@example.com")
session.add(new_user)
session.commit()
except:
# エラーが発生したら最初の状態に戻す
session.rollback()SQLAlchemyの最適化
これまでご紹介した方法でもSQLAlchemyを使うことはできますが、ここからは「さらにSQLAlchemyのベストプラクティスを追い求めたい」場合に参考にしていただけそうな知識をまとめます。
事前ロードにより高速化できる場合がある
SQLAlchemy ORMは遅延ロードがデフォルトです。
遅延ロードとはデータが必要になるまでロードを先送りにすることで、通常はこれによりパフォーマンスUPが期待できます。
ところが必要な情報がたくさんあり、SQLに対する命令が小出しに行われる場合には非効率です。俗に”N+1問題“と呼ばれるものです。
このような場合には一度のデータベースへのリクエストで必要なデータを事前に用意しておいた方がパフォーマンスが良くなる場合があります。
SQLAlchemyではjoinedloadやsubqueryloadといったものを用いて事前ロードの実装ができます。
joinedloadはLEFT OUTER JOINを実行します。
from sqlalchemy.orm import joinedload
# Userクラスのaddressesを一気にロード
query = session.query(User).options(joinedload(User.addresses))
for user in query:
print(user.name, user.addresses)関連する要素を一度に取得するので、要素の配置状態によってはデータ量が大きくなる場合があります。（1つの親エンティティに対して多数の子エンティティがある場合など）
一方でsubqueryloadでは次のようになります。
from sqlalchemy.orm import subqueryload
# Userクラスのaddressesを一気にロード
query = session.query(User).options(subqueryload(User.addresses))
for user in query:
print(user.name, user.addresses)こちらはjoinedloadと違ってカラムごとにSQLを発行するので列数だけクエリがされますが、1つの親エンティティに対して多数の子エンティティが存在する場合にはパフォーマンスのアップが期待できます。
デバッグのヒント
create_engineメソッドの引数としてecho=Trueを指定することで、すべてのSQLクエリが標準出力されます。
from sqlalchemy import create_engine
engine = create_engine("sqlite:///example.db", echo=True)便利なログ出力機能ですが、本番環境ではFalseにしておきましょう。
すべてのクエリを出力することでアプリケーションのパフォーマンスが落ちたり、機密情報がログに含まれてしまうことがあるためです。
例外処理
様々な例外が準備されているので、適切なものを使います。
from sqlalchemy.exc import SQLAlchemyError
try:
# SQLAlchemyの操作を実行するコード
# ...
except NoResultFound:
# レコードが見つからなかった場合の処理
# ...
except MultipleResultsFound:
# 複数のレコードが見つかった場合の処理
# ...
except InvalidRequestError:
# 無効な要求が行われた場合の処理
# ...
except OperationalError:
# データベース接続などの操作エラーが発生した場合の処理
# ...
except IntegrityError:
# データベースの整合性制約に違反した場合の処理
# ...
except DataError:
# データベースに関連するデータエラーが発生した場合の処理
# ...
except StatementError:
# SQLステートメントの実行中にエラーが発生した場合の処理
# ...
except SQLAlchemyError:
# SQLAlchemyの他のエラーをキャッチするための一般的な例外処理
# ...
except Exception:
# その他の一般的な例外をキャッチするためのブロック
# ...
finally:
# 必要に応じてクリーンアップ処理を実行するブロック
# ...詳しくは公式ドキュメントを参照してください。
まとめ
ざっくりではありますが、SQLAlchemyの概略から基本的な使い方まで横断的にお伝えしてきました。
少しずつでもコードを書いていくことでデータベースを触る感覚が掴めてくると思います。
ぜひサンプルコードなども参考にしつつ、開発中のアプリケーションに導入してみてください！
コメント