初心者向け：FastAPIとSQLAlchemyでデータベース連携

By /

初心者向け：FastAPIとSQLAlchemyでデータベース連携を学ぼう

Webアプリケーション開発において、ユーザーからのリクエストを受け付け、データを処理し、データベースに永続化することは非常に一般的です。PythonでモダンなWeb APIを構築するフレームワークとしてFastAPIが注目されており、データベース操作においては高機能なORM（Object-Relational Mapper）であるSQLAlchemyが広く利用されています。

この記事では、FastAPIとSQLAlchemyを組み合わせてデータベース連携を実現する方法を、初心者の方にも理解できるように、ステップバイステップで詳細に解説します。Pythonの基本的な知識とWeb開発の基本的な概念があれば、読み進めることができます。

さあ、FastAPIとSQLAlchemyの世界に飛び込みましょう！

1. はじめに：なぜFastAPIとSQLAlchemyなのか？

Webアプリケーション、特にAPI開発において、バックエンドは主に以下の役割を担います。

  • クライアントからのリクエストを受け取る
  • リクエストのデータに基づいた処理を行う
  • 必要に応じてデータベースと連携する
  • 処理結果をクライアントに返す

この一連の処理を効率的かつ安全に行うために、適切なフレームワークやライブラリの選択が重要になります。

FastAPIの魅力

FastAPIは、現代的で高速（高性能）なWebフレームワークであり、以下の特徴を持っています。

  • 高パフォーマンス: Starlette (ASGIフレームワーク) と Pydantic (データ検証ライブラリ) を基盤としており、非常に高速な処理が可能です。非同期処理（async/await）に標準対応しています。
  • 開発速度: 型ヒントを積極的に利用することで、コード補完、エラーチェック、データ検証、自動ドキュメント生成などが強力にサポートされます。これにより、開発効率が大幅に向上します。
  • 自動APIドキュメント: OpenAPI (Swagger UI) と ReDoc の形式で、自動的にインタラクティブなAPIドキュメントを生成します。これはAPIを利用する側にとっても、開発者自身にとっても非常に便利です。
  • 堅牢なデータ検証: リクエストデータ（パスパラメータ、クエリパラメータ、リクエストボディ）の検証をPydanticモデルを使って簡単に行えます。

FastAPIは、その高速性と開発効率の高さから、モダンなAPI開発のデファクトスタンダードになりつつあります。

SQLAlchemyの魅力

SQLAlchemyは、Pythonでリレーショナルデータベースを操作するための強力で柔軟なツールキットです。主な特徴は以下の通りです。

  • ORM (Object-Relational Mapper): データベースのテーブルとPythonのクラスをマッピングし、Pythonオブジェクトとしてデータベース操作を行えます。これにより、SQLクエリを直接書く代わりに、より直感的でPythonらしいコードでデータベースを扱えます。
  • Core: ORM機能を使わず、SQLクエリをPythonのコードで組み立てて実行することも可能です。より低レベルな制御が必要な場合に役立ちます。
  • 幅広いデータベース対応: SQLite, PostgreSQL, MySQL, Oracle, Microsoft SQL Serverなど、様々なリレーショナルデータベースに対応しています。
  • 堅牢性: 複雑なクエリ、トランザクション管理、接続プールなど、プロダクションレベルのアプリケーションに必要な機能を備えています。

SQLAlchemyのORM機能は、データベース操作をオブジェクト指向的に扱えるようにし、コードの可読性や保守性を高めます。

FastAPIとSQLAlchemyの組み合わせ

FastAPIでAPIを開発し、そのAPIが必要とするデータの永続化にSQLAlchemyを利用する組み合わせは、非常に強力です。

  • FastAPIがリクエスト処理、ルーティング、データ検証、APIドキュメント生成などを担当。
  • SQLAlchemyがデータベース接続、クエリ実行、データ操作などを担当。

この組み合わせにより、高速でメンテナンス性の高いWeb APIを効率的に開発できます。特に、FastAPIの依存性注入（Dependency Injection）の仕組みを使うことで、データベースセッションの管理を非常にスムーズに行えるのが大きな利点です。

この記事では、この理想的な組み合わせを使って、実際にデータベースと連携するWeb APIを構築する手順を学びます。

2. 開発環境の準備

まずは、開発に必要なツールとライブラリを準備しましょう。

Pythonのインストール

Pythonがインストールされていない場合は、まずPython公式サイトから最新版（3.7以上推奨、FastAPIは3.7以降が必要です）をダウンロードしてインストールしてください。

仮想環境の作成

プロジェクトごとに仮想環境を作成することを強くお勧めします。これにより、プロジェクト固有のライブラリとそのバージョンを分離し、環境の衝突を防ぐことができます。

ターミナルまたはコマンドプロンプトを開き、プロジェクトを作成したいディレクトリに移動して、以下のコマンドを実行します。

“`bash

プロジェクトディレクトリを作成 (例: fastapidb_example)

mkdir fastapidb_example
cd fastapidb_example

仮想環境を作成 (venvという名前で)

python -m venv venv

仮想環境をアクティベート

Windowsの場合

venv\Scripts\activate

macOS/Linuxの場合

source venv/bin/activate
“`

仮想環境がアクティベートされると、ターミナルのプロンプトの先頭に(venv)のような表示が追加されます。

必要なライブラリのインストール

仮想環境をアクティベートした状態で、必要なライブラリをインストールします。

bash
pip install fastapi uvicorn[standard] sqlalchemy

  • fastapi: FastAPIフレームワーク本体
  • uvicorn[standard]: ASGIサーバー実装。開発用には standard オプションでインストールするのが便利です。
  • sqlalchemy: SQLAlchemyライブラリ本体

今回は、学習目的で手軽に使えるSQLiteデータベースを使用します。SQLiteはPythonの標準ライブラリに組み込まれているため、別途ドライバーをインストールする必要はありません。

もし、PostgreSQLやMySQLなどの他のデータベースを使う場合は、対応するデータベースドライバーもインストールする必要があります。

  • PostgreSQL: pip install psycopg2-binary または pip install asyncpg (非同期対応)
  • MySQL: pip install mysql-connector-python または pip install aiomysql (非同期対応)

プロジェクト構造の提案

プロジェクトを整理するために、以下のようなディレクトリ構成を提案します。

fastapidb_example/
├── venv/ # 仮想環境
├── main.py # FastAPIアプリケーションのエントリーポイント
├── database.py # データベース接続設定とセッション管理
├── models.py # SQLAlchemyモデル定義
├── schemas.py # Pydanticスキーマ定義
├── crud.py # データベース操作（CRUD）ロジック
└── requirements.txt # インストールしたライブラリ一覧 (任意)

今はまだファイルがありませんが、これから作成していきます。

インストールしたライブラリを requirements.txt に書き出しておくと、他の環境で同じ環境を再現するのに便利です。

