Merge pull request #7 from rfrezino/refactor-code

Fix code to work with nested dependencies

README.md

@@ -43,9 +43,10 @@ Example, for this folder structure:
     └── create_person_usecase.py
 ```
 The file:
+
 ```python
 from hexagonal.domain.hexagonal_layer import HexagonalLayer
-from hexagonal.main import hexagonal_config
+from hexagonal.hexagonal_config import hexagonal_config
 
 infrastructure_layer = HexagonalLayer(name='infrastructure', directories=['infrastructure'])
 use_cases_layer = HexagonalLayer(name='use_cases', directories=['usecases'])

pyproject.toml

@@ -1,12 +1,12 @@
 [tool.poetry]
 name = "hexagonal-sanity-check"
-version = "0.0.20"
+version = "0.0.21"
 description = """Hexagonal Sanity Check"""
 readme = 'README.md'
 authors = ["rfrezino <rodrigofrezino@gmail.com>"]
 packages = [{ include = "hexagonal", from = "src"}]
 keywords = ["hexagonal architecture", "onion architecture", "enforce rules"]
-exclude = ["src/hexagonal/hexagonal_config.py"]
+exclude = ["src/hexagonal/sanity_check_hexagonal_config.py"]
 
 [tool.poetry.dependencies]
 python = "^3.7"

src/hexagonal/domain/hexagonal_layer.py

@@ -6,3 +6,14 @@
 class HexagonalLayer:
     name: str
     directories: List[str]
+
+    def __post_init__(self):
+        self._valid_dirs()
+
+    def _valid_dirs(self):
+        for dir in self.directories:
+            if not dir.startswith('/'):
+                raise Exception(f'Hexagonal Layer directory "{self.name}" must start with /. Example: "/domain"')
+
+            if dir.endswith('/'):
+                raise Exception(f'Hexagonal Layer directory "{self.name}" must not finish with /. Example: "/domain"')

src/hexagonal/domain/hexagonal_project/hexagonal_project.py

@@ -0,0 +1,27 @@
+from dataclasses import dataclass
+from typing import List, Optional
+
+from hexagonal.domain.hexagonal_project.hexagonal_project_layer import HexagonalProjectLayer
+from hexagonal.domain.python_file import PythonFile
+
+
+@dataclass
+class HexagonalProject:
+    project_path: str
+    layers: List[HexagonalProjectLayer]
+    files_not_in_layers: List[PythonFile]
+
+    def get_layer_for_file_path(self, file_full_path: str) -> Optional[HexagonalProjectLayer]:
+        for layer in self.layers:
+            for python_file in layer.python_files:
+                if python_file.file_full_path == file_full_path:
+                    return layer
+        return None
+
+    def get_python_file(self, file_full_path: str) -> PythonFile:
+        for layer in self.layers:
+            for python_file in layer.python_files:
+                if python_file.file_full_path == file_full_path:
+                    return python_file
+
+        raise Exception(f'File not found {file_full_path}')

src/hexagonal/domain/python_project.py → src/hexagonal/domain/hexagonal_project/hexagonal_project_layer.py

@@ -3,8 +3,9 @@
 
 from hexagonal.domain.python_file import PythonFile
 
-
 @dataclass
-class PythonProject:
-    full_path: str
+class HexagonalProjectLayer:
+    index: int
+    name: str
+    directories: List[str]
     python_files: List[PythonFile]

src/hexagonal/domain/python_file.py

@@ -1,13 +1,15 @@
 from dataclasses import dataclass
-from typing import Optional, List
-
-from hexagonal.domain.python_module import PythonModule
+from typing import List
 
 
 @dataclass
 class PythonFile:
-    full_path: str
-    relative_path_from_source_module: str
-    layer_name: str
-    layer_index: Optional[int]
-    imported_modules: List[PythonModule]
+    file_full_path: str
+    file_name: str
+    file_folder_full_path: str
+    # For instance if the project is located at: /usr/project/
+    # and the file is /usr/project/services/another/file.py
+    # the project_relative_folder_path is "services/another"
+    relative_folder_path_from_project_folder: str
+    project_folder_full_path: str
+    imported_modules: List[str]

src/hexagonal/domain/raw_python_file.py

@@ -0,0 +1,14 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class RawPythonFile:
+    file_full_path: str
+    file_name: str
+    file_folder_full_path: str
+    # For instance if the project is located at: /usr/project/
+    # and the file is /usr/project/services/another/file.py
+    # the project_relative_folder_path is "services/another"
+    relative_folder_path_from_project_folder: str
+    project_folder_full_path: str
+

