【2026年版】FastAPIで作るマイクロサービス実践ガイド：設計パターンからデプロイまで

Tech Trends AI

2026年2月12日 - 4 minutes read - 716 words

はじめに

FastAPIは、Python製のモダンなWebフレームワークとして急速に普及し、2026年現在ではAI/MLサービスのAPIバックエンドとして最も選ばれるフレームワークの一つです。

型ヒントベースの自動バリデーション、非同期処理のネイティブサポート、自動APIドキュメント生成など、開発効率と本番品質を両立する特徴を持っています。

本記事では、FastAPIを使ったマイクロサービスの設計パターンから本番デプロイまでを実践的に解説します。

FastAPIの強み

特徴	説明	メリット
型ヒントベース	Pydanticによる自動バリデーション	バグの早期発見
非同期対応	async/awaitネイティブサポート	高いスループット
自動ドキュメント	OpenAPI/Swagger自動生成	API連携の効率化
高速	Starlette + Uvicorn	Node.js並の性能
依存性注入	DIシステム内蔵	テスタビリティ向上

プロジェクト構成

推奨ディレクトリ構造

my-service/
├── app/
│   ├── __init__.py
│   ├── main.py              # アプリケーションエントリポイント
│   ├── config.py             # 設定管理
│   ├── api/
│   │   ├── __init__.py
│   │   ├── v1/
│   │   │   ├── __init__.py
│   │   │   ├── router.py     # ルーターの集約
│   │   │   ├── endpoints/
│   │   │   │   ├── users.py
│   │   │   │   ├── items.py
│   │   │   │   └── health.py
│   │   │   └── dependencies.py
│   │   └── v2/               # バージョニング
│   ├── models/
│   │   ├── __init__.py
│   │   ├── user.py           # SQLAlchemyモデル
│   │   └── item.py
│   ├── schemas/
│   │   ├── __init__.py
│   │   ├── user.py           # Pydanticスキーマ
│   │   └── item.py
│   ├── services/
│   │   ├── __init__.py
│   │   ├── user_service.py   # ビジネスロジック
│   │   └── item_service.py
│   ├── repositories/
│   │   ├── __init__.py
│   │   └── user_repository.py # データアクセス層
│   └── core/
│       ├── __init__.py
│       ├── database.py       # DB接続設定
│       ├── security.py       # 認証・認可
│       └── exceptions.py     # カスタム例外
├── tests/
│   ├── conftest.py
│   ├── test_users.py
│   └── test_items.py
├── Dockerfile
├── docker-compose.yml
├── pyproject.toml
└── .env

実装：基本的なCRUD API

アプリケーションのエントリポイント

# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.api.v1.router import api_router
from app.core.database import engine, Base

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    await engine.dispose()

app = FastAPI(
    title="My Microservice",
    version="1.0.0",
    lifespan=lifespan,
)

app.include_router(api_router, prefix="/api/v1")

Pydanticスキーマ

# app/schemas/user.py
from pydantic import BaseModel, EmailStr
from datetime import datetime

class UserCreate(BaseModel):
    name: str
    email: EmailStr

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    created_at: datetime

    model_config = {"from_attributes": True}

class UserList(BaseModel):
    users: list[UserResponse]
    total: int

エンドポイント

# app/api/v1/endpoints/users.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.schemas.user import UserCreate, UserResponse, UserList
from app.services.user_service import UserService
from app.api.v1.dependencies import get_db, get_current_user

router = APIRouter(prefix="/users", tags=["users"])

@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(user: UserCreate, db: AsyncSession = Depends(get_db)):
    service = UserService(db)
    return await service.create(user)

@router.get("/", response_model=UserList)
async def list_users(
    skip: int = 0,
    limit: int = 20,
    db: AsyncSession = Depends(get_db),
):
    service = UserService(db)
    users, total = await service.list(skip=skip, limit=limit)
    return UserList(users=users, total=total)

@router.get("/{user_id}", response_model=UserResponse)
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    service = UserService(db)
    user = await service.get(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

サービス層

# app/services/user_service.py
from sqlalchemy.ext.asyncio import AsyncSession
from app.repositories.user_repository import UserRepository
from app.schemas.user import UserCreate

class UserService:
    def __init__(self, db: AsyncSession):
        self.repo = UserRepository(db)

    async def create(self, data: UserCreate):
        return await self.repo.create(data)

    async def get(self, user_id: int):
        return await self.repo.get_by_id(user_id)

    async def list(self, skip: int = 0, limit: int = 20):
        users = await self.repo.get_all(skip=skip, limit=limit)
        total = await self.repo.count()
        return users, total

認証・認可

JWT認証の実装

# app/core/security.py
from datetime import datetime, timedelta
from jose import JWTError, jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from app.config import settings

security = HTTPBearer()

def create_access_token(data: dict, expires_delta: timedelta = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(hours=1))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, settings.SECRET_KEY, algorithm="HS256")

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
):
    try:
        payload = jwt.decode(
            credentials.credentials,
            settings.SECRET_KEY,
            algorithms=["HS256"],
        )
        user_id: int = payload.get("sub")
        if user_id is None:
            raise HTTPException(status_code=401, detail="Invalid token")
        return user_id
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

テスト

テスト設定

# tests/conftest.py
import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app

@pytest.fixture
async def client():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac

エンドポイントテスト

# tests/test_users.py
import pytest

@pytest.mark.anyio
async def test_create_user(client):
    response = await client.post(
        "/api/v1/users/",
        json={"name": "テスト太郎", "email": "test@example.com"},
    )
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "テスト太郎"
    assert data["email"] == "test@example.com"

Docker化とデプロイ

Dockerfile

FROM python:3.12-slim AS builder
WORKDIR /app
COPY pyproject.toml ./
RUN pip install --no-cache-dir .

FROM python:3.12-slim
WORKDIR /app
COPY --from=builder /usr/local/lib/python3.12/site-packages /usr/local/lib/python3.12/site-packages
COPY --from=builder /usr/local/bin /usr/local/bin
COPY app/ app/
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

docker-compose.yml

services:
  api:
    build: .
    ports:
      - "8000:8000"
    env_file:
      - .env
    depends_on:
      db:
        condition: service_healthy

  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: mydb
      POSTGRES_USER: user
      POSTGRES_PASSWORD: password
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U user"]
      interval: 5s
      timeout: 5s
      retries: 5

volumes:
  pgdata:

まとめ

FastAPIによるマイクロサービス開発のポイント：

クリーンアーキテクチャ: エンドポイント→サービス→リポジトリの3層構造
型安全性: Pydanticによる自動バリデーションでバグを防止
非同期処理: async/awaitで高いスループットを実現
テスタビリティ: DIと分離された層構造でテストが容易
本番運用: Docker + Uvicornで信頼性の高いデプロイ

FastAPIは、AI/MLサービスのAPIバックエンドとして最適な選択肢です。

カテゴリー

AI開発ガイド

タグ

FastAPI Python マイクロサービス REST API 非同期処理バックエンド開発