概要
本記事では、FastAPIを使ったRESTful APIの実装例として、アカウント登録エンドポイントを段階的に構築していきます。
こちらの記事で整えた開発環境をベースに、実際のAPIエンドポイントを作成しながら、以下の内容を解説していきます。
本記事で解説する内容
- レイヤードアーキテクチャの実践: 責任を明確に分離したディレクトリ構成
- SQLModelによるテーブル設計: 共通基底クラスとタイムスタンプ管理
- Pydanticスキーマの活用: リクエスト/レスポンスの型安全な定義
- カスタムバリデーション: エラーメッセージの一元管理とハンドリング
- 依存性注入(DI)の実装: FastAPIのDependsを使った疎結合な設計
- 処理層の設計: サービス、プロセッサー、クエリの役割分担
想定読者
- 環境構築を完了している方
- FastAPIの基本的な機能を理解している方
- 保守性の高いAPI設計パターンを学びたい方
- 実際のコードを通じてアーキテクチャを理解したい方
本記事では、テーブル作成からエンドポイント実装、動作確認まで、7つのステップで段階的に進めていきます。各ステップで「なぜこう設計するのか」を解説しながら、実践的なコーディングパターンを身につけていきましょう。
実装例
1. アーキテクチャ
下記のような構成になるようにディレクトリを切ります。
以降は基本的に /src/app 配下に必要なファイルを作成していきます。
myproject
├─ .venv
├─ .vscode
| └─ settings.json
├─ src/
| └─ app/
| ├─ constants/ # 定数定義(設定値など)
| ├─ dependencies/ # FastAPI依存関係(DB接続、認証など)
| ├─ documents/ # APIドキュメント用のスキーマや説明
| ├─ endpoints/ # APIエンドポイント(ルーター)の実装
| ├─ exceptions/ # カスタム例外クラスの定義
| ├─ models/ # データベースモデル(SQLModel/SQLAlchemy)
| ├─ processors/ # データ処理・変換ロジック(ビジネスロジック)
| ├─ queries/ # データベースクエリ関数(CRUD操作)
| ├─ schemas/ # Pydanticスキーマ(リクエスト/レスポンスの型定義)
| | ├─ requests/ # リクエストボディのスキーマ
| | └─ responses/ # レスポンスのスキーマ
| ├─ services/ # エンドポイントの処理ハンドラー
| ├─ utilities/ # 汎用ユーティリティ関数(ハッシュ化、日付処理など)
| ├─ validations/ # カスタムバリデーションロジック
| ├─ database.py # データベース初期化・設定
| └─ main.py # FastAPIアプリケーションのエントリーポイント
├─ .env
├─ .env.example
├─ .gitignore
├─ .python-version
├─ docker-compose.yml
├─ Dockerfile
├─ pyproject.toml
├─ README.md
└─ uv.lock
fastAPIでよく用いられる、レイヤードアーキテクチャをベースに、初学者でも扱いやすい構造にしています。
厳密なアーキテクチャではありませんが、大規模開発でなければ十分耐えられる設計かと思います。
# ディレクトリの一括追加
# database.pyの追加
mkdir -p src/app/{constants,dependencies,documents,endpoints,exceptions,models,processors,queries,schemas/{requests,responses},services,utilities,validations} && \
touch src/app/database.py
下記リストに含まれるディレクトリは、基本的に同一のファイル構成にします。
- constants
- documents
- endpoints
- models
- processors
- queries
- schemas/requests
- schemas/responses
- services
- utilities
- validations
/src/app/<上記リスト内のディレクトリ名>/
├─ authentication.py
├─ common.py
├─ health.py
├─ statistics.py
︙
全てのエンドポイントに対し、あらかじめ種別 (authenticationなど) を定義して分類しておき、レイヤー (ディレクトリ名) と種別の組み合わせでソースの格納場所を特定しやすくするイメージです。
種別の分割単位については、プロジェクト規模に合わせて変更してください。
1ファイルが大きくなりすぎないよう、適切な設計を実装前に行うことを推奨します。
例外的に dependencies/ と exceptions/ は下記の構成にします。
# app配下のファイル名またはディレクトリ名と同一の命名
/src/app/dependencies/
├─ database.py
├─ services.py
︙
# HTTPステータス毎の例外で分類して命名
/src/app/exceptions/
├─ unauthorized.py # HTTP 401
├─ locked.py # HTTP 423
︙
2. アプリ初期化処理
アプリに必要な諸々の初期化処理を追記していきます。
main.py を下記のように書き換えます。
from contextlib import asynccontextmanager
from fastapi import FastAPI
from src.app.database import create_db_and_tables_async
@asynccontextmanager
async def lifespan(app: FastAPI):
await create_db_and_tables_async()
yield
app = FastAPI(
title="My Project",
description="A FastAPI project",
version="0.1.0",
lifespan=lifespan,
)
@asynccontextmanager はPython標準ライブラリ contextlib の機能であり、
「非同期対応のコンテキストマネージャ」を簡単に定義するためのデコレーターです。
FastAPIではアプリのライフサイクル管理 (起動 / 終了処理) に使われます。
-
アプリ起動時:
⤷ lifespan(app) が呼ばれる
⤷ create_db_and_tables_async() が await される(テーブル作成など)
⤷ yield の手前まで実行し、FastAPIに制御を返す
⤷ yield の後、FastAPI サーバーが起動 -
アプリ終了時:
⤷ yield の後のクリーンアップ処理(ここでは省略)が実行される
database.py にcreate_db_and_tables_asyncを追加してください。
from sqlmodel import SQLModel
from src.app.dependencies.database import async_engine
async def create_db_and_tables_async():
async with async_engine.begin() as conn:
await conn.run_sync(SQLModel.metadata.create_all)
アプリ初回起動時、DB内にSQLModelで定義されたテーブルを作成するための設定です。
この設定は開発用であり、本番環境ではAlembicによるマイグレーション管理に置き換えるのが一般的です。
app/dependencies/database.py に下記を追加してください。
import os
from sqlalchemy.ext.asyncio import create_async_engine
POSTGRES_USER = os.getenv("POSTGRES_USER")
POSTGRES_PASSWORD = os.getenv("POSTGRES_PASSWORD")
POSTGRES_DB = os.getenv("POSTGRES_DB")
POSTGRES_HOST = os.getenv("POSTGRES_HOST")
POSTGRES_PORT = os.getenv("POSTGRES_PORT")
ASYNC_DATABASE_URL = f"postgresql+asyncpg://{POSTGRES_USER}:{POSTGRES_PASSWORD}@{POSTGRES_HOST}:{POSTGRES_PORT}/{POSTGRES_DB}"
async_engine = create_async_engine(
ASYNC_DATABASE_URL,
echo=False,
pool_pre_ping=True,
pool_size=5,
max_overflow=10,
)
SQLAlchemyの非同期接続エンジンを生成しています。
SQLModelは内部的にSQLAlchemyを使用しているため、このエンジンが実際のDB通信を担っています。
ここまでで、アプリ初期化処理は完了です。
少しややこしい話が多くなってしまいましたが、まとめるとここで記述したコードは下記の処理を行なっています。
-
async_engineが作られる
⤷ DBとの非同期接続を管理 -
create_db_and_tables_async()
⤷async_engineで接続を開き、テーブルを作成 -
lifespan()
⤷ アプリ起動時に上のDB初期化関数を実行 -
FastAPI(lifespan=lifespan)
⤷ 起動時・終了時にライフサイクルイベントを統一管理
つまり、
FastAPIアプリの起動時に、
非同期SQLAlchemyエンジンを使ってDBと接続し、
SQLModelで定義されたテーブルを自動生成してから
サーバーを起動する
という根本の設定が完了した状態です。
3. エンドポイント実装
例として、ユーザーのアカウント新規登録を行うAPIを、次の順で実装していきます。
- アカウントを登録するテーブルの作成
- リクエスト&レスポンスのスキーマ定義
- リクエストバリデーションと、返却するカスタム例外の定義
- 処理層 (サービス→プロセッサーまたはクエリ)の作成
- 処理層とエンドポイントの接続
- エンドポイントの作成
- 動作確認
3.1. テーブルの作成
アカウントを登録するためのテーブルを用意します。
src/
└─ app/
└─ models/
├─ authentication.py
└─ common.py
/src/app/models/common.py に全てのテーブルで共通的に使用するidやcreated_atなどを定義します。
from datetime import UTC, datetime
from sqlalchemy import event
from sqlmodel import Field, SQLModel
class TimestampedBase(SQLModel):
id: int = Field(primary_key=True)
created_at: datetime = Field(default_factory=lambda: datetime.now(UTC), nullable=False)
updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC), nullable=False)
deleted_at: datetime | None = Field(default=None, nullable=True)
def soft_delete(self):
self.deleted_at = datetime.now(UTC)
@event.listens_for(TimestampedBase, "before_update", propagate=True)
def set_updated_at(mapper, connection, target):
target.updated_at = datetime.now(UTC)
@event.listens_for(TimestampedBase, "before_insert", propagate=True)
def set_created_at(mapper, connection, target):
now = datetime.now(UTC)
if not target.created_at:
target.created_at = now
if not target.updated_at:
target.updated_at = now
/src/app/models/authentication.py に、TimestampedBase を継承した Account テーブルを定義します。
from typing import ClassVar
from sqlmodel import Field
from src.app.models.common import TimestampedBase
class Account(TimestampedBase, table=True):
__tablename__: ClassVar[str] = "accounts"
email: str = Field(
unique=True,
title="Email",
description="ユーザーのemail",
)
hashed_password: str = Field(
title="Hashed Password",
description="ハッシュ化したユーザーのパスワード",
)
3.2. スキーマの作成
アカウント登録APIが受け取るリクエスト、返却するレスポンスを定義します。
テーブルではないため、SQLModelではなく、pydanticの BaseModel を継承して作成します。
src/
└─ app/
└─ schemas/
├─ requests/
| └─ authentication.py
└─ responses/
└─ common.py
from pydantic import BaseModel, ConfigDict, Field
class PostRegisterAccountRequest(BaseModel):
model_config = ConfigDict(
json_schema_extra={
"example": {
"email": "user@example.com",
"password": "Password123!",
},
}
)
email: str = Field(
title="Email Address",
description="ユーザーIDとして使用するemail",
examples=["user@example.com"],
)
password: str = Field(
...,
title="Password",
description="ユーザーのパスワード",
examples=["Password123!"],
)
アカウント登録APIは空のボディを返すため、共通的に使用するレスポンススキーマを定義して使用します。
from pydantic import BaseModel
class CommonEmptyResponse(BaseModel):
pass
3.3. リクエストバリデーション
先ほど作成した PostRegisterAccountRequest にバリデーションをかけます。
PostRegisterAccountRequest がインスタンス化される前に、validate_post_register_account_request を呼び出し、バリデーションエラーの場合にHTTPステータス422のカスタム例外を返却する設定をしていきます。
src/
└─ app/
├─ constants/
| ├─ authentication.py
| └─ common.py
├─ exceptions/
| ├─ base.py
| └─ unprocessable_entity.py
└─ validations/
└─ authentication.py
カスタム例外のベースを作成します。
from typing import Any
from fastapi import status
class AppException(Exception):
status_code: int = status.HTTP_400_BAD_REQUEST
message: str = "An unexpected error occurred"
def __init__(self, message: str | None = None):
self.message = message or self.message
super().__init__(self.message)
def to_response(self) -> dict[str, Any]:
return {
"message": self.message,
}
バリデーションエラーのメッセージを一元管理するためのEnumを作成し、
AppExceptionを継承してバリデーションエラー用のカスタム例外を作成します。
from enum import Enum
class UnprocessableEntityErrorTypes(str, Enum):
EMAIL_REQUIRED = "email_required"
EMAIL_INVALID_TYPE = "email_invalid_type"
EMAIL_INVALID_FORMAT = "email_invalid_format"
PASSWORD_REQUIRED = "password_required"
PASSWORD_INVALID_TYPE = "password_invalid_type"
PASSWORD_INVALID_FORMAT = "password_invalid_format"
from dataclasses import dataclass
from typing import Any
from fastapi import status
from src.app.constants.authentication import PASSWORD_MIN_LENGTH
from src.app.constants.common import UnprocessableEntityErrorTypes
from .base import AppException
@dataclass(frozen=True)
class UnprocessableEntityErrorDetail:
message: str
def to_dict(self) -> dict[str, str]:
return {
"message": self.message,
}
UNPROCESSABLE_ENTITY_ERROR_CATALOG: dict[UnprocessableEntityErrorTypes, UnprocessableEntityErrorDetail] = {
UnprocessableEntityErrorTypes.EMAIL_REQUIRED: UnprocessableEntityErrorDetail(
message="メールアドレスは必須項目です。",
),
UnprocessableEntityErrorTypes.EMAIL_INVALID_TYPE: UnprocessableEntityErrorDetail(
message="メールアドレスは文字列である必要があります。",
),
UnprocessableEntityErrorTypes.EMAIL_INVALID_FORMAT: UnprocessableEntityErrorDetail(
message="メールアドレスは有効な形式である必要があります。",
),
UnprocessableEntityErrorTypes.PASSWORD_REQUIRED: UnprocessableEntityErrorDetail(
message="パスワードは必須項目です。",
),
UnprocessableEntityErrorTypes.PASSWORD_INVALID_TYPE: UnprocessableEntityErrorDetail(
message="パスワードは文字列である必要があります。",
),
UnprocessableEntityErrorTypes.PASSWORD_INVALID_FORMAT: UnprocessableEntityErrorDetail(
message=f"パスワードは最低{PASSWORD_MIN_LENGTH}文字以上である必要があります。",
),
}
class UnprocessableEntityException(AppException):
status_code = status.HTTP_422_UNPROCESSABLE_ENTITY
message = "Unprocessable entity."
def __init__(self, error_type: UnprocessableEntityErrorTypes) -> None:
self.error_type = error_type
if error_type in UNPROCESSABLE_ENTITY_ERROR_CATALOG:
self.message = UNPROCESSABLE_ENTITY_ERROR_CATALOG[error_type].message
super().__init__(self.message)
def to_response(self) -> dict[str, Any]:
return {
"message": self.message,
}
これらを使用して、実際のバリデーションメソッドを記述します。
PASSWORD_MIN_LENGTH = 8
from src.app.constants.authentication import PASSWORD_MIN_LENGTH
from src.app.constants.common import UnprocessableEntityErrorTypes
from src.app.exceptions.unprocessable_entity import UnprocessableEntityException
def validate_post_register_account_request(values: dict):
email = values.get("email")
password = values.get("password")
if email is None:
raise UnprocessableEntityException(UnprocessableEntityErrorTypes.EMAIL_REQUIRED)
if not isinstance(email, str):
raise UnprocessableEntityException(UnprocessableEntityErrorTypes.EMAIL_INVALID_TYPE)
if "@" not in email:
raise UnprocessableEntityException(UnprocessableEntityErrorTypes.EMAIL_INVALID_FORMAT)
if password is None:
raise UnprocessableEntityException(UnprocessableEntityErrorTypes.PASSWORD_REQUIRED)
if not isinstance(password, str):
raise UnprocessableEntityException(UnprocessableEntityErrorTypes.PASSWORD_INVALID_TYPE)
if len(password) < PASSWORD_MIN_LENGTH:
raise UnprocessableEntityException(UnprocessableEntityErrorTypes.PASSWORD_INVALID_FORMAT)
return values
PostRegisterAccountRequest インスタンス作成前に呼び出されるように設定します
from pydantic import BaseModel, ConfigDict, Field, model_validator
from src.app.validations.authentication import validate_post_register_account_request
class PostRegisterAccountRequest(BaseModel):
...
# クラスの下部に追記
@model_validator(mode="before")
def validate_values(cls, values: dict) -> dict:
return validate_post_register_account_request(values=values)
3.4. 処理層の作成
上記で記述したリクエストを受け取り、レスポンスを返すための処理を記述します。
まずはエンドポイントから直接呼ばれるサービスの作成です。
from sqlmodel.ext.asyncio.session import AsyncSession
from src.app.processors.authentication import PostRegisterAccountProcessor
from src.app.queries.authentication import AuthenticationQuery
from src.app.schemas.requests.authentication import PostRegisterAccountRequest
from src.app.schemas.responses.common import CommonEmptyResponse
class AuthenticationService:
def __init__(self, session: AsyncSession):
self.session = session
self.authentication_query = AuthenticationQuery(session)
async def post_register_account(
self,
request_body: PostRegisterAccountRequest,
) -> CommonEmptyResponse:
processor = PostRegisterAccountProcessor()
account_create_data = processor.build_account_create_data(
email=request_body.email,
password=request_body.password,
)
await self.authentication_query.create_account(
create_data=account_create_data,
)
return CommonEmptyResponse()
サービス内の関数はあくまでハンドラーとして使用するため、単一責任原則を満たすために個別の処理やクエリは別レイヤーを呼び出して実行します。
class PostRegisterAccountProcessor:
def build_account_create_data(self, email: str, password: str) -> dict:
hashed_password = self.hash_password(password)
return {
"email": email,
"hashed_password": hashed_password,
}
def hash_password(self, password: str) -> str:
# HACK: プロジェクトに合ったハッシュ化の関数を作成してください
return "hashed_" + password
from typing import Any
from sqlmodel.ext.asyncio.session import AsyncSession
from src.app.models.authentication import Account
class AuthenticationQuery:
def __init__(self, session: AsyncSession) -> None:
self.session = session
async def create_account(self, create_data: dict[str, Any], auto_commit: bool = True) -> Account | None:
account = Account(**create_data)
self.session.add(account)
if auto_commit:
await self.session.commit()
await self.session.refresh(account)
return account
3.5. 処理層とエンドポイントの接続
エンドポイントから処理層を呼び出すために、FastAPIの Depends で依存性を注入します。
from fastapi import Depends
from src.app.dependencies.database import get_async_session
from src.app.services.authentication import AuthenticationService
def get_authentication_service(
session=Depends(get_async_session),
) -> AuthenticationService:
return AuthenticationService(session=session)
from collections.abc import AsyncGenerator
from sqlmodel.ext.asyncio.session import AsyncSession as SQLModelAsyncSession
async def get_async_session() -> AsyncGenerator[SQLModelAsyncSession, None]:
async with SQLModelAsyncSession(async_engine) as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
3.6. エンドポイントの作成
アカウント関連のエンドポイントをまとめるために、main.py にrouterを追記します。
また、バリデーションが起きた際のハンドラーを追記します。
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from src.app.database import create_db_and_tables_async
from src.app.endpoints import authentication
from src.app.exceptions.unprocessable_entity import UnprocessableEntityException
@asynccontextmanager
async def lifespan(app: FastAPI):
await create_db_and_tables_async()
yield
app = FastAPI(
title="My Project",
description="A FastAPI project",
version="0.1.0",
lifespan=lifespan,
)
@app.exception_handler(UnprocessableEntityException)
async def unprocessable_entity_exception_handler(request: Request, exc: UnprocessableEntityException):
return JSONResponse(
status_code=422,
content={"detail": str(exc), "error_type": exc.error_type if hasattr(exc, "error_type") else None},
)
API_V1_PREFIX = "/api/v1"
AUTHENTICATION_PREFIX = API_V1_PREFIX + "/auth"
app.include_router(
router=authentication.router,
prefix=AUTHENTICATION_PREFIX,
tags=["authentication"],
)
これで、アカウント関連のエンドポイントは全て/api/v1/auth/から始まるようになりました。
endpoints/authentication.py を作成し、ここにエンドポイントを記載します。
from fastapi import APIRouter, Depends, status
from src.app.dependencies.services import get_authentication_service
from src.app.schemas.requests.authentication import PostRegisterAccountRequest
from src.app.schemas.responses.common import CommonEmptyResponse
from src.app.services.authentication import AuthenticationService
router = APIRouter()
@router.post(
path="/account",
response_model=CommonEmptyResponse,
status_code=status.HTTP_201_CREATED,
)
async def post_register_account(
request_body: PostRegisterAccountRequest,
service: AuthenticationService = Depends(get_authentication_service),
) -> CommonEmptyResponse:
return await service.post_register_account(
request_body=request_body,
)
これでPOST - /api/v1/auth/accountというエンドポイントの完成です。
3.7. 動作確認
Postmanなどで、http://localhost:8000/api/v1/auth/account に空のリクエストを送信してみます。
意図通り、バリデーションエラーが返っていることが確認できます。
次は正常系の確認をしてみます。
{
"email": "test@emample.com",
"password": "test_password"
}
成功しました。
実際にDBに登録されているか確認してみます。
正常に登録されていることが確認できました。
以上でアカウント登録APIは完成です。
まとめ
本記事では、FastAPIを使ったアカウント登録APIの実装を通じて、保守性の高いバックエンド設計の実践例を紹介しました。
実装したアーキテクチャの特徴
今回採用したレイヤードアーキテクチャは、以下のような責任分離を実現しています。
各層の役割
- endpoints → ルーティングとリクエスト/レスポンスのハンドリング
- services → ビジネスロジックの調整(オーケストレーション)
- processors → データ変換や計算などの具体的な処理
- queries → データベース操作(CRUD)
- models → データ構造の定義
- schemas → API入出力の型定義
- validations → カスタムバリデーションロジック
- exceptions → エラーハンドリングの一元管理
この構成により、以下のメリットが得られます:
- ✅ 可読性の向上: 各ファイルの責任が明確で、コードの所在が特定しやすい
- ✅ テスタビリティ: 各層を独立してテスト可能
- ✅ 保守性: 変更が特定の層に限定され、影響範囲が明確
- ✅ 拡張性: 新機能追加時も既存の構造に沿って実装できる
主要な設計パターン
依存性注入(DI)
FastAPIの Depends により、DBセッションやサービスクラスを自動管理
トランザクションの自動 commit / rollback
テスト時のモック注入が容易
カスタムバリデーション
エラーメッセージの一元管理(Enum + データクラス)
統一されたエラーレスポンス形式
多言語対応への拡張性
種別による整理
機能ごとに同じファイル名で統一(例:authentication.py)
レイヤーと種別の組み合わせでファイルパスが予測可能
チーム開発での認知負荷を低減
今後の拡張について
本記事の構成をベースに、以下のような機能を段階的に追加できます:
認証・認可: JWT、OAuth2、RBAC
DB管理: Alembic、リレーションシップ
テスト: 単体テスト、統合テスト
監視: 構造化ログ、APMツール連携
最後に
本記事で紹介したアーキテクチャは、厳密なクリーンアーキテクチャではなく、実用性と学習コストのバランスを重視した設計です。
中小規模のプロジェクトであれば、この構成で十分な保守性と拡張性を確保できます。
FastAPIの型システムと依存性注入を活用することで、長期的に運用可能なバックエンドシステムを構築できます。
本記事が実践的なAPI開発の参考になれば幸いです。
参考


