create project

Author: t-tiger

.gitignore

@@ -0,0 +1,4 @@
+.idea
+tmp
+venv
+__pycache__

Makefile

@@ -0,0 +1,2 @@
+test:
+	python -m unittest

README.md

@@ -0,0 +1,18 @@
+# Python Architecture Example
+
+This project is a sample Python web application adapting Clean Architecture. By using typing module, development experience similar to static type language is achieved.
+
+Dependency injection and Dependency inversion principle, which are representative feature in Clean Architecture, can be implemented without any obstacles.
+
+In order to experience the separation of concerning and improvement of testability by Clean Architecture along with the high productivity by Python, I've published this sample project.
+
+Also in `tests` directory, there are some test samples with partial mock. If you are interested, please take a look this.
+
+# How to run
+
+```bash
+$ pip install -r requirements/requirements
+$ python -m flask run
+```
+
+Then open `http://localhost:5000` in your browser.

app.py

@@ -0,0 +1,18 @@
+from flask import Flask
+
+from src.driver.article_driver import ArticleDriverImpl
+from src.interactor.article_interactor import ArticleInteractor
+from src.repository.article_repository import ArticleRepositoryImpl
+from src.rest.article_resource import ArticleResource
+
+app = Flask(__name__)
+
+article_resource = ArticleResource(
+    article_usecase=ArticleInteractor(
+        article_repository=ArticleRepositoryImpl(
+            article_driver=ArticleDriverImpl()
+        )
+    )
+)
+
+app.add_url_rule('/', view_func=article_resource.index)

requirements/requirements.txt

@@ -0,0 +1,2 @@
+flask
+aiohttp

src/domain/article.py

@@ -0,0 +1,14 @@
+from dataclasses import dataclass
+
+from src.domain.collection import Collection
+
+
+@dataclass(frozen=True)
+class Article:
+    id: str
+    body: str
+
+
+@dataclass(frozen=True)
+class Articles(Collection[Article]):
+    values: [Article]

src/domain/collection.py

@@ -0,0 +1,13 @@
+from typing import Generic, TypeVar
+
+from attr import dataclass
+
+T = TypeVar('T')
+
+
+@dataclass(frozen=True)
+class Collection(Generic[T]):
+    values: [T]
+
+    def map(self, func) -> map:
+        return map(func, self.values)

src/driver/article_driver.py

@@ -0,0 +1,15 @@
+import json
+import aiohttp
+from src.interface.driver.article_driver import ArticleDriver
+
+
+class ArticleDriverImpl(ArticleDriver):
+    async def get_articles(self, page: int) -> dict:
+        response = await aiohttp.request('GET', 'https://qiita.com/api/v2/items?page=1&per_page=20').coro
+        articles = json.loads(await response.text())
+        return {
+            "articles": [{
+                "id": a["id"],
+                "body": a["body"]
+            } for a in articles]
+        }

src/interactor/article_interactor.py

@@ -0,0 +1,13 @@
+from src.domain.article import Articles
+from src.interface.repository.article_repository import ArticleRepository
+from src.interface.usecase.article_usecase import ArticleUsecase
+
+
+class ArticleInteractor(ArticleUsecase):
+    article_repository: ArticleRepository
+
+    def __init__(self, article_repository: ArticleRepository):
+        self.article_repository = article_repository
+
+    async def get_list(self, page: int) -> Articles:
+        return await self.article_repository.get_list(page)

src/interface/driver/article_driver.py

@@ -0,0 +1,7 @@
+from abc import ABCMeta, abstractmethod
+
+
+class ArticleDriver(metaclass=ABCMeta):
+    @abstractmethod
+    async def get_articles(self, page: int) -> dict:
+        raise NotImplementedError

src/interface/repository/article_repository.py

@@ -0,0 +1,9 @@
+from abc import ABCMeta, abstractmethod
+
+from src.domain.article import Articles
+
+
+class ArticleRepository(metaclass=ABCMeta):
+    @abstractmethod
+    async def get_list(self, page: int) -> Articles:
+        raise NotImplementedError