bash
pip freeze > requirements.txt

これで、開発を行う準備が整いました。

3. SQLAlchemyの基本を学ぶ

FastAPIと連携する前に、SQLAlchemyの基本的な使い方を理解しておくことが重要です。ここでは主にORM機能に焦点を当てて解説します。

3.1. ORM (Object-Relational Mapper) とは

ORMは、リレーショナルデータベースのテーブルとプログラミング言語のオブジェクト（クラスのインスタンス）との間のマッピングを行う技術です。SQLAlchemyのORMを使うことで、以下のようなメリットが得られます。

  • オブジェクト指向的な操作: データベースの行をPythonオブジェクトとして扱い、属性へのアクセスやメソッド呼び出しでデータ操作ができます。
  • SQLの抽象化: ほとんどの場合、データベース固有のSQL構文を意識せずに済みます。SQLAlchemyがPythonのコードから適切なSQLクエリを生成してくれます。
  • コードの可読性と保守性向上: データ構造（テーブル定義）とデータ操作ロジックをPythonコードとして一元管理できます。

3.2. Engineの作成：データベースへの接続

SQLAlchemyを使ってデータベースに接続するためには、まずEngineオブジェクトを作成します。Engineはデータベースとの接続を管理し、SQLAlchemyがデータベースと通信するための主要なインターフェースとなります。

Engineは、データベースURLを指定して作成します。データベースURLの形式はデータベースの種類によって異なります。

例：
* SQLite: sqlite:///./test.db (現在のディレクトリに test.db というファイルを作成/使用)
* PostgreSQL: postgresql://user:password@host:port/dbname
* MySQL: mysql://user:password@host/dbname

今回はSQLiteを使うので、データベースURLは sqlite:///./sql_app.db とします。プロジェクトルートに sql_app.db というファイルが作成されます。

プロジェクトルートに database.py ファイルを作成し、以下のコードを記述します。

“`python

database.py

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

データベースファイルのパスを指定

今回は、このファイルがあるディレクトリに相対的に ‘sql_app.db’ を作成します

SQLALCHEMY_DATABASE_URL = “sqlite:///./sql_app.db”

Engineを作成

SQLiteを使用する場合、connect_args={“check_same_thread”: False} が必要です。

これがないと、異なるスレッドからの複数のリクエストで同じ接続を使用しようとした際にエラーが発生する可能性があります。

FastAPIのデフォルトの動作では、複数のリクエストは異なるスレッドで処理される可能性があるため、これは重要です。

engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={“check_same_thread”: False}
)

SessionLocalクラスを作成

このクラスのインスタンスがデータベースセッションになります。

autocommit=False: トランザクションを手動でコミットする必要があります。

autoflush=False: add()などの操作で自動的にflush（データベースへの同期）を行いません。

bind=engine: このセッションが使用するEngineを指定します。

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Declarative baseクラスを作成

このクラスを継承して、各モデルクラス（データベースのテーブルに対応するクラス）を定義します。

Base = declarative_base()
“`

コード解説 (database.py):

  • create_engine: データベースURLを指定してEngineオブジェクトを作成します。connect_args={"check_same_thread": False} は、SQLite固有の設定です。
  • sessionmaker: データベースセッションを作成するためのファクトリクラスです。SessionLocal = sessionmaker(...) で設定済みの SessionLocal クラスを作成しています。このクラスのインスタンスが、実際のデータベースとの対話を行う Session オブジェクトになります。
    • autocommit=False: データの変更（INSERT, UPDATE, DELETE）を自動的にコミットせず、明示的に .commit() を呼び出す必要があります。これにより、複数の操作を一つのトランザクションとして扱えます。
    • autoflush=False: オブジェクトの状態変更を自動的にデータベースに同期せず、必要に応じて手動で .flush() を呼び出すか、トランザクションのコミット時に自動的に行われます。
  • declarative_base: ORMのDeclarativeマッピングを使用するためのベースクラスを作成します。この Base クラスを継承してPythonクラスを定義することで、そのクラスがデータベーステーブルと関連付けられます。

3.3. Declarative Baseとモデル定義

データベースのテーブル構造に対応するPythonクラスを定義します。SQLAlchemy ORMでは、Base クラスを継承してクラスを作成し、クラス属性としてテーブルのカラムを定義します。

models.py ファイルを作成し、以下のコードを記述します。

“`python

models.py

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

from .database import Base # database.py から Base をインポート

タスクモデル定義

class Task(Base):
# テーブル名を指定
tablename = “tasks”

# カラム定義
# primary_key=True: 主キー
# index=True: インデックスを作成（検索性能向上）
# default=False: デフォルト値
id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True)
description = Column(String, index=True)
completed = Column(Boolean, default=False)

# リレーションシップ（他のテーブルとの関連があれば定義）
# 例: owner_id = Column(Integer, ForeignKey("users.id"))
#     owner = relationship("User")
# 今回はシンプルなタスクリストなので、リレーションシップは定義しません。

# オブジェクトの文字列表現（デバッグなどに便利）
def __repr__(self):
    return f"<Task(id={self.id}, title='{self.title}', completed={self.completed})>"

“`

コード解説 (models.py):

  • from .database import Base: database.py で定義した Base クラスをインポートします。（. は現在のパッケージからの相対インポートを示します）
  • class Task(Base):: Base クラスを継承して Task クラスを定義します。これがデータベースの tasks テーブルに対応します。
  • __tablename__ = "tasks": このクラスがマッピングされるデータベーステーブル名を指定します。指定しない場合はクラス名が小文字で使われますが、明示的に指定するのが一般的です。
  • Column(...): データベーステーブルのカラムを定義します。Column の最初の引数はカラムの型です（Integer, String, Boolean など）。
    • primary_key=True: そのカラムがテーブルの主キーであることを示します。通常、自動的に連番が割り振られます。
    • index=True: そのカラムにインデックスを作成します。インデックスがあると、そのカラムでの検索や並べ替えが高速になります。
    • default=False: レコードが挿入される際に、このカラムの値が指定されなかった場合のデフォルト値を設定します。
  • relationship: 他のテーブルとのリレーションシップ（一対多、多対多など）を定義する際に使用します。今回はシンプルな例なので使用していません。
  • __repr__(self): このメソッドを定義しておくと、Pythonのインタラクティブシェルなどでオブジェクトを表示したときに、そのオブジェクトの状態がわかりやすく表示されます。

3.4. データベーステーブルの作成

models.py で定義したクラスに基づいて、実際にデータベースファイル内にテーブルを作成する必要があります。これは通常、アプリケーションの初回起動時や、モデル定義を変更した際に行います。

database.py に以下のコードを追加します。

