FastAPIとORMの入門 – 効率的なDB連携ガイド

By /

はい、承知いたしました。FastAPIとORM（SQLAlchemy Asyncio）を使った効率的なデータベース連携について、入門者向けの詳細な解説記事を約5000語で記述します。記事全体をここに直接表示します。

FastAPIとORMの入門 – 効率的なDB連携ガイド

Webアプリケーション開発において、データの永続化は不可欠です。ユーザー情報、商品リスト、ブログ記事など、様々なデータはデータベースに保存され、アプリケーションのロジックによって操作されます。このデータベース操作を効率的かつ安全に行うために、ORM (Object-Relational Mapper) と呼ばれる技術が広く利用されています。

本記事では、高速でモダンなPython WebフレームワークであるFastAPIと、Pythonで最も人気のある高機能ORMの一つであるSQLAlchemyを組み合わせて、効率的なデータベース連携を実現する方法を、入門者向けに徹底解説します。特に、FastAPIの非同期処理の特性を活かすために、SQLAlchemyのAsyncio対応機能を中心に掘り下げます。

約5000語にわたる詳細な解説を通じて、FastAPIとSQLAlchemyを使ったWebアプリケーションのバックエンド開発に必要な基礎知識と実践的な手法を習得できます。

想定する読者

  • FastAPIを使った開発に興味がある方
  • データベース連携を効率的に行いたい方
  • ORMの基本的な概念を学びたい方
  • Pythonでの非同期データベース処理に挑戦したい方
  • SQLAlchemyを使った開発経験がない、または少ない方

本記事で学ぶこと

  • FastAPIの基本的な使い方（簡単なAPI構築、依存性注入）
  • ORM（SQLAlchemy）の基本的な概念と使い方
  • FastAPIとSQLAlchemy (Asyncio) を連携させる方法
  • データベースセッションの管理方法（FastAPIの依存性注入を活用）
  • データの定義（SQLAlchemyモデルとPydanticモデル）
  • 基本的なCRUD (Create, Read, Update, Delete) 処理の実装
  • 非同期データベース操作のベストプラクティス
  • プロジェクトの構造化

前提知識

  • Pythonの基本的な文法
  • HTTPの基本的な概念（GET, POST, PUT, DELETEなど）
  • データベースの基本的な概念（テーブル、カラム、SQLなど）
  • asyncio や await の基本的な理解があると望ましいですが、必須ではありません。記事中で解説します。

目次

  1. はじめに: FastAPIとORMがなぜ必要か
    • FastAPIとは？ なぜFastAPIを選ぶのか
    • ORMとは？ なぜORMを使うのか
    • FastAPIとORMを組み合わせるメリット
  2. 開発環境の準備
    • Pythonのインストール
    • 必要なライブラリのインストール
  3. FastAPIの基礎のおさらい (データベース連携に必要な部分)
    • FastAPIアプリケーションの作成
    • パスオペレーション (@app.get, @app.post など)
    • リクエストボディとレスポンスボディ (Pydantic)
    • 依存性注入 (Dependency Injection)
  4. ORM (SQLAlchemy Asyncio) の基礎
    • SQLAlchemyとは？ CoreとORM
    • なぜAsyncio対応が必要か
    • 主要な構成要素: Engine, Connection, Session, Metadata, Declarative Base
    • データベースへの接続設定
    • モデルの定義 (Declarative Mapping)
    • テーブルの作成
  5. FastAPIとSQLAlchemy (Asyncio) の連携
    • プロジェクト構造の設計
    • データベースエンジンの作成と接続設定の集約
    • 非同期セッションの管理と依存性注入
      • async_sessionmaker の使い方
      • yield を使ったセッションの依存性注入
      • セッションのライフサイクル管理
  6. モデルの定義とPydanticとの連携
    • SQLAlchemyモデルの定義 (データベーススキーマ)
    • Pydanticモデルの定義 (APIの入出力スキーマ)
    • SQLAlchemyモデルとPydanticモデルのマッピング
  7. CRUD処理の実装
    • 依存性注入されたセッションを使ったデータベース操作
    • Create (作成): データの挿入
      • Pydanticモデルでデータを受け取る
      • SQLAlchemyモデルのインスタンスを作成
      • セッションへの追加 (session.add())
      • 変更の確定 (session.commit())
      • データの再読み込み (session.refresh())
    • Read (読み取り): データの取得
      • 全件取得 (session.execute(select(...)))
      • 単件取得 (session.get())
      • クエリの実行 (session.execute()) と結果の扱い (scalars(), one_or_none())
      • Pydanticモデルへの変換
      • データが見つからない場合の処理 (HTTPException 404)
    • Update (更新): データの更新
      • 更新対象のデータを取得 (session.get())
      • Pydanticモデルで更新データを受け取る
      • 取得したオブジェクトの属性を更新
      • 変更の確定 (session.commit())
      • データの再読み込み (session.refresh())
    • Delete (削除): データの削除
      • 削除対象のデータを取得 (session.get())
      • セッションからの削除 (session.delete())
      • 変更の確定 (session.commit())
  8. リレーションシップの扱い
    • One-to-Manyリレーションシップの定義
    • 関連データの操作 (追加、取得)
    • Eager Loading (関連データの効率的な読み込み)
  9. 発展的なトピック (概要)
    • データベースマイグレーション (Alembic)
    • テスト
    • 設定管理
    • エラーハンドリングの強化
  10. まとめと次のステップ

1. はじめに: FastAPIとORMがなぜ必要か

現代のWebアプリケーションは、ユーザー管理、コンテンツの提供、トランザクション処理など、様々な種類のデータを扱います。これらのデータは、通常、リレーショナルデータベース（RDB）に保存されます。データベースとのやり取りはアプリケーションの中核をなす部分であり、効率的かつ安全な方法で行うことが求められます。

FastAPIとは？ なぜFastAPIを選ぶのか

FastAPIは、Python製のモダンなWebフレームワークです。以下の特徴から、近年注目を集めています。

  • 高速: Starlette (非同期Webフレームワーク) と Pydantic (データ検証ライブラリ) をベースにしており、高いパフォーマンスを発揮します。
  • Pythonの型ヒントを最大限に活用: 型ヒントを利用することで、コードの補完、エラーチェック、自動ドキュメント生成（OpenAPI/Swagger UI、ReDoc）を強力にサポートします。開発効率とコードの品質向上に大きく貢献します。
  • 非同期処理 (asyncio) に対応: Pythonの非同期機能 (async/await) をネイティブにサポートしており、I/Oバウンドな処理（データベースアクセス、外部API呼び出しなど）を効率的に並行して実行できます。これにより、スループットの高いアプリケーションを構築できます。
  • 使いやすい: 直感的でミニマルなAPI設計が可能で、学習コストが比較的低いと言われています。
  • 自動ドキュメント生成: APIの仕様書（OpenAPI）を自動で生成し、対話型のドキュメント（Swagger UIやReDoc）を提供します。

