How to implement Hexagonal Architecture in Python and valide it using your CI workflow.

Illustration photo of Hexagonal Architecture

Featured Image - Hexagonal Architecture

Hexagonal architecture has gained popularity in recent years. Unfortunately, there are few practical courses available online that show how to implement it with code snippets. In this course, we will introduce the Hexagonal Architecture, build a Python application using this architecture, and create a GitHub workflow that ensures developers code respects the architectural decisions made by the architects.

A brief introduction to Hexagonal Architecture:

Hexagonal Architecture, also known as the Ports and Adapters design pattern, has a main goal of separating domain logic from external dependencies. The following image illustrates the basic components of the Hexagonal Architecture. In this section, we will dig deeper into each part of the pattern and provide Python code examples.

Components of Hexagonal architecture (aka ports and adapters pattern)

Ports

Ports are the contracts the application settles for the interaction with the external actors. There as two type of ports: Driver ports define how external components or actors interact with the application. Driven ports define how the application will interact with the external actors.

Adapters

Adapters are components that connect the core application to external systems or services. They allow the core domain logic to remain focused on business rules, while the technical details are handled separately at the edges of the application.

Domain Logic

The domain layer contains the business logic and must run without any dependency of external code. We can test the entire HA application without setting up external component like DBs..

Configurator

As we will see later the domain logic use Interfaces which are not concrete implementation, so to be able to run our application we need to wire everything together, this is why we need the configurator. An example of a configurator could be a FastAPI or a Flask application setup code.

Hexagonal Architecture implementation in Python

As I mentioned earlier, there are many resources on the Internet that discuss this architecture, but they do not provide enough code examples. This is why many developers cannot fully understand it. In this section I will try to implement it in a pythonic way, if you have any concerns, questions or suggestions please send us an email to contact@workflows.guru

We will implement a task scheduler application that we will call tscheduler. The application allow the user to create a task, read a task and execute a task. The full code is available in github at https://github.com/workflows-guru/hexagonal-architecture

Let's start by defining our ports. We should have a storage system to store the tasks, so the domain logic will communicate with this storage system using a port. Let's call this port StorageInterface. This interface has two abstract methods that must be implemented inside the storage adapters.

1from abc import ABC, abstractmethod
2
3class StorageInterface(ABC):
4    """Storage interface that should be implemented by our storage systems"""
5
6    @abstractmethod
7    def create_task(self, name: str, description: str, data: str) -> dict:
8        raise NotImplementedError
9
10    @abstractmethod
11    def read_task(self, name) -> dict | None:
12        raise NotImplementedError

To be able to use a storage system in our application, we should create an adapters that implement this interface. This LocalStorage class implement the StorageInterface.

1from tscheduler.ports.storage import StorageInterface
2
3class LocalStorage(StorageInterface):
4    def __init__(self):
5        self.tasks = []
6
7    def create_task(self, name: str, description: str, data: str) -> dict:
8        task = {"name": name, "description": description, "data": data}
9        self.tasks.append(task)
10        return task
11
12    def read_task(self, name: str) -> dict | None:
13        for item in self.tasks:
14            if item.get("name") == name:
15                return item

To execute the tasks we need a runner or executor. So let's create another port for this external component and let's call it ExecutorInterface.

1from abc import ABC, abstractmethod
2
3class ExecutorInterface(ABC):
4    """Executor interface that should be implemented by our executors"""
5
6    @abstractmethod
7    def execute_task(self, data: str) -> str:
8        raise NotImplementedError

A possible adapter for this port could be a celery or airflow executor. In our example we will just implement a local executor that will print the task data and return a string as follow:

1import uuid
2
3from tscheduler.ports.executor import ExecutorInterface
4
5class LocalExecutor(ExecutorInterface):
6
7    def execute_task(self, data: str) -> str:
8          print("Executing:", data)
9          execution_id = uuid.uuid4().hex
10          return execution_id