src/interface/usecase/article_usecase.py

@@ -0,0 +1,9 @@
+from abc import ABCMeta, abstractmethod
+
+from src.domain.article import Articles
+
+
+class ArticleUsecase(metaclass=ABCMeta):
+    @abstractmethod
+    async def get_list(self, page: int) -> Articles:
+        raise NotImplementedError

src/repository/article_repository.py

@@ -0,0 +1,19 @@
+from src.domain.article import Article, Articles
+from src.interface.driver.article_driver import ArticleDriver
+from src.interface.repository.article_repository import ArticleRepository
+
+
+class ArticleRepositoryImpl(ArticleRepository):
+    article_driver: ArticleDriver
+
+    def __init__(self, article_driver: ArticleDriver):
+        self.article_driver = article_driver
+
+    async def get_list(self, page: int) -> Articles:
+        res = await self.article_driver.get_articles(page)
+        return Articles(
+            values=[
+                Article(id=a["id"], body=a["body"])
+                for a in res["articles"]
+            ]
+        )

src/rest/article_resource.py

@@ -0,0 +1,26 @@
+import asyncio
+
+from flask import jsonify
+
+from src.domain.article import Articles
+from src.interface.usecase.article_usecase import ArticleUsecase
+
+
+class ArticleResource:
+    article_usecase: ArticleUsecase
+
+    def __init__(self, article_usecase: ArticleUsecase):
+        self.article_usecase = article_usecase
+
+    def index(self):
+        main_group: Articles = asyncio.run(self.article_usecase.get_list(0))
+        return jsonify(articles_response(main_group))
+
+
+def articles_response(articles: Articles) -> dict:
+    return {
+        "items": [{
+            "id": article.id,
+            "body": article.body
+        } for article in articles.values]
+    }

tests/__init__.py

@@ -0,0 +1,8 @@
+import asyncio
+
+
+def async_test(coro):
+    def wrapper(*args, **kwargs):
+        loop = asyncio.new_event_loop()
+        return loop.run_until_complete(coro(*args, **kwargs))
+    return wrapper

tests/repository/__init__.py

@@ -0,0 +1,31 @@
+from unittest import TestCase
+
+from asyncmock import AsyncMock
+
+from src.domain.article import Article, Articles
+from src.interface.driver.article_driver import ArticleDriver
+from src.repository.article_repository import ArticleRepositoryImpl
+from tests import async_test
+
+
+class DriverMock(ArticleDriver):
+    async def get_articles(self, page: int) -> dict:
+        raise NotImplementedError
+
+
+class TestArticleRepository(TestCase):
+    @async_test
+    async def test_get(self):
+        get_main_group_mock = AsyncMock(return_value={"articles": [
+            {"id": "1", "body": "test"}
+        ]})
+
+        driver = DriverMock()
+        driver.get_articles = get_main_group_mock
+        repository = ArticleRepositoryImpl(article_driver=driver)
+
+        self.assertEqual(
+            await repository.get_list(1),
+            Articles(values=[Article(id="1", body="test")])
+        )
+        get_main_group_mock.assert_called_with(1)

tests/repository/test_article_repository.py

@@ -0,0 +1,26 @@
+from unittest import TestCase
+
+from asyncmock import AsyncMock
+
+from src.domain.article import Article, Articles
+from src.interactor.article_interactor import ArticleInteractor
+from src.interface.repository.article_repository import ArticleRepository
+from tests import async_test
+
+
+class RepositoryMock(ArticleRepository):
+    async def get_list(self, page: str) -> Articles:
+        raise NotImplementedError
+
+
+class TestArticleInteractor(TestCase):
+    @async_test
+    async def test_get(self):
+        get_mock = AsyncMock(return_value=Article(id="1", body='test'))
+
+        repo = RepositoryMock()
+        repo.get_list = get_mock
+        usecase = ArticleInteractor(article_repository=repo)
+
+        self.assertEqual(await usecase.get_list(1), Article(id="1", body="test"))
+        get_mock.assert_called_with(1)