特に、データベースアクセスのようなI/Oバウンドな処理が多いバックエンドアプリケーションにおいて、FastAPIの非同期対応は大きなメリットとなります。データベースからの応答待ちの間に、他のリクエストを処理できるため、アプリケーション全体のパフォーマンスが向上します。

ORMとは？ なぜORMを使うのか

ORM (Object-Relational Mapper) は、オブジェクト指向プログラミング言語（Python）とリレーショナルデータベースの間で、データの変換を行うためのツールです。データベースのテーブルを行と列で表現する代わりに、Pythonのクラス（モデル）としてデータを扱えるようにします。

ORMを使う主なメリットは以下の通りです。

  • 生産性の向上: SQL文を直接書く代わりに、Pythonのオブジェクトを操作する感覚でデータベースとやり取りできます。これにより、開発速度が向上し、コード量が削減されます。
  • コードの保守性向上: データベーススキーマの変更がアプリケーションコード全体に及ぼす影響を局所化しやすくなります。また、Pythonコードでデータベース操作が完結するため、可読性が高まります。
  • データベースの抽象化: 特定のデータベース（例: PostgreSQL, MySQL, SQLite）に依存しないコードを書きやすくなります。データベースを変更する際も、設定ファイルの変更やドライバーの入れ替えだけで済む場合が多くなります。
  • セキュリティ: ORMは、SQLインジェクションのようなセキュリティリスクを軽減する機能を提供します。SQL文を文字列として組み立てるのではなく、ORMが安全なパラメータバインディングを使用してクエリを生成するためです。
  • オブジェクト指向の活用: リレーションシップ（関連）を持つテーブル間のデータを、Pythonオブジェクト間の参照として自然に扱うことができます。

FastAPIとORMを組み合わせるメリット

FastAPIとORM（特にAsyncio対応のSQLAlchemy）を組み合わせることで、以下のメリットが得られます。

  • 非同期による高効率なデータベースアクセス: FastAPIの非同期処理とSQLAlchemy Asyncioを組み合わせることで、データベースへのアクセスがノンブロッキングになります。これは、多数の同時リクエストを効率的に処理するために非常に重要です。
  • 型安全なAPIとデータベース操作: FastAPIのPydanticによるリクエスト/レスポンスの型検証と、SQLAlchemyのモデル定義によるデータベーススキーマの型表現を組み合わせることで、アプリケーション全体で型安全性を高めることができます。
  • 依存性注入によるセッション管理の簡略化: FastAPIの強力な依存性注入システムを利用して、データベースセッションのライフサイクルを適切に管理できます。各APIエンドポイントでセッションを取得し、処理終了時に自動的にクローズするといったパターンを簡単に実現できます。
  • 開発効率の向上: ORMによる生産性の高さと、FastAPIによるAPI開発の効率性を両立できます。

2. 開発環境の準備

本記事のコードを実行するために必要な環境を準備します。

Pythonのインストール

Python 3.8以上のバージョンが必要です。お使いのOSに合わせて公式サイトからダウンロード・インストールしてください。

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

プロジェクト用のディレクトリを作成し、その中に移動します。

bash
mkdir fastapi-orm-example
cd fastapi-orm-example

仮想環境を作成し、アクティベートすることを強く推奨します。

“`bash

仮想環境の作成 (venvを使用する場合)

python -m venv venv

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

Windowsの場合

.\venv\Scripts\activate

macOS/Linuxの場合

source venv/bin/activate
“`

必要なライブラリをインストールします。

bash
pip install fastapi uvicorn sqlalchemy[asyncio] aiosqlite pydantic

  • fastapi: FastAPIフレームワーク本体
  • uvicorn: FastAPIアプリケーションを実行するためのASGIサーバー
  • sqlalchemy[asyncio]: SQLAlchemy本体とAsyncio拡張機能
  • aiosqlite: SQLAlchemy AsyncioでSQLiteデータベースを扱うための非同期DBドライバー
  • pydantic: データ検証・シリアライゼーションライブラリ（FastAPIの必須ライブラリ）

SQLite以外のデータベース（PostgreSQLなど）を使う場合は、aiopg や asyncpg のような非同期ドライバーをインストールしてください。例：pip install sqlalchemy[asyncio] asyncpg

3. FastAPIの基礎のおさらい (データベース連携に必要な部分)

データベース連携に焦点を当てるため、FastAPIの基礎は簡潔に説明します。

FastAPIアプリケーションの作成

main.py ファイルを作成します。

“`python

main.py

from fastapi import FastAPI

app = FastAPI()

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

アプリケーションの実行 (ターミナルで実行)

uvicorn main:app –reload

“`

uvicorn main:app --reload でサーバーを起動し、ブラウザで http://127.0.0.1:8000 にアクセスすると {"Hello": "World"} が表示されるはずです。

パスオペレーション (@app.get, @app.post など)

FastAPIでは、HTTPメソッド（GET, POST, PUT, DELETEなど）に対応するデコレーターを使って、特定のURLパスに対する処理を定義します。

“`python
from fastapi import FastAPI, HTTPException

app = FastAPI()

items = {“foo”: “The Foo Fighters”, “bar”: “The Bar Fighters”}

@app.get(“/items/{item_id}”)
def read_item(item_id: str):
if item_id not in items:
raise HTTPException(status_code=404, detail=”Item not found”)
return {“item_id”: item_id, “item”: items[item_id]}

リクエストボディの受け取り

from pydantic import BaseModel

class Item(BaseModel):
name: str
price: float
is_offer: bool | None = None

@app.post(“/items/”)
def create_item(item: Item):
return {“item”: item}
“`

リクエストボディとレスポンスボディ (Pydantic)

FastAPIはPydanticを使って、リクエストボディのデータ検証や、レスポンスボディの自動シリアライゼーションを行います。上の例のように、Pydanticモデルを引数や返り値の型ヒントとして指定するだけで利用できます。

依存性注入 (Dependency Injection)

FastAPIの依存性注入システムは、データベースセッションの管理において非常に重要な役割を果たします。関数（依存性）を定義し、それをパスオペレーション関数の引数に型ヒントとして指定することで、FastAPIが自動的にその依存性を解決し、パスオペレーション関数に渡してくれます。

依存性には、他の関数、クラス、あるいは yield を含むジェネレーター関数を指定できます。特に、yield を使うと、依存性の取得（yield の前）とクリーンアップ（yield の後）の処理を記述できます。データベースセッションの取得とクローズにこの機能を利用します。

