FastAPI入門｜Pythonで高速WebAPIを構築する実践チュートリアル

WebAPIの構築は、現代のシステム開発において必須のスキルです。FastAPIは、Pythonで高速かつ型安全なWebAPIを構築するためのモダンなフレームワークで、自動ドキュメント生成や非同期処理のサポートなど、開発効率を大幅に向上させる機能を備えています。

本記事では、FastAPIの環境構築からCRUD APIの実装、データベース連携、デプロイまでを実践的なチュートリアル形式で解説します。Flask等のフレームワーク経験がなくても理解できるよう、基本から丁寧に進めていきます。

FastAPIとは？特徴と他フレームワークとの比較

FastAPIは、2018年にSebastian Ramirezによって開発されたPython製のWebフレームワークです。Python 3.7以降の型ヒントを活用し、高速なAPI開発を実現します。

FastAPIの主な特徴

高速：Starlette（ASGI）とPydanticをベースとし、NodeJSやGoに匹敵するパフォーマンス
自動ドキュメント生成：Swagger UIとReDocによるAPIドキュメントが自動生成される
型安全：Pydanticモデルによる自動バリデーションと型チェック
非同期処理：async/awaitによる非同期処理をネイティブサポート
標準準拠：OpenAPI（旧Swagger）とJSON Schemaに完全準拠

Flask・Djangoとの比較

項目	FastAPI	Flask	Django
パフォーマンス	非常に高速	中程度	中程度
型チェック	ネイティブ対応	拡張が必要	一部対応
自動ドキュメント	標準搭載	拡張が必要	拡張が必要
非同期処理	ネイティブ対応	限定的	3.1以降対応
学習コスト	低い	低い	中〜高
適した規模	API中心の開発	小〜中規模	大規模Web

環境構築と最初のAPI

プロジェクトのセットアップ

# プロジェクトディレクトリの作成
mkdir fastapi-tutorial
cd fastapi-tutorial

# 仮想環境の作成と有効化
python3 -m venv .venv
source .venv/bin/activate

# 必要なパッケージのインストール
pip install fastapi uvicorn[standard] pydantic

Hello World API

main.pyを作成します。

from fastapi import FastAPI

app = FastAPI(title="My API", version="1.0.0")


@app.get("/")
def read_root():
    return {"message": "Hello, FastAPI!"}


@app.get("/health")
def health_check():
    return {"status": "ok"}

開発サーバーの起動

# 開発サーバーの起動（ホットリロード有効）
uvicorn main:app --reload --host 0.0.0.0 --port 8000

ブラウザで以下のURLにアクセスして動作を確認します。

http://localhost:8000 - APIレスポンス
http://localhost:8000/docs - Swagger UI（自動生成ドキュメント）
http://localhost:8000/redoc - ReDoc（別形式のドキュメント）

パスパラメータとクエリパラメータ

APIのエンドポイント設計において、パスパラメータとクエリパラメータの使い分けは重要です。

パスパラメータ

@app.get("/users/{user_id}")
def get_user(user_id: int):
    """ユーザーIDを指定してユーザー情報を取得する"""
    return {"user_id": user_id, "name": f"ユーザー{user_id}"}


@app.get("/items/{item_id}")
def get_item(item_id: int):
    """アイテムIDを指定してアイテム情報を取得する"""
    return {"item_id": item_id}

パスパラメータに型を指定するだけで、FastAPIが自動的に型変換とバリデーションを行います。/users/abcのような不正なリクエストには、自動的に422エラーが返されます。

クエリパラメータ

from typing import Optional

@app.get("/items")
def list_items(
    skip: int = 0,
    limit: int = 10,
    category: Optional[str] = None,
    sort_by: str = "created_at",
):
    """アイテム一覧を取得する（ページネーション対応）"""
    return {
        "skip": skip,
        "limit": limit,
        "category": category,
        "sort_by": sort_by,
    }

