はじめに
前回の記事「FastAPI を Docker コンテナで動かしてみた」では、FastAPI アプリケーションを Docker 化して効率的な開発環境を構築しました。
今回は、そのプロジェクトに PostgreSQL データベースを追加し、ユーザー登録機能を実装してみます。SQLAlchemy を使用した ORM と、Pydantic によるバリデーションを組み合わせて、本格的な Web API を構築していきます。
概要
本記事では、以下の内容を解説します：
- PostgreSQL データベースの Docker 化
- SQLAlchemy による ORM の実装
- ユーザー登録 API エンドポイントの作成
- pgAdmin によるデータベース管理
- パスワードハッシュ化によるセキュリティ対策
プロジェクト構成
今回のプロジェクトでは、前回作成した FastAPI アプリをベースに、PostgreSQL データベースとユーザー登録機能の実装を追加しています。
また、アプリケーションの可読性と保守性を高めるために、モデル・スキーマ・サービス・ルーターを適切に分離した構成としました。
fastapi-test/
├── config.py # データベース接続設定
├── docker-compose.yml
├── Dockerfile
├── main.py
├── models/ # SQLAlchemyモデル定義
│ └── user.py
├── routers/ # APIルーター
│ └── users.py
├── schemas/ # Pydanticスキーマ
│ └── users.py
├── services/ # ビジネスロジック（サービス層）
│ └── users.py
├── requirements.txt
└── tests/ # テストコード（※今回は省略）
├── __init__.py
├── test_main.py
└── test_users.py
1. モデル作成とデータベース連携
1-1. 依存モジュールを追加
requirement.txt
psycopg2-binary==2.9.9
sqlalchemy==2.0.23
pydantic-settings==2.0.0
1-2. データベース接続設定
config.py
"""データベース接続設定
最低限のSQLAlchemy設定
"""
from typing import Generator
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
# データベース接続URL
DATABASE_URL = "postgresql://user:password@db:5432/fastapi-test"
# SQLAlchemyエンジンの作成
engine = create_engine(DATABASE_URL)
# セッションファクトリーの作成
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Baseクラス
Base = declarative_base()
def get_db() -> Generator[Session, None, None]:
"""データベースセッションを取得する依存性注入関数"""
db = SessionLocal()
try:
yield db
finally:
db.close()
def create_tables():
"""テーブルを作成する関数"""
# モデルをインポートしてテーブルを作成
from models.user import User
Base.metadata.create_all(bind=engine)
1-3. ユーザーモデルの作成
models/user.py
"""ユーザーモデル
ユーザー情報を管理するSQLAlchemyモデルです。
"""
from sqlalchemy import Column, Integer, String, Index
from config import Base
class User(Base):
"""ユーザーモデルクラス
Attributes:
id: プライマリキー
name: ユーザー名（ユニーク）
email: メールアドレス（ユニーク）
password: パスワード（ハッシュ化済み）
"""
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(50), unique=True, nullable=False)
email = Column(String(100), unique=True, nullable=False)
password = Column(String(255), nullable=False)
# インデックスを定義
__table_args__ = (
Index("idx_users_name", "name"),
Index("idx_users_email", "email"),
)
def __repr__(self):
"""文字列表現を返す
Returns:
str: ユーザーの文字列表現
"""
return f"<User(id={self.id}, name='{self.name}', email='{self.email}')>"
1-4. DB接続処理を追加
main.py
# 既存のimport...
from config import create_tables
# アプリケーション起動時にテーブルを作成
@app.on_event("startup")
async def startup_event():
create_tables()
# 既存の処理...
1-5. DBコンテナを追加
docker-compose.yml
version: "3.8"
services:
fastapi:
build: .
ports:
- "8000:8000"
volumes:
- .:/app
environment:
- PYTHONUNBUFFERED=1
restart: unless-stopped
db:
image: postgres:15
environment:
- POSTGRES_DB=fastapi-test
- POSTGRES_USER=user
- POSTGRES_PASSWORD=password
ports:
- "5432:5432"
volumes:
- postgres_data:/var/lib/postgresql/data
restart: unless-stopped
pgadmin:
image: dpage/pgadmin4:latest
environment:
- PGADMIN_DEFAULT_EMAIL=admin@example.com
- PGADMIN_DEFAULT_PASSWORD=password
ports:
- "8080:80"
depends_on:
- db
restart: unless-stopped
volumes:
postgres_data:
1-6. コンテナを起動
# 一度コンテナを停止&ボリュームを削除し、コンテナを起動
docker compose down -v
docker compose up -d --build
1-7. pgAdminでテーブル確認
-
http://localhost:8080にアクセス -
.ymlに記載したEmail,Passwordでログイン
-
サーバーを追加してDBに接続
-
fastapi-test → スキーマ → public → テーブル で users テーブルが作成されていることを確認
2. ユーザー登録処理の追加
2-1. 依存モジュールを追加
requirements.txt
passlib[bcrypt]>=1.7.4
2-2. スキーマの作成
schemas/users.py
"""ユーザー関連のPydanticスキーマ
リクエスト/レスポンスのバリデーションとシリアライゼーションを行います。
"""
from pydantic import BaseModel, Field
class UserCreate(BaseModel):
"""ユーザー作成用スキーマ
POSTリクエストで受け取るデータの形式を定義します。
Attributes:
name: ユーザー名（必須、3-50文字）
email: メールアドレス（必須、有効なメール形式）
password: パスワード（必須、8文字以上）
"""
name: str = Field(..., min_length=3, max_length=50, description="ユーザー名")
email: str = Field(..., description="メールアドレス")
password: str = Field(..., min_length=8, description="パスワード")
class Config:
"""設定クラス"""
# JSON Schema用のサンプルデータ
json_schema_extra = {
"example": {
"name": "user",
"email": "user@example.com",
"password": "P@ssw0rd",
}
}
class UserResponse(BaseModel):
"""ユーザーレスポンス用スキーマ
APIレスポンスで返すデータの形式を定義します。
セキュリティのためパスワードは含めません。
Attributes:
id: ユーザーID
name: ユーザー名
email: メールアドレス
"""
id: int
name: str
email: str
class Config:
"""設定クラス"""
from_attributes = True # SQLAlchemyモデルからの変換を有効化
# JSON Schema用のサンプルデータ
json_schema_extra = {
"example": {"id": 1, "name": "user", "email": "user@example.com"}
}
2-3. サービスの作成
services/users.py
"""ユーザーサービス
ユーザー関連のビジネスロジックを処理します。
"""
from typing import Optional
from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError
from passlib.context import CryptContext
from models.user import User
from schemas.users import UserCreate
# パスワードハッシュ化用の設定
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
class UserService:
"""ユーザーサービスクラス
データベース操作とビジネスルールの実装を担当。
"""
@staticmethod
def create_user(db: Session, user_data: UserCreate) -> User:
"""ユーザーを作成する
Args:
db: データベースセッション
user_data: ユーザー作成データ
Returns:
User: 作成されたユーザーオブジェクト
Raises:
IntegrityError: ユーザー名またはメールアドレスが重複している場合
"""
# パスワードをハッシュ化
hashed_password = pwd_context.hash(user_data.password)
# SQLAlchemyモデルインスタンスの作成
db_user = User(
name=user_data.name, email=user_data.email, password=hashed_password
)
try:
# データベースに追加
db.add(db_user)
db.commit()
db.refresh(db_user) # 自動生成されたIDなどを取得
return db_user
except IntegrityError:
db.rollback()
raise
@staticmethod
def get_user_by_id(db: Session, user_id: int) -> Optional[User]:
"""IDでユーザーを取得する
Args:
db: データベースセッション
user_id: ユーザーID
Returns:
Optional[User]: ユーザーオブジェクト（存在しない場合はNone）
"""
return db.query(User).filter(User.id == user_id).first()
@staticmethod
def get_user_by_name(db: Session, name: str) -> Optional[User]:
"""ユーザー名でユーザーを取得する
Args:
db: データベースセッション
name: ユーザー名
Returns:
Optional[User]: ユーザーオブジェクト（存在しない場合はNone）
"""
return db.query(User).filter(User.name == name).first()
@staticmethod
def get_user_by_email(db: Session, email: str) -> Optional[User]:
"""メールアドレスでユーザーを取得する
Args:
db: データベースセッション
email: メールアドレス
Returns:
Optional[User]: ユーザーオブジェクト（存在しない場合はNone）
"""
return db.query(User).filter(User.email == email).first()
@staticmethod
def is_name_taken(db: Session, name: str) -> bool:
"""ユーザー名が既に使用されているかチェックする
Args:
db: データベースセッション
name: チェックするユーザー名
Returns:
bool: 使用済みの場合True、利用可能な場合False
"""
return UserService.get_user_by_name(db, name) is not None
@staticmethod
def is_email_taken(db: Session, email: str) -> bool:
"""メールアドレスが既に使用されているかチェックする
Args:
db: データベースセッション
email: チェックするメールアドレス
Returns:
bool: 使用済みの場合True、利用可能な場合False
"""
return UserService.get_user_by_email(db, email) is not None
@staticmethod
def verify_password(plain_password: str, hashed_password: str) -> bool:
"""パスワードを検証する
Args:
plain_password: 平文パスワード
hashed_password: ハッシュ化済みパスワード
Returns:
bool: パスワードが一致する場合True
"""
return pwd_context.verify(plain_password, hashed_password)
2-4. ルーターの作成
routers/users.py
"""ユーザー関連のAPIエンドポイント
PostgreSQLを使用したユーザー管理のためのREST APIエンドポイントです。
ユーザー情報、認証情報などの構造化データを管理します。
"""
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError
from config import get_db
from schemas.users import UserCreate, UserResponse
from services.users import UserService
# ユーザー管理用ルーター
router = APIRouter(
prefix="/api/users",
tags=["users"],
responses={404: {"description": "Not found"}},
)
@router.post(
"/",
response_model=UserResponse,
status_code=status.HTTP_201_CREATED,
summary="ユーザー登録",
description="新しいユーザーを登録します。パスワードは自動的にハッシュ化されます。",
response_description="作成されたユーザー情報",
)
async def create_user(
user_data: UserCreate, # リクエストボディ（Pydanticで自動バリデーション）
db: Session = Depends(get_db), # データベースセッション（依存性注入）
) -> UserResponse:
"""ユーザーを作成する
Args:
user_data: ユーザー作成データ（自動的にバリデーション済み）
db: データベースセッション（依存性注入）
Returns:
UserResponse: 作成されたユーザー情報
Raises:
HTTPException: ユーザー名またはメールアドレスが重複している場合（400）
"""
# 1. ユーザー名の重複チェック
if UserService.is_name_taken(db, user_data.name):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"ユーザー名 '{user_data.name}' は既に使用されています",
)
# 2. メールアドレスの重複チェック
if UserService.is_email_taken(db, user_data.email):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"メールアドレス '{user_data.email}' は既に使用されています",
)
# 3. ユーザー作成
try:
db_user = UserService.create_user(db, user_data)
return UserResponse.model_validate(db_user)
except IntegrityError:
# データベースレベルでの制約違反（念のため）
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="ユーザー名またはメールアドレスが既に使用されています",
)
2-5. ルーターを追加
main.py
# 既存のimport...
from routers import users
# 既存の処理...
# ルーターの登録
app.include_router(users.router)
2-6. コンテナを再起動
docker compose restart
2-7. 動作確認
-
http://localhost:8000/docsにアクセス -
Swagger UIで
POST /api/users/エンドポイントが表示されていることを確認
-
"Try it out" をクリックしてユーザー登録をテスト
{
"email": "user01@example.com",
"name": "user01",
"password": "P@ssw0rd"
}
-
pgAdminでデータ確認
まとめ
今回は、前回作成した FastAPI アプリに PostgreSQL データベースとユーザー登録機能を追加しました。
今回やったこと：
- Docker Compose で PostgreSQL と pgAdmin のコンテナを追加
- SQLAlchemy でユーザーモデルとテーブル定義を実装
- Pydantic によるリクエストバリデーションの導入
- パスワードのハッシュ化によるセキュリティ対策
-
/api/users/エンドポイントで新規ユーザー登録を実装 - pgAdmin でデータの可視化・確認
次回は、今回作成したユーザー情報を活用して、ログイン認証機能を追加し、セッション管理を導入していきます。