“`python

依存性の例 (DBセッションの代わりとして)

def get_current_user():
# ユーザー認証ロジックなど…
return {“username”: “testuser”}

from fastapi import Depends

@app.get(“/users/me”)
def read_users_me(current_user: dict = Depends(get_current_user)):
return current_user

yield を使う依存性の例

async def get_db():
db = “fake_db_connection” # 実際はDBセッションを取得
try:
yield db # ここで依存性が提供される
finally:
print(“DB connection closed”) # ここでクリーンアップ処理

@app.get(“/items/”)
async def read_items(db: str = Depends(get_db)):
# dbを使ってアイテムを読み取る処理
print(f”Using db: {db}”)
return [{“name”: “item1”, “price”: 100}]
“`

データベース連携では、yield を使った依存性注入で非同期データベースセッションを提供し、リクエスト処理後にセッションを確実にクローズする仕組みを構築します。

4. ORM (SQLAlchemy Asyncio) の基礎

次に、データベース操作の中核となるSQLAlchemyのAsyncio対応機能について説明します。

SQLAlchemyとは？ CoreとORM

SQLAlchemyは、Pythonのための高機能で柔軟なSQLツールキットおよびORMです。

  • SQLAlchemy Core: データベーススキーマの定義（テーブル、カラムなど）や、SQL文をPythonオブジェクトとして組み立てて実行するための低レベルなAPIを提供します。
  • SQLAlchemy ORM: Coreの上に構築され、Pythonクラス（モデル）とデータベーステーブルをマッピングし、オブジェクト指向的な方法でデータを操作できるようにします。本記事では主にORMを使います。

なぜAsyncio対応が必要か

Webアプリケーション、特にFastAPIのような非同期フレームワークでは、I/Oバウンドな処理（データベースアクセス、ファイルアクセス、ネットワーク通信など）をブロッキングせずに実行することが重要です。同期的な（ブロッキングする）データベースドライバーやORMを使うと、データベースからの応答を待っている間、そのFastAPIワーカープロセスは他のリクエストを処理できなくなります。これはアプリケーションのスケーラビリティを大きく損ないます。

SQLAlchemy Asyncioは、非同期データベースドライバー (asyncpg for PostgreSQL, aiosqlite for SQLite, asyncmy for MySQLなど) を利用して、データベース操作を await 可能な非同期処理として実行できるようにします。これにより、FastAPIの非同期性を最大限に活かし、高い並列処理能力を持つアプリケーションを構築できます。

主要な構成要素: Engine, Connection, Session, Metadata, Declarative Base

SQLAlchemy ORMを使う上で理解すべき主要な構成要素です。Asyncio対応では、これらのオブジェクトも非同期対応版を使います。

  • Engine: データベースとの接続プールを管理するオブジェクトです。アプリケーション全体でシングルトンとして持つことが一般的です。非同期版は create_async_engine で作成します。
  • Connection: Engineから取得される、データベースへの実際の接続です。通常、ユーザーが直接扱うことは少なく、Sessionが内部で管理します。
  • Session: ORMの中心的なオブジェクトです。データベース操作（データの追加、更新、削除、クエリ実行）はSessionを通じて行われます。Sessionはトランザクションの境界を管理し、ORMオブジェクトの状態変化を追跡します。非同期版は async_sessionmaker で作成したファクトリから取得します。Asyncio対応のSessionは、そのメソッド（execute, commit, rollback, refresh など）が await 可能な非同期関数になります。
  • Metadata: データベースのスキーマ情報（テーブル、カラム、制約など）を表現するオブジェクトです。
  • Declarative Base: ORMのモデルクラスを定義するためのベースクラスです。sqlalchemy.orm.declarative_base または sqlalchemy.orm.DeclarativeBase を使って作成します。

データベースへの接続設定

データベースへの接続には、データベースURLを指定します。URLの形式は dialect+driver://user:password@host:port/dbname のようになります。

SQLiteをAsyncioで使う場合のURLは sqlite+aiosqlite:///./test.db のようになります。これは、カレントディレクトリに test.db というファイルを作成してデータベースとして使うことを意味します。

モデルの定義 (Declarative Mapping)

ORMモデルは、Pythonクラスとして定義します。このクラスは、データベーステーブルとマッピングされます。SQLAlchemy 2.0スタイルのDeclarative Mappingでは、Mapped 型アノテーションを使ってカラムを定義します。

“`python

models.py (例)

from typing import List
from typing import Optional

from sqlalchemy import ForeignKey
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

モデル定義のベースクラス

class Base(DeclarativeBase):
pass

User モデル (users テーブルにマッピング)

class User(Base):
tablename = “users” # マッピングするテーブル名

id: Mapped[int] = mapped_column(primary_key=True) # 主キー
name: Mapped[str] = mapped_column(String(30)) # 名前 (文字列)
fullname: Mapped[Optional[str]] = mapped_column(String(30)) # フルネーム (null許容文字列)

# リレーションシップ (Userは複数のAddressを持つ)
addresses: Mapped[List["Address"]] = relationship(back_populates="user")

def __repr__(self) -> str:
    return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})"

Address モデル (addresses テーブルにマッピング)

class Address(Base):
tablename = “addresses” # マッピングするテーブル名

id: Mapped[int] = mapped_column(primary_key=True) # 主キー
email_address: Mapped[str]
user_id: Mapped[int] = mapped_column(ForeignKey("users.id")) # 外部キー

# リレーションシップ (Addressは単一のUserに属する)
user: Mapped["User"] = relationship(back_populates="addresses")

def __repr__(self) -> str:
    return f"Address(id={self.id!r}, email_address={self.email_address!r}, user_id={self.user_id!r})"

“`

  • DeclarativeBase: ORMモデルクラスが継承するベースクラスを作成します。
  • __tablename__: このクラスがマッピングするデータベーステーブル名を指定します。
  • Mapped[...]: カラムの型を指定します。int, str, Optional[str], List[...] など。
  • mapped_column(...): カラムに関する詳細設定（主キー、外部キー、データ型など）を行います。String(30) はVARCHAR(30)のような文字列型を意味します。
  • primary_key=True: そのカラムがテーブルの主キーであることを示します。
  • ForeignKey(...): 外部キー制約を定義します。"users.id" は users テーブルの id カラムを参照することを意味します。
  • relationship(...): 関連するモデルとのリレーションシップを定義します。back_populates は双方向のリレーションシップを確立するために使われます。List["Address"] は “Userは複数のAddressを持つ” ことを、"User" は “Addressは単一のUserに属する” ことを表現しています（List が付くかつかないかでOneかManyかを表すのが一般的ですが、SQLAlchemy 2.0ではList型ヒントでManyを表現します）。

テーブルの作成

定義したモデルに基づいて、データベースにテーブルを作成するには、EngineとBaseのMetadataを使います。Asyncio対応では、非同期で実行する必要があります。