src/hexagonal/hexagonal_config.py

@@ -1,9 +1,27 @@
+from typing import List
+
 from hexagonal.domain.hexagonal_layer import HexagonalLayer
-from hexagonal.main import hexagonal_config
+from hexagonal.services.hexagonal_composition import HexagonalComposition
+
+
+class HexagonalConfig:
+    excluded_dirs: List[str]
+    _layers: HexagonalComposition
+
+    def __init__(self):
+        self._layers = HexagonalComposition()
+        self.excluded_dirs = []
+
+    def add_inner_layer(self, layer: HexagonalLayer) -> 'HexagonalConfig':
+        self._layers.append(layer)
+        return self
+
+    def clear_layers(self):
+        self._layers.clear()
+
+    @property
+    def layers(self) -> HexagonalComposition:
+        return self._layers
 
-infrastructure_layer = HexagonalLayer(name='infrastructure', directories=['infrastructure'])
-use_cases_layer = HexagonalLayer(name='use_cases', directories=['use_cases'])
-services_layer = HexagonalLayer(name='services', directories=['services'])
-domain_layer = HexagonalLayer(name='domain', directories=['domain'])
 
-hexagonal_config + infrastructure_layer >> use_cases_layer >> services_layer >> domain_layer
+hexagonal_config = HexagonalConfig()

src/hexagonal/infrastructure/cli.py

@@ -6,8 +6,8 @@
 import click
 
 from hexagonal.domain.hexagonal_error import HexagonalError
-from hexagonal.main import hexagonal_config
-from hexagonal.use_cases.check_project_sanity_usecase import CheckProjectSanityUseCase
+from hexagonal.hexagonal_config import hexagonal_config
+from hexagonal.use_cases.check_project_sanity_usecase import CheckProjectSanityUseCase, HexagonalCheckResponse
 from hexagonal.use_cases.generate_diagram_usecase import GenerateDiagramUseCase
 
 
@@ -20,9 +20,9 @@ def cli():
 @click.option('--source_path', help='Where main source folder is located.', required=True)
 @click.option('--hexagonal_config_file', default='hexagonal_config.py', help="Hexagonal configuration file's name.")
 def diagram(source_path, hexagonal_config_file):
-    _process_cli_arguments(source_path=source_path, hexagonal_config_file=hexagonal_config_file)
-
     try:
+        _process_cli_arguments(source_path=source_path, hexagonal_config_file=hexagonal_config_file)
+
         hexa_diagram = GenerateDiagramUseCase()
         hexa_diagram.execute(project_name='Hexagonal Architecture Diagram', hexagonal_composition=hexagonal_config,
                              show=True)
@@ -35,22 +35,43 @@ def diagram(source_path, hexagonal_config_file):
 @click.option('--source_path', help='Where main source folder is located.', required=True)
 @click.option('--hexagonal_config_file', default='hexagonal_config.py', help="Hexagonal configuration file's name.")
 def check(source_path, hexagonal_config_file):
-    _process_cli_arguments(source_path=source_path, hexagonal_config_file=hexagonal_config_file)
+    def _build_response_message() -> str:
+        return f'Hexagonal Architecture: Checked a project with {len(response.hexagonal_project.layers)} ' \
+               f'hexagonal layers, {len(response.python_files)} python files ' \
+               f'and found {len(response.errors)} errors.'
 
     try:
-        checker = CheckProjectSanityUseCase()
-        response = checker.check(composition=hexagonal_config, source_folder=source_path)
+        _process_cli_arguments(source_path=source_path, hexagonal_config_file=hexagonal_config_file)
+
+        checker = CheckProjectSanityUseCase(hexagonal_config=hexagonal_config, source_folder=source_path)
+        response = checker.check()
+        _print_check_response(response)
     except Exception as error:
         logging.error('Error while processing project', exc_info=error)
         click.echo(f'Error while processing project: "{error}"')
         exit(1)
 
     [_print_error(index, error) for index, error in enumerate(response.errors)]
     if len(response.errors) > 0:
-        logging.error('Hexagonal Architecture: Errors found in dependencies flow.')
+        logging.error(_build_response_message())
         exit(1)
 