We have created two ports and their implementation in the adapters. Let's now see how we will use this in our business logic.

1from tscheduler.domain.models import Task
2from tscheduler.ports.executor import ExecutorInterface
3from tscheduler.ports.storage import StorageInterface
4
5class TaskController:
6    # Here we use Constructor injection we can as well use method injection
7    def __init__(
8        self,
9        storage: StorageInterface | None = None,
10        executor: ExecutorInterface | None = None,
11    ):
12        self.storage = storage
13        self.executor = executor
14
15    def execute_task(self, name: str) -> str:
16        task = self.storage.read_task(name=name)
17        task = Task(**task)
18
19        # Our business logic is implemented here
20
21        # For demonstration purpose our executor accept any str and return any str
22        execution_id = self.executor.execute_task(data=task.data)
23        return execution_id
24

As you can see, the domain code does not depends on external component, instead it use the interfaces defined in ports.

Let's now wire everything together and run our application.

1
2from tscheduler.adapters.local_executor import LocalExecutor
3from tscheduler.adapters.local_storage import LocalStorage
4from tscheduler.domain.core import TaskController
5
6controller = TaskController(storage=LocalStorage(), executor=LocalExecutor())
7execution_id = controller.execute_task(name="my_task")
8

This script show how to execute a task, but you must first create the task. In the gihtub repository you will find an example on how to set up the application using FastAPI.

Testing the application is similar to this simple script, we only need to setup our application with the test class. For example: InMemoryStorage and LocalExecutor

Validate Hexagonal Architecture with fitness functions

Fitness Functions are used to ensure that a system's architecture continuously evolves in a controlled manner.

These functions evaluates whether an architecture is meeting its desired characteristics and help teams continuously validate architectural decisions rather than relying on manual PR reviews.

In this section, we will create fitness functions and add these functions to our CI workflow. The application code is hosted on github, so we will use github workflow to automate our fitness functions.

The are some core principales that we must follow when implementing this architecture. Some of these principales are:

Decouple Business Logic from External Concerns: We should not mix business logic with infrastructure concerns
Use Ports: We should not expose domain models to external layers
Respect Dependency Flow: We should not allow the domain layer to depend on external services

To check that we don't breaks decoupling, we will implement a fitness function to ensure that we don't add forbidden dependencies in our code. The following test will fail if the domain core depends on adapters repository.

1import ast
2
3def is_dependant(source_file: str, destination: str):
4    with open(source_file, "r") as f:
5        source = ast.parse(f.read())
6
7        for statement in source.body:
8            if isinstance(statement, ast.Import) or isinstance(
9                statement, ast.ImportFrom
10            ):
11                if hasattr(statement, "module") and destination in statement.module:
12                    return True
13                if any(name for name in statement.names if destination in name.name):
14                    return True
15
16    return False
17
18
19def test_domain_core_independant_of_all_adapters():
20    assert is_dependant("./tscheduler/domain/core.py", "adapters") is False
21

Feel free to add more functions to check other principales.

We have created our fitness functions. We will now add these functions to our GitHub workflow. In your .github/workflows directory, create a YAML file and add the following job to your pipeline

1
2jobs:
3  architecture:
4    runs-on: ubuntu-latest
5
6    steps:
7    - uses: actions/checkout@v4
8    - name: Set up Python 3.11
9      uses: actions/setup-python@v3
10      with:
11        python-version: "3.11"
12    - name: Install pytest
13      run: |
14        python -m pip install --upgrade pip
15        pip install pytest
16    - name: Fitness functions to validate Hexagonal Architecture
17      run: |
18        pytest tests/test_check_architecture.py
19

Conclusion

In this course, we introduced Hexagonal Architecture, created a demonstration application, and added fitness functions to our CI workflow to validate the architecture. Fitness functions are a great way to ensure that everything works as expected and to evaluate whether the architecture is meeting its desired characteristics.