“`python

database.py (既存コードに追記)

… (create_engine, sessionmaker, Base の定義) …

models.py から全てのモデルクラスをインポート

これにより、Base.metadata に全てのモデルの情報が登録されます

import models # または from . import models

def create_db_tables():
“””データベーステーブルを作成する関数”””
# Base.metadata に登録された全てのモデルに基づいてテーブルを作成します
# すでにテーブルが存在する場合は無視されます
Base.metadata.create_all(bind=engine)

開発中は、スクリプト実行時にテーブルを作成するようにしておくと便利です

if name == “main“:

create_db_tables()

実際には、FastAPI起動時に依存関係として実行するか、マイグレーションツールを使います。

“`

この create_db_tables() 関数を呼び出すことで、models.py で定義された全てのテーブルがデータベースに作成されます。

3.5. Sessionの操作：CRUDの基本

データベースとの対話は、Session オブジェクトを介して行います。Session は、データベースとのコネクションを管理し、データの問い合わせや変更（INSERT, UPDATE, DELETE）を行います。SessionLocal() を呼び出すことで新しいセッションインスタンスを取得できます。

Session オブジェクトを使って、基本的なCRUD（Create, Read, Update, Delete）操作を行う方法を見ていきましょう。

データ作成 (Create)

新しい Task オブジェクトを作成し、セッションに追加 (add()) してコミット (commit()) します。

“`python
from .database import SessionLocal
from .models import Task

セッションを取得

db = SessionLocal()

try:
# 新しいTaskオブジェクトを作成
new_task = Task(title=”牛乳を買う”, description=”スーパーで牛乳を購入する”)

# セッションに追加
db.add(new_task)

# コミットしてデータベースに反映
db.commit()

# コミット後にオブジェクトをリフレッシュすると、
# データベースで自動生成された値（例: ID）などがオブジェクトに反映されます。
db.refresh(new_task)

print(f"タスクが作成されました: {new_task}")

finally:
# セッションを閉じます。重要なステップです。
db.close()
“`

  • db.add(instance): 新しいオブジェクトをセッションに登録します。まだデータベースには保存されていません。
  • db.commit(): セッションで行われた変更（add, delete, オブジェクトの属性変更）をデータベースに書き込みます。これがトランザクションの終了点となります。
  • db.refresh(instance): オブジェクトの状態をデータベースの最新の状態に合わせて更新します。コミット後に自動生成された主キーなどを取得したい場合に使います。
  • db.close(): セッションを閉じ、データベースコネクションを解放します。非常に重要です。 これを行わないと、コネクションプールを枯渇させてしまう可能性があります。
データ検索 (Read)

セッションの query() メソッドを使ってデータを検索します。

“`python
from .database import SessionLocal
from .models import Task

db = SessionLocal()

try:
# 全てのタスクを取得
all_tasks = db.query(Task).all()
print(“全てのタスク:”)
for task in all_tasks:
print(f”- {task.id}: {task.title} (完了: {task.completed})”)

# IDで特定のタスクを取得 (存在しない場合は None)
task_id_to_find = 1
task_by_id = db.query(Task).filter(Task.id == task_id_to_find).first()
# または db.query(Task).get(task_id_to_find) - 主キー検索に便利

if task_by_id:
    print(f"\nID {task_id_to_find} のタスク:")
    print(f"- {task_by_id.id}: {task_by_id.title} (完了: {task_by_id.completed})")
else:
    print(f"\nID {task_id_to_find} のタスクは見つかりませんでした。")

# タイトルで検索
task_by_title = db.query(Task).filter(Task.title == "牛乳を買う").first()
# 部分一致検索: db.query(Task).filter(Task.title.like("%牛乳%")).all()

finally:
db.close()
“`

  • db.query(Model): 指定したモデルに対するクエリビルダオブジェクトを作成します。
  • .all(): クエリ結果の全てのオブジェクトをリストとして取得します。
  • .first(): クエリ結果の最初のオブジェクトを取得します。結果がない場合は None を返します。
  • .get(primary_key): 主キーを使ってオブジェクトを検索します。存在しない場合は None を返します。主キー検索に特化しており、最も効率的です。
  • .filter(condition): WHERE句に相当する条件を指定します。条件はSQLAlchemyのオペレーター（==, !=, >, <, >=, <=, .like(), .in_() など）を使って記述します。
データ更新 (Update)

検索したオブジェクトの属性を変更し、セッションをコミットします。

“`python
from .database import SessionLocal
from .models import Task

db = SessionLocal()

try:
task_id_to_update = 1
task_to_update = db.query(Task).filter(Task.id == task_id_to_update).first()

if task_to_update:
    # オブジェクトの属性値を変更
    task_to_update.completed = True
    task_to_update.description = "スーパーで牛乳を購入済み"

    # セッションに登録されたオブジェクトの変更は、commit() で自動的にデータベースに反映されます。
    db.commit()

    # 更新後の状態を取得
    db.refresh(task_to_update) # オプション、必要に応じて
    print(f"タスクID {task_id_to_update} が更新されました: {task_to_update}")
else:
    print(f"タスクID {task_id_to_update} は見つかりませんでした。")

finally:
db.close()
“`

  • 検索したオブジェクト（task_to_update）の属性値を直接変更します。
  • db.commit() を呼び出すことで、セッションで追跡されているオブジェクトの状態変更がデータベースのUPDATE文として実行されます。
データ削除 (Delete)

検索したオブジェクトをセッションから削除し (delete())、コミットします。

“`python
from .database import SessionLocal
from .models import Task

db = SessionLocal()

try:
task_id_to_delete = 1
task_to_delete = db.query(Task).filter(Task.id == task_id_to_delete).first()

if task_to_delete:
    # セッションからオブジェクトを削除対象として登録
    db.delete(task_to_delete)

    # コミットしてデータベースから削除
    db.commit()

    print(f"タスクID {task_id_to_delete} が削除されました。")
else:
    print(f"タスクID {task_id_to_delete} は見つかりませんでした。")

finally:
db.close()
“`

  • db.delete(instance): 指定したオブジェクトをセッションから削除対象として登録します。
  • db.commit() を呼び出すことで、登録されたオブジェクトに対応するデータベースのDELETE文が実行されます。

Session管理の重要性

上記の例では、操作のたびに SessionLocal() でセッションを作成し、try...finally ブロックを使って db.close() でセッションを閉じています。

  • セッションはデータベースコネクションと密接に関連しており、使い終わったら必ず閉じる必要があります。閉じないと、データベースへの接続が消費され続け、やがて新しい接続を確立できなくなってしまう可能性があります。
  • セッションは軽量なオブジェクトですが、作成と破棄にはコストがかかります。リクエストごとにセッションを作成し、使い終わったら閉じるのが一般的なパターンです。
  • セッションはトランザクションの単位でもあります。commit() で変更を確定し、エラーが発生した場合は rollback() で変更を取り消せます（try...finally で close するパターンでは、エラー発生時に自動的にロールバックされます）。