-    click.echo('Hexagonal Architecture: No errors found.')
+    click.echo(_build_response_message())
+
+
+def _print_check_response(response: HexagonalCheckResponse):
+    hexa_project = response.hexagonal_project
+    logging.info(f'Checked information for project: {hexa_project.project_path}')
+
+    for layer in hexa_project.layers:
+        logging.info(f'#### Files in layer "{layer.name}"')
+
+        for idx, file in enumerate(layer.python_files):
+            logging.info(f' {idx + 1}) File {file.file_full_path}')
+
+    logging.warning(f'#### Files out site of layers')
+    for idx, file in enumerate(hexa_project.files_not_in_layers):
+        logging.warning(f' {idx + 1}) File {file.file_full_path}')
 
 
 def _print_error(error_index: int, error: HexagonalError):
@@ -80,7 +101,7 @@ def _import_hexagonal_config_file(hexagonal_config_file: str):
     if not os.path.isfile(hexagonal_config_file):
         click.echo('Project configuration file not found.')
         exit(1)
-    hexagonal_config.clear()
+    hexagonal_config.clear_layers()
     run_path(hexagonal_config_file)
 
 
@@ -97,7 +118,7 @@ def _process_cli_arguments(source_path: str, hexagonal_config_file: str):
 
 
 def main():
-    logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', stream=sys.stdout, level=logging.WARNING)
+    logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', stream=sys.stdout, level=logging.INFO)
     cli()
 
 

src/hexagonal/sanity_check_hexagonal_config.py

@@ -0,0 +1,7 @@
+from hexagonal.domain.hexagonal_layer import HexagonalLayer
+from hexagonal.hexagonal_config import hexagonal_config
+
+hexagonal_config.add_inner_layer(HexagonalLayer(name='infrastructure', directories=['/infrastructure']))
+hexagonal_config.add_inner_layer(HexagonalLayer(name='use_cases', directories=['/use_cases']))
+hexagonal_config.add_inner_layer(HexagonalLayer(name='services', directories=['/services']))
+hexagonal_config.add_inner_layer(HexagonalLayer(name='domain', directories=['/domain']))

src/hexagonal/services/hexagonal_composition.py

@@ -5,13 +5,6 @@
 
 class HexagonalComposition(List[HexagonalLayer]):
 
-    def get_layer_index_by_module_name(self, module: str) -> Optional[int]:
-        for idx, layer in enumerate(self):
-            if module in layer.directories:
-                return idx + 1
-
-        return None
-
     def __add__(self, next_layer) -> 'HexagonalComposition':
         self.clear()
         self.append(next_layer)

src/hexagonal/services/hexagonal_dependency_flow_checker.py

@@ -0,0 +1,55 @@
+from dataclasses import dataclass
+from typing import List
+
+from hexagonal.domain.hexagonal_project.hexagonal_project import HexagonalProject
+from hexagonal.domain.hexagonal_project.hexagonal_project_layer import HexagonalProjectLayer
+from hexagonal.domain.python_file import PythonFile
+
+
+@dataclass
+class DependencyFlowError:
+    source_file: PythonFile
+    source_file_layer: HexagonalProjectLayer
+    imported_module: PythonFile
+    imported_module_layer: HexagonalProjectLayer
+
+
+@dataclass
+class DependencyFlowResponse:
+    errors: List[DependencyFlowError]
+
+
+class HexagonalDependencyFlowChecker:
+    _hexagonal_project: HexagonalProject
+
+    def __init__(self, hexagonal_project: HexagonalProject):
+        self._hexagonal_project = hexagonal_project
+
+    def check(self) -> DependencyFlowResponse:
+        errors = []
+        for layer in self._hexagonal_project.layers:
+            errors.extend(self._check_layer(layer=layer))
+        return DependencyFlowResponse(errors=errors)
+
+    def _check_layer(self, *, layer: HexagonalProjectLayer) -> List[DependencyFlowError]:
+        result = []
+        for python_file in layer.python_files:
+            result.extend(self._check_file(layer=layer, python_file=python_file))
+        return result
+
+    def _check_file(self, *, layer: HexagonalProjectLayer, python_file: PythonFile) -> List[DependencyFlowError]:
+        result = []
+        for imported_module in python_file.imported_modules:
+            module_layer = self._hexagonal_project.get_layer_for_file_path(file_full_path=imported_module)
+            if not module_layer:
+                continue
+
+            if layer.index < module_layer.index:
+                error = DependencyFlowError(
+                    source_file=python_file,
+                    source_file_layer=layer,
+                    imported_module=self._hexagonal_project.get_python_file(file_full_path=imported_module),
+                    imported_module_layer=module_layer)
+                result.append(error)
+
+        return result