Zenn
株式会社XronotechPublicationへの投稿
⚡

FastAPIで学ぶデータベース操作の実践ガイド(SQLAlchemy ORM)

mizukich
2024/09/22に公開
ORM
FastAPI
DB
SQLAlchemy
tech

この記事では、Pythonの強力なORMであるSQLAlchemy ORMを使って、データベースからデータを取得するさまざまな方法を解説します。UserテーブルとAuthテーブルを例に取り、具体的なユースケースに基づいて多様な書き方を学んでいきましょう。

SQLAlchemy ORMとは？

SQLAlchemy ORMは、Pythonでデータベース操作をオブジェクト指向の形で行えるライブラリです。直接SQL文を書く代わりに、Pythonのコードでデータベースを操作できるため、生産性とコードの可読性が向上します。

テーブルの定義

まずは、例として使用するUserテーブルとAuthテーブルを定義します。Userはユーザー情報を、Authは認証情報を持つとします。

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

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    auth = relationship('Auth', back_populates='user')

class Auth(Base):
    __tablename__ = 'auths'
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.id'))
    email = Column(String)
    email_verified_at = Column(DateTime)
    user = relationship('User', back_populates='auth')
  • UserテーブルとAuthテーブルは一対一の関係を持ちます。
  • auth属性を通じて関連する認証情報にアクセスできます。

ユースケース別データ取得方法

ユースケース1：ユーザー名でユーザーを取得

方法1：filterを使用

def get_user_by_name(session, name):
    return session.query(User).filter(User.name == name).first()

実行結果

user = get_user_by_name(session, 'Alice')
print(user.name)  # 出力: Alice

方法2：filter_byを使用

def get_user_by_name(session, name):
    return session.query(User).filter_by(name=name).first()

実行結果

user = get_user_by_name(session, 'Bob')
print(user.name)  # 出力: Bob

解説

  • filterは条件式を受け取り、filter_byはキーワード引数を受け取ります。
  • どちらも同じ結果を返しますが、filter_byはシンプルな条件の場合に便利です。

ユースケース2：メールアドレスで認証情報を取得

def get_auth_by_email(session, email):
    return session.query(Auth).filter(Auth.email == email).first()

実行結果

auth = get_auth_by_email(session, 'alice@example.com')
print(auth.email)  # 出力: alice@example.com

解説

  • Authテーブルをクエリし、emailが一致する最初のレコードを取得します。

ユースケース3：ユーザーとその認証情報を同時に取得

方法1：joinを使用

def get_user_and_auth(session, user_id):
    return session.query(User).join(Auth).filter(User.id == user_id).first()

実行結果

user = get_user_and_auth(session, 1)
print(user.name)          # 出力: Alice
print(user.auth.email)    # 出力: alice@example.com

方法2：optionsとjoinedloadを使用

from sqlalchemy.orm import joinedload

def get_user_and_auth(session, user_id):
    return session.query(User).options(joinedload(User.auth)).filter(User.id == user_id).first()

実行結果

user = get_user_and_auth(session, 2)
print(user.name)          # 出力: Bob
print(user.auth.email)    # 出力: bob@example.com

解説

  • joinはデフォルトで内部結合を行います。
  • joinedloadは関連オブジェクトを一度のクエリでロードするため、パフォーマンスが向上します。

ユースケース4：メール確認済みのユーザーを取得

方法1：joinとフィルタリングを使用

def get_verified_users(session):
    return session.query(User).join(Auth).filter(Auth.email_verified_at != None).all()

実行結果

users = get_verified_users(session)
for user in users:
    print(user.name)
# 出力:
# Alice
# Charlie

方法2：existsを使用

from sqlalchemy.sql import exists

def get_verified_users(session):
    return session.query(User).filter(
        exists().where(
            (Auth.user_id == User.id) & (Auth.email_verified_at != None)
        )
    ).all()

実行結果