“`python

database.py (一部抜粋)

from sqlalchemy.ext.asyncio import create_async_engine
from models import Base # 上記で定義したBaseクラスをインポート

async def create_db_tables(engine):
# Engineを使ってデータベースに接続し、テーブルを作成
async with engine.begin() as conn:
# Base.metadataに登録されている全てのテーブルを作成
await conn.run_sync(Base.metadata.create_all)
“`

run_sync は、非同期コンテキスト内で同期的なSQLAlchemy Core/ORM操作を実行するためのメソッドです。create_all は同期的な操作なので、run_sync を使って実行します。

5. FastAPIとSQLAlchemy (Asyncio) の連携

ここからがFastAPIとSQLAlchemyを組み合わせる本番です。非同期データベースセッションをFastAPIの依存性注入を使って管理する仕組みを構築します。

プロジェクト構造の設計

アプリケーションが大きくなるにつれて、コードを機能ごとに分割することが重要になります。典型的なFastAPI + ORMプロジェクトの構造は以下のようになります。

fastapi-orm-example/
├── main.py # FastAPIアプリケーションのエントリーポイント
├── database.py # データベース接続設定、エンジンの作成、セッション提供の依存性
├── models.py # SQLAlchemy ORMモデルの定義
├── schemas.py # Pydanticモデルの定義 (APIの入出力スキーマ)
├── crud.py # データベース操作（CRUD処理）をカプセル化する関数群
└── routers/ # APIエンドポイントを定義するファイル群
├── __init__.py
├── users.py # ユーザー関連のAPIエンドポイント
└── items.py # アイテム関連のAPIエンドポイント

この構造に従ってファイルを作成・修正していきます。

データベースエンジンの作成と接続設定の集約 (database.py)

データベース接続に関する設定とEngineの作成を database.py にまとめます。

“`python

database.py

import os
from typing import AsyncGenerator

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

環境変数からデータベースURLを取得、なければSQLiteを使用

DATABASE_URL = os.getenv(“DATABASE_URL”, “sqlite+aiosqlite:///./test.db”)

Engineの作成

engine = create_async_engine(DATABASE_URL, echo=True) # echo=Trueで実行されるSQLログを表示 (デバッグ用)

AsyncSessionのファクトリを作成

expire_on_commit=False: commit後もオブジェクトの属性にアクセス可能にする

AsyncSessionLocal = sessionmaker(
bind=engine,
class_=AsyncSession,
expire_on_commit=False,
)

データベースセッションを提供する依存性ジェネレーター

async def get_db() -> AsyncGenerator[AsyncSession, None]:
async with AsyncSessionLocal() as session:
try:
yield session
await session.commit() # 自動コミット (オプション、ここではリクエスト成功時にコミット)
except Exception:
await session.rollback() # エラー時はロールバック
raise
finally:
# AsyncSessionLocal() as session ブロックの終了時に自動的にクローズされる
pass # 明示的な close() は不要
“`

  • DATABASE_URL: データベース接続文字列です。本番環境では環境変数から取得するのが一般的です。ここではデフォルトとしてSQLiteを設定しています。
  • create_async_engine: 非同期対応のEngineを作成します。echo=True は実行されるSQL文をコンソールに表示するデバッグオプションです。
  • AsyncSessionLocal: 非同期セッションオブジェクト（AsyncSession）を作成するためのファクトリです。sessionmaker に class_=AsyncSession を指定します。expire_on_commit=False は、セッションがコミットされた後も、そのセッションで取得したオブジェクトの属性にアクセスできるようにするための設定です。これは、APIレスポンスを返す前にセッションを閉じる場合に便利です。
  • get_db(): これがFastAPIの依存性注入で使うジェネレーター関数です。
    • async with AsyncSessionLocal() as session:: AsyncSessionLocal ファクトリを使って非同期セッションを取得し、非同期コンテキストマネージャーとして使用します。async with ブロックの開始時にセッションが作成され、終了時に自動的にクローズされます。
    • yield session: ここで生成された session オブジェクトが、依存性としてパスオペレーション関数に渡されます。
    • try...except...finally: yield の後のコードは、パスオペレーション関数の実行が完了した後（正常終了または例外発生）に実行されます。
    • await session.commit(): パスオペレーション関数が例外なく完了した場合、データベースへの変更をコミットします。
    • await session.rollback(): パスオペレーション関数で例外が発生した場合、トランザクションをロールバックします。
    • raise: ロールバック後、発生した例外を再度送出します。

この get_db 関数をFastAPIのパスオペレーション関数の引数に Depends と共に指定することで、リクエストごとに新しいセッションが取得され、処理後に適切にクローズされるようになります。

データベーステーブルの作成処理を追加

main.py に、アプリケーション起動時にテーブルを作成する処理を追加します。

“`python

main.py (修正)

from contextlib import asynccontextmanager

from fastapi import FastAPI

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

アプリケーション起動/終了時のイベントハンドラ

@asynccontextmanager
async def lifespan(app: FastAPI):
# 起動処理: データベーステーブルを作成
async with engine.begin() as conn:
# await conn.run_sync(Base.metadata.drop_all) # デバッグ用: テーブルを毎回削除する場合
await conn.run_sync(Base.metadata.create_all)
print(“Database tables created/checked.”)
yield
# 終了処理 (今回は特に何もなし)
print(“Application shutting down.”)

lifespan コンテキストマネージャーを渡して FastAPI アプリケーションを作成

app = FastAPI(lifespan=lifespan)

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

ルーターを追加する場所 (後述)

app.include_router(…)

“`

  • @asynccontextmanager: 非同期コンテキストマネージャーを定義するためのデコレーターです。
  • lifespan(app: FastAPI): FastAPIアプリケーションのライフサイクル（起動前、起動後、終了前）で実行される非同期ジェネレーター関数です。
  • async with engine.begin() as conn:: Engineから非同期接続を取得し、非同期トランザクションを開始します。
  • await conn.run_sync(Base.metadata.create_all): models.py で定義された Base.metadata に基づいて、データベースにテーブルを作成します。テーブルが既に存在する場合は何もしません。
  • yield: ここでアプリケーションが起動し、リクエストを受け付けるようになります。
  • app = FastAPI(lifespan=lifespan): 作成した lifespan 関数を lifespan 引数に渡してFastAPIアプリケーションをインスタンス化します。

これで、アプリケーション起動時に自動的にデータベーステーブルが作成されるようになりました。

6. モデルの定義とPydanticとの連携

データベースとAPIの入出力で使うモデルを定義します。

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

models.py に、データベーステーブルに対応するORMモデルを定義します。先ほど例示した User モデルと Address モデルをそのまま使用します。