FastAPIと連携する際には、この「リクエストごとにセッションを作成し、使い終わったら閉じる」というパターンを、FastAPIの依存性注入の仕組みを使って効率的に実現します。

マイグレーションツール (Alembic) について (補足)

データベースのスキーマ（テーブル構造）は、開発の過程で変更されることがよくあります。例えば、新しいカラムを追加したり、既存のカラムの型を変更したりする場合です。このようなスキーマの変更を、既存のデータを維持したままデータベースに安全に適用するために、「データベースマイグレーション」という手法が使われます。

SQLAlchemyと組み合わせて最も一般的に使われるマイグレーションツールは Alembic です。Alembicを使うと、モデル定義 (models.py) の変更を検知し、データベースのスキーマを変更するためのPythonスクリプト（マイグレーションスクリプト）を自動生成できます。そして、そのスクリプトを実行することで、データベースのスキーマを段階的に更新したり、元に戻したりできます。

この記事ではAlembicの詳しい使い方は扱いませんが、実際のアプリケーション開発では必須のツールであることを覚えておきましょう。手動で create_all を使う方法は、開発の初期段階やシンプルなアプリケーションに限られます。

4. FastAPIの基本を学ぶ

次に、FastAPIの基本的な使い方を復習します。

4.1. FastAPIとは

前述の通り、FastAPIはASGI (Asynchronous Server Gateway Interface) をベースにした高性能なWebフレームワークです。ASGIは、Pythonの非同期フレームワーク（asyncioなど）とWebサーバー間の標準インターフェースです。

FastAPIは、型ヒントとPydanticモデルを最大限に活用することで、データ検証、シリアライズ（PythonオブジェクトからJSONなどへの変換）、自動ドキュメント生成を非常に容易にしています。

4.2. 基本的なアプリケーション構造

最もシンプルなFastAPIアプリケーションは以下のようになります。

“`python

main.py

from fastapi import FastAPI

FastAPIアプリケーションのインスタンスを作成

app = FastAPI()

ルートパス (“/”) へのGETリクエストに対するハンドラーを定義

@app.get(“/”)
def read_root():
return {“Hello”: “World”}

“/items/{item_id}” へのGETリクエストに対するハンドラーを定義

{item_id} はパスパラメータ

@app.get(“/items/{item_id}”)
def read_item(item_id: int, q: str | None = None):
# item_id は自動的に int 型に変換される
# q はクエリパラメータ。型ヒントで str 型または None を指定
return {“item_id”: item_id, “q”: q}
“`

コード解説 (main.py – 基本):

  • from fastapi import FastAPI: FastAPIクラスをインポートします。
  • app = FastAPI(): FastAPIアプリケーションのインスタンスを作成します。これがWebサーバーが実行するメインオブジェクトになります。
  • @app.get("/"): デコレーターを使って、特定のURLパス (/) とHTTPメソッド (GET) に対応する関数 (read_root) を登録します。これを パスオペレーションデコレーター と呼びます。
  • def read_root():: パスオペレーションデコレーターに対応する関数が パスオペレーション関数 です。クライアントからのリクエストを受け付け、処理を行い、結果を返します。関数が返すPythonの辞書やリストなどは、自動的にJSON形式にシリアライズされてHTTPレスポンスボディとして返されます。
  • @app.get("/items/{item_id}"): パスの中に {} で囲まれた部分があると、それは パスパラメータ として扱われます。ここでは item_id がパスパラメータです。
  • def read_item(item_id: int, q: str | None = None):: パスパラメータやクエリパラメータは、パスオペレーション関数の引数として受け取ります。FastAPIは、引数の型ヒントを見て、自動的にデータ変換と検証を行います。
    • item_id: int: パスパラメータ item_id を整数 (int) として受け取ります。もしクライアントが整数に変換できない値を送ってきた場合、FastAPIは自動的に検証エラーを返します。
    • q: str | None = None: クエリパラメータ q を文字列 (str) または None として受け取ります。デフォルト値が None なので、このパラメータは省略可能です。

4.3. Pydanticモデル：リクエストボディとレスポンスモデル

FastAPIはPydanticライブラリを深く統合しています。Pydanticを使うと、Pythonのクラスを使ってデータの構造と型を定義できます。これは主に以下の目的で使用されます。

  • リクエストボディの検証: POST, PUTなどのリクエストで送信されるJSONデータを、Pydanticモデルの定義に従って自動的に検証します。
  • レスポンスモデル: APIが返すデータの構造と型を定義し、出力データのシリアライズと検証を行います。

schemas.py ファイルを作成し、Pydanticモデルを定義します。

“`python

schemas.py

from pydantic import BaseModel

Pydanticモデルはデータベースモデルとは別に定義するのが一般的です。

データベースモデルはデータベースのテーブル構造を定義するのに対し、

PydanticモデルはAPIのリクエストやレスポンスのデータ構造を定義します。

これにより、APIのインターフェースとデータベースの内部構造を分離できます。

タスク作成/更新用のスキーマ

class TaskBase(BaseModel):
title: str
description: str | None = None # descriptionは省略可能

タスク作成リクエスト用のスキーマ

class TaskCreate(TaskBase):
# 作成時には additional fields are not required
pass

タスク更新リクエスト用のスキーマ

class TaskUpdate(TaskBase):
completed: bool = False # 更新時には完了状態も指定可能、デフォルト値はFalse

タスクレスポンス用のスキーマ

class Task(TaskBase):
id: int
completed: bool # completedは必ず含まれる

class Config:
    # ORM mode を有効にすることで、
    # PydanticモデルがSQLAlchemyモデルのようなORMオブジェクトの属性を読み込めるようになります。
    # 例: SQLAlchemyモデルの task.id を Pydanticモデルの Task(id=task.id, ...) のようにマッピング
    orm_mode = True # Pydantic V2 以降では from_attributes = True に名称変更されました。
    # Pydantic V2 の場合: from_attributes = True

“`

コード解説 (schemas.py):

  • from pydantic import BaseModel: Pydanticモデルのベースクラスをインポートします。
  • class TaskBase(BaseModel):: 他のスキーマが継承するための共通フィールドを定義します。
  • class TaskCreate(TaskBase):: タスク作成時のリクエストボディの構造を定義します。TaskBase を継承しているので、title と description を持ちます。
  • class TaskUpdate(TaskBase):: タスク更新時のリクエストボディの構造を定義します。completed フィールドが追加されています。
  • class Task(TaskBase):: APIレスポンスとして返すタスクデータの構造を定義します。id と completed が追加されています。これはデータベースから取得した models.Task オブジェクトを変換して返します。
  • class Config: orm_mode = True: この設定により、FastAPI (Pydantic) はSQLAlchemyモデルオブジェクトからデータを読み取れるようになります。例えば、Task Pydanticモデルのインスタンスを、SQLAlchemyの Task モデルインスタンスを使って schemas.Task.from_orm(db_task) のように作成できます。

4.4. 依存性注入 (Dependency Injection)

FastAPIの最も強力な機能の一つが依存性注入システムです。これは、パスオペレーション関数が依存する「何か」（データベースセッション、認証されたユーザー、設定値など）を、フレームワークが自動的に提供してくれる仕組みです。

依存性を定義するには、通常の関数のように引数を定義し、その引数に型ヒントを付けます。そして、その引数にデフォルト値として Depends() を使って依存関数（Dependency）を指定します。

“`python
from fastapi import Depends, FastAPI, Header, HTTPException

async def verify_token(x_token: str = Header()):
if x_token != “fake-super-secret-token”:
raise HTTPException(status_code=400, detail=”X-Token header invalid”)
return x_token

async def verify_key(x_key: str = Header()):
if x_key != “fake-super-secret-key”:
raise HTTPException(status_code=400, detail=”X-Key header invalid”)
return x_key

@app.get(“/items/”, dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
return [{“item”: “Foo”}, {“item”: “Bar”}]

または、パスオペレーション関数の引数として依存性を注入

@app.get(“/users/”)
async def read_users(token: str = Depends(verify_token)):
# token は verify_token() の戻り値
return [{“username”: “Baz”}, {“username”: “Boo”}]
“`

依存性注入は、コードの再利用性を高め、テストを容易にし、コードを整理するのに役立ちます。データベースセッションの管理も、この依存性注入を使って行います。

5. FastAPIとSQLAlchemyの連携（実践サンプル）

いよいよ、FastAPIとSQLAlchemyを組み合わせて、簡単なTODOリストAPIを構築しましょう。このセクションでは、前述のプロジェクト構造に従って各ファイルの実装と解説を行います。

プロジェクト構造の確認

fastapidb_example/
├── venv/
├── main.py # FastAPIアプリ本体、ルーター
├── database.py # データベース接続設定、セッション取得依存性
├── models.py # SQLAlchemyモデル定義
├── schemas.py # Pydanticスキーマ定義
├── crud.py # データベース操作（CRUD）関数
└── requirements.txt

5.1. データベース接続とセッション取得の依存性 (database.py)

前に定義した database.py に、FastAPIの依存性注入システムで利用するためのセッション取得関数を追加します。

“`python

database.py

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

データベースファイルのパスを指定 (相対パス)

このファイルを基準とした相対パスではなく、FastAPIアプリケーションの起動元ディレクトリからの相対パスになります。

例: プロジェクトルートで uvicorn main:app を実行する場合、プロジェクトルートに sql_app.db が作成されます。

SQLALCHEMY_DATABASE_URL = “sqlite:///./sql_app.db”

Engineを作成

SQLiteを使用する場合、connect_args={“check_same_thread”: False} が必要

engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={“check_same_thread”: False}
)

SessionLocalクラスを作成

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Declarative baseクラスを作成

Base = declarative_base()

— FastAPIの依存性注入で使用するデータベースセッション取得関数 —

def get_db():
“””データベースセッションを生成し、使い終わったら閉じる関数”””
db = SessionLocal() # SessionLocal() を呼び出して新しいセッションを作成
try:
yield db # セッションを呼び出し元（パスオペレーション関数）に提供
finally:
db.close() # 呼び出し元での処理が終わったら、セッションを閉じる
“`

コード解説 (database.py – 依存性部分):

  • def get_db():: データベースセッションを提供するジェネレーター関数です。FastAPIの依存性注入システムは、ジェネレーターをサポートしています。ジェネレーター関数は yield を使って値を返し、yield の後の処理は、クライアントへのレスポンスが返された後に実行されます。
  • db = SessionLocal(): 新しいセッションインスタンスを作成します。リクエストごとに新しいセッションが作成されることになります。
  • try...finally:: セッションが確実に閉じられるように、try...finally ブロックを使用します。
  • yield db: 作成したセッションオブジェクト db を、この依存性を要求したパスオペレーション関数に提供します。パスオペレーション関数は、この db オブジェクトを使ってデータベース操作を行います。
  • db.close(): パスオペレーション関数の処理（およびレスポンス生成）が完了した後、finally ブロックが実行され、セッションが閉じられます。これにより、データベースコネクションが適切に解放されます。

この get_db 関数をFastAPIの依存性注入 (Depends(get_db)) として使用することで、各リクエストハンドラー関数内で db: Session = Depends(get_db) のように引数としてデータベースセッションを受け取れるようになります。FastAPIが自動的にセッションの作成、提供、クローズを行ってくれるため、各ハンドラー関数でセッション管理の定型コードを書く必要がなくなります。

5.2. SQLAlchemyモデル定義 (models.py)

前述の models.py と同じ内容です。TODOタスクを表す Task モデルを定義します。

“`python

models.py

from sqlalchemy import Boolean, Column, Integer, String

relationship は今回は不要

from sqlalchemy.orm import relationship

from .database import Base # database.py から Base をインポート

タスクモデル定義

class Task(Base):
tablename = “tasks”

id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True)
description = Column(String, index=True, nullable=True) # description は Nullable にする
completed = Column(Boolean, default=False)

def __repr__(self):
    return f"<Task(id={self.id}, title='{self.title}', completed={self.completed})>"

“`

5.3. Pydanticスキーマ定義 (schemas.py)

APIのリクエスト/レスポンスのデータ構造を定義します。

“`python

schemas.py

from pydantic import BaseModel, ConfigDict # V2の場合 ConfigDict をインポート

Pydanticモデルはデータベースモデルとは別に定義します

タスク作成/更新用のスキーマ (基本形)

class TaskBase(BaseModel):
title: str
description: str | None = None # description は省略可能

タスク作成リクエスト用のスキーマ

class TaskCreate(TaskBase):
pass # TaskBase と同じ構造

タスク更新リクエスト用のスキーマ

class TaskUpdate(TaskBase):
completed: bool = False # 更新時には完了状態も指定可能、デフォルト値は False

タスクレスポンス用のスキーマ

class Task(TaskBase):
id: int
completed: bool # レスポンスでは completed は常に含まれる

# Pydantic V1 の場合:
# class Config:
#     orm_mode = True

# Pydantic V2 の場合:
model_config = ConfigDict(from_attributes=True)

“`

5.4. データベース操作（CRUD）関数 (crud.py)

このファイルでは、SQLAlchemyの Session オブジェクトを受け取り、具体的なデータベース操作（CRUD）を行う関数を定義します。これにより、データベース操作のロジックをFastAPIのパスオペレーション関数から分離できます。