users = get_verified_users(session)
for user in users:
    print(user.name)
# 出力:
# Alice
# Charlie

解説

  • joinを使うと、関連テーブルの条件でフィルタリングできます。
  • existsはサブクエリを使用して、より効率的にデータを取得します。

ユースケース5：複数の条件でユーザーを検索

def search_users(session, name=None, email=None):
    query = session.query(User).join(Auth)
    if name:
        query = query.filter(User.name.like(f"%{name}%"))
    if email:
        query = query.filter(Auth.email.like(f"%{email}%"))
    return query.all()

実行結果

users = search_users(session, name='A', email='@example.com')
for user in users:
    print(user.name, user.auth.email)
# 出力:
# Alice alice@example.com

解説

  • 条件を動的に追加することで、柔軟な検索が可能です。
  • likeを使用して部分一致検索を行います。

ユースケース6：ページネーション（ページング）を実装

方法1：手動でのページネーション

def get_users_paginated(session, page=1, per_page=2):
    return session.query(User).offset((page - 1) * per_page).limit(per_page).all()

実行結果

users = get_users_paginated(session, page=1, per_page=2)
for user in users:
    print(user.name)
# 出力:
# Alice
# Bob

users = get_users_paginated(session, page=2, per_page=2)
for user in users:
    print(user.name)
# 出力:
# Charlie
# David

方法2：paginateを使用したページネーション

fastapi-paginationライブラリを使用すると、ページネーションをより簡潔に実装できます。

必要なパッケージのインストール

pip install fastapi-pagination

コード例

from fastapi_pagination import Page, add_pagination
from fastapi_pagination.ext.sqlalchemy import paginate
from pydantic import BaseModel
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session

app = FastAPI()

# ユーザーのスキーマを定義
class UserSchema(BaseModel):
    id: int
    name: str

    class Config:
        orm_mode = True

# データベースセッションの依存関係
def get_db():
    # セッションの取得方法を実装してください
    pass

@app.get("/users", response_model=Page[UserSchema])
def get_users(db: Session = Depends(get_db)):
    return paginate(db, db.query(User))

add_pagination(app)

実行結果

GET /users?page=1&size=2

{
  "total": 4,
  "page": 1,
  "size": 2,
  "items": [
    {"id": 1, "name": "Alice"},
    {"id": 2, "name": "Bob"}
  ]
}

解説

  • fastapi_paginationのpaginate関数を使用して、SQLAlchemyのクエリに対してページネーションを適用します。
  • response_modelにPage[UserSchema]を指定することで、レスポンスがページネーションされた形式になります。
  • add_pagination(app)を呼び出すことで、ページネーションに必要な設定が自動的に追加されます。

ユースケース7：ユーザーをソートして取得

def get_users_sorted(session, sort_by='name', descending=False):
    order = getattr(User, sort_by)
    if descending:
        order = order.desc()
    else:
        order = order.asc()
    return session.query(User).order_by(order).all()

実行結果

users = get_users_sorted(session, descending=True)
for user in users:
    print(user.name)
# 出力:
# David
# Charlie
# Bob
# Alice

解説

  • getattrを使用して、動的にソート対象のカラムを指定します。
  • asc()とdesc()で昇順・降順を制御します。

ユースケース8：特定のフィールドのみを取得

def get_user_names(session):
    return session.query(User.name).all()

実行結果

names = get_user_names(session)
for name, in names:
    print(name)
# 出力:
# Alice
# Bob
# Charlie
# David

解説

  • 必要なカラムだけを指定することで、データ転送量を削減します。

ユースケース9：集計関数を使用してユーザー数を取得

from sqlalchemy import func

def get_user_count(session):
    return session.query(func.count(User.id)).scalar()

実行結果

count = get_user_count(session)
print(f"ユーザー数: {count}")
# 出力:
# ユーザー数: 4

解説

  • func.countを使用して、ユーザーの総数を取得します。
  • scalar()で単一の値を取得します。

