クリーンアーキテクチャに入門してバッチシステムをリニューアルした

10ヶ月ほどやってきた案件がS-inを迎えたのでやってきたことを備忘録として残します。
現場のベテランエンジニアに教えてもらいながらクリーンアーキテクチャを導入してバッチシステムをリニューアルしました。

背景

全国の事業者から届くあるデータ(水道の検針データとして話を進めます)を集計して計算するバッチシステムです。Jenkins(AWS EC2)でWorkFlowを管理し、そこでShellとPythonのスクリプトが動作する環境でした。
実装ルールが定まっておらず、似たような実装が点在していたりテストコードが書かれていない箇所が散見されました。
レイヤー分けをして責務を明確にする、そしてクラス間を疎結合にすることでテストコードを容易に書けるようにクリーンアーキテクチャを導入することとなりました。

方針

The Clean Code Blogを参考に依存関係を円の外側から内側に向けます。内側のレイヤーが外側のレイヤーに依存しないようにします。

UsecaseとRepositoryはIFとなっていて、UsecaseはInteractor, RepositoryはInfrastructure(DB, APIなど物理的なもの)にそれぞれ実装が存在します。DIでIFと実装を紐づけます。
疎結合にするために基本的にクラスにしてグローバルで実行するコードはなくします。

こちらはアプリケーションが起動してからの呼び出し順序です。

やってきたことを時系列で

フィージビリティスタディ

私のプロジェクトでは基盤にAWSを採用しており、AWSサービスの中からWorkFlow管理のためのJenkinsに代わるミドルウェアを探しました。StepFunctionsやAirFlowが候補として上がり、触ってみた結果StepFunctions良さそうって話になりました。ステートは基本Lambda(Python)で作る想定です。

ドメイン洗い出し

そもそも水道の検針データから水道料金ってどんな計算式で求めるの？って状態だったので水道局のページを見たり、過去に社内の人が作成していた計算ドリルを解くことからはじめました。
そこで学んだことや業務に詳しい人に話を聞いた中から、ドメインになりそうな言葉を表にまとめていきました。

ここからクラス図にしてDTO,ValueObject,Entityに分けていきました。これは仕様変更だったり、後からやっぱり違ったなとかで何度も繰り返し修正してます。

ユースケース洗い出し

ドメインに定義したDTO,ValueObject,Entityを使って実現したいことを書いていきます。
クリーンアーキテクチャで実装することを想定し、以下のフォーマットで記載するとその後の実装が捗りました。(ユースケース記述を参考にしました。)

• Usecaseを組み合わせて成し遂げたいこと(検針票を作成してファイルに出力する)
  • Usecase単位の処理内容(検針票を作成する)
    • 事前条件
      • 検針データが所定の場所にあること
    • 事後条件
      • 検針票の情報が作成されていること
    • 基本フロー
      1. 検針データを読み込む
      2. .....
  • 検針票の情報をファイルに出力する
    • 事前条件
      .....

スケルトン実装

洗い出したビジネスルールをソースコードに反映していきました。
まずは中身の実装は書かずに、必要なプロパティやメソッドを定義していきます。
書くのが難しいところも後で書き直せばいいやという気持ちで書いていきました。

Domain

dataclassを用いて定義します。
ドメインロジックにはエンタープライズビジネスルールを置きます。
システム都合ではないコアなルールです。

supply_water.py

@dataclass
class SupplyWater:
    id: str
    usage: int

    def calc_cost(self, rate: Rate) -> SupplyWaterCost:
        pass

テスト駆動開発を意識して開発するといいよというアドバイスをもらい、ドメインロジックやユースケースは中身の実装に入る前のこの段階で書けるところはテストコードを書きました。わかりやすいように計算式などもコメントに書いたりしてました。
プロダクトコードの中身がないのでこの段階ではテストが通りませんが、仕様の理解が深まりました。
参考: https://www.waterworks.metro.tokyo.lg.jp/tetsuduki/ryokin/keisan_23.html

test_supply_water.py

import factory

class SupplyWaterFactory(factory.Factory):
    class Meta:
        model = SupplyWater

    id = factory.Sequence(lambda n: f"{n:010}")
    usage = 30

class TestSupplyWater:
    def test_calc_cost_水道使用量が2か月で59m³(self):
        """
        計算式は（基本料金＋従量料金）×1.10（1円未満の端数は、切り捨てます。）
　　　　　　　    （基本料金 1,170円×2か月分 ＋ 従量料金 3,020円+2,857円）×1.10＝9,038円
        """
        sut = SupplyWaterFactory.build(usage=59)
        rate = RateFactory.build()
        assert sut.calc_cost(rate).cost == 9038

テストデータの作成にはfactory_boyを使うと便利です。

Usecase

IFを定義します。PythonはInterfaceがないので抽象クラスで定義します。
ここにはアプリケーションビジネスルールを置きます。システムにしたことによって発生するロジックです。
使用しないRepositoryが初期化されることを防ぐために多数のRepositoryに依存しないような設計にすると良いです。

water_bill_usecase.py

class WaterBillUsecase(ABC):
    @abstractmethod
    def gen_water_bill(self, id: str) -> WaterBill:
        raise NotImplementedError

Repository

Domainの永続化、取得を行うIFを定義します。

user_repository.py

class UserRepository(ABC):
    @abstractmethod
    def find_by_id(self, id: str) -> User:
        raise NotImplementedError

条件のフィルタなどをUsecaseでやるかRepositoryでやるかの判断が難しかったです。
インフラが変わっても同じ処理ができることを前提に、Repositoryに負荷がかかりそうな場面ではUsecase側でフィルタ処理をするといったことも考えて決めていきました。

ここまでで以下のようになりました。
まだまだ曖昧な部分は多いですが、レイヤー分けはされました。

ビジネスルール詳細化

ドメインロジックとユースケースを実装します。
ユースケースの実装はインタラクターですね。
ドメインロジックに書くべきロジックをインタラクターに書いてしまう失敗を何度もやってしまいました。こうすると似たようなロジックがインタラクターに散らばり保守性が下がります。
システム都合ではないコアなルールをドメインロジックにする。そしてシステムにしたことによって発生するロジックをユースケースに実装します。
参考: https://qiita.com/os1ma/items/25725edfe3c2af93d735

Interactor

Usecaseを実装します。
DomainとRepositoryに定義されたクラスを使って実装します。

water_bill_usecase_impl.py

class WaterBillUsecaseImpl(WaterBillUsecase):
    @inject.autoparams()
    def __init__(self, user_respository: UserRepository):
        self.user_respository = user_respository

    def gen_water_bill(self, id: str) -> WaterBill:
        user = self.user_respository.find_by_id(id)
        ...

ドメインとユースケースは詳細が実装されました。
ドメインロジックがユースケースに漏れないようにすることで、境目が綺麗になりました🌸

ビジネスルール外の実装

ここからはDB,API,FWなど技術の詳細が登場します。
これまでの手順はモブで取り組むことも効果的でしたが、ここからは決まっているIFに対して実装を進めるために、担当者を割り振って並列に実装を進める方が効率的だと思います。

Controller

Usecaseを組み合わせてやりたいことを実現します。
外部からの入力をUsecaseに渡して、Usecaseからの戻りを外部に返します。
TheCleanCodeBlogにあるPresenterの役割はControllerに含めています。

water_bill_controller.py

class WaterBillController:
    @inject.autoparams()
    def __init__(self, water_bill_usecase: WaterBillUsecase):
        self.water_bill_usecase = water_bill_usecase

    def gen_water_bill(self, param) -> WaterBill:
        ...
        return self.water_bill_usecase.gen_water_bill(id)

Infrastructure

Repositoryを実装します。技術の詳細が外に漏れないようにします。

RDS

SQLAlchemyのORMを利用しました。
テーブルのスキーマを定義したものとしてModelを作成しますがここで注意が必要です。
Repositoryが返すべきものはこのModelではなくあくまでDomainだということです。
Modelを定義した後、DomainとModelの関係をマッピングします。

user_repository_impl.py

from sqlalchemy.orm import Session, registry

mapper_registry = registry()

@mapper_registry.mapped
@dataclass
class UserModel(User):
    __table__ = Table(
        "UserTable",
        mapper_registry.metadata,
        Column("id",String(10), primary_key=True, key="id"),
        Column("name", Date, key="name"),
        Column("address", Date, key="address"),
    )
 
    id: str
    name: str
    address: str


class UserRepositoryImpl(UserRepository):
    @inject.autoparams()
    def __init__(self, utils: Utils):
        self.engine = utils.create_engine()

    def find_by_id(self, id: str) -> User:
        with Session(self.engine, future=True) as session:
            return session.execute(select(user).where(user.id == id)).one()

SQLAlchemyに用意された型以外の型を利用したい場合は、TypeDecoratorを作りました。
例えば日付型をarrowで扱いたくて作成したりしましたね。

SQLAlchemyは開発スピードが早くて、開発中にもバージョンアップで新しい書き方に変えたりなどしました。
1.3から1.4に移行した時のメモ:https://qiita.com/nishinishidayo/items/10fb5e03129f4765a041
今はもう1.4.27なんですね😂

DynamoDB

PynamoDBを利用しました。DynamoDBをORMのように使用できます。

user_repository_impl.py

from pynamodb.attributes import NumberAttribute, UnicodeAttribute
from pynamodb.models import Model

class UserRepositoryImpl(UserRepository):
    class UserModel(Model, User):

        class Meta:
            table_name = "UserTable"

        id = NumberAttribute(hash_key=True, attr_name="id")
        name = UnicodeAttribute(attr_name="name")
        address = UnicodeAttribute(attr_name="address")
        

    def find_by_id(self, id: str) -> User:
        return self.UserModel.get(id)