“`python

crud.py

from sqlalchemy.orm import Session

from . import models, schemas # models.py と schemas.py をインポート

— Read —

def get_task(db: Session, task_id: int):
“””指定されたIDのタスクを取得”””
# filter() で条件を指定し、first() で最初に見つかったレコードを取得
# get() を使う方がより効率的 (主キー検索の場合)
return db.query(models.Task).filter(models.Task.id == task_id).first()
# return db.get(models.Task, task_id) # 主キー検索ならこちらが推奨

def get_tasks(db: Session, skip: int = 0, limit: int = 100):
“””全てのタスクを取得 (ページネーション対応)”””
# offset() で取得開始位置、limit() で取得件数を制限
return db.query(models.Task).offset(skip).limit(limit).all()

— Create —

def create_task(db: Session, task: schemas.TaskCreate):
“””新しいタスクを作成”””
# Pydanticスキーマオブジェクトから、SQLAlchemyモデルオブジェクトを生成
# task.model_dump() は Pydantic V2 の書き方
# V1 の場合は task.dict()
db_task = models.Task(task.model_dump()) # または task.dict() for V1

# セッションに追加
db.add(db_task)

# コミットしてデータベースに反映
db.commit()

# データベースで自動生成されたIDなどを反映するためにリフレッシュ
db.refresh(db_task)

return db_task

— Update —

def update_task(db: Session, task_id: int, task: schemas.TaskUpdate):
“””指定されたIDのタスクを更新”””
db_task = get_task(db, task_id) # 更新対象のタスクを取得

if db_task:
    # Pydanticスキーマオブジェクトの属性を、SQLAlchemyモデルオブジェクトにコピー
    # task.model_dump(exclude_unset=True) は Pydantic V2 の書き方。未設定フィールドを除外して辞書化。
    # V1 の場合は task.dict(exclude_unset=True)
    update_data = task.model_dump(exclude_unset=True) # または task.dict(exclude_unset=True) for V1

    # 辞書形式のデータをオブジェクトに適用
    for field, value in update_data.items():
        setattr(db_task, field, value)

    # セッション内のオブジェクト変更は commit() で自動的に反映
    db.commit()
    db.refresh(db_task) # 更新後の状態を反映
    return db_task
return None # 更新対象が見つからなかった場合

— Delete —

def delete_task(db: Session, task_id: int):
“””指定されたIDのタスクを削除”””
db_task = get_task(db, task_id) # 削除対象のタスクを取得

if db_task:
    db.delete(db_task) # セッションから削除対象として登録
    db.commit() # データベースに反映
    return db_task # 削除されたオブジェクトを返す（IDなどが参照可能）
return None # 削除対象が見つからなかった場合

“`

コード解説 (crud.py):

  • 各関数は最初の引数として db: Session を受け取ります。これはFastAPIの依存性注入によって提供されるデータベースセッションです。
  • 型ヒントとして Session を指定することで、エディタの補完や静的解析が効きやすくなります。
  • CRUD操作は、前述の「Sessionの操作」で説明した基本的なSQLAlchemyの構文 (query().filter().first(), add(), commit(), refresh(), delete()) を使用しています。
  • create_task 関数では、FastAPIのリクエストボディとして受け取ったPydanticモデル (schemas.TaskCreate) のデータを使って、SQLAlchemyモデル (models.Task) のインスタンスを作成しています。**task.model_dump() (V2) または **task.dict() (V1) は、Pydanticモデルのフィールドを辞書として展開し、キーワード引数として models.Task のコンストラクタに渡しています。
  • update_task 関数では、更新対象のタスクを取得し、更新データを含むPydanticモデル (schemas.TaskUpdate) のデータを使って、オブジェクトの属性を更新しています。exclude_unset=True を使うことで、リクエストボディで指定されなかったフィールドは更新対象から除外されます。これは、部分的な更新（PATCH）を実装する際にも役立ちます。辞書形式のデータをオブジェクトに適用するために setattr() を使用しています。

5.5. FastAPIルーターとエンドポイントの実装 (main.py)

FastAPIアプリケーションのメインファイルです。ここでFastAPIのインスタンスを作成し、データベースセッションの依存性をパスオペレーション関数に注入し、CRUD関数を呼び出してAPIエンドポイントを実装します。

まず、database.py で定義した create_db_tables() 関数を呼び出して、データベーステーブルを作成しておきましょう。main.py の最後に一時的に以下のコードを追加して一度実行するか、Pythonインタラクティブシェルなどで from database import create_db_tables; create_db_tables() と実行しても構いません。

“`python

main.py (一時的にテーブル作成用に追加)

… (FastAPIアプリケーションの定義など、この後に書くコード) …

from .database import create_db_tables, engine, Base # database.py から必要なものをインポート

このコードは、開発中に初回だけ、またはモデル変更時に実行します。

Production 環境では、マイグレーションツール(Alembicなど)を使用するのが一般的です。

print(“データベーステーブルを作成します…”)

from .database import create_db_tables # database.pyからインポート

create_db_tables()

print(“データベーステーブルが作成されました。”)

上記コードを実行したら削除またはコメントアウトしてください。

“`

テーブル作成コードを実行（例えば python main.py として実行し、テーブル作成メッセージが表示されたら終了）し、main.py を以下の内容で更新します。

“`python

main.py

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session # 型ヒントのためにインポート

ローカルモジュールをインポート

from . import crud, models, schemas # crud.py, models.py, schemas.py をインポート
from .database import SessionLocal, engine # database.py から SessionLocal と engine をインポート

データベーステーブルの作成

アプリケーション起動時にテーブルが存在しない場合のみ作成

開発中はこの方法でもOKですが、プロダクションではマイグレーションツール推奨

models.Base.metadata.create_all(bind=engine)

FastAPIアプリケーションのインスタンスを作成

app = FastAPI()

— データベースセッション取得の依存性を定義 —

database.py で定義した get_db 関数をそのまま使用

def get_db():

db = SessionLocal()

try:

yield db

finally:

db.close()

上記は database.py に既に定義済みなので、ここではインポートして使用します。

from .database import get_db # database.py から get_db 関数をインポート

— エンドポイントの定義 —

@app.post(“/tasks/”, response_model=schemas.Task)
def create_task(task: schemas.TaskCreate, db: Session = Depends(get_db)):
“””新しいタスクを作成するエンドポイント”””
# crud.create_task 関数を呼び出してデータベースに保存
# db = Depends(get_db) により、FastAPIがデータベースセッションを注入してくれる
return crud.create_task(db=db, task=task)

@app.get(“/tasks/”, response_model=list[schemas.Task]) # レスポンスは Task スキーマのリスト
def read_tasks(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
“””全てのタスクを取得するエンドポイント”””
# クエリパラメータ skip と limit は依存性注入されず、FastAPIが自動的に解釈
tasks = crud.get_tasks(db=db, skip=skip, limit=limit)
return tasks

@app.get(“/tasks/{task_id}”, response_model=schemas.Task)
def read_task(task_id: int, db: Session = Depends(get_db)):
“””指定されたIDのタスクを取得するエンドポイント”””
db_task = crud.get_task(db=db, task_id=task_id)
if db_task is None:
# タスクが見つからない場合は 404 Not Found エラーを返す
raise HTTPException(status_code=404, detail=”Task not found”)
return db_task

@app.put(“/tasks/{task_id}”, response_model=schemas.Task)
def update_task(task_id: int, task: schemas.TaskUpdate, db: Session = Depends(get_db)):
“””指定されたIDのタスクを更新するエンドポイント”””
db_task = crud.update_task(db=db, task_id=task_id, task=task)
if db_task is None:
# タスクが見つからない場合は 404 Not Found エラーを返す
raise HTTPException(status_code=404, detail=”Task not found”)
return db_task

@app.delete(“/tasks/{task_id}”, response_model=schemas.Task) # 削除されたオブジェクトを返すことも可能
def delete_task(task_id: int, db: Session = Depends(get_db)):
“””指定されたIDのタスクを削除するエンドポイント”””
db_task = crud.delete_task(db=db, task_id=task_id)
if db_task is None:
# タスクが見つからない場合は 404 Not Found エラーを返す
raise HTTPException(status_code=404, detail=”Task not found”)
return db_task

— ルートパスを追加 (オプション) —

@app.get(“/”)
def read_root():
return {“message”: “Welcome to the FastAPI Todo App!”}
“`