# リクエスト例: GET /items?skip=0&limit=20&category=electronics

Pydanticモデルによるリクエスト・レスポンスの定義

FastAPIの強みの一つが、Pydanticモデルを使ったデータのバリデーションとシリアライゼーションです。

リクエストボディの定義

from pydantic import BaseModel, Field, EmailStr
from typing import Optional
from datetime import datetime


class UserCreate(BaseModel):
    """ユーザー作成リクエスト"""
    name: str = Field(..., min_length=1, max_length=100, examples=["田中太郎"])
    email: EmailStr = Field(..., examples=["tanaka@example.com"])
    age: Optional[int] = Field(None, ge=0, le=150)
    department: str = Field(..., min_length=1)


class UserResponse(BaseModel):
    """ユーザーレスポンス"""
    id: int
    name: str
    email: str
    age: Optional[int]
    department: str
    created_at: datetime


class UserUpdate(BaseModel):
    """ユーザー更新リクエスト（部分更新）"""
    name: Optional[str] = Field(None, min_length=1, max_length=100)
    email: Optional[EmailStr] = None
    age: Optional[int] = Field(None, ge=0, le=150)
    department: Optional[str] = None

モデルを使ったエンドポイント

@app.post("/users", response_model=UserResponse, status_code=201)
def create_user(user: UserCreate):
    """新しいユーザーを作成する"""
    new_user = {
        "id": 1,
        "name": user.name,
        "email": user.email,
        "age": user.age,
        "department": user.department,
        "created_at": datetime.now(),
    }
    return new_user

Pydanticモデルを定義するだけで、以下が自動的に実現されます。

リクエストボディのバリデーション（必須チェック、型チェック、値の範囲チェック）
不正なリクエストに対する詳細なエラーメッセージの返却
Swagger UIのドキュメントへのスキーマ反映

CRUD APIの実装

ここでは、インメモリのデータストアを使ったCRUD APIの全体像を実装します。

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime

app = FastAPI(title="タスク管理API", version="1.0.0")

# インメモリのデータストア
tasks_db: dict[int, dict] = {}
task_counter = 0


class TaskCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    description: Optional[str] = None
    priority: int = Field(default=1, ge=1, le=5)


class TaskUpdate(BaseModel):
    title: Optional[str] = Field(None, min_length=1, max_length=200)
    description: Optional[str] = None
    priority: Optional[int] = Field(None, ge=1, le=5)
    done: Optional[bool] = None


class TaskResponse(BaseModel):
    id: int
    title: str
    description: Optional[str]
    priority: int
    done: bool
    created_at: datetime
    updated_at: datetime


# CREATE
@app.post("/tasks", response_model=TaskResponse, status_code=201)
def create_task(task: TaskCreate):
    global task_counter
    task_counter += 1
    now = datetime.now()
    new_task = {
        "id": task_counter,
        "title": task.title,
        "description": task.description,
        "priority": task.priority,
        "done": False,
        "created_at": now,
        "updated_at": now,
    }
    tasks_db[task_counter] = new_task
    return new_task


# READ（一覧）
@app.get("/tasks", response_model=list[TaskResponse])
def list_tasks(done: Optional[bool] = None, skip: int = 0, limit: int = 20):
    tasks = list(tasks_db.values())
    if done is not None:
        tasks = [t for t in tasks if t["done"] == done]
    return tasks[skip : skip + limit]


# READ（単一）
@app.get("/tasks/{task_id}", response_model=TaskResponse)
def get_task(task_id: int):
    if task_id not in tasks_db:
        raise HTTPException(status_code=404, detail="タスクが見つかりません")
    return tasks_db[task_id]


# UPDATE
@app.patch("/tasks/{task_id}", response_model=TaskResponse)
def update_task(task_id: int, task_update: TaskUpdate):
    if task_id not in tasks_db:
        raise HTTPException(status_code=404, detail="タスクが見つかりません")

    task = tasks_db[task_id]
    update_data = task_update.model_dump(exclude_unset=True)
    for key, value in update_data.items():
        task[key] = value
    task["updated_at"] = datetime.now()

    return task


