FastAPIとSQLAlchemyでDB経由のCRUD操作をやってみた！

背景

こんにちは、白々さじきです！
前回はFastAPIのインラインデータベースを使って簡単なCRUD操作を試しましたが、今回はPythonのSQL操作用ライブラリ「SQLAlchemy」を使い、実際のデータベースを操作してみました。FastAPIとSQLAlchemyを組み合わせることで、効率的なAPI開発が可能になります。

なお、まだFastAPIの環境構築を行っていない方は、前回の記事を参考に準備を整えてください。それでは早速始めましょう！

この記事で使用したコードは以下のリポジトリで公開していますので、ぜひ参考にしてください。 [GitHubリポジトリはこちら]

SQLAlchemyとは？

SQLAlchemyは、Python用のSQLツールキットおよびオブジェクト関係マッピング（ORM）ライブラリです。これにより、データベース操作を直接SQL文で書くのではなく、Pythonのコードで行うことができます。公式サイトでは、以下のように紹介されています：

SQLAlchemy is the Python SQL toolkit and Object Relational Mapper that gives application developers the full power and flexibility of SQL.

SQLAlchemy

SQLAlchemyの特徴

SQLAlchemyの主な特徴を簡単にまとめると以下の通りです：

強力なORM機能: Pythonのクラスとデータベースのテーブルをマッピングし、オブジェクト指向の方法でデータベース操作が可能です。
柔軟なSQL生成: 複雑なSQLクエリもPythonコードで組み立てられます。
データベースの抽象化: 複数のデータベースエンジン（例: SQLite, PostgreSQL, MySQLなど）に対応しており、コードの再利用性が高まります。
高性能: 効率的なデータベースアクセスを実現するためのエンタープライズレベルのパターンが組み込まれています。

事前準備

ここでは、SQLAlchemyのインストールとファイル構成の準備を行います。

SQLAlchemyのインストール

まずはSQLAlchemyをインストールしましょう。以下のコマンドを実行してください：

pip install sqlalchemy

以下のようになっていればOKです。

今回のファイル構成

今回は以下のような構成でプロジェクトを作成します：

project/
├── __init__.py         # このフォルダをパッケージとして認識させる
├── main.py             # FastAPIアプリケーション
├── database.py         # データベース接続と設定
├── models.py           # SQLAlchemyモデル定義
└── crud.py             # CRUD操作の関数定義

ちなみに__init.py__がない状態で他のファイルの関数をmain.pyに呼び出そうとすると以下のようなエラーが出ます。

データベース接続とセッション管理

まず、database.pyにデータベース接続の設定を記述します：

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

# データベースセッション依存関係
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

SQLAlchemyモデル定義

次に、データベースのテーブルを表現するモデルを定義します。models.pyに以下のコードを記述してください：

from sqlalchemy import Column, Integer, String
from database import Base

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    email = Column(String, unique=True, index=True)

データベース操作を定義

CRUD操作に必要な関数をcrud.pyに記述します。

ユーザー作成

この関数は、新しいユーザーを作成してデータベースに保存します。usernameとemailを引数として受け取り、作成したユーザー情報を返します。

from sqlalchemy.orm import Session
from models import User

# ユーザー作成
def create_user(db: Session, username: str, email: str) -> User:
    new_user = User(username=username, email=email)
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user

全ユーザーを取得

データベース内に登録されているすべてのユーザーを取得する関数です。ユーザー一覧を返します。

# 全ユーザーを取得
def get_users(db: Session):
    return db.query(User).all()

特定ユーザーの取得

この関数は、特定のユーザーIDを使って、対応するユーザーをデータベースから取得します。該当するユーザーが見つからない場合はNoneを返します。

# 特定のユーザーを取得
def get_user(db: Session, user_id: int) -> User:
    return db.query(User).filter(User.id == user_id).first()

ユーザー情報の更新

ユーザーのusernameやemailを更新する関数です。対象のユーザーIDを指定し、新しい値を引数として渡します。更新後のユーザー情報を返します。