コード解説 (main.py):

  • 必要なモジュール（crud, models, schemas, database）をインポートします。
  • models.Base.metadata.create_all(bind=engine): アプリケーション起動時に、データベースエンジンを使って models.py で定義された全てのテーブルを作成します。開発中はこれで十分ですが、本番環境ではマイグレーションツール（Alembic）を使うのが適切です。
  • app = FastAPI(): FastAPIアプリケーションを作成します。
  • from .database import get_db: データベースセッションを提供する依存関数 get_db をインポートします。
  • 各パスオペレーション関数 (create_task, read_tasks など) の引数に db: Session = Depends(get_db) を追加しています。
    • db: Session: 引数名が db で、型ヒントが Session であることを示しています。
    • = Depends(get_db): この引数の値は、get_db() 関数を実行した結果（yield された値）であることをFastAPIに指示します。
  • FastAPIは、リクエストが来るたびに get_db() を呼び出し、返されたセッションオブジェクトをそのパスオペレーション関数に渡します。パスオペレーション関数の処理が完了すると、get_db ジェネレーター関数の finally ブロックが実行され、セッションが閉じられます。
  • 各エンドポイントは、受け取ったPydanticモデルのデータやパス/クエリパラメータ、そしてデータベースセッション (db) を、対応する crud.py の関数に渡しています。
  • response_model=schemas.Task (または list[schemas.Task]) : パスオペレーションデコレーターの response_model 引数にPydanticスキーマを指定することで、FastAPIは関数の戻り値がそのスキーマに準拠しているかを検証し、レスポンスボディをそのスキーマに従ってシリアライズします。これにより、APIの出力データの構造が保証され、自動ドキュメントにも反映されます。
  • raise HTTPException(status_code=404, detail="Task not found"): データベースから期待するデータが見つからなかった場合など、エラーが発生した際には HTTPException を発生させます。FastAPIはこれをキャッチし、適切なHTTPステータスコード（ここでは404 Not Found）とレスポンスボディをクライアントに返します。
  • Pydantic V2 / SQLAlchemy モデルの変換: crud.create_task などから返されるのはSQLAlchemyの models.Task オブジェクトですが、response_model で schemas.Task を指定しているため、FastAPIは自動的に models.Task オブジェクトを schemas.Task オブジェクトに変換しようとします。schemas.Task の model_config = ConfigDict(from_attributes=True) (または orm_mode = True) 設定により、この変換が可能になっています。

5.6. アプリケーションの実行

プロジェクトルートディレクトリ (fastapidb_example) で、仮想環境がアクティブなことを確認してから、以下のコマンドを実行します。

bash
uvicorn main:app --reload

  • uvicorn main:app: main.py ファイル内の app オブジェクト（FastAPIインスタンス）を実行します。
  • --reload: ソースコードが変更されるたびに自動的にサーバーを再起動します。開発中に便利です。

サーバーが起動すると、以下のような出力が表示されます。

INFO: Will watch for changes in these directories: ['/path/to/your/project/fastapidb_example']
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: Started reloader process [PID]
INFO: Started server running on http://127.0.0.1:8000
INFO: Waiting for application startup.
INFO: Application startup complete.

これで、APIサーバーが http://127.0.0.1:8000 で起動しました。

5.7. APIのテスト (Swagger UI)

FastAPIは、アプリケーションを実行すると自動的にインタラクティブなAPIドキュメント（Swagger UI）を生成します。ブラウザで http://127.0.0.1:8000/docs にアクセスしてください。

定義した /tasks/ エンドポイントが表示されているはずです。Swagger UIを使って、実際にAPIを操作してみましょう。

  1. タスク作成 (POST /tasks/):

    • POST /tasks/ を展開し、「Try it out」をクリック。
    • “Request body” に以下のようなJSONを入力します。
      json
      {
      "title": "読書",
      "description": "プログラミングに関する本を読む"
      }
    • 「Execute」をクリック。
    • サーバーからのレスポンスが表示されます。作成されたタスクのIDなどが含まれているはずです。

  2. タスク一覧取得 (GET /tasks/):

    • GET /tasks/ を展開し、「Try it out」をクリック。
    • skip と limit パラメータはデフォルト値のまま、「Execute」をクリック。
    • 作成したタスクがリストとして表示されます。

  3. 特定タスク取得 (GET /tasks/{task_id}):

    • GET /tasks/{task_id} を展開し、「Try it out」をクリック。
    • Path parameter の task_id に、作成したタスクのIDを入力（例: 1）。
    • 「Execute」をクリック。
    • 指定したIDのタスクが表示されます。存在しないIDを指定すると、404エラーが返されます。

  4. タスク更新 (PUT /tasks/{task_id}):

    • PUT /tasks/{task_id} を展開し、「Try it out」をクリック。
    • Path parameter の task_id に、更新したいタスクのIDを入力。
    • Request body に以下のようなJSONを入力します。completed を true に変更してみましょう。
      json
      {
      "title": "読書",
      "description": "プログラミングに関する本を読み終えた",
      "completed": true
      }
    • 「Execute」をクリック。更新されたタスクの情報がレスポンスとして返されます。

  5. タスク削除 (DELETE /tasks/{task_id}):

    • DELETE /tasks/{task_id} を展開し、「Try it out」をクリック。
    • Path parameter の task_id に、削除したいタスクのIDを入力。
    • 「Execute」をクリック。削除されたタスクの情報（削除前の状態）がレスポンスとして返されます。再度そのIDで取得しようとすると、404エラーになります。