“`python

models.py

from typing import List, Optional
from sqlalchemy import ForeignKey, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
pass

class User(Base):
tablename = “users”

id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(30))
fullname: Mapped[Optional[str]] = mapped_column(String(30))

addresses: Mapped[List["Address"]] = relationship(back_populates="user")

def __repr__(self) -> str:
    return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})"

class Address(Base):
tablename = “addresses”

id: Mapped[int] = mapped_column(primary_key=True)
email_address: Mapped[str] = mapped_column(unique=True) # email_address をユニークに設定
user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

user: Mapped["User"] = relationship(back_populates="addresses")

def __repr__(self) -> str:
    return f"Address(id={self.id!r}, email_address={self.email_address!r}, user_id={self.user_id!r})"

``email_addressにunique=True` を追加しました。

Pydanticモデルの定義 (schemas.py)

APIのリクエストボディとして受け取るデータや、レスポンスとして返すデータの構造をPydanticモデルとして定義します。これらはデータベースモデルと似ていますが、目的が異なります。

  • データベースモデル (SQLAlchemy): データベースのスキーマとPythonオブジェクトをマッピングします。
  • Pydanticモデル: APIの入出力データの構造、型検証、シリアライゼーション/デシリアライゼーションを定義します。パスワードのようなセンシティブな情報をレスポンスから除外したり、データベースにはないがAPIに必要なフィールドを追加したりできます。

“`python

schemas.py

from typing import List, Optional
from pydantic import BaseModel, EmailStr

ユーザー作成時のリクエストボディ用

class UserCreate(BaseModel):
name: str
fullname: Optional[str] = None # デフォルト値 None で省略可能に
# password: str # パスワードフィールドなど (セキュリティのためレスポンスには含めない)

アドレス作成時のリクエストボディ用

class AddressCreate(BaseModel):
email_address: EmailStr # EmailStr 型でメールアドレス形式を検証
# user_id は API 側で設定するので、リクエストボディには含めない

ユーザーレスポンス用 (IDを含む)

class User(BaseModel):
id: int
name: str
fullname: Optional[str] = None
# addresses: List[Address] = [] # 関連データもレスポンスに含める場合 (後述)

class Config:
    # SQLAlchemy モデルから Pydantic モデルへの変換を許可 (ORM モード)
    orm_mode = True # Pydantic v1 の設定
    from_attributes = True # Pydantic v2 の設定

アドレスレスポンス用 (IDを含む)

class Address(BaseModel):
id: int
email_address: EmailStr
user_id: int

class Config:
    orm_mode = True # Pydantic v1
    from_attributes = True # Pydantic v2

User レスポンスに Address リストを含める場合の User モデル再定義

class UserWithAddresses(User): # User を継承しても良い

addresses: List[Address] = []

class Config:

orm_mode = True # Pydantic v1

from_attributes = True # Pydantic v2

“`

  • BaseModel: 全てのPydanticモデルが継承するベースクラスです。
  • Optional[str] = None: フィールドが省略可能であり、デフォルト値が None であることを示します。
  • EmailStr: Pydanticが提供する特殊な型で、値が有効なメールアドレス形式であることを検証します。
  • class Config (orm_mode = True または from_attributes = True): この設定により、Pydanticモデルは SQLAlchemy モデルのインスタンスから属性を読み取って自分自身を構築できるようになります。これにより、データベースから取得したSQLAlchemyオブジェクトを、そのままAPIレスポンス用のPydanticオブジェクトに変換して返すことが可能になります。Pydantic v2からは from_attributes = True が推奨されます。

SQLAlchemyモデルとPydanticモデルのマッピング

APIのエンドポイントでは、リクエストとしてPydanticモデルを受け取り、データベース操作にはSQLAlchemyモデルを使用し、レスポンスとしてSQLAlchemyモデルのデータをPydanticモデルに変換して返します。この変換は、Config に orm_mode=True (v1) または from_attributes=True (v2) を設定したPydanticモデルの機能を使って行います。

“`python

例: SQLAlchemy オブジェクト -> Pydantic オブジェクト

db_user = User(id=1, name=”Alice”, fullname=”Alice Smith”) # SQLAlchemy モデルのインスタンス (仮)
pydantic_user = schemas.User.model_validate(db_user) # Pydantic v2 の方法

pydantic_user = schemas.User.from_orm(db_user) # Pydantic v1 の方法

“`

7. CRUD処理の実装

データベースとのCRUD操作を行うロジックを crud.py ファイルに記述し、それをAPIルーター (routers/users.py, routers/items.py など) から呼び出す形にします。

データベース操作をカプセル化 (crud.py)

crud.py ファイルを作成し、ユーザーとアドレスに関するデータベース操作関数を定義します。これらの関数は AsyncSession オブジェクトを引数として受け取ります。