# ユーザー情報更新
def update_user(db: Session, user_id: int, username: str = None, email: str = None) -> User:
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        return None  # ユーザーが見つからない場合はNoneを返す
    if username:
        user.username = username
    if email:
        user.email = email
    db.commit()
    db.refresh(user)
    return user

ユーザー情報削除

特定のユーザーをデータベースから削除する関数です。対象のユーザーが見つからなかった場合はFalseを返し、正常に削除された場合はTrueを返します。

# ユーザー情報削除
def delete_user(db: Session, user_id: int) -> bool:
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        return False  # ユーザーが見つからない場合はFalseを返す
    db.delete(user)
    db.commit()
    return True

FastAPIエンドポイント

main.pyでFastAPIエンドポイントを定義します。ここでは、CRUD操作を実行するAPIエンドポイントを作成します。それぞれのエンドポイントがどのような役割を持ち、どのように動作するのかを解説していきます。

データベース連携準備

まず、main.pyに以下のコードを記述します。これにより、FastAPIアプリケーションとデータベースの連携が可能になります。

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from database import get_db, create_tables
from crud import create_user, get_user, get_users

app = FastAPI()

# サーバー起動時にテーブル作成
create_tables()

ユーザー作成

このエンドポイントは、新しいユーザーを作成します。usernameとemailをリクエストで受け取り、データベースに保存します。重複するユーザー名がある場合はエラーを返します。

# ユーザー作成エンドポイント
@app.post("/users/")
def create_user_endpoint(username: str, email: str, db: Session = Depends(get_db)):
    try:
        user = create_user(db, username=username, email=email)
        return {"id": user.id, "username": user.username, "email": user.email}
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

ポイント:

新しいユーザーをデータベースに追加します。
エラー処理を実装しており、重複ユーザー名の登録を防ぎます。

ユーザー全体取得

このエンドポイントは、データベースに保存されたすべてのユーザーを取得します。結果はリスト形式で返されます。

# 全ユーザーを取得
@app.get("/users/")
def read_users_endpoint(db: Session = Depends(get_db)):
    users = get_users(db)
    return [{"id": user.id, "username": user.username, "email": user.email} for user in users]

ポイント:

データベース内のすべてのユーザー情報を取得して返します。
ユーザー情報はリスト形式のJSONとして整形されています。

特定のユーザー取得

このエンドポイントは、指定したユーザーIDに対応するユーザーを取得します。該当するユーザーが見つからない場合、404エラーを返します。

@app.get("/users/{user_id}")
def read_user_endpoint(user_id: int, db: Session = Depends(get_db)):
    user = get_user(db, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return {"id": user.id, "username": user.username, "email": user.email}

ポイント:

ユーザーIDを指定してデータベースを検索します。
該当ユーザーがいない場合、404エラーで「User not found」を返します。

ユーザー削除

このエンドポイントは、指定したユーザーIDのユーザーをデータベースから削除します。該当するユーザーがいない場合、エラーを返します。

# ユーザー削除
@app.delete("/users/{user_id}")
def delete_user_endpoint(user_id: int, db: Session = Depends(get_db)):
    # ユーザーを削除
    try:
        result = delete_user(db, user_id)
        if not result:
            # 見つからない場合はエラーを返す
            raise HTTPException(status_code=404, detail="User not found")
        # 見つかった場合は削除
        return "delete success"
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

ポイント:

成功時には「delete success」のメッセージを返します。
ユーザーが存在しない場合、404エラーを返します。

ユーザー情報更新

このエンドポイントは、指定したユーザーIDのusernameまたはemailを更新します。入力が不正な場合や、ユーザーが見つからない場合はエラーを返します。

# ユーザー情報更新
@app.put("/users/{user_id}")
def update_user_endpoint(user_id: int, username: str, email: str, db: Session = Depends(get_db)):
    try:
        user = update_user(db, user_id=user_id, username=username, email=email)
        if user is None:
            # 見つからない場合はエラーを返す
            raise HTTPException(status_code=404, detail="User not found")
        # 見つかった場合は、更新
        return {"id": user.id, "username": user.username, "email": user.email}
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

ポイント: