🚀 ニフティ’s Notion

バックエンドの設計にはレイヤードアーキテクチャ(あるいはクリーンアーキテクチャ)を用いることが多くなっています。例を元に考えていきましょう。

シンプルな2層での分割を考えます。

題材は単純なユーザ登録です。

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:
    __db: AsyncSession
    
    def __init__(self, db: AsyncSession):
        self.__db = db
    
    # 全件取得
    async def get_all(self) -> list[User]:
        user_records = await 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 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
       )
       db.add(user_record)
       await db.commit()
       await db.refresh(user_record)
       
       return User(
           id=user_record.id,
           name=user_record.name
       )

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

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

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

作成したUserRepositoryをDIコンテナから呼び出せるようにします。

FastAPIの場合は単に関数を作るだけです。

(SQLAlchemyの初期化処理が入っているので少々複雑)

from collections.abc import AsyncGenerator

from sqlalchemy import exc
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine


engine = crreate_async_engine('{データベースURL}')

# DBセッションを作ってUserRepositoryに渡し、最後にcommit/rollbackする
# この辺はジェネレータ構文を理解する必要があるので、今はあまり気にしなくてOK
async def get_user_repository() -> AsyncGenerator[UserRepository, None]:
    factory = async_sessionmaker(engine)
    async with factory() as session:
        try:
            yield UserRepository(session)
            await session.commit()
        except exc.SqlAlchemyError:
            await session.rollback()
            raise

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

Controllerの実装はフレームワークに依存します。クラスとして定義される場合もありますが、FastAPIを使う場合はただの関数になります。

UserRepositoryはFastAPIのDIコンテナ機能である Depends() から取得します。

まずは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(repository: UserRepository = Depends(get_user_repository)) -> list[UserJson]:
    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, repository: UserRepository = Depends(get_user_repository)) -> UserJson:
    user = await repository.get(user)
    
    return UserJson.model_validate(user)

# 新規作成
@router.post("/users", response_model=UserJson)
async def create_user(request_json: CreateUserRequestJson, repository: UserRepository = Depends(get_user_repository)) -> User:
    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, repository: UserRepository = Depends(get_user_repository)) -> None:
    try:
        await repository.delete(id)
    except KeyError:
        raise HTTPException(status_code=404, detail="user not found")

ポイントは以下です。

• FastAPIのDIコンテナ機能を使って取得した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ビルダー)を使うことも多くなっています。