ユースケース10：サブクエリを使用した高度なクエリ

def get_users_with_recent_verification(session, days=7):
    from datetime import datetime, timedelta
    cutoff_date = datetime.utcnow() - timedelta(days=days)
    subquery = session.query(Auth.user_id).filter(Auth.email_verified_at >= cutoff_date).subquery()
    return session.query(User).filter(User.id.in_(subquery)).all()

実行結果

users = get_users_with_recent_verification(session, days=30)
for user in users:
    print(user.name)
# 出力:
# Charlie

解説

  • 最近メール確認したユーザーを取得します。
  • サブクエリを使用して、効率的にデータをフィルタリングします。

ユースケース11：複数テーブルを結合してデータを取得

def get_user_and_auth_info(session):
    return session.query(User.name, Auth.email).join(Auth).all()

実行結果

results = get_user_and_auth_info(session)
for name, email in results:
    print(f"{name}: {email}")
# 出力:
# Alice: alice@example.com
# Bob: bob@example.com
# Charlie: charlie@example.com
# David: david@example.com

解説

  • joinを使用して、UserとAuthを結合します。
  • 必要なフィールドだけを選択して取得します。

ユースケース12：条件によって異なるクエリを実行

def get_users_dynamic_query(session, verified_only=False):
    query = session.query(User).join(Auth)
    if verified_only:
        query = query.filter(Auth.email_verified_at != None)
    return query.all()

実行結果

users = get_users_dynamic_query(session, verified_only=True)
for user in users:
    print(user.name)
# 出力:
# Alice
# Charlie

解説

  • 引数verified_onlyがTrueの場合、メール確認済みのユーザーのみを取得します。
  • 動的にクエリを構築できます。

ユースケース13：トランザクションを使用した安全なデータ取得

def get_user_in_transaction(session, user_id):
    with session.begin():
        user = session.query(User).filter(User.id == user_id).first()
    return user

実行結果

user = get_user_in_transaction(session, 1)
print(user.name)
# 出力:
# Alice

解説

  • with session.begin()を使用して、トランザクション内でクエリを実行します。
  • データの一貫性を保つために有用です。

ユースケース14：エイリアスを使用して複数回の結合

from sqlalchemy.orm import aliased

def get_users_with_multiple_auths(session):
    AuthAlias = aliased(Auth)
    return session.query(User).join(Auth).join(AuthAlias, User.id == AuthAlias.user_id).all()

実行結果

users = get_users_with_multiple_auths(session)
for user in users:
    print(user.name)
# 出力:
# Alice
# Bob
# Charlie
# David

解説

  • aliasedを使用して、同じテーブルを複数回結合できます。
  • 複雑なリレーションを扱う場合に便利です。

ユースケース15：生のSQLを使用

def get_users_raw_sql(session):
    result = session.execute("SELECT * FROM users")
    return result.fetchall()

実行結果

users = get_users_raw_sql(session)
for row in users:
    print(row.name)
# 出力:
# Alice
# Bob
# Charlie
# David

解説

  • executeを使用して、生のSQLクエリを実行できます。
  • ORMで対応しきれない特殊なクエリを実行する場合に有用です。

まとめ

  • SQLAlchemy ORMは、さまざまな方法でデータを取得する柔軟性を提供します。
  • ユースケースに応じて最適なクエリ方法を選択することで、効率的なデータベース操作が可能です。
  • リレーション、フィルタリング、ソート、ページネーションなど、多様な機能を組み合わせて活用しましょう。

この記事を通じて、SQLAlchemy ORMを使用したデータ取得の多様な方法と、その実行結果を確認いただけたと思います。これらの知識を活用して、より高度なデータベース操作に挑戦してみてください。

株式会社Xronotech
株式会社XronotechPublication

福岡でシステム開発を行うXronotechの公式アカウントです。 Webアプリやスマホアプリを開発しています。

Discussion