# DELETE
@app.delete("/tasks/{task_id}", status_code=204)
def delete_task(task_id: int):
    if task_id not in tasks_db:
        raise HTTPException(status_code=404, detail="タスクが見つかりません")
    del tasks_db[task_id]

データベース連携（SQLAlchemy）

本番環境ではインメモリではなくデータベースを使用します。ここではSQLAlchemyを使ったSQLite連携を紹介します。

# 追加パッケージのインストール
pip install sqlalchemy

データベース設定（database.py）

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

DATABASE_URL = "sqlite:///./tasks.db"

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


class Base(DeclarativeBase):
    pass


def get_db():
    """データベースセッションの依存性注入"""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

モデル定義（models.py）

from sqlalchemy import Column, Integer, String, Boolean, DateTime
from datetime import datetime
from database import Base


class Task(Base):
    __tablename__ = "tasks"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(200), nullable=False)
    description = Column(String(1000), nullable=True)
    priority = Column(Integer, default=1)
    done = Column(Boolean, default=False)
    created_at = Column(DateTime, default=datetime.now)
    updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)

エンドポイントの修正（main.py）

from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy.orm import Session
from database import engine, get_db, Base
from models import Task

# テーブルの作成
Base.metadata.create_all(bind=engine)

app = FastAPI(title="タスク管理API", version="2.0.0")


@app.post("/tasks", response_model=TaskResponse, status_code=201)
def create_task(task: TaskCreate, db: Session = Depends(get_db)):
    db_task = Task(
        title=task.title,
        description=task.description,
        priority=task.priority,
    )
    db.add(db_task)
    db.commit()
    db.refresh(db_task)
    return db_task


@app.get("/tasks", response_model=list[TaskResponse])
def list_tasks(
    done: Optional[bool] = None,
    skip: int = 0,
    limit: int = 20,
    db: Session = Depends(get_db),
):
    query = db.query(Task)
    if done is not None:
        query = query.filter(Task.done == done)
    return query.offset(skip).limit(limit).all()

エラーハンドリングとミドルウェア

カスタム例外ハンドラ

from fastapi import Request
from fastapi.responses import JSONResponse


class AppException(Exception):
    def __init__(self, status_code: int, detail: str):
        self.status_code = status_code
        self.detail = detail


@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": exc.detail, "path": str(request.url)},
    )

CORSミドルウェア

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # フロントエンドのオリジン
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

ログミドルウェア

import time
import logging

logger = logging.getLogger(__name__)


@app.middleware("http")
async def log_requests(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    duration = time.time() - start_time
    logger.info(f"{request.method} {request.url.path} - {response.status_code} ({duration:.3f}s)")
    return response

デプロイとプロジェクト構成

推奨プロジェクト構成

fastapi-project/
├── app/
│   ├── __init__.py
│   ├── main.py           # FastAPIアプリケーション
│   ├── database.py       # DB設定
│   ├── models.py         # SQLAlchemyモデル
│   ├── schemas.py        # Pydanticスキーマ
│   ├── routers/
│   │   ├── __init__.py
│   │   ├── tasks.py      # タスク関連エンドポイント
│   │   └── users.py      # ユーザー関連エンドポイント
│   └── dependencies.py   # 共通の依存性
├── tests/
│   ├── __init__.py
│   └── test_tasks.py
├── requirements.txt
├── Dockerfile
└── docker-compose.yml

Dockerを使ったデプロイ

# Dockerfile
FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY ./app ./app

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml
version: "3.8"
services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://user:pass@db:5432/mydb
    depends_on:
      - db
  db:
    image: postgres:16
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: mydb
    volumes:
      - postgres_data:/var/lib/postgresql/data

volumes:
  postgres_data: