【2025年完全版】FastAPI完全マスターガイド：API開発からデプロイまで

🆕 2025年11月最新版！
エラー10選追加、パフォーマンス最適化、本番デプロイガイドを反映。

5分でわかる：FastAPI完全マスターガイド

高速で直感的なAPIを構築したい。そんなあなたのために、FastAPIの全てを実践的に解説します。

なぜこのガイドが必要なのか

本記事で学べること

1. 基礎理解（第1章）：FastAPIとは、環境構築
2. 実装パターン（第2章）：CRUD、Pydantic、非同期
3. 認証・セキュリティ（第3章）：JWT、OAuth2
4. トラブル解決（第4章）：よくあるエラー10選
5. 最適化（第5章）：パフォーマンスチューニング
6. デプロイ（第6章）：本番環境構築

第1章：FastAPI完全理解

1.1 FastAPIとは

FastAPIの特徴:
├── 高速パフォーマンス
│   └── NodeJS・Go並み（3,000+req/s）
│
├── 開発速度向上
│   └── Flask比200-300%向上
│
├── 自動ドキュメント生成
│   └── Swagger UI・ReDoc
│
├── 型安全性
│   └── Pydanticでバグ40%削減
│
└── 非同期サポート
    └── asyncio/await標準対応

FastAPIが選ばれる理由:
✅ 高速・軽量
✅ 直感的な設計
✅ 自動バリデーション
✅ OpenAPI準拠
✅ Python標準型ヒント活用

1.2 10行で動くAPI

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

@app.post("/items/")
async def create_item(item: Item):
    return {"item": item, "total": item.price * 1.1}

# 起動: uvicorn main:app --reload
# ドキュメント: http://localhost:8000/docs

第2章：実践的実装パターン

パターン1：CRUD完全実装

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from typing import List, Optional
from uuid import uuid4

app = FastAPI()

# モデル定義
class Item(BaseModel):
    name: str
    price: float
    description: Optional[str] = None

class ItemDB(Item):
    id: str

# インメモリDB
items_db = {}

# Create
@app.post("/items/", response_model=ItemDB, status_code=201)
async def create_item(item: Item):
    item_id = str(uuid4())
    item_data = ItemDB(id=item_id, **item.dict())
    items_db[item_id] = item_data
    return item_data

# Read (List)
@app.get("/items/", response_model=List[ItemDB])
async def list_items(skip: int = 0, limit: int = 10):
    items = list(items_db.values())
    return items[skip:skip+limit]

# Read (Single)
@app.get("/items/{item_id}", response_model=ItemDB)
async def get_item(item_id: str):
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return items_db[item_id]

# Update
@app.put("/items/{item_id}", response_model=ItemDB)
async def update_item(item_id: str, item: Item):
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    
    updated_item = ItemDB(id=item_id, **item.dict())
    items_db[item_id] = updated_item
    return updated_item

# Delete
@app.delete("/items/{item_id}", status_code=204)
async def delete_item(item_id: str):
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    
    del items_db[item_id]
    return None

パターン2：Pydantic活用

from pydantic import BaseModel, Field, field_validator
from typing import List, Optional
from datetime import datetime

# 基本モデル
class UserBase(BaseModel):
    email: str = Field(..., pattern=r"^[\w\.-]+@[\w\.-]+\.\w+$")
    username: str = Field(..., min_length=3, max_length=50)

# 作成用（パスワード含む）
class UserCreate(UserBase):
    password: str = Field(..., min_length=8)

# 表示用（パスワード除外）
class User(UserBase):
    id: int
    is_active: bool = True
    created_at: datetime

# カスタムバリデータ
class Event(BaseModel):
    title: str
    start_date: datetime
    end_date: datetime
    
    @field_validator('end_date')
    @classmethod
    def validate_dates(cls, v, info):
        start = info.data.get('start_date')
        if start and v <= start:
            raise ValueError('終了日は開始日より後')
        return v

# ネストモデル
class Address(BaseModel):
    street: str
    city: str
    postal_code: str

class Customer(BaseModel):
    name: str
    addresses: List[Address]

パターン3：非同期処理

import asyncio
import httpx
from fastapi import FastAPI

app = FastAPI()

# 並列API呼び出し
@app.get("/parallel/")
async def parallel_requests():
    async with httpx.AsyncClient() as client:
        tasks = [
            client.get("https://api1.example.com/data"),
            client.get("https://api2.example.com/data"),
            client.get("https://api3.example.com/data")
        ]
        responses = await asyncio.gather(*tasks)
    
    return {"results": [r.json() for r in responses]}

# データベース非同期
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"
engine = create_async_engine(DATABASE_URL)
async_session = sessionmaker(engine, class_=AsyncSession)

async def get_db():
    db = async_session()
    try:
        yield db
    finally:
        await db.close()

@app.get("/users/{user_id}")
async def get_user(
    user_id: int,
    db: AsyncSession = Depends(get_db)
):
    result = await db.execute(
        select(User).where(User.id == user_id)
    )
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="Not found")
    return user

第3章：認証・セキュリティ

3.1 JWT認証

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta
from pydantic import BaseModel

app = FastAPI()

# 設定
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"])
security = HTTPBearer()

# モデル
class Token(BaseModel):
    access_token: str
    token_type: str

class User(BaseModel):
    username: str
    email: str

# パスワードハッシュ化
def hash_password(password: str):
    return pwd_context.hash(password)

def verify_password(plain, hashed):
    return pwd_context.verify(plain, hashed)

# トークン生成
def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

# 認証ミドルウェア
async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security)
):
    try:
        payload = jwt.decode(
            credentials.credentials,
            SECRET_KEY,
            algorithms=[ALGORITHM]
        )
        username: str = payload.get("sub")
        if username is None:
            raise HTTPException(status_code=401, detail="Invalid token")
        return User(username=username, email="")
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

# エンドポイント
@app.post("/token", response_model=Token)
async def login(username: str, password: str):
    # ユーザー確認（実際はDB）
    if username != "testuser" or password != "testpass":
        raise HTTPException(status_code=401, detail="Invalid credentials")
    
    token = create_access_token(data={"sub": username})
    return {"access_token": token, "token_type": "bearer"}

@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

第4章：よくあるエラー10選と解決策

エラー1：ValidationError

エラー:
422 Unprocessable Entity
{
  "detail": [
    {
      "loc": ["body", "email"],
      "msg": "field required",
      "type": "value_error.missing"
    }
  ]
}

原因:
必須フィールドが未送信

解決策:
```python
# ❌ 間違い
class User(BaseModel):
    username: str
    email: str  # 必須

# リクエスト: {"username": "test"}  # emailなし → エラー

# ✅ 正しい
class User(BaseModel):
    username: str
    email: Optional[str] = None  # オプションに

# または
# リクエスト: {"username": "test", "email": "test@example.com"}

### エラー2-10のクイックリファレンス

| エラー | 原因 | 解決策 |
|--------|------|--------|
| **2. CORS error** | CORS未設定 | CORSMiddleware追加 |
| **3. Circular import** | 循環インポート | router分離 |
| **4. Event loop closed** | 同期/非同期混在 | async/await統一 |
| **5. Database connection** | 接続管理不適切 | Depends使用 |
| **6. File upload failed** | サイズ・型制限 | UploadFile設定 |
| **7. Token expired** | JWT有効期限切れ | リフレッシュトークン |
| **8. Memory leak** | セッション未クローズ | finally句で閉じる |
| **9. Slow response** | 同期処理 | async化 |
| **10. 413 Request entity** | リクエストサイズ | max_size設定 |

## 第5章：パフォーマンス最適化

### 5.1 最適化テクニック

```python
# テクニック1：キャッシング
from functools import lru_cache

@lru_cache(maxsize=128)
def expensive_computation(param: str):
    # 重い計算
    return result

# テクニック2：非同期バッチ処理
async def batch_process(items: List[Item]):
    tasks = [process_item(item) for item in items]
    results = await asyncio.gather(*tasks)
    return results

# テクニック3：接続プーリング
from sqlalchemy.pool import QueuePool

engine = create_async_engine(
    DATABASE_URL,
    poolclass=QueuePool,
    pool_size=20,
    max_overflow=10
)

# テクニック4：ストリーミングレスポンス
from fastapi.responses import StreamingResponse

async def generate_large_data():
    for i in range(1000000):
        yield f"data{i}\n"

@app.get("/stream")
async def stream():
    return StreamingResponse(generate_large_data())

# テクニック5：バックグラウンドタスク
from fastapi import BackgroundTasks

def send_email(email: str):
    # メール送信処理
    pass

@app.post("/users/")
async def create_user(
    email: str,
    background_tasks: BackgroundTasks
):
    background_tasks.add_task(send_email, email)
    return {"message": "User created"}

5.2 ベストプラクティス

# 1. 依存性注入の活用
from fastapi import Depends

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

@app.get("/items/")
def read_items(db: Session = Depends(get_db)):
    return db.query(Item).all()

# 2. エラーハンドリング
from fastapi.exceptions import RequestValidationError

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    return JSONResponse(
        status_code=400,
        content={"detail": exc.errors()}
    )

# 3. ロギング
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@app.middleware("http")
async def log_requests(request, call_next):
    logger.info(f"Request: {request.method} {request.url}")
    response = await call_next(request)
    logger.info(f"Response: {response.status_code}")
    return response

# 4. レート制限
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

@app.get("/limited")
@limiter.limit("5/minute")
async def limited_endpoint(request: Request):
    return {"message": "OK"}

# 5. ヘルスチェック
@app.get("/health")
async def health_check():
    return {"status": "healthy"}

第6章：本番環境デプロイ

6.1 Docker化

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

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

COPY . .

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:15
    environment:
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=pass
      - POSTGRES_DB=mydb
    volumes:
      - postgres_data:/var/lib/postgresql/data

volumes:
  postgres_data:

6.2 本番設定

# config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    app_name: str = "FastAPI App"
    admin_email: str
    database_url: str
    secret_key: str
    
    class Config:
        env_file = ".env"

settings = Settings()

# main.py
from fastapi import FastAPI
from config import settings

app = FastAPI(title=settings.app_name)

まとめ：FastAPIで高速APIを構築

FastAPIは、高速で直感的なAPI開発に最適なフレームワークです。

✅ 主要ポイント

FastAPIの強み:
1. 高速パフォーマンス
   → 3,000+req/s

2. 開発速度向上
   → Flask比200-300%

3. 自動ドキュメント
   → Swagger UI自動生成

4. 型安全性
   → Pydanticでバグ削減

5. 非同期サポート
   → async/await標準

使い分け:
- REST API → FastAPI
- シンプルWeb → Flask
- フルスタック → Django
- マイクロサービス → FastAPI

今すぐFastAPIで高速APIを構築しよう。

※本記事の情報は2025年11月時点のものです。最新情報はFastAPI公式ドキュメントでご確認ください。