こちらもPynamodbに用意された型以外の型を利用したい場合は、カスタム属性で独自に作成する必要がありました。

DI

各レイヤーの依存関係を管理するためにpython-injectライブラリを利用します。(500行ほどなので一度実装を読んでみることをおすすめします。)
このconfigはHandlerでimportします。遅延評価させて、使用したいクラスだけ初期化できるようにします。
シングルトンなのでアプリケーションが落ちるまでインスタンスは使い回されます。

di.py

import inject

def config(binder: inject.Binder):
    from interactors.water_bill_usecase_impl import WaterBillUsecaseImpl
    from usecases.water_bill_usecase import WaterBillUsecase
    binder.bind_to_constructor(WaterBillUsecase, WaterBillUsecaseImpl)
    
    from controllers.water_bill_controller import WaterBillController
    binder.bind_to_constructor(WaterBillController, WaterBillController)
    ...

Handler

Controllerを呼び出します。こちらはLambdaのHandlerの例です。

handler.py

import inject
import di

inject.configure(di.config)

def handler(event, context) -> Any:
    input = event["input"]
    controller = inject.instance(WaterBillController)
    controller.gen_water_bill(input)

余談ですが、Lambdaの場合だとグローバルで生成されたインスタンスはLambdaコンテナが落ちるまで生き続けるので、DIの結果をキャッシュでき、より効率的に扱えます。
コンテナイメージのLambdaを作ってみるとLambdaのライフサイクルについて学べると思います。(難しい・・・)
https://docs.aws.amazon.com/ja_jp/lambda/latest/dg/lambda-runtimes.html

これで完成しました。

感想

単体テストが書きやすい

レイヤーに分けてクラスを疎結合にすることでモックが使いやすくなります。

test_user_usecase.py

import pytest
from tests.user_factory import UserFactory

class TestUserUsecaseImpl:
    @pytest.fixture
    def sut(self, mocker):
        import interactors.user_usecase_impl as usecase
        repository = mocker.MagicMock()
        return usecase.UserUsecaseImpl(repository)
    
    def test_get_user(self, sut):
        sut.user_repository.return_value = UserFactory.build()
        user = sut.get_user()
        assert user == "Taro"

テストコードの見通しが悪いと書く気持ちが失われます。特に他の人が書いたテストコードはプロダクトコード以上に読みづらいですよね。
しかし、単体テストを積み上げることが品質の向上につながると思いますのでとにかくシンプルで書きやすい環境を作ることが大切だと感じました。
カバレッジの高いテストコードがあればリファクタリングもかなりやりやすいですよね。

シンプルなテストコード作成に活用できますので以下のライブラリもおすすめです。
https://docs.pytest.org/en/6.2.x/tmpdir.html
https://docs.pytest.org/en/6.2.x/monkeypatch.html

使用技術の変更に強い

あるステートがLambdaで作られていたのですが、性能試験してみると処理に15分以上かかりタイムアウトになるという問題が発生しました。
そんな場合でも中身のビジネスロジックには手を入れることなくLambda用のHandlerをFargate用のHandlerに変更することで対応可能です。技術の詳細が外側にあり付け替え可能であるというアーキテクチャのメリットを実感しました。

要件の追加・変更に強い

水道料金を試験用に計算して表示するツールとか作れないかな？といった要望がありました。
こんな時でも水道料金を計算するためのビジネスルール(Domain, Usecase)といった部品を利用して
コントロールする部分だけ変更すればいいですね。入力と出力を変えるだけです。
Domain, Usecaseは部品であり、やりたいことを実現するためにControllerで部品を組み合わせるという考え方がしっくりきますね！

タスクを他の人に渡しやすい

普段スクラムで開発をしているのですが、ここはスウォーミングするべきかな？スイッチングコストがかかりそう・・・みたいな話があります。
きちんとレイヤー分けされているのでレイヤーごとに独立して開発できてスイッチングコストはあまりかからないと思っています。実際に今回の開発でも他のメンバーにスウォーミングしてもらった例がありました。

コードが読みやすい、実装の指針になる

どこに何が書いてあるかルールがあるので読みやすいし実装しやすいです。

今後の改善点

コード例にあるように現在はトランザクション管理をRepositoryで行っています。しかし、そうするとRepositoryの実装がファットになってしまう可能性があるのでこのトランザクション管理をUsecaseで行えるようにしたいと考えています。同じ内容の質問もありました。
Repositoryにトランザクション管理のIFを用意してinfrastructureでDBの種類ごとにそれを実装する。(デコレータにでもしよう)それをUsecaseで利用するように実装しようと考えています。

まとめ

確かに最初は多くのクラス、ファイルを作成する必要があり実装量は多かったです。
ただ一度基盤を作ってしまえば、部品を再利用できて実装速度、品質が上がることを実感できました。