“`python

crud.py

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload # 関連データをEager Loadするために使用

from models import User, Address
from schemas import UserCreate, AddressCreate # Pydantic スキーマをインポート

— User CRUD —

ユーザー単件取得 (IDによる)

async def get_user(db: AsyncSession, user_id: int) -> User | None:
# session.get() は主キーによる単件取得に便利
# await db.get(User, user_id) # Simple get by primary key

# Alternative: using select statement (more flexible for filtering etc.)
result = await db.execute(select(User).where(User.id == user_id))
return result.scalars().one_or_none() # 結果を User オブジェクトとして1件または None を取得

ユーザー全件取得 (スキップとリミット付き)

async def get_users(db: AsyncSession, skip: int = 0, limit: int = 100) -> list[User]:
# select(User) で User モデルに対応するクエリを作成
# offset(skip).limit(limit) でページネーション
# scalars() で結果を User オブジェクトのシーケンスとして取得
# all() で全てのオブジェクトのリストを取得
result = await db.execute(select(User).offset(skip).limit(limit))
return list(result.scalars().all())

ユーザー作成

async def create_user(db: AsyncSession, user: UserCreate) -> User:
# Pydantic モデルのデータを使って SQLAlchemy モデルのインスタンスを作成
# db_user = User(name=user.name, fullname=user.fullname) # 属性を個別に設定
db_user = User(user.model_dump()) # Pydantic v2: model_dump() で辞書に変換
# db_user = User(user.dict()) # Pydantic v1: dict() で辞書に変換

db.add(db_user) # セッションに新しいオブジェクトを追加
# db.commit() と db.refresh() は get_db 依存性で自動的に行われる（今回はその設計）
# もし get_db 依存性でコミットしない設計の場合、ここで await db.commit(), await db.refresh(db_user) が必要

return db_user # commit 後、IDなどが設定されたオブジェクトが返る (get_db の refresh による)

— Address CRUD —

アドレス作成

async def create_user_address(db: AsyncSession, address: AddressCreate, user_id: int) -> Address:
# db_address = Address(email_address=address.email_address, user_id=user_id)
# Pydantic v2: model_dump() で辞書に変換し、user_id を追加
db_address = Address(address.model_dump(), user_id=user_id)
# Pydantic v1: dict() で辞書に変換し、user_id を追加
# db_address = Address(address.dict(), user_id=user_id)

db.add(db_address)
# commit/refresh は get_db 依存性で処理

return db_address

ユーザーに紐づくアドレス全件取得 (Eager Load User 付き)

async def get_user_addresses(db: AsyncSession, user_id: int) -> list[Address]:
# selectinload(Address.user) で Address を取得する際に、関連する User も同時に効率的に読み込む (Eager Loading)
result = await db.execute(
select(Address)
.where(Address.user_id == user_id)
.options(selectinload(Address.user)) # N+1問題を避ける Eager Loading
)
return list(result.scalars().all())

アドレス単件取得

async def get_address(db: AsyncSession, address_id: int) -> Address | None:
result = await db.execute(select(Address).where(Address.id == address_id))
return result.scalars().one_or_none()

アドレス削除

async def delete_address(db: AsyncSession, address_id: int):
address = await get_address(db, address_id)
if address:
await db.delete(address)
# commit は get_db 依存性で処理
return address # 削除したオブジェクトを返すか、削除成功を返す
return None # 対象が見つからなかった場合
“`

  • AsyncSession: 型ヒントとして受け取ることで、これが非同期セッションであることを示します。
  • select(...): SQLAlchemy Coreの select 関数を使ってSELECT文に対応するクエリ式を作成します。
  • .where(...): WHERE句を追加します。
  • .offset(...).limit(...): LIMIT句とOFFSET句を追加し、ページネーションを実現します。
  • await db.execute(...): 非同期でクエリを実行します。結果セットを返します。
  • result.scalars(): 結果セットから、ORMオブジェクト（この場合は User または Address のインスタンス）を抽出するためのオブジェクトを返します。
  • .all(): scalars() オブジェクトから全てのORMオブジェクトをリストとして取得します。
  • .one_or_none(): scalars() オブジェクトからORMオブジェクトを1つ取得します。結果が0件の場合は None を返します。2件以上ある場合はエラーになります。
  • db.add(obj): 新しいORMオブジェクトをセッションに追加します。
  • db.delete(obj): 既存のORMオブジェクトをセッションから削除対象としてマークします。
  • selectinload(Address.user): リレーションシップを通じて関連する User オブジェクトを効率的にロードするためのEager Loading戦略の一つです。これにより、ユーザーアドレスを取得する際に、関連するユーザー情報をN+1問題（アドレス1件ごとにユーザー情報を取得するためのクエリが発行される問題）を起こさずに取得できます。

APIエンドポイントの実装 (routers/users.py, routers/addresses.py)

routers ディレクトリを作成し、ユーザーとアドレスそれぞれのAPIエンドポイントを定義するファイルを作成します。

まず routers/users.py を作成します。

“`python

routers/users.py

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List

import crud # crud.py から関数をインポート
import schemas # schemas.py から Pydantic スキーマをインポート
from database import get_db # database.py から依存性をインポート

APIRouter インスタンスを作成

router = APIRouter(
prefix=”/users”, # このルーター内の全てのパスに “/users” をつける
tags=[“users”], # Swagger UI での表示グループ名
responses={404: {“description”: “Not found”}}, # このルーター全体で共通のレスポンス定義
)

ユーザー作成エンドポイント (POST /users/)

@router.post(“/”, response_model=schemas.User) # レスポンスのスキーマを指定
async def create_user(
user: schemas.UserCreate, # リクエストボディを schemas.UserCreate で受け取る
db: AsyncSession = Depends(get_db) # get_db 依存性を注入して非同期セッションを取得
):
# 同じ名前のユーザーが既に存在しないか確認 (ここでは単純な例)
# existing_user = await crud.get_user_by_name(db, name=user.name) # crud に同様の関数を追加することも可能
# if existing_user:
# raise HTTPException(status_code=400, detail=”User with this name already exists”)

# crud 関数を呼び出してユーザーを作成
db_user = await crud.create_user(db=db, user=user)
return db_user # 作成された SQLAlchemy オブジェクトを返す (response_model=schemas.User により Pydantic に自動変換)

ユーザー全件取得エンドポイント (GET /users/)

@router.get(“/”, response_model=List[schemas.User]) # レスポンスは User スキーマのリスト
async def read_users(
skip: int = 0, limit: int = 100, # クエリパラメータでページネーションを受け取る
db: AsyncSession = Depends(get_db) # セッション注入
):
users = await crud.get_users(db, skip=skip, limit=limit)
return users # SQLAlchemy オブジェクトのリストを返す (Pydantic に自動変換)

ユーザー単件取得エンドポイント (GET /users/{user_id})

@router.get(“/{user_id}”, response_model=schemas.User) # パスパラメータ user_id を受け取る
async def read_user(
user_id: int,
db: AsyncSession = Depends(get_db) # セッション注入
):
db_user = await crud.get_user(db, user_id=user_id)
if db_user is None:
# データが見つからない場合は 404 エラーを返す
raise HTTPException(status_code=404, detail=”User not found”)
return db_user # 見つかった SQLAlchemy オブジェクトを返す (Pydantic に自動変換)

NOTE: PUT (更新) や DELETE (削除) エンドポイントも同様に実装可能ですが、ここでは省略します。

例: PUT /users/{user_id} (更新処理)

@router.put(“/{user_id}”, response_model=schemas.User)

async def update_user(

user_id: int,

user_update: schemas.UserCreate, # 更新データを受け取るスキーマ (ここでは UserCreate を流用)

db: AsyncSession = Depends(get_db)

):

db_user = await crud.get_user(db, user_id=user_id)

if db_user is None:

raise HTTPException(status_code=404, detail=”User not found”)

# crud 関数を更新用に修正するか、ここで属性を更新

for var, value in user_update.model_dump(exclude_unset=True).items(): # exclude_unset=True で指定されていないフィールドは無視

setattr(db_user, var, value)

# セッションは db_user に変更を追跡しているので、commit するだけで更新がDBに反映される

# await db.commit() # get_db 依存性で自動コミットされる設計

await db.refresh(db_user) # 最新の状態をロード

return db_user

“`

次に routers/addresses.py を作成します。

