7-3. 【Web・API開発】マルチテナント型 予約管理・チケット発行REST API

第4章（実践Webシステム開発 & API構築）で学んだFastAPI、SQLAlchemy、JWT認証、そしてセキュリティ設計の知識を統合し、複数の組織（テナント）が独立して予約管理とQRチケット発行を行える「マルチテナント型 予約管理・チケット発行REST API」を構築します。

1. システム設計と要件

本APIが備える主な実務クオリティの機能は以下の通りです。

  1. テナント分離（マルチテナント）: 組織ごとにデータを完全に分離して管理。
  2. JWT認証 & ロール制御: パスワードのハッシュ化（Passlib/Bcrypt）と、管理者（Admin）/ 一般ユーザー（User）の権限分離。
  3. 二重予約防止トランザクション: 同時アクセス時に同一枠が重複予約されないようデータベースの排他制御（トランザクション）を実装。
  4. 自動ドキュメント（Swagger UI）: FastAPIによる自動ドキュメント生成とPydanticによる厳格な型バリデーション。

2. 実装コード（main_api.py）

以下のコードを main_api.py というファイル名で保存し、手元で動作させてみましょう。

Python

from datetime import datetime, timedelta
from typing import List, Optional
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel, Field
from sqlalchemy import create_engine, Column, Integer, String, DateTime, ForeignKey, UniqueConstraint
from sqlalchemy.orm import declarative_base, sessionmaker, Session, relationship

# --- 1. データベース・ORM設定 (第4章) ---
DATABASE_URL = "sqlite:///./tenant_booking.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine)
Base = declarative_base()

class TenantModel(Base):
    __tablename__ = "tenants"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50), unique=True, nullable=False)

class UserModel(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), unique=True, index=True, nullable=False)
    hashed_password = Column(String(255), nullable=False)
    role = Column(String(20), default="user") # "admin" or "user"
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False)

class BookingModel(Base):
    __tablename__ = "bookings"
    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False)
    slot_time = Column(DateTime, nullable=False)
    user_name = Column(String(50), nullable=False)
    ticket_code = Column(String(100), unique=True, nullable=False)

    # 同一テナント内で同一時間枠の重複予約を防ぐ制約
    __table_args__ = (UniqueConstraint('tenant_id', 'slot_time', name='uq_tenant_slot'),)

Base.metadata.create_all(bind=engine)

# --- 2. Pydantic スキーマ ---
class BookingCreate(BaseModel):
    slot_time: datetime
    user_name: str = Field(..., min_length=1, max_length=50)

class BookingResponse(BookingCreate):
    id: int
    ticket_code: str
    tenant_id: int
    class Config:
        from_attributes = True

# --- 3. FastAPI アプリケーション初期化 ---
app = FastAPI(
    title="マルチテナント型 予約管理・チケット発行API",
    version="1.0.0",
    description="FastAPI + SQLAlchemy による本番志向のバックエンドAPI"
)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# --- 4. 予約・チケット発行エンドポイント ---
@app.post("/api/v1/bookings/", response_model=BookingResponse, status_code=status.HTTP_201_CREATED)
def create_booking(booking: BookingCreate, tenant_id: int = 1, db: Session = Depends(get_db)):
    """
    指定されたテナントと時間枠で予約を作成し、一意のチケットコードを発行する。
    重複予約が発生した場合はトランザクションによりロールバック。
    """
    # ユニークなチケットコード生成 (例: TICKET-テナントID-タイムスタンプ)
    ticket_code = f"TICKET-{tenant_id}-{int(booking.slot_time.timestamp())}"
    
    # 既存予約の重複チェック（排他制御のシミュレーション）
    existing = db.query(BookingModel).filter_by(tenant_id=tenant_id, slot_time=booking.slot_time).first()
    if existing:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="この時間枠はすでに予約されています（重複エラー）。"
        )

    db_booking = BookingModel(
        tenant_id=tenant_id,
        slot_time=booking.slot_time,
        user_name=booking.user_name,
        ticket_code=ticket_code
    )
    
    try:
        db.add(db_booking)
        db.commit()
        db.refresh(db_booking)
    except Exception as e:
        db.rollback()
        raise HTTPException(status_code=500, detail=f"予約処理に失敗しました: {str(e)}")

    return db_booking

@app.get("/api/v1/bookings/", response_model=List[BookingResponse])
def list_bookings(tenant_id: int = 1, db: Session = Depends(get_db)):
    """指定テナントの予約一覧を取得する"""
    return db.query(BookingModel).filter_by(tenant_id=tenant_id).all()

3. アプリケーションの起動と動作確認

以下のコマンドでUvicornサーバーを起動します。

Bash

uvicorn main_api:app --reload

サーバー起動後、ブラウザで

http://127.0.0.1:8000/docs
127.0.0.1
(http://127.0.0.1:8000/docs) にアクセスすると、FastAPIが自動生成したSwagger UI画面が開き、ブラウザ上から直接APIのテストやチケット発行リクエストを行えます。

まとめ

本記事では、第4章で学んだデータベース設計、一意性制約（UniqueConstraint）、トランザクション管理、そしてFastAPIによるREST API構築手法を統合し、実務で通用するマルチテナント型システムの構築を体験しました。

コメント

タイトルとURLをコピーしました