Pythonでクリーンアーキテクチャ実装
クリーンアーキテクチャとは
端的に言うと、アプリケーションを作成する際に自分たちの(アプリケーション作成側)のロジックを中心におく事を目的としたアーキテクチャのことです。
ではここでいうロジックとは何でしょうか？
ロジックとは...
ロジックについて話す前に用語を説明しましょう。
| 用語 | 説明 | 層 |
|---|---|---|
| Controller | requestを受け取りUseCaseへ渡すところ。requestからユーザーの入力情報を展開してInputDtoという形式でUseCase層へデータを渡す役割 | Presentation層 Infrastructure層から(View側から)の要望に応える領域 |
| Presenter | UseCase層から渡されたOutputDtoを処理し、ViewModelという形式に変換しView側へ渡す処理をする | Presentation層 |
| ViewModel | View側がどのような形式で受け取るのかなど、View側に合わせた形式 | Presentation層 |
| InputDto(InputData) | Controller -> UseCaseへデータを渡す際のデータ形式 | Application層 ビジネスのロジック(後で説明)を記述する領域。このアプリではこういったリクエストがあってそれをどう処理してどう返すかを決定する。Entityなどアプリで使用されるオブジェクトを通してリクエストに応える |
| InputBoundary | UseCaseのインターフェース(Controllerがこのインターフェースを使用) | Application層 |
| OutputBoundary | Presenterのインターフェース(UseCaseがpresenterのインターフェースを使用) | Application層 |
| Repository(DataAccessInterface) | DataAccessが実装しているインターフェース(UseCaseがこのインターフェースを使用している) | Application層 |
| Entity | ロジック(後で説明)の中で、Application層が使用する形式。これを中心として処理を書いて行く | Domain層 アプリケーションで使用するオブジェクト(Enitity)を定義する層 |
| DataAccess | データベースへアクセスし生のデータを、EntityなどApplication層が扱いやすいオブジェクトへ変換してUse Caseへ渡す役割。Repositoryを実装 | Infrastructure層 外部とのやりとりをする際に最初に通過する領域 |
| DataBase | DBを使用するところ。sqlを書いたり、ORMを使用してデータを扱うような役割 | Infrastructure層 |
| View | いわゆるフロント部分である。json形式などでデータを受け取ってブラウザやスマホ画面に表示させるところ | Infrastructure層 |
Entityについて(Domain層)
他のサイトではビジネスロジックなどと抽象的に述べられているが、ここでは具体例を出して説明していきます。
例えば、オンライン書店の注文処理システムを考えてみます。
本を注文する際にまずコード上で"本"というオブジェクトが必要なのでまず本オブジェクトを作ります
import datetime
class Author:
def __init__(self, name: str, birth: datetime) -> None:
self.name = name
self.birth = birth
class Book:
def __init__(self, title: str, price: int, author: Author) -> None:
self.title = title
self.price = price
self.author = author
ここでAuthorがあるのは、アプリケーションを開発する前にある程度、"Bookがあるならその著者もいるか"などと決めておいて作るためここでもAuthorというクラスを定義しておきました。
次にこれらのアプリケーションで使用するオブジェクトに機能を持たせましょう。
import datetime
from typing import List, Dict, Any
class Author:
def __init__(self, name: str, birth: datetime, is_active: bool) -> None:
self.name = name
self.birth = birth
self.is_active = is_active
def deactivate(self) -> None:
self.is_active = False
def activate(self) -> None:
self.active = True
def introduce(self) -> Dict[str, Any]:
if not self.is_active:
return {}
return {
"name": self.name,
"birth": self.birth,
}
class Book:
def __init__(self, title: str, price: int, author: Author) -> None:
self.title = title
self.price = price
self.author = author
def set_price(self, price) -> None:
if price <= 0:
raise ValueError("price must be positive")
self.price = price
def info(self) -> Dict[str, Any]:
return {
"title": self.title,
"price": self.price,
"author": self.author.introduce()
}
と、このように機能を付け加えました。実際、アプリケーション作成中に度々機能は追加していくものなので、必ずしも機能を予め全て考えておく必要はありません。しかし、アプリケーション作成前にAという機能を作るにはこのclassにこのメソッドは持たせておく必要があるなど予め考えておくほうが良いでしょう。
ここで重要なのは何か機能やオブジェクトをアプリケーションに組み込みたい時はこのロジックのもとになるオブジェクトから作成し始める必要があるという点出る。
クリーンアーキテクチャの文脈において、このようなオブジェクトをEntity(実態)と呼びます。
UseCaseについて(Application層、こいつがロジック)
次にユースケースとは何か?そもそもユースケースとは日本語では利用用途という意味でsy。ならば、アプリケーションにおけるユースケースとは”そのアプリケーションにおける利用用途”と解釈して間違い無いでしょう。先ほどのオンライン書店の注文処理システムの利用用途について具体的に考えてみましょう。一つ良い例があります。ある著者が書いた本の一覧を取得するケースです。
from typing import List, Dict, Any
from .i_get_book_usecase import IGetBooksUseCase
from .book_repository import BookRepository
from .author_repository import AuthorRepository
class GetBooksDetailUseCase(IGetBooksUseCase):
def __init__(self,
book_repository: BookRepository, # interface
author_repository: AuthorRepository, # interface
presenter: IGetBooksPresenter # interface
) -> None:
self.book_repository = book_repository
self.author_repository = author_repository
self.presenter = presenter
def handle(self, input_dto: GetBooksInputDto) -> Response:
try:
author = self.author_repository.get_author_by_id(author_id=input_dto.author_id)
books: List[Book] = self.book_repository.get_by_author(author=author)
output_dto: List[GetBooksOutputDto] = [GetBooksOutputDto(**book.info()) for book in books]
return self.presenter.present(output_dto)
except Exception as e:
return self.presenter.present_error(e)
このようにUseCaseを記述しているが、book_repositoryとpresenterはインタフェースを参照しているので実態はまだありません。ただ、authorやbooksなどのオブジェクトがself.repository.get_by_nameメソッドから得られるということがUseCaseから分かるだけです。presenterも同様にインターフェースです。
言葉で語るより実際のコードをみてインターフェースを理解していきます
import abc
from .get_books_input_dto import GetBooksInputDto
from .get_books_output_dto import GetBooksOutputDto
from typing import List, Dict, Any
class IGetBooksUseCase(metaclass=abc.ABCMeta):
@abstractmethod
def handle(self, input_dto: GetBooksInputDto) -> MockResponse:
raise NotImplementedError
from typing import List, Dict, Any
from abc import ABCMeta, abstractmethod
from .get_books_output_dto import GetBooksOutputDto
class IGetBooksPresenter(metaclass=abc.ABCMeta):
@abstractmethod
def present(self, output_dto: List[GetBooksOutputDto]) -> MockResponse:
raise NotImplementedError
@abstractmethod
def present_error(self, error: Exception) -> MockResponse:
raise NotImplementedError
from .entity import Author, Book
from abc import ABCMeta, abstractmethod
class BookRepository(metaclass=abc.ABCMeta):
@abstractmethod
def get_by_author(self, author: Author) -> Book:
raise NotImplementedError
class AuthorRepository(metaclass=abc.ABCMeta):
@abstractmethod
def get_by_id(self, author_id: str) -> Author:
raise NotImplementedError
Dto(DataTransferObject)の実装は以下の通りです
from typing import List, Dict, Any
class GetBooksInputDto:
def __init__(self, author_id: str):
self.author_id = author_id
class GetBooksOutputDto:
def __init__(self, title: str, price: int, author: Dict[str, Any]):
self.title = title
self.price = price
self.author = author
def to_json(self):
return {
"title": self.title,
"price": self.price,
"author": self.author
}
このようにインターフェースを定義することで、UseCase層ではインターフェースだけを参照してロジックを記述することができます。今回の例では、著者が書いた本の一覧を取得するとき
- ユーザーからinput_dtoとして、著者に関する情報(今回はid)を取得する
- 著者idから著者Entity(Author)のインスタンスを取得する(autor_repositoryのインターフェースからAuthorオブジェクトが取得できることが分かる)
- そのEntityをself.book_repositoryに渡す事でList[Book]という形で本の一覧を取得する
- List[Book]をpresenterが受け取るオブジェクト(output_dto)へ変換してpresenterに渡す
これが著者情報から本の一覧を取得するというユースケースに対応するロジックです。
ロジックとしては、UseCase層で入力形式としてGetBooksInputDtoを受け取り、presenter(presentation層にある)への入力形式としてGetBooksOutputDtoを渡すという流れになっています。
DataAccessと DB(Infra層)
先ほどロジックを形作るApplication層を実装しました。このままではまだ抽象的概念を記述したにすぎません。なぜ抽象的かというと、repositoryとpresenterはインターフェースであるが故に、あくまで、例えばAというオブジェクトがinputとして入力された時に、Bというオブジェクトを返却するというルールがあると仮定しているすぎないからです。実際にインタフェースであるrepositoryやpresenterに何かinputを入力しても、そもそも返却する仕組みを記述していないのだから何も返却はされません。
言葉だけではわかりづらいので、先ほどのコードを再考してみましょう。
class BookRepository(metaclass=abc.ABCMeta):
@abstractmethod
def get_by_author(self, author: Author) -> Book:
raise NotImplementedError
このBookRepositoryのget_by_authorメソッドには入力としてauthorというEntityを受け取りBookというEntityを返却するとしているが、実際にこのBookRepositoryのインスタンを作ってget_by_authorメソッドを呼び出してもBookオブジェクトは返却されません。これは、ただこのメソッドのinputとoutputの型を定義付けているだけでその仕組み(inputを受け取って、さまざまな処理を行いBookを返却する仕組み)を記述していないからです。ここで、DataAccessの出番というわけです。
DataAccessクラスでは先ほどの中身も何もない抽象的なBookRepositoryの中身を実装するのが目的です。この際DataAccessクラスはBookRepositoryを継承する必要があります。こうすることで、DataAccessクラスに、強制的にBookRepository内で書かれているメソッドの実装を行わせます。実装を行わないとインスタンス化できないというエラーが生じます。このある種の縛りが強力な柔軟性を生みます。
その理由
- Application層で記述したGetBooksUseCaseはBookRepositoryのメソッドしか見ていいない -> 逆に言えばこのBookRepository通りのメソッドをDataAccessでしっかり記述していればApplication層からするとその記述の仕方はどうでもよい
- DataAccessからすれば、記述の仕方は自由であるのでここはDataAccessクラスが自由に変更できる
ここでも先ほどの例を基に実例を示します
from my_database import my_db
from .entity import Book, Author
class InMemoryDB:
def __init__(self):
self.authors = {} # author_id をキーとする辞書
self.books = [] # Book のリスト
def add_author(self, author_id: str, author: Author):
if author_id in self.authors:
raise ValueError("Author with this ID already exists.")
self.authors[author_id] = author
def get_author_by_id(self, author_id: str) -> Author:
if author_id not in self.authors:
raise ValueError("Author not found.")
return self.authors[author_id]
def add_book(self, book: Book):
if not any(author for author in self.authors.values() if author.name == book.author.name):
raise ValueError("Author for this book does not exist.")
self.books.append(book)
def get_books_by_author_name(self, author_name: str) -> List[Book]:
return [book for book in self.books if book.author.name == author_name]
my_db = InMemoryDB()
your_db = InMemoryDB()
class MyBookDataAccess(BookRepository):
def __init__(self, db: InMemoryDB):
self.db = db
def get_by_author(self, author: Author) -> List[Book]:
books = self.db.get_books_by_author_name(author_name=author.name)
if not books:
raise ValueError(f"No books found for author {author.name}")
return books
class YourBookDataAccess(BookRepository):
def __init__(self):
self.db = your_db
def get_by_author(self, author: Author) -> Book:
book_data = self.db.book.get({"name": author.name})
# ここでデータベースから取得したbook_dataから必要な情報を取り出しBookEntityを作成して返却する
return Book(title=book_data["title"], price=book_data["price"], author=author)
このようにいずれのBookDataAccessもBookRepositoryを実装しているので使用するデータベースが異なれど、Application層からすると全く問題ないのです。これが柔軟性です。
Application層のUseCaseではインターフェースであるRepositoryを参照しており、そのインターフェースの内部実装を担っているのがInfra層のDataAccessです。図だと、下図にあたる
UseCaseがDataAccessInterface(Repository)を参照しており、DataAccessがRepositoryを実装しています。(下図の黒矢印は参照を意味し、白矢印は実装を意味します。)処理の流れはUseCase-> DataAccessであるが、依存関係はInfra層のDataAccessが Application層のRepositoryに縛られるため、DataAccessがUseCaseに依存している形になる。これを依存関係の逆転という。(難しそうに聞こえるが大したことはやっていません)。このメカニズムがPresentation層とApplication層の境界でも使用されています。
次にPresentation層の説明に行きます
ControllerとPresenter(Presentation層)
ControllerとPresenterの役割について解説していきます。
Controllerはフロントからのrequestを受け取って、リクエストからデータを受け取りユースケースに適したオブジェクトへ変換し、UseCase層へそのオブジェクトを渡す役割があります。ユースーケース層へ渡されるオブジェクトは先ほど定義したInputDtoです。GetBooksという本一覧を取得するユースケースでは、author情報(author_id)をrequestオブジェクトから取得しこれをGetBooksInputDtoというオブジェクトへ変換してGetBooksUseCaseへそのオブジェクトを渡します。この際、requestのbodyまたはheader、urlparamsに今回のユースケースに必要とされる情報があるか？や型があっているか？などvalidationの処理も書きます。
次にPresenterの役割について説明します。これはControllerとは逆で、requestを送信してきたフロントの期待通りの情報をjsonなど適切な形式にしてresponseを返す役割があります。また、errorが生じた際に何を返却するかなどもここで処理します。(400 Bad Requestなのか、404 Not Found、500 Internal Server Erroなど)
実際にコードを見て理解しましょう
from .i_get_books_use_case import IGetBooksUseCase
from .get_books_dto import GetBooksInputDto
class BooksController:
def __init__(self, get_books_use_case: IGetBooksUseCase) -> None:
self.get_books_use_case = get_books_use_case
def get(self, request: Request) -> Response:
'''
validation処理など必要に応じて処理
'''
input_dto = GetBooksInputDto(**request.data)
return self.get_books_use_case.handle(input_dto)
from typing import List
from .i_get_books_presenter import IGetBooksPresenter
from .get_books_output_dto import GetBooksOutputDto
class GetBooksPresenter(IGetBooksPresenter):
def present(self, output_dto: List[GetBooksOutputDto]) -> Response:
res_data = output_dto.to_json()
return Response(res_data, status=200)
def present_error(self, e: Exception) -> Response:
return Response(e.message(), status=400)
BooksControllerではgetメソッドが実装されていて、requestを受け取り、requestオブジェクトからdataを取得し、input_dtoに変換しています。これによってUseCase層へデータを渡すことができるようになります。ここでも、Controller層はUseCaseを直接参照しているのではなくInterfaceを通して間接的にUseCase層を参照しています。
PresenterはインターフェースであるIGetBooksPresenterを実装しています。実際に受け取ったOutputDtoをjsonに変換して、Responseオブジェクトに渡せる形式に変換してResponseを返しています。さらにpresent_errorメソッドもインターフェースの縛りによって実装されています(されなければなりません)。これによってエラーが生じた場合はエラーを返却するようになっています。
DI(Dependency Injection)
最後に、今までインターフェースを参照している箇所に、そのインターフェースを実装しているクラスのインスタンスを渡す必要があります。具体的に以下のようなことをしなければなりません。
from abc import ABCMeta, abstractmethod
class InterfaceReorio(metaclass=ABCMeta):
@abstractmethod
def speak(self) -> str:
raise NotImplementedError
class Reorio(InterfaceReorio):
def speak(self) -> str:
return "レオリオ：薄汚いクルタ族とかを絶やしてやるぜ"
class Kurapika:
def __init__(self, reorio: InterfaceReorio) -> None:
self.reorio = reorio
def speak(self) -> str:
return "クラピカ：品性は金で買えないよ、レオリオ。 " + self.reorio.speak()
reorio = Reorio()
kurapika = Kurapika(reorio)
kurapika.speak()
これと同じようなことを今までの実装を全て記述して、DIも行ってみます。
import datetime
from typing import List, Dict, Any
import abc
from requests import Response, Request
from abc import ABCMeta, abstractmethod
# モックリクエストとレスポンスを定義 (実際のフレームワークに応じて変更可能)
class MockRequest:
def __init__(self, data: Dict[str, Any]):
self.data = data
class MockResponse(Response):
def __init__(self, json_data: Any, status: int):
self.json_data = json_data
self.status_code = status
def json(self):
return self.json_data
class Author:
def __init__(self, name: str, birth: datetime, is_active: bool) -> None:
self.name = name
self.birth = birth
self.is_active = is_active
def deactivate(self) -> None:
self.is_active = False
def activate(self) -> None:
self.active = True
def introduce(self) -> Dict[str, Any]:
if not self.is_active:
return {}
return {
"name": self.name,
"birth": self.birth,
}
class Book:
def __init__(self, title: str, price: int, author: Author) -> None:
self.title = title
self.price = price
self.author = author
def set_price(self, price) -> None:
if price <= 0:
raise ValueError("price must be positive")
self.price = price
def info(self) -> Dict[str, Any]:
return {
"title": self.title,
"price": self.price,
"author": self.author.introduce()
}
class GetBooksInputDto:
def __init__(self, author_id: str):
self.author_id = author_id
class GetBooksOutputDto:
def __init__(self, title: str, price: int, author: Dict[str, Any]):
self.title = title
self.price = price
self.author = author
def to_json(self):
return {
"title": self.title,
"price": self.price,
"author": self.author
}
class IGetBooksUseCase(metaclass=abc.ABCMeta):
@abstractmethod
def handle(self, input_dto: GetBooksInputDto) -> MockResponse:
raise NotImplementedError
class IGetBooksPresenter(metaclass=abc.ABCMeta):
@abstractmethod
def present(self, output_dto: List[GetBooksOutputDto]) -> MockResponse:
raise NotImplementedError
@abstractmethod
def present_error(self, error: Exception) -> MockResponse:
raise NotImplementedError
class BookRepository(metaclass=abc.ABCMeta):
@abstractmethod
def get_by_author(self, author: Author) -> Book:
raise NotImplementedError
class AuthorRepository(metaclass=abc.ABCMeta):
@abstractmethod
def get_by_id(self, author_id: str) -> Author:
raise NotImplementedError
class GetBooksDetailUseCase(IGetBooksUseCase):
def __init__(self,
book_repository: BookRepository, # interface
author_repository: AuthorRepository, # interface
presenter: IGetBooksPresenter # interface
) -> None:
self.book_repository = book_repository
self.author_repository = author_repository
self.presenter = presenter
def handle(self, input_dto: GetBooksInputDto) -> Response:
try:
author = self.author_repository.get_author_by_id(author_id=input_dto.author_id)
books: List[Book] = self.book_repository.get_by_author(author=author)
output_dto: List[GetBooksOutputDto] = [GetBooksOutputDto(**book.info()) for book in books]
return self.presenter.present(output_dto)
except Exception as e:
return self.presenter.present_error(e)
class InMemoryDB:
def __init__(self) -> None:
self.authors = {} # author_id をキーとする辞書
self.books = [] # Book のリスト
def add_author(self, author_id: str, author: Author):
if author_id in self.authors:
raise ValueError("Author with this ID already exists.")
self.authors[author_id] = author
def get_author_by_id(self, author_id: str) -> Author:
if author_id not in self.authors:
raise ValueError("Author not found.")
return self.authors[author_id]
def add_book(self, book: Book):
if not any(author for author in self.authors.values() if author.name == book.author.name):
raise ValueError("Author for this book does not exist.")
self.books.append(book)
def get_books_by_author_name(self, author_name: str) -> List[Book]:
return [book for book in self.books if book.author.name == author_name]
#dbのインスタンス化(実際はfirebaseのfirestoreのdbなどを使用することも)
my_db = InMemoryDB()
your_db = InMemoryDB()
class MyBookDataAccess(BookRepository):
def __init__(self, db: InMemoryDB):
self.db = db
def get_by_author(self, author: Author) -> List[Book]:
books = self.db.get_books_by_author_name(author_name=author.name)
if not books:
raise ValueError(f"No books found for author {author.name}")
return books
class YourBookDataAccess(BookRepository):
def __init__(self):
self.db = your_db
def get_by_author(self, author: Author) -> Book:
book_data = self.db.book.get({"name": author.name})
# ここでデータベースから取得したbook_dataから必要な情報を取り出しBookEntityを作成して返却する
return Book(title=book_data["title"], price=book_data["price"], author=author)
# Controllerの実装
class BooksController:
def __init__(self, get_books_use_case: IGetBooksUseCase) -> None:
self.get_books_use_case = get_books_use_case
def get(self, request: MockRequest) -> MockResponse:
'''
validation処理など必要に応じて処理
'''
input_dto = GetBooksInputDto(**request.data)
return self.get_books_use_case.handle(input_dto)
# Presenter の具体実装
class GetBooksPresenter(IGetBooksPresenter):
def present(self, output_dto: List[GetBooksOutputDto]) -> Response:
res_data = [dto.to_json() for dto in output_dto]
return MockResponse(res_data, status=200)
def present_error(self, e: Exception) -> Response:
return MockResponse({"error": str(e)}, status=400)
次はDIを行います。
from requests.models import Request, Response
# モックリクエストとレスポンスを定義
class MockRequest:
def __init__(self, data: Dict[str, Any]):
self.data = data
class MockResponse(Response):
def __init__(self, json_data: Any, status: int):
self.json_data = json_data
self.status_code = status
def json(self):
return self.json_data
# データセットアップ
author_1 = Author(name="Author A", birth=datetime.datetime(1980, 1, 1), is_active=True)
author_2 = Author(name="Author B", birth=datetime.datetime(1990, 1, 1), is_active=True)
book_1 = Book(title="Book 1", price=1000, author=author_1)
book_2 = Book(title="Book 2", price=1500, author=author_1)
book_3 = Book(title="Book 3", price=2000, author=author_2)
# データベース初期化
my_db = InMemoryDB()
my_db.add_author("1", author_1)
my_db.add_author("2", author_2)
my_db.add_book(book_1)
my_db.add_book(book_2)
my_db.add_book(book_3)
# 各クラスの初期化
book_repository = MyBookDataAccess(my_db)
author_repository = my_db # InMemoryDBをAuthorRepositoryとして利用
presenter = GetBooksPresenter()
# ユースケースの初期化
get_books_use_case = GetBooksDetailUseCase(
book_repository=book_repository,
author_repository=author_repository,
presenter=presenter
)
# コントローラーの初期化
controller = BooksController(get_books_use_case=get_books_use_case)
# リクエストを作成して実行
mock_request = MockRequest(data={"author_id": "1"})
response = controller.get(request=mock_request)
# 結果を出力
print(f"Status Code: {response.status_code}")
print(f"Response JSON: {response.json()}")
結局何が嬉しいの?
クリーンアーキテクチャを採用するメリット
- 実装する際にEntity -> Application -> Infra または presenterという順に実装するためコードの中心にEntityやApplicationといったロジックが置かれる
- interfaceでpresentationとInfrastructureとapplicationを分離しているため、interfaceにさえ準拠していれば、InfraだけテストするとかMockを使用するとかできる
- UseCaseごとにロジックを記述するため、特定のユースケースの変更が別のユースケースの変更に影響を与えづらい(Entityを変更するとかだと影響する)
- 技術変更が容易。MySQLからPostgreSQLへの変更など
- 依存関係の明確化によって、全体構造を把握しやすい
私が個人的に考えるデメリットは以下です
デメリット - コードが冗長的になる。ユースケースごとに記述するため一つのユースケースに対するコード量が増す。->短期的な開発効率は落ちる
- 開発者の思想(層の分離など)を遵守しないと、依存関係が曖昧になりがち
- 学習コスト。今までMVCなどで開発していた人がいきなりクリーンアーキテクチャに則って開発するのは難しいかも
- MTVやMVCに基づいたフレームワーク(Djangoなど)に導入しようとすると少し面倒
以上のことを踏まえると、必ずしもクリーンアーキテクチャが良いとは言えません。作成者がどの程度の規模のアプリケーションをどの程度の期間で作りたいかなどケースバイケースであるということです。しかし、クリーンアーキテクチャの依存関係の明確化という概念は、interfaceやDIなどをフルに活用していて、pythonだけでなくtypescriptなどの言語でよく出てくるので理解しておくことは大切だと思います。
今回はここまでです。DIの部分などpythonのinjectorライブラリなど使用すればもっと綺麗にコードを書くことができるのですが、ここでは割愛します。
何か解釈違いなどあれば遠慮なく指摘していただけると幸いです。github上にもソースコードを載せていく予定なので、随時本記事は更新されていくかと思います。質問などあればお気軽に。
Discussion