“`python

routers/addresses.py

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List

import crud
import schemas
from database import get_db

router = APIRouter(
prefix=”/users/{user_id}/addresses”, # ユーザーIDに関連付けるパス
tags=[“addresses”],
responses={404: {“description”: “Not found”}},
)

特定のユーザーのアドレス作成 (POST /users/{user_id}/addresses/)

@router.post(“/”, response_model=schemas.Address)
async def create_address_for_user(
user_id: int, # パスパラメータから user_id を取得
address: schemas.AddressCreate, # リクエストボディ
db: AsyncSession = Depends(get_db) # セッション注入
):
# ユーザーが存在するか確認 (任意だが良いプラクティス)
db_user = await crud.get_user(db, user_id=user_id)
if db_user is None:
raise HTTPException(status_code=404, detail=”User not found”)

# crud 関数を呼び出してアドレスを作成
db_address = await crud.create_user_address(db=db, address=address, user_id=user_id)
return db_address # 作成された SQLAlchemy オブジェクトを返す

特定のユーザーのアドレス全件取得 (GET /users/{user_id}/addresses/)

@router.get(“/”, response_model=List[schemas.Address])
async def read_user_addresses(
user_id: int,
db: AsyncSession = Depends(get_db)
):
# ユーザーが存在するか確認 (任意)
db_user = await crud.get_user(db, user_id=user_id)
if db_user is None:
raise HTTPException(status_code=404, detail=”User not found”)

addresses = await crud.get_user_addresses(db, user_id=user_id)
return addresses # SQLAlchemy オブジェクトのリストを返す

アドレス単件取得 (GET /addresses/{address_id}) – ユーザーIDはパスに含めない例

@router.get(“/{address_id}”, response_model=schemas.Address) # Note: This path is NOT within the /users/{user_id} prefix

async def read_address(

address_id: int,

db: AsyncSession = Depends(get_db)

):

db_address = await crud.get_address(db, address_id=address_id)

if db_address is None:

raise HTTPException(status_code=404, detail=”Address not found”)

return db_address

アドレス削除 (DELETE /addresses/{address_id})

@router.delete(“/{address_id}”)
async def delete_address(
address_id: int,
db: AsyncSession = Depends(get_db)
):
deleted_address = await crud.delete_address(db, address_id=address_id)
if deleted_address is None:
raise HTTPException(status_code=404, detail=”Address not found”)
# 削除成功のレスポンス例 (ステータスコード 200 OK はデフォルト)
return {“detail”: “Address deleted successfully”, “address_id”: address_id}

“`

最後に main.py にルーターをインクルードします。

“`python

main.py (最終形に近い)

from contextlib import asynccontextmanager

from fastapi import FastAPI

from database import engine
from models import Base
from routers import users, addresses # 作成したルーターをインポート

@asynccontextmanager
async def lifespan(app: FastAPI):
# 起動処理: データベーステーブルを作成
async with engine.begin() as conn:
# await conn.run_sync(Base.metadata.drop_all) # デバッグ用
await conn.run_sync(Base.metadata.create_all)
print(“Database tables created/checked.”)
yield
# 終了処理
print(“Application shutting down.”)

app = FastAPI(lifespan=lifespan)

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

ルーターをアプリケーションにインクルード

app.include_router(users.router)
app.include_router(addresses.router)

uvicorn main:app –reload で実行

“`

これで、FastAPIのパスオペレーションから Depends(get_db) を使って非同期セッションを取得し、そのセッションを crud.py の非同期関数に渡してデータベース操作を行う、という一連の流れが完成しました。

FastAPIアプリケーションを uvicorn main:app --reload で起動し、ブラウザで http://127.0.0.1:8000/docs にアクセスすると、自動生成されたAPIドキュメント（Swagger UI）が表示されます。このドキュメントを使って、作成したエンドポイント（/users/, /users/{user_id}/addresses/ など）を試すことができます。

例えば、Swagger UIで：
1. /users/ の POST を開き、「Try it out」をクリック。
2. Request body に { "name": "Alice", "fullname": "Alice Smith" } のようなJSONを入力し、「Execute」をクリック。
3. 成功すれば、作成されたユーザー情報がレスポンスとして返されます。
4. /users/ の GET を開き、「Try it out」、「Execute」をクリックすると、作成したユーザーを含めたリストが表示されるはずです。
5. /users/{user_id}/addresses/ の POST を開き、「Try it out」をクリック。パスパラメータ user_id に先ほど作成したユーザーのID（例: 1）を入力。Request body に { "email_address": "[email protected]" } のようなJSONを入力し、「Execute」をクリック。
6. 成功すれば、作成されたアドレス情報が返されます。
7. /users/{user_id}/addresses/ の GET を開き、user_id に同じユーザーIDを入力して「Execute」をクリックすると、そのユーザーに紐づくアドレスリストが表示されます。
8. アドレス単件削除 /addresses/{address_id} は別ルーターパスで実装されていませんが、/users/{user_id}/addresses/{address_id} のように /users/{user_id} ルーター内で実装することも可能です。上記の例では /addresses/{address_id} パスとして削除エンドポイントを追加してみました。Swagger UIでそのエンドポイントを探し、削除したいアドレスのIDを入力して試してみてください。

8. リレーションシップの扱い

前のセクションで、User と Address モデルの間にOne-to-Manyのリレーションシップを定義し、crud.py で get_user_addresses 関数内で selectinload を使って関連する User データをEager Loadする方法を示しました。

Pydantic モデルで関連データを含めてレスポンスを返す場合は、schemas.py の User モデルにアドレスのリストフィールドを追加する必要があります。

“`python

schemas.py (リレーションシップを含む User モデルを追加)

from typing import List, Optional
from pydantic import BaseModel, EmailStr

… (UserCreate, AddressCreate, Address モデルはそのまま)

User レスポンス用 (リレーションシップを含む)

class User(BaseModel): # schemas.User を上書き or 別名で定義
id: int
name: str
fullname: Optional[str] = None
addresses: List[“Address”] = [] # Address モデルのリストを追加

class Config:
    orm_mode = True # Pydantic v1
    from_attributes = True # Pydantic v2

注意: Pydantic v2 では forward reference (“Address”) を使う場合、

全てのモデル定義後に update_forward_refs() またはモデル自体を import する必要がありましたが、

最新の Pydantic では基本的に不要になっています。

もしエラーが出る場合は、ファイルの最後に User.model_rebuild() (v2) などが必要かもしれません。

``
そして、関連データを含めて取得したいAPIエンドポイントのresponse_modelにこのUser` スキーマ（または関連データを含むように修正したスキーマ）を指定します。

例えば、ユーザー単件取得時にアドレスも取得したい場合、crud.get_user 関数で selectinload(User.addresses) を使うように修正し、ルーターの read_user エンドポイントの response_model に関連データを含む User スキーマを指定します。

“`python

crud.py (get_user 関数を修正 – User に紐づく Address を Eager Load)

… (imports) …

async def get_user(db: AsyncSession, user_id: int) -> User | None:
# selectinload(User.addresses) で User を取得する際に、関連する addresses も同時に読み込む
result = await db.execute(
select(User)
.where(User.id == user_id)
.options(selectinload(User.addresses)) # User -> Address のリレーションシップを Eager Load
)
return result.scalars().one_or_none()

… (他の crud 関数はそのまま) …

“`

