🚀 ニフティ’s Notion

バックエンドの設計にはレイヤードアーキテクチャ(あるいはクリーンアーキテクチャ)を用いることが多くなっている。

例を元に考えていく。シンプルな2層での分割を考える。

• 題材は単純なユーザ登録
  • 入力されたユーザの登録・削除・参照を行えるようにする
    Copy

    class User:
        id: str
        name: str

まずはこのシステムで取り扱うデータを定義する。

このとき、以下の点に注意する。

• データは特定の外部要素(DBなど)に依存してはならない
  • レイヤをまたいで使うデータになるので、特定のレイヤに依存した機能を持っていてはならない
  • Pythonの標準言語機能だけで表現できるものにする
    • Plain Old Python Object(=POPO) と呼んだりする
• 可能な限り不変にする
  • 値の更新はバグの元となり、可能な限り避けるべきとされる
  • 新しいデータを作るときは既存オブジェクトを更新するのではなく、新しいオブジェクトを作る

今回の場合、User型を dataclass で作る。

from dataclasses import dataclass

@dataclass(frozen=True):
class User:
    id: str
    name: str

frozen=True を指定すると、値を変更しようとするとエラーを出す。

user = User(
    id="1",
    name="Yamada Taro"
)
user.name = "Yamada Jiro" # FrozenInstanceError

# もし一部の値だけ書き換えたいなら、新しいオブジェクトを作る
user2 = User(
    id=user.id,
    name="New Name"
)
# またはreplace()を使うと、指定フィールドだけ書き換えた新しいオブジェクトを作れる
user2 = replace(user, name="New Name")

また、Userを新規作成する場合はidがないデータモデルが必要になるので、これも作成する。

from dataclasses import dataclass

@dataclass(frozen=True):
class CreateUserRequest:
    name: str

Infrastructure層を作っていく。

データ操作を担当するクラスはRepositoryという命名がなされることが多いので、UserReopsitoryを定義する。

(今回はインターフェースを省略)

SQLAlchemyを前提とするとデータベース用のモデルを作る必要があるので、まず定義する。

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class UserTable(DeclarativeBase):
	"""userテーブルの定義"""
	
	__tablename__ = "users"
	id: Mapped[str] = mapped_column(primary_key=True, autoincrement=True)
	name: Mapped[str] = mapped_column(nullable=False)

そしてこれを使ってDBを操作するクラスを作る。

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

class UserRepository:
    def __init__(self, db: AsyncSession):
        self.db = db
    
    # 全件取得
    async def get_all(self) -> list[User]:
        user_records = await self.db.scalars(select(UserTable))
        return [
            User(id=record.id, name=record.name)
            for record in user_records
        ]

    # 1件取得
    async def get(self, id: str) -> User:
        user_record = await self.db.get(User, id)
        return User(
            id=user_record.id,
            name=user_record.name
        )

    # 新規作成
    async def create(self, request: CreateUserRequest) -> User:
       user_record = UserTable(name=request.name)
       self.db.add(user_record)
       await self.db.commit()
       await self.db.refresh(user_record)
       
       return User(
           id=user_record.id,
           name=user_record.name
       )

    # 削除
    async def delete(self, id: str) -> None:
        user_record = await self.db.get(User, id)
        if not user_record:
            raise KeyError('Not found')
        
        await self.db.delete(user_record)
        await self.db.commit()
        return None

ここでのポイントは以下。

• Repositoryの入出力にはSQLAlchemyのデータモデルは使用しない
  • DB操作はこのクラス内部で完結するので、SQLAlchemy用のORMモデルも外に出してはならない
  • 別途定義した共通モデルに詰め込み直すことになる

あとはPresentation層を定義します。Webシステムの場合はControllerと呼ばれることが多い。

Controllerの実装はフレームワークに依存する。

クラスとして定義される場合もあるが、FastAPIを使う場合はただの関数になる。

まずはJSONデータを取り扱うためのクラスを定義する。

from pydantic import BaseModel, ConfigDict

class UserJson(BaseModel):
    """ユーザのJSONモデル"""
    id: str
    name: str
    
    # 簡単に変換できる機能を有効化する
    # 同じ構造を持っているオブジェクトからmodel_validate()で変換できるようになる
    model_config = ConfigDict(from_attributes=True)


class CreateUserRequestJson(BaseModel):
    """ユーザ作成リクエストのJSONモデル"""
    name: str
    
    # 同上
    model_config = ConfigDict(from_attributes=True)

そしてこれを使ってControllerの処理を書いていく。

from fastapi import APIRouter, Depends, HTTPException, status

router = APIRouter()

# 全件取得
@router.get("/users", response_model=list[UserJson])
async def get_all_users(db: AsyncSession = Depends(get_db)) -> list[UserJson]:
    repository = UserRepository(db)
    users = await repository.get_all()
    return [UserJson.model_validate(user) for user in users]

# 1件取得
@router.get("/users/{id}", response_model=UserJson)
async def get_user(id: str, db: AsyncSession = Depends(get_db)) -> UserJson:
    repository = UserRepository(db)
    user = await repository.get(user)
    return UserJson.model_validate(user)

# 新規作成
@router.post("/users", response_model=UserJson)
async def create_user(request_json: CreateUserRequestJson, db: AsyncSession = Depends(get_db)) -> User:
    repository = UserRepository(db)
    request = CreateUserRequest(name=request_json.name)
    user = await repository.create(request)
    return UserJson.model_validate(user)

# 削除
@router.delete("/users/{id}")
async def delete_user(id: str, db: AsyncSession = Depends(get_db)) -> None:
    try:
        repository = UserRepository(db)
        await repository.delete(id)
    except KeyError:
        raise HTTPException(status_code=404, detail="user not found")

ポイントは以下。

• Repositoryにデータ操作を任せる
  • ControllerでDBを意識するコードを書くことはない
• 外部から入力されたJSONデータ(pydanticモデル)はそのままRepositoryに渡さず、共通のデータモデルに詰め直して渡す
• Repositoryから受け取ったデータは共通データモデルなので、JSONデータ(pydantic)に詰め直してユーザに返す

はい、その通りです。

DBテーブルのモデルをそのまま返すような単純なAPIでは、このような明確なレイヤ分割・責務分離を行う必要はあまりありません。

テスト観点から見ても、クラス単位でのユニットテストを捨てて、結合テストのみで担保する形で問題ない規模です。

これを行うかどうかは、以下の利点に価値を見出すかどうかです。

• DBテーブル=JSONデータではなくなる場合に対応できる
  • DBテーブルには管理情報を含めたいが、ユーザには見せたくないなどのケースです
  • やりとりするデータはUser型で変わらず、RepositoryまたはControllerのいずれかだけで変更が完結します
  • ex) 削除をDBレコードの削除ではなく、削除フラグで行う(論理削除)場合
• ライブラリの仕様変更に引っ張られづらくなる
  • DB用のライブラリ(SQLAlchemy)またはJSON用のライブラリ(pydantic)に大幅な仕様変更が入り、改修が必要になるケースです
  • ライブラリが関わる範囲がレイヤ内に限定されているため、改修範囲を限定できます

はい、その通りです。

ORMとはDBのテーブルをクラスにマッピングして扱うための仕組みですが、ORM用のクラスは特定のORMライブラリに依存しているので、レイヤ外に持ち出すことができません。

結果としてさらに別のクラスに自力でマッピングすることになるので、二重管理になってしまいます。

このため、ORMを利用せず、SQL文を安全に生成するだけのライブラリ(SQLビルダー)を使うことも多くなっています。