アーキテクチャとは バックエンドとレイヤードアーキテクチャ なぜ層を分けるのか 4層の構成 依存の方向 Domain層の定義 値オブジェクト エンティティ リポジトリインターフェース エラークラス Infrastructure層: Repositoryの実装 Application層: UseCaseの定義 Presentation層: Controllerの定義 例外ハンドラー おまけ 3つも同じようなデータモデル用意するの無駄じゃない？ ORMの意味なくない？
アーキテクチャとは
ソフトウェアアーキテクチャとは「コードをどう構造化するか」の設計方針のこと。
💡
設計入門回でやりましたね？
どのファイルに何を書くか、ファイル同士がどう依存するか、変更が起きたときにどこまで影響するか。
これらを事前に決めておくのがアーキテクチャの役割。
アーキテクチャを決めずに開発を進めると、機能追加のたびに「どこに書けばいいか分からない」「1つ直したら別の場所が壊れた」が起きる。人数が増えると「人によって書く場所がバラバラ」になり、コードの見通しが悪化する。
バックエンドでは以下のようなアーキテクチャパターンがよく使われる。
| パターン | 特徴 |
|---|---|
| レイヤードアーキテクチャ | コードを責務ごとの「層」に分ける。依存の方向を一方向に制御する |
| クリーンアーキテクチャ | レイヤードの発展形。ビジネスロジックを中心に据え、外側（DB、Web）に依存しない |
| ヘキサゴナルアーキテクチャ | ポートとアダプタで外部とのやり取りを抽象化する |
いずれも「関心の分離」と「依存方向の制御」が共通する考え方。
本講義ではレイヤードアーキテクチャ(あるいはクリーンアーキテクチャ)の4層構成を採用して解説する。
バックエンドとレイヤードアーキテクチャ
バックエンドの設計にはレイヤードアーキテクチャ(あるいはクリーンアーキテクチャ)を用いることが多くなっている。
例を元に考えていく。4層での分割を考える。
なぜ層を分けるのか
レイヤ分割をしていない route.py の実装を例に考える。記事削除の処理はこうなっている。
@router.delete("/articles/{id}", status_code=status.HTTP_204_NO_CONTENT, tags=["article"])
async def delete_article(id: int, db: AsyncSession = Depends(get_db)) -> None:
article = await db.get(ArticleTable, id)
if not article:
raise HTTPException(status_code=404, detail="article not found")
await db.delete(article)この関数は1つの中で以下を全て行っている。
- HTTPリクエストのパスパラメータを受け取る（HTTP処理）
- DBから記事を取得する（DB操作）
- 存在しなければ404を返す（ビジネス判断 + HTTPレスポンス生成）
- DBから削除する（DB操作）
シンプルなCRUDであればこれで十分だが、要件が増えると問題が出る。
例えば以下の要件が追加された場合を考える。
- 記事削除時に紐づくコメントも削除する
- 削除ではなくステータスを"deleted"に変更する（論理削除）
- 削除済みの記事にアクセスしたら404ではなく410を返す
- 管理者のみ削除可能にする
これを全て1つの関数に書くとこうなる。
@router.delete("/articles/{id}", status_code=status.HTTP_204_NO_CONTENT, tags=["article"])
async def delete_article(id: int, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)) -> None:
if not current_user.is_admin:
raise HTTPException(status_code=403, detail="forbidden")
article = await db.get(ArticleTable, id)
if not article:
deleted_article = await db.scalars(select(ArticleTable).filter(ArticleTable.id == id, ArticleTable.status == "deleted"))
if deleted_article.first():
raise HTTPException(status_code=410, detail="article already deleted")
raise HTTPException(status_code=404, detail="article not found")
comments = await db.scalars(select(CommentTable).filter(CommentTable.article_id == id))
for comment in comments:
comment.deleted_at = datetime.now()
article.status = "deleted"
article.deleted_at = datetime.now()
await db.flush()何が問題か
- 認可チェック、DB操作、ビジネスルール、HTTPレスポンスが全て混在している
- 「公開中の記事は削除不可」というルールを追加したい場合、この関数を読み解いて適切な位置に挿入する必要がある
- テストするにはHTTPサーバーとDBを両方立てないと動かない
- 別のエンドポイント（管理画面API等）で同じ削除ロジックを使いたい場合、コピペするしかない
層を分けると、それぞれが独立した責務を持つ。
Infrastructure層（DB操作だけ）
class ArticleRepositoryImpl:
async def delete(self, article_id: int) -> None:
record = await self.session.get(ArticleTable, article_id)
record.status = "deleted"
record.deleted_at = datetime.now()
await self.session.flush()
async def find_by_id_including_deleted(self, article_id: int) -> ArticleEntity | None:
record = await self.session.get(ArticleTable, article_id)
if not record:
return None
return ArticleEntity(
id=record.id,
title=record.title,
detail=record.detail,
category=record.category,
img_url=record.img_url,
status=ArticleStatus(record.status),
published_at=record.published_at,
created_at=record.created_at,
updated_at=record.updated_at,
)Application層（ビジネス判断だけ）
class ArticleUseCase:
async def delete_handler(self, article_id: int) -> None:
article = await self.article_repo.find_by_id_including_deleted(article_id)
if not article:
raise NotFoundError("Article not found.")
if article.status == ArticleStatus.DELETED:
raise AlreadyDeletedError("Article already deleted.")
await self.comment_repo.delete_by_article_id(article_id)
await self.article_repo.delete(article_id)Presentation層（HTTP処理だけ）
@router.delete("/articles/{id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_article(id: int, db: AsyncSession = Depends(get_db)) -> None:
repo = ArticleRepositoryImpl(db)
use_case = ArticleUseCase(repo)
await use_case.delete_handler(id)各層が1つのことだけを担当している。
- Repositoryは「DBをどう操作するか」だけを知っている。記事が削除可能かどうかは知らない
- UseCaseは「いつ削除してよいか」だけを判断する。SQLの書き方は知らない
- Controllerは「HTTPリクエストを受け取ってUseCaseに渡す」だけ。ビジネスルールもDB操作も知らない
結果として:
- 「公開ルールが変わった」→ Application層だけ修正
- 「DBをMySQLからPostgreSQLに変える」→ Infrastructure層だけ修正
- 「エラーレスポンスの形式を変える」→ Presentation層だけ修正
- テスト時にDBをモックに差し替えて、ビジネスロジックだけテストできる
4層の構成
| 層 | 責務 | やっていいこと | やってはいけないこと |
|---|---|---|---|
| Presentation | HTTPの入出力変換 | バリデーション、レスポンス整形 | DBアクセス、ビジネス判断 |
| Application | ビジネスフローの調整 | ユースケースの組み立て | SQL発行、HTTPレスポンス生成 |
| Domain | ビジネスルールの定義 | エンティティ定義、インターフェース定義 | 具体的なDB操作、フレームワーク依存 |
| Infrastructure | 技術的な実装 | SQL発行、ORM操作 | ビジネス判断 |
依存の方向
Presentation → Application → Domain ← Infrastructure- 題材はニュース記事のステータス管理
- 記事の作成・公開・非公開を行えるようにする
- 記事作成時のカテゴリは許可リストでバリデーションする
class ArticleEntity: id: int title: str detail: str category: str img_url: str status: ArticleStatus published_at: datetime | None
Domain層の定義
まずはこのシステムで取り扱うデータを定義する。
このとき、以下の点に注意する。
- データは特定の外部要素(DBなど)に依存してはならない
- レイヤをまたいで使うデータになるので、特定のレイヤに依存した機能を持っていてはならない
- Pythonの標準言語機能だけで表現できるものにする
-
Plain Old Python Object(=POPO)と呼んだりする
-
- 可能な限り不変にする
- 値の更新はバグの元となり、可能な限り避けるべきとされる
- 新しいデータを作るときは既存オブジェクトを更新するのではなく、新しいオブジェクトを作る
値オブジェクト
ステータスを文字列リテラル("draft"等)で扱うとタイポに気づけない。Enumで定義することで、無効な値を型レベルで排除できる。
from enum import StrEnum
class ArticleStatus(StrEnum):
"""記事のステータスを表す値オブジェクト."""
DRAFT = "draft"
PUBLISHED = "published"
UNPUBLISHED = "unpublished"
ALLOWED_CATEGORIES = frozenset({"エンタがビタミン", "インターネット", "新着ニュース", "海外セレブ", "テクノロジー", "スポーツ"})エンティティ
今回の場合、ArticleEntity型をdataclassで作る。
frozen=Trueを指定すると、値を変更しようとするとエラーを出す。
from datetime import datetime
from pydantic.dataclasses import dataclass
@dataclass(frozen=True)
class ArticleEntity:
"""記事エンティティ."""
id: int
title: str
detail: str
category: str
img_url: str
status: ArticleStatus
published_at: datetime | None
created_at: datetime
updated_at: datetime
@property
def is_published(self) -> bool:
"""記事が公開中か判定する."""
return self.status == ArticleStatus.PUBLISHEDarticle = ArticleEntity(id=1, title="テスト", ...)
article.status = ArticleStatus.PUBLISHED # FrozenInstanceErrorまた、記事を新規作成する場合はid・status・日時がまだないデータモデルが必要になるので、これも作成する。
@dataclass(frozen=True)
class ArticleEntityForCreate:
"""記事作成用エンティティ."""
title: str
detail: str
category: str
img_url: strリポジトリインターフェース
Infrastructure層が実装すべき操作を、抽象クラスで定義する。
Domain層は「何が必要か」だけを宣言し、「どう実現するか」は知らない。
from abc import ABC, abstractmethod
class ArticleRepositoryInterface(ABC):
@abstractmethod
async def find_by_id(self, article_id: int) -> ArticleEntity | None:
pass
@abstractmethod
async def create(self, entity: ArticleEntityForCreate) -> ArticleEntity:
pass
@abstractmethod
async def update_status(self, article_id: int, status: ArticleStatus) -> ArticleEntity | None:
passエラークラス
ビジネスルール違反を表すエラーもDomain層に定義する。HTTPステータスコードには依存しない。
class ApplicationError(Exception):
"""ビジネスルール違反エラー."""
def __init__(self, message: str) -> None:
self.message = message
super().__init__(message)
class NotFoundError(Exception):
"""リソース未存在エラー."""
def __init__(self, message: str) -> None:
self.message = message
super().__init__(message)Infrastructure層: Repositoryの実装
Infrastructure層を作っていく。
Domain層で定義したインターフェースを、SQLAlchemyを使って具体的に実装する。
SQLAlchemyを前提とするとデータベース用のモデルを作る必要があるので、まず定義する。
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
class ArticleTable(DeclarativeBase):
"""articlesテーブルの定義"""
__tablename__ = "articles"
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
title: Mapped[str] = mapped_column(nullable=False)
detail: Mapped[str] = mapped_column(nullable=False)
category: Mapped[str] = mapped_column(nullable=False)
img_url: Mapped[str] = mapped_column(nullable=False)
status: Mapped[str] = mapped_column(nullable=False, default="draft")
published_at: Mapped[datetime | None] = mapped_column(nullable=True, default=None)
created_at: Mapped[datetime] = mapped_column(nullable=False, default=datetime.now)
updated_at: Mapped[datetime] = mapped_column(nullable=False, default=datetime.now, onupdate=datetime.now)そしてこれを使ってDBを操作するクラスを作る。
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
class ArticleRepositoryImpl(ArticleRepositoryInterface):
def __init__(self, session: AsyncSession) -> None:
self.session = session
async def find_by_id(self, article_id: int) -> ArticleEntity | None:
record = await self.session.get(ArticleTable, article_id)
if not record:
return None
return ArticleEntity(
id=record.id,
title=record.title,
detail=record.detail,
category=record.category,
img_url=record.img_url,
status=ArticleStatus(record.status),
published_at=record.published_at,
created_at=record.created_at,
updated_at=record.updated_at,
)
async def create(self, entity: ArticleEntityForCreate) -> ArticleEntity:
record = ArticleTable(
title=entity.title,
detail=entity.detail,
category=entity.category,
img_url=entity.img_url,
)
self.session.add(record)
await self.session.flush()
return ArticleEntity(
id=record.id,
title=record.title,
detail=record.detail,
category=record.category,
img_url=record.img_url,
status=ArticleStatus(record.status),
published_at=record.published_at,
created_at=record.created_at,
updated_at=record.updated_at,
)
async def update_status(self, article_id: int, status: ArticleStatus) -> ArticleEntity | None:
record = await self.session.get(ArticleTable, article_id)
if not record:
return None
record.status = status.value
if status == ArticleStatus.PUBLISHED:
record.published_at = datetime.now()
await self.session.flush()
return ArticleEntity(
id=record.id,
title=record.title,
detail=record.detail,
category=record.category,
img_url=record.img_url,
status=ArticleStatus(record.status),
published_at=record.published_at,
created_at=record.created_at,
updated_at=record.updated_at,
)ここでのポイントは以下。
- Repositoryの入出力にはSQLAlchemyのデータモデルは使用しない
- DB操作はこのクラス内部で完結するので、SQLAlchemy用のORMモデルも外に出してはならない
- 別途定義したエンティティに詰め込み直すことになる
-
commitではなくflushを使う-
flushはSQL文を発行するが、トランザクションは確定しない(rollback可能) - 最終的な
commitはget_dbが一括で行う
-
なぜリポジトリ内で commit せず、get_db に任せるのか:
- 1つのリクエスト内で複数のリポジトリ操作がある場合（例: 記事削除 + コメント一括削除）、全てが成功してから確定したい
- どこか1つでも失敗したら、それまでの操作を全て取り消す（rollback）必要がある
- リポジトリが個別に
commitすると、途中で失敗した場合に一部だけ確定されてデータが不整合になる
from collections.abc import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
engine = create_async_engine(DATABASE_URL)
async def get_db() -> AsyncGenerator[AsyncSession]:
factory = async_sessionmaker(engine)
async with factory() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raiseApplication層: UseCaseの定義
ビジネスフローを組み立てる層。
Presentation層から受け取ったデータに対して、ビジネスルールに基づく判断を行い、Repositoryを呼び出す。
ここではカテゴリのバリデーション（ビジネスルール）と、公開/非公開の状態遷移ルールを実装する。
class ArticleUseCase:
def __init__(self, article_repo: ArticleRepositoryInterface) -> None:
self.article_repo = article_repo
async def create_handler(self, entity: ArticleEntityForCreate) -> ArticleEntity:
"""記事を作成する."""
if entity.category not in ALLOWED_CATEGORIES:
raise ApplicationError(f"Category '{entity.category}' is not allowed.")
return await self.article_repo.create(entity)
async def publish_handler(self, article_id: int) -> ArticleEntity:
"""記事を公開する."""
article = await self.article_repo.find_by_id(article_id)
if not article:
raise NotFoundError(f"Article {article_id} not found.")
if article.status not in (ArticleStatus.DRAFT, ArticleStatus.UNPUBLISHED):
raise ApplicationError(
f"Cannot publish article with status '{article.status}'. "
"Only draft or unpublished articles can be published."
)
result = await self.article_repo.update_status(article_id, ArticleStatus.PUBLISHED)
return result
async def unpublish_handler(self, article_id: int) -> ArticleEntity:
"""記事を非公開にする."""
article = await self.article_repo.find_by_id(article_id)
if not article:
raise NotFoundError(f"Article {article_id} not found.")
if not article.is_published:
raise ApplicationError(
f"Cannot unpublish article with status '{article.status}'. "
"Only published articles can be unpublished."
)
result = await self.article_repo.update_status(article_id, ArticleStatus.UNPUBLISHED)
return resultなぜカテゴリのバリデーションはPresentation層(Pydantic)ではなくここで行うのか？
→ 「どのカテゴリを許可するか」はビジネス要件であり、HTTPリクエストの形式チェックとは性質が異なるため。
Presentation層: Controllerの定義
あとはPresentation層を定義する。Webシステムの場合はControllerと呼ばれることが多い。
Controllerの実装はフレームワークに依存する。
クラスとして定義される場合もあるが、FastAPIを使う場合はただの関数になる。
まずはJSONデータを取り扱うためのクラスを定義する。
from pydantic import BaseModel, ConfigDict, field_validator
class ArticleJson(BaseModel):
"""記事レスポンスJSON."""
id: int
title: str
detail: str
category: str
img_url: str
status: str
published_at: datetime | None
created_at: datetime
updated_at: datetime
model_config = ConfigDict(from_attributes=True)
TITLE_MAX_LENGTH = 100
class CreateArticleRequestJson(BaseModel):
"""記事作成リクエストJSON."""
title: str
detail: str
category: str
img_url: str
@field_validator("title")
@classmethod
def title_must_not_be_empty(cls, v: str) -> str:
if not v.strip():
raise ValueError("Title must not be empty")
if len(v) > TITLE_MAX_LENGTH:
raise ValueError(f"Title must be {TITLE_MAX_LENGTH} characters or less")
return v入力形式のバリデーション（空文字チェック、文字数上限）はPresentation層で行う。
これは「入力の形が正しいか」のチェックであり、ビジネスルールとは無関係なため。
そしてこれを使ってControllerの処理を書いていく。
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
router = APIRouter()
@router.post("/articles", response_model=ArticleJson, status_code=status.HTTP_201_CREATED, tags=["article"])
async def create_article(
request: CreateArticleRequestJson, db: AsyncSession = Depends(get_db)
) -> ArticleJson:
repo = ArticleRepositoryImpl(db)
use_case = ArticleUseCase(repo)
entity = ArticleEntityForCreate(
title=request.title,
detail=request.detail,
category=request.category,
img_url=request.img_url,
)
article = await use_case.create_handler(entity)
return ArticleJson.model_validate(article)
@router.post("/articles/{article_id}/publish", response_model=ArticleJson, tags=["article"])
async def publish_article(article_id: int, db: AsyncSession = Depends(get_db)) -> ArticleJson:
repo = ArticleRepositoryImpl(db)
use_case = ArticleUseCase(repo)
article = await use_case.publish_handler(article_id)
return ArticleJson.model_validate(article)
@router.post("/articles/{article_id}/unpublish", response_model=ArticleJson, tags=["article"])
async def unpublish_article(article_id: int, db: AsyncSession = Depends(get_db)) -> ArticleJson:
repo = ArticleRepositoryImpl(db)
use_case = ArticleUseCase(repo)
article = await use_case.unpublish_handler(article_id)
return ArticleJson.model_validate(article)ポイントは以下。
- UseCaseにビジネスフローを任せる
- ControllerでDBを意識するコードを書くことはない
- 外部から入力されたJSONデータ(pydanticモデル)はそのままRepositoryに渡さず、エンティティに詰め直して渡す
- UseCaseから受け取ったエンティティは、JSONデータ(pydantic)に詰め直してユーザに返す
例外ハンドラー
Domain/Application層で定義したエラーは、HTTPのことを知らない。
これをHTTPステータスコードに変換するのはPresentation層の仕事。
from fastapi import Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
@app.exception_handler(NotFoundError)
async def not_found_handler(_request: Request, exc: NotFoundError) -> JSONResponse:
return JSONResponse(status_code=404, content={"message": exc.message})
@app.exception_handler(ApplicationError)
async def application_error_handler(_request: Request, exc: ApplicationError) -> JSONResponse:
return JSONResponse(status_code=422, content={"message": exc.message})
@app.exception_handler(RequestValidationError)
async def request_validation_handler(_request: Request, exc: RequestValidationError) -> JSONResponse:
errors = [{k: str(v) if k == "ctx" else v for k, v in err.items()} for err in exc.errors()]
return JSONResponse(status_code=400, content={"message": errors})おまけ
3つも同じようなデータモデル用意するの無駄じゃない？
はい、その通りです。
DBテーブルのモデルをそのまま返すような単純なAPIでは、このような明確なレイヤ分割・責務分離を行う必要はあまりありません。
テスト観点から見ても、クラス単位でのユニットテストを捨てて、結合テストのみで担保する形で問題ない規模です。
これを行うかどうかは、以下の利点に価値を見出すかどうかです。
- DBテーブル=JSONデータではなくなる場合に対応できる
- DBテーブルには管理情報を含めたいが、ユーザには見せたくないなどのケースです
- やりとりするデータはUser型で変わらず、RepositoryまたはControllerのいずれかだけで変更が完結します
- ex) 削除をDBレコードの削除ではなく、削除フラグで行う(論理削除)場合
- ライブラリの仕様変更に引っ張られづらくなる
- DB用のライブラリ(SQLAlchemy)またはJSON用のライブラリ(pydantic)に大幅な仕様変更が入り、改修が必要になるケースです
- ライブラリが関わる範囲がレイヤ内に限定されているため、改修範囲を限定できます
ORMの意味なくない？
はい、その通り。
ORMとはDBのテーブルをクラスにマッピングして扱うための仕組みだが、ORM用のクラスは特定のORMライブラリに依存しているので、レイヤ外に持ち出すことができない。
結果としてさらに別のクラスに自力でマッピングすることになるので、二重管理になってしまう。
このため、ORMを利用せず、SQL文を安全に生成するだけのライブラリ(SQLビルダー)を使うことも多くなっている。