1. FastAPI のクリーンなアーキテクチャ
Clean Architecture (Uncle Bob) はアプリケーションを独立したレイヤーに分離し、テスト、保守、拡張を容易にします。
┌───────────────────────────────────────────────┐
│ Presentation Layer │
│ (FastAPI Routes, Schemas, Dependencies) │
├───────────────────────────────────────────────┤
│ Application Layer │
│ (Use Cases, Services, DTOs) │
├───────────────────────────────────────────────┤
│ Domain Layer │
│ (Entities, Value Objects, Interfaces) │
├───────────────────────────────────────────────┤
│ Infrastructure Layer │
│ (Database, External APIs, File System, Cache) │
└───────────────────────────────────────────────┘
Rule: Dependencies point INWARD only
Outer layers depend on inner layers, never the reverse
2. モジュール型プロジェクト構造
app/
├── __init__.py
├── main.py # FastAPI app factory
├── config.py # Pydantic settings
│
├── core/ # Shared infrastructure
│ ├── __init__.py
│ ├── database.py # SQLAlchemy engine, session
│ ├── security.py # JWT, password hashing
│ ├── cache.py # Redis cache
│ ├── exceptions.py # Base exceptions
│ └── dependencies.py # Shared dependencies
│
├── modules/ # Feature modules
│ ├── __init__.py
│ │
│ ├── users/ # User module
│ │ ├── __init__.py
│ │ ├── router.py # API routes
│ │ ├── schemas.py # Pydantic schemas
│ │ ├── models.py # SQLAlchemy models
│ │ ├── repository.py # Data access
│ │ ├── service.py # Business logic
│ │ ├── dependencies.py # Module-specific deps
│ │ └── exceptions.py # Module exceptions
│ │
│ ├── posts/ # Post module
│ │ ├── __init__.py
│ │ ├── router.py
│ │ ├── schemas.py
│ │ ├── models.py
│ │ ├── repository.py
│ │ └── service.py
│ │
│ └── auth/ # Auth module
│ ├── __init__.py
│ ├── router.py
│ ├── schemas.py
│ ├── service.py
│ └── dependencies.py
│
├── middleware/ # Custom middleware
│ ├── __init__.py
│ ├── logging.py
│ └── security.py
│
└── utils/ # Shared utilities
├── __init__.py
├── pagination.py
└── validators.py
3. アプリケーション ファクトリ パターン
# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.config import settings
def create_app() -> FastAPI:
"""Application factory - tạo và cấu hình FastAPI app."""
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup
from app.core.database import engine, Base
from app.core.cache import init_cache
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
await init_cache()
yield
# Shutdown
await engine.dispose()
app = FastAPI(
title=settings.app_name,
version=settings.app_version,
lifespan=lifespan,
docs_url="/docs" if settings.debug else None,
redoc_url="/redoc" if settings.debug else None,
)
# Middleware
_setup_middleware(app)
# Routes
_setup_routes(app)
# Exception handlers
_setup_exception_handlers(app)
return app
def _setup_middleware(app: FastAPI) -> None:
from app.middleware.logging import LoggingMiddleware
from app.middleware.security import SecurityHeadersMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.add_middleware(SecurityHeadersMiddleware)
app.add_middleware(LoggingMiddleware)
def _setup_routes(app: FastAPI) -> None:
from app.modules.auth.router import router as auth_router
from app.modules.users.router import router as users_router
from app.modules.posts.router import router as posts_router
app.include_router(auth_router, prefix="/api/v1")
app.include_router(users_router, prefix="/api/v1")
app.include_router(posts_router, prefix="/api/v1")
def _setup_exception_handlers(app: FastAPI) -> None:
from app.core.exceptions import AppException, app_exception_handler
app.add_exception_handler(AppException, app_exception_handler)
# Create app instance
app = create_app()
4. モジュールパターン
# app/modules/users/models.py
from datetime import datetime
from sqlalchemy import String, Boolean
from sqlalchemy.orm import Mapped, mapped_column
from app.core.database import Base
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100))
email: Mapped[str] = mapped_column(String(255), unique=True)
hashed_password: Mapped[str] = mapped_column(String(255))
is_active: Mapped[bool] = mapped_column(Boolean, default=True)
created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
# app/modules/users/schemas.py
from pydantic import BaseModel, Field
from datetime import datetime
class UserCreate(BaseModel):
name: str = Field(..., min_length=2, max_length=100)
email: str
password: str = Field(..., min_length=8)
class UserUpdate(BaseModel):
name: str | None = None
email: str | None = None
class UserResponse(BaseModel):
id: int
name: str
email: str
is_active: bool
created_at: datetime
model_config = {"from_attributes": True}
# app/modules/users/repository.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from .models import User
class UserRepository:
def __init__(self, session: AsyncSession):
self.session = session
async def get_by_id(self, id: int) -> User | None:
result = await self.session.execute(select(User).where(User.id == id))
return result.scalar_one_or_none()
async def get_by_email(self, email: str) -> User | None:
result = await self.session.execute(select(User).where(User.email == email))
return result.scalar_one_or_none()
async def create(self, user: User) -> User:
self.session.add(user)
await self.session.flush()
await self.session.refresh(user)
return user
# app/modules/users/service.py
from fastapi import HTTPException, status
from app.core.security import hash_password
from .models import User
from .repository import UserRepository
from .schemas import UserCreate
class UserService:
def __init__(self, repo: UserRepository):
self.repo = repo
async def create_user(self, data: UserCreate) -> User:
if await self.repo.get_by_email(data.email):
raise HTTPException(status.HTTP_409_CONFLICT, "Email taken")
user = User(
name=data.name,
email=data.email,
hashed_password=hash_password(data.password),
)
return await self.repo.create(user)
async def get_user(self, user_id: int) -> User:
user = await self.repo.get_by_id(user_id)
if not user:
raise HTTPException(status.HTTP_404_NOT_FOUND, "User not found")
return user
# app/modules/users/dependencies.py
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
from .repository import UserRepository
from .service import UserService
def get_user_service(session: AsyncSession = Depends(get_db)) -> UserService:
repo = UserRepository(session)
return UserService(repo)
# app/modules/users/router.py
from fastapi import APIRouter, Depends, status
from .dependencies import get_user_service
from .schemas import UserCreate, UserResponse
from .service import UserService
router = APIRouter(prefix="/users", tags=["Users"])
@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
data: UserCreate,
service: UserService = Depends(get_user_service),
):
return await service.create_user(data)
@router.get("/{user_id}", response_model=UserResponse)
async def get_user(
user_id: int,
service: UserService = Depends(get_user_service),
):
return await service.get_user(user_id)
5. API のバージョン管理
# URL-based versioning (recommended)
app.include_router(users_v1_router, prefix="/api/v1")
app.include_router(users_v2_router, prefix="/api/v2")
# Header-based versioning
from fastapi import Header
@router.get("/users/")
async def get_users(api_version: str = Header("v1", alias="X-API-Version")):
if api_version == "v2":
return get_users_v2()
return get_users_v1()
概要
この記事では、以下を構築しました。
- クリーンなアーキテクチャ: 4つの明確に分離されたレイヤー
- モジュール構造: 機能ベースのモジュール
- アプリファクトリー: 柔軟でテストしやすいアプリを作成する
- モジュールパターン: ルーター → サービス → リポジトリ → モデル
- API のバージョニング: URL ベースのバージョニング
次のレッスンはテストです。