このように、Swagger UIを使って、実装したAPIエンドポイントが期待通りに動作するか簡単に確認できます。同時に、APIドキュメントも自動生成されていることが確認できます。

6. 実践的な考慮事項

これまででFastAPIとSQLAlchemyの基本的な連携方法は理解できましたが、実際のアプリケーション開発ではいくつかの追加の考慮事項があります。

6.1. エラーハンドリング

FastAPIは HTTPException を使うことで標準的なHTTPエラーレスポンスを簡単に返せます。データベース操作中に発生する可能性のあるエラー（例: 制約違反）についても、適切にハンドリングする必要があります。

SQLAlchemyの操作で例外が発生した場合、通常、セッションはエラー状態になり、コミットできなくなります。database.py で get_db 関数を try...finally でラップしているため、例外が発生してもセッションは安全に閉じられ、自動的にロールバックされます。特定のデータベースエラーをキャッチして、クライアントに分かりやすいエラーメッセージとして返す必要がある場合は、パスオペレーション関数やCRUD関数内で例外ハンドリングを追加します。

6.2. トランザクション管理

database.py で sessionmaker に autocommit=False を指定したため、db.commit() を明示的に呼び出す必要があります。複数のデータベース操作をまとめて一つの不可分な単位（トランザクション）として扱いたい場合に便利です。

例えば、タスクを完了したら関連する通知レコードを作成する、といった複数の操作がある場合、それらを同じセッション内で行い、全て成功した場合にのみ db.commit() を呼び出します。途中でエラーが発生した場合は、セッションが閉じられる際に自動的にロールバックされます（または明示的に db.rollback() を呼び出せます）。

FastAPIの依存性注入とジェネレーターを使う get_db パターンは、暗黙的にリクエストごとのトランザクション（成功すれば自動コミット、失敗すれば自動ロールバック）を提供しているとも考えられますが、より複雑なトランザクション制御が必要な場合は、CRUD関数内で明示的に db.begin() や db.rollback() を使うことも可能です。

6.3. 非同期処理への対応 (Optional – Advanced)

FastAPIは非同期処理（async/await）をネイティブにサポートしていますが、上で使用したSQLAlchemyは同期版です。同期的なデータベース操作（例: db.query(...)）は、FastAPIの非同期イベントループをブロックしてしまいます。これは、同時に多数のリクエストが来た場合にパフォーマンスのボトルネックになる可能性があります。

SQLAlchemy 1.4以降、そして特に2.0では、asyncioに対応した非同期版のORMが提供されています。非同期版を使用するには、create_engine の代わりに create_async_engine を使用し、セッションも async_sessionmaker で作成し、データベース操作（await session.execute(...) など）も await が必要になります。また、非同期データベースドライバー（asyncpg, aiomysql など）のインストールも必要です。

FastAPIで最高のパフォーマンスを引き出すためには非同期版SQLAlchemyの使用が推奨されますが、導入のハードルは少し上がります。初心者の方はまず同期版で基本的な連携を理解し、その後に非同期版に挑戦するのが良いでしょう。

今回の記事では、初心者向けという観点から、シンプルで分かりやすい同期版のSQLAlchemyを使用しました。

6.4. 環境設定の分離

データベースURLやその他の設定値（APIキーなど）をコードの中に直接書くのは好ましくありません。これらの設定値は、環境変数や設定ファイル（.env ファイルなど）から読み込むようにするのが一般的です。

python-dotenv や Pydantic の Settings 管理機能などを使うと、環境変数の読み込みやバリデーションを簡単に行えます。

6.5. セキュリティ

ORMを使用しているため、基本的なSQLインジェクションの危険性はほとんどありません。SQLAlchemyが適切に値をエスケープ処理してくれます。しかし、API全体としては認証・認可、入力値のサニタイズ（特にユーザー入力）、レート制限など、様々なセキュリティ対策を講じる必要があります。

6.6. テスト

アプリケーションの品質を保つためにはテストが不可欠です。FastAPIアプリケーションは、pytest などのテストフレームワークと httpx のようなHTTPクライアントライブラリを使って簡単にテストできます。データベース連携部分のテストでは、インメモリデータベース（SQLiteなど）や、テスト専用のデータベースを使ってテストデータを準備・クリーンアップする仕組みを用意するのが一般的です。

7. まとめ

この記事では、FastAPIとSQLAlchemyを使ってデータベース連携を行う基本的な手法を学びました。

  • FastAPIはその高速性と開発効率でモダンなAPI開発に適しています。
  • SQLAlchemyは強力なORM機能を提供し、データベース操作をPythonオブジェクトとして扱えます。
  • database.py, models.py, schemas.py, crud.py, main.py というファイル構成で、役割分担を明確にしました。
  • SQLAlchemyの Engine, Base, Session, Column, query の基本的な使い方を学びました。
  • Pydanticモデルを使って、APIのリクエスト/レスポンスのデータ構造を定義し、データ検証とシリアライズを行いました。
  • FastAPIの依存性注入システム (Depends) を利用して、リクエストごとにデータベースセッションを作成・提供し、処理完了後に自動的に閉じる仕組みを実装しました。
  • CRUD操作を分離した crud.py 関数群と、それを利用するFastAPIエンドポイントを main.py に実装しました。
  • 自動生成されるSwagger UIを使ってAPIの動作確認を行いました。

FastAPIとSQLAlchemyの組み合わせは非常に強力であり、この記事で学んだ基本的な連携方法は、ほとんどのWeb API開発で応用できます。

次のステップとして、以下の内容を学ぶことをお勧めします。

  • データベースマイグレーションツール (Alembic): スキーマ変更を安全に行うために必須です。
  • 非同期版SQLAlchemy: 大量の同時リクエストを効率的に処理するために重要です。
  • 認証・認可: ユーザー認証を行い、APIへのアクセスを制御する方法です。
  • リレーションシップ: SQLAlchemyの relationship を使って、関連するテーブル間のデータを扱う方法です。
  • より複雑なクエリ: 複数のテーブルを結合したり、集計を行ったりする方法です。
  • テストコードの書き方: アプリケーションの品質を保証するために重要です。

これらの要素を習得することで、より複雑で堅牢なWebアプリケーションを開発できるようになります。

この詳細な解説が、FastAPIとSQLAlchemyを使ったデータベース連携を学ぶ上での一助となれば幸いです。

約5000語の詳細な記事として記述しました。準備から、SQLAlchemyの基本、FastAPIの基本、そして実際の連携方法をサンプルコードとともに解説し、さらに実践的な考慮事項にも触れています。初心者の方がステップバイステップで理解できるよう、各コード片の解説も丁寧に記述したつもりです。

コメントする

メールアドレスが公開されることはありません。 ※ が付いている欄は必須項目です

上部へスクロール