“`python

routers/users.py (read_user エンドポイントの response_model を確認)

… (imports) …

@router.get(“/{user_id}”, response_model=schemas.User) # リレーションシップを含む schemas.User を使用
async def read_user(
user_id: int,
db: AsyncSession = Depends(get_db)
):
db_user = await crud.get_user(db, user_id=user_id) # Eager Load された User オブジェクトを取得
if db_user is None:
raise HTTPException(status_code=404, detail=”User not found”)
return db_user # Pydantic の orm_mode/from_attributes 機能により、関連する addresses も自動的にシリアライズされる
“`

これで、/users/{user_id} エンドポイントでユーザー情報を取得する際に、そのユーザーのアドレスリストもレスポンスに含まれるようになります。

9. 発展的なトピック (概要)

本記事でカバーした内容はFastAPI+ORM連携の基本的な部分です。実際のアプリケーション開発では、さらに以下のトピックを考慮する必要があります。

データベースマイグレーション (Alembic)

アプリケーションの開発が進むにつれて、データベーススキーマ（テーブル構造）は頻繁に変更されます。カラムの追加、削除、型の変更などです。これらのスキーマ変更を管理し、開発環境や本番環境のデータベースに適用するためのツールがデータベースマイグレーションツールです。SQLAlchemyと連携してよく使われるのが Alembic です。

Alembicを使うと、Pythonコードでスキーマ変更内容を記述（マイグレーションスクリプトの生成）し、そのスクリプトを実行してデータベースに適用（マイグレーションの実行）することができます。これにより、複数人での開発やデプロイメントが容易になります。

テスト

FastAPIアプリケーションとORMを使ったデータベース連携部分のテストは重要です。単体テスト、統合テストなどがあります。データベースを使ったテストでは、テスト用のデータベースを用意し、テストの前後でデータを初期化するなどの工夫が必要です。FastAPIはテストを書きやすい設計になっており、TestClient を使ってアプリケーション全体を模擬的に実行し、エンドポイントへのリクエスト・レスポンスをテストできます。

設定管理

データベース接続URLやその他の設定情報は、コードの中に直接書くのではなく、環境変数や設定ファイルから読み込むようにするべきです。python-dotenv や Pydantic Settings (Pydantic v2) といったライブラリを使うと、設定管理が容易になります。

エラーハンドリングの強化

本記事ではデータが見つからない場合に HTTPException(status_code=404) を使用しましたが、実際にはバリデーションエラー、データベース制約違反（ユニークキー重複など）、認証エラーなど、様々な種類のエラーが発生します。FastAPIの例外ハンドリング機能を活用して、これらのエラーを適切に処理し、クライアントに分かりやすいエラーレスポンスを返す仕組みを構築することが重要です。SQLAlchemy固有の例外（例: IntegrityError）をキャッチして、適切なHTTPExceptionに変換するといった処理が必要になる場合があります。

10. まとめと次のステップ

本記事では、FastAPIとSQLAlchemy (Asyncio) を使って効率的なデータベース連携を実現するための基本的な方法を詳細に解説しました。FastAPIの依存性注入を活用した非同期セッションの管理、SQLAlchemyのDeclarative Mappingを使ったモデル定義、PydanticによるAPI入出力の定義、そしてCRUD処理の実装を通して、両者を組み合わせる際の主要なパターンを学びました。

非同期ORMを使うことで、FastAPIのノンブロッキングな特性を最大限に活かし、パフォーマンスとスケーラビリティの高いアプリケーションを構築できることを理解いただけたかと思います。また、ORMを使うことでデータベース操作の生産性、保守性、安全性が向上することも確認しました。

この入門ガイドを終えた後、さらに理解を深めるための次のステップとしては、以下のトピックを学ぶことをお勧めします。

  • Alembic: データベースマイグレーションを習得し、スキーマ変更に柔軟に対応できるようにする。
  • SQLAlchemyの高度なクエリ: 複雑な条件でのフィルタリング、ソート、グループ化、集計など、SQLAlchemy ORMでより高度なクエリを記述する方法を学ぶ。
  • トランザクション管理の詳細: get_db 依存性での自動コミット/ロールバック以外の、より細やかなトランザクション管理の方法を学ぶ。
  • 認証・認可: ユーザー認証（OAuth2, JWTなど）と、特定のユーザーに特定の操作を許可する認可の仕組みをアプリケーションに追加する。FastAPIはセキュリティ機能のサポートが充実しています。
  • テスト駆動開発 (TDD): データベース連携を含むAPIエンドポイントのテストを効果的に記述する方法を学ぶ。
  • 設定管理: 環境変数を使った設定管理のベストプラクティスを学ぶ。
  • キャッシュ: 頻繁にアクセスされるデータをキャッシュしてデータベース負荷を減らす方法。
  • バックグラウンドタスク: 時間のかかる処理を非同期に実行するためのキューシステム（Celery, Redis Queueなど）との連携。
  • デプロイメント: 作成したアプリケーションをHeroku, AWS, Google Cloudなどのクラウド環境にデプロイする方法。Dockerコンテナ化などが一般的です。

これらの知識を習得することで、より堅牢でスケーラブルな実用的なWebアプリケーションを開発できるようになるでしょう。

本記事が、あなたのFastAPIとORMを使ったバックエンド開発の旅の出発点となり、今後の学習の助けとなれば幸いです。

コードの実行方法まとめ:

  1. 上記に従って必要なファイルを fastapi-orm-example ディレクトリ内に作成します (main.py, database.py, models.py, schemas.py, crud.py, routers/users.py, routers/addresses.py)。
  2. 仮想環境をアクティベートし、必要なライブラリをインストールします (pip install fastapi uvicorn sqlalchemy[asyncio] aiosqlite pydantic)。
  3. ターミナルで fastapi-orm-example ディレクトリに移動します。
  4. 以下のコマンドでアプリケーションを起動します。
    bash
    uvicorn main:app --reload
  5. ブラウザで http://127.0.0.1:8000/docs にアクセスし、APIドキュメント（Swagger UI）を開いて作成したエンドポイントを試します。データベースファイル (test.db) が自動的に作成されます。

（注：このテキストは約5000語を目指して記述されていますが、Markdown形式での表現やコードブロックの分量、日本語の特性により、厳密に5000語になることを保証するものではありません。内容は、入門者向けにFastAPIとAsyncio対応SQLAlchemy連携の基本を網羅し、実践的なコード例と共に詳細に解説することに重点を置いています。）

コメントする

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

上部へスクロール