Implementing Clean Architecture in Python

Clean Architecture is a variant of Alistair Cockburn's hexagonal architecture. The main idea is to separate the business logic from the infrastructure. Proposed by Robert C. Martin in 2012, it combines the principles of hexagonal architecture, onion architecture and other variants, defining more precisely the responsibility of each layer and how they should communicate with each other.

A project is divided into different layers:

• 🟡 Entities: Business variables/constants/classes/objects. The core of the application. For example, User, Product, Invoice, etc.
• 🔴 Use Cases: The implementation of the business logic. Mainly functions. For example, the logic to create a new user, calculate the total of an invoice, etc.
• 🟢 Gateways: The interfaces that the Use Cases need to interact with the outside world. For example, the database interface, the file system interface, etc.
• 🔵 External Interfaces: The applications that interact with the outside world.
  • Database
  • Frameworks
  • ORM
  • File system
  • Devices
  • External APIs
  • UI
    • HTTP (website)
    • CLI (command line interface)
    • API (REST API)

The main general rule is that inner layers cannot depend on outer ones: dependencies always point inwards. When an inner layer needs something from the outside world (a database, an external API), it does not import it directly: it receives an interface provided by the outer layer, and it only returns simple structures (in Python we will use dictionaries).

⬆️ Calls come in through interfaces | ⬇️ Results go out as simple structures

And we will never break these rules. Even when an exception occurs, we will handle it and return a structure.

Advantages

Let's go over some of the advantages why we should implement clean architecture in our projects.

• Maintenance and modularity: The separations make it easy to make changes without affecting other parts of the code.
• Easier testing: Since all components are independent, we can test them in isolation. We know what we receive and what we must return, regardless of the interface or technology behind it.
• Code reuse: We can reuse use cases in different interfaces, or entities in different use cases.
• Technology isolation: We can change elements such as frameworks, databases, UIs, etc. without affecting the business logic.
• Scalability: We can add new features without affecting the existing ones.
• Documentation: Having a well-defined structure, the documentation will be easier to write.

All of this results in more robust, cleaner and easier to maintain code.

Disadvantages

Not everything is rosy. Clean architecture also has its drawbacks:

• Increases initial development time: Having to define the structure and the interfaces, the initial development time can be longer. Thorough upfront planning is required.
• Learning curve: At first it can be difficult to understand how the different layers communicate. The team must be trained and know the architecture.
• Abstraction overhead: Creating many interfaces and layers can result in abstraction overhead. Be careful, it is not necessary to create an interface for every function, only for those that interact with the outside world.
• File fragmentation: Having different layers, each in its own folder, can result in a considerable file tree.

However, with practice and training, these disadvantages can be mitigated.

Simple implementation example in Python

Let's look at an example of a feature that calculates the installation price of a turbine. We will need, from the user, the number of turbines to install. The price for each turbine installation will be a constant.

The folder structure will be as follows.

• mi_proyecto: The main folder. The project name.
  • core: Business logic.
    • entities
      • constants.py
    • use_cases
      • turbine
        • calculate_turbine_installation.py
  • infra: Infrastructure or external interfaces.
    • cli
      • click
        • src
    • api
      • flask
        • src
      • fastapi
        • src
    • http
      • django

As we said before, the installation price will be a constant. We will create a constants.py file in the entities folder.

# mi_proyecto/core/entities/constants.py
INSTALLATION_PRICE = 1250

Business classes and objects also live in entities. In this article the constant is enough, but if the project grew, the turbine would have its own entity.

# mi_proyecto/core/entities/turbine.py
from dataclasses import dataclass


@dataclass
class Turbine:
    model: str
    power_kw: float

The use case will be in the use_cases folder.

# mi_proyecto/core/use_cases/turbine/calculate_turbine_installation.py
from mi_proyecto.core.entities.constants import INSTALLATION_PRICE


def calculate_turbine_cost_use_case(number_of_turbines: int) -> dict:
    total = number_of_turbines * INSTALLATION_PRICE
    return {"total": total}

The first external interface will be a REST API. In the example we will use Flask for simplicity.

We can have different implementations in different frameworks. Therefore, each repository (not to be confused with a versioning repository like Git) will have its own folder. In this situation, we will have a folder for Flask inside the api folder.

mi_proyecto/infra/api/flask/

We use the following code. We will get the input through a parameter present in the URL.

# mi_proyecto/infra/api/flask/src/app.py
from flask import Flask, jsonify
from mi_proyecto.core.use_cases.turbine.calculate_turbine_installation import (
    calculate_turbine_cost_use_case,
)

app = Flask(__name__)


@app.route("/calculate-turbine-cost/<int:number_of_turbines>", methods=["GET"])
def calculate_turbine_cost(number_of_turbines):
    result = calculate_turbine_cost_use_case(number_of_turbines)
    return jsonify(result)


if __name__ == "__main__":
    app.run()

Remember: the main idea is to keep the business logic separated from the external interfaces.

Now a new need appears: creating an API to interact with a mobile application. We are asked to use FastAPI for it.

Inside the api folder of infra we will create another one named fastapi.

mi_proyecto/infra/api/fastapi/

# mi_proyecto/infra/api/fastapi/src/main.py

from fastapi import FastAPI
from mi_proyecto.core.use_cases.turbine.calculate_turbine_installation import (
    calculate_turbine_cost_use_case,
)

app = FastAPI()


@app.post("/calculate-turbine-cost")
def calculate_turbine_cost(number_of_turbines: int):
    return calculate_turbine_cost_use_case(number_of_turbines)

Next, we are asked for a web page that shows the result in HTML to a human. We will use Django. Since it is a website, its folder will be in http.

mi_proyecto/infra/http/django/

# mi_proyecto/infra/http/django/views.py
from django.shortcuts import render
from mi_proyecto.core.use_cases.turbine.calculate_turbine_installation import (
    calculate_turbine_cost_use_case,
)


def calculate_turbine_cost(request, number_of_turbines: int):
    result = calculate_turbine_cost_use_case(number_of_turbines)
    return render(request, "turbine.html", {"total": result["total"]})

<!-- mi_proyecto/infra/http/django/templates/turbine.html -->
<h1>Total: {{ total }} €</h1>

# mi_proyecto/infra/http/django/urls.py
from django.urls import path

from .views import calculate_turbine_cost

urlpatterns = [
    path("calculate-turbine-cost/<int:number_of_turbines>/", calculate_turbine_cost),
]

Finally, why not a terminal client? We will use Click. Since it is a CLI, its folder will be in cli.

mi_proyecto/infra/cli/click/

# mi_proyecto/infra/cli/click/src/main.py
import click
from mi_proyecto.core.use_cases.turbine.calculate_turbine_installation import (
    calculate_turbine_cost_use_case,
)


@click.command()
@click.option("--number_of_turbines", type=int, required=True)
def calculate_turbine_cost(number_of_turbines):
    result = calculate_turbine_cost_use_case(number_of_turbines)
    click.echo(result["total"])


if __name__ == "__main__":
    calculate_turbine_cost()

Seen in a diagram, the four interfaces converge on the same point:

flowchart LR
    subgraph infra["infra (external interfaces)"]
        F["API with Flask"]
        FA["API with FastAPI"]
        D["Web with Django"]
        C["CLI with Click"]
    end
    subgraph core["core (business logic)"]
        U["calculate_turbine_cost_use_case"]
    end
    F --> U
    FA --> U
    D --> U
    C --> U

Now we have 4 different interfaces for the business logic: an API with Flask, another API with FastAPI, an HTML website with Django and a CLI with Click. We can change the user interface without touching the heart of the application. Impressive, isn't it?

Another implementation example in Python with several external interfaces

Let's look at the previous example with a new need. The price will no longer come from a constant alone. Now the calculation needs 2 pieces of data that live in the outside world.

• tax. We will get it from a public API. We will use Spain as an example.
• Average salary of all our workers. We will get it from a database.

This creates a problem. The use case needs to talk to an API and to a database, but remember the rule: inner layers cannot depend on outer ones. How do we solve it? With dependency inversion. The use case will receive those pieces from the outside, as interfaces, without knowing what technology is behind them.

We will restructure the use case. Now we will use the number of turbines, the tax percentage and the average salary of the workers.

# mi_proyecto/core/use_cases/turbine/calculate_turbine_installation.py
from mi_proyecto.core.entities.constants import INSTALLATION_PRICE


def calculate_turbine_cost_use_case(
    number_of_turbines: int, tax: float, average_salary: float
) -> dict:
    total = number_of_turbines * INSTALLATION_PRICE
    total += total * tax
    total += average_salary
    return {"total": total}

We will create a new folder inside the infra folder named gateways.

mi_proyecto/infra/gateways/

Let's create the interface for the tax API. We will use the requests library to make the request, calling a fictional API that returns a tax filtered by country.

# mi_proyecto/infra/gateways/tax.py

import requests


def get_tax() -> float:
    response = requests.get("https://tax.com/", params={"country": "Spain"}, timeout=10)
    return response.json()["tax"]

The next step is to create the interface for the database. We will use SQLite for the example.

# mi_proyecto/infra/database/sqlite_repo.py
import sqlite3
from typing import Any


class SQLiteRepo:
    def __init__(self, connection_string: str):
        self.connection = sqlite3.connect(connection_string)
        # Rows will behave like dictionaries
        self.connection.row_factory = sqlite3.Row

    def fetch_one(self, table: str, key: str, columns: list[str] | None = None) -> Any:
        """
        Fetch one row from a table

        :param table: The table name
        :param key: The key to fetch
        :param columns: The columns to fetch. If None, fetch all columns
        :return: A row
        """
        columns_str = "*" if columns is None else ",".join(columns)
        return self.connection.execute(
            f"SELECT {columns_str} FROM {table} WHERE key = ?", (key,)
        ).fetchone()

    def fetch_all(self, table: str, columns: list[str] | None = None) -> list[Any]:
        """
        Fetch all rows from a table

        :param table: The table name
        :param columns: The columns to fetch. If None, fetch all columns
        :return: A list of rows
        """
        columns_str = "*" if columns is None else ",".join(columns)
        return self.connection.execute(f"SELECT {columns_str} FROM {table}").fetchall()

Two details about the repository: we enable sqlite3.Row so that each row behaves like a dictionary, and the value of key is passed as a query parameter, never interpolated into the SQL, to avoid injections.

Now we will modify the use case to use the interfaces.

# mi_proyecto/core/use_cases/turbine/calculate_turbine_installation.py
from mi_proyecto.core.entities.constants import INSTALLATION_PRICE


def calculate_turbine_cost_use_case(repo, get_tax, number_of_turbines: int) -> dict:
    tax = get_tax()
    salaries = repo.fetch_all("workers", ["salary"])
    average_salary = sum(row["salary"] for row in salaries) / len(salaries)
    total = number_of_turbines * INSTALLATION_PRICE
    total += total * tax
    total += average_salary
    return {"total": total}

We have added two new parameters to the use case: repo, the database interface, and get_tax, the tax API interface. Notice that the use case no longer imports anything from infra: it receives its dependencies from the outside. This is known as dependency inversion, and it is what will allow us to change the database or the tax provider without affecting the business logic.

You may wonder where the interface is defined. In Python there is no need to declare it formally: any object with a fetch_all method works as a repository (duck typing). If you prefer an explicit contract, you can define a Protocol inside core and use it as the type of the repo parameter. The infra implementations will fulfill it without inheriting from anything.

# mi_proyecto/core/gateways/repository.py
from typing import Any, Protocol


class Repository(Protocol):
    def fetch_one(
        self, table: str, key: str, columns: list[str] | None = None
    ) -> Any: ...

    def fetch_all(self, table: str, columns: list[str] | None = None) -> list[Any]: ...

When we call the use case from an interface, we will pass it the database and tax API interfaces.

# mi_proyecto/infra/api/flask/src/app.py
import os

from flask import Flask, jsonify
from mi_proyecto.core.use_cases.turbine.calculate_turbine_installation import (
    calculate_turbine_cost_use_case,
)
from mi_proyecto.infra.database.sqlite_repo import SQLiteRepo
from mi_proyecto.infra.gateways.tax import get_tax

app = Flask(__name__)


@app.route("/calculate-turbine-cost/<int:number_of_turbines>", methods=["GET"])
def calculate_turbine_cost(number_of_turbines):
    repo = SQLiteRepo(os.environ.get("CONNECTION_STRING"))
    result = calculate_turbine_cost_use_case(repo, get_tax, number_of_turbines)
    return jsonify(result)

And this is how the terminal client with Click from the first example evolves: only what we inject changes.

# mi_proyecto/infra/cli/click/src/main.py
import os

import click
from mi_proyecto.core.use_cases.turbine.calculate_turbine_installation import (
    calculate_turbine_cost_use_case,
)
from mi_proyecto.infra.database.sqlite_repo import SQLiteRepo
from mi_proyecto.infra.gateways.tax import get_tax


@click.command()
@click.option("--number_of_turbines", type=int)
def calculate_turbine_cost(number_of_turbines):
    repo = SQLiteRepo(os.environ.get("CONNECTION_STRING"))
    result = calculate_turbine_cost_use_case(repo, get_tax, number_of_turbines)
    click.echo(result)


if __name__ == "__main__":
    calculate_turbine_cost()

We do not break the rule: the inner layers still know nothing about the outer ones. The use case receives its dependencies as interfaces and returns simple structures.

This is the complete journey of a request, from the outside in and back:

sequenceDiagram
    autonumber
    participant U as User
    participant F as Flask (external interface)
    participant C as Use case (core)
    participant T as Tax gateway (infra)
    participant R as SQLite repository (infra)

    U->>F: GET /calculate-turbine-cost/10
    F->>C: calculate_turbine_cost_use_case(repo, get_tax, 10)
    C->>T: get_tax()
    T-->>C: 0.21
    C->>R: fetch_all("workers", ["salary"])
    R-->>C: rows with the salaries
    C-->>F: {"total": 16625.0}
    F-->>U: JSON with the total

Notice that the call arrows always point inwards and that the use case only talks to repo and get_tax: it does not know that behind them there is SQLite and a public API, nor does it care. And what comes back out is always a simple structure, a dictionary.

Now it turns out that we must replace the repository with a JSON file on disk. We will only have to change the implementation of the interface.

# mi_proyecto/infra/database/json_repo.py
import json
from typing import Any


class JSONRepo:
    def __init__(self, file_path: str):
        self.file_path = file_path

    def fetch_one(self, table: str, key: str, columns: list[str] | None = None) -> Any:
        """
        Fetch one row from a table

        :param table: The table name
        :param key: The key to fetch
        :param columns: The columns to fetch. If None, fetch all columns
        :return: A row
        """
        with open(self.file_path, "r") as file:
            data = json.load(file)
            return data[table][key]

    def fetch_all(self, table: str, columns: list[str] | None = None) -> list[Any]:
        """
        Fetch all rows from a table

        :param table: The table name
        :param columns: The columns to fetch. If None, fetch all columns
        :return: A list of rows
        """
        with open(self.file_path, "r") as file:
            data = json.load(file)
            return list(data[table].values())

The use case does not change at all. We only change the implementation that we inject from the external interface.

# mi_proyecto/infra/api/flask/src/app.py
import os

from flask import Flask, jsonify
from mi_proyecto.core.use_cases.turbine.calculate_turbine_installation import (
    calculate_turbine_cost_use_case,
)
from mi_proyecto.infra.database.json_repo import JSONRepo  # New
from mi_proyecto.infra.gateways.tax import get_tax

app = Flask(__name__)


@app.route("/calculate-turbine-cost/<int:number_of_turbines>", methods=["GET"])
def calculate_turbine_cost(number_of_turbines):
    repo = JSONRepo(os.environ.get("JSON_DB_PATH"))  # Change
    result = calculate_turbine_cost_use_case(repo, get_tax, number_of_turbines)
    return jsonify(result)

We have only modified the database implementation, without touching a single line of the business logic.

Error handling

We have been working with controlled cases, where we receive perfectly structured data. Reality is messier.

In clean architecture we cannot return Python exceptions, besides the fact that we would break the rule of only returning dictionaries. For this we will slightly modify the return structure.

If everything went well, we will return a dictionary with the type key set to Success. If there was an error, we will return a dictionary with the type key set to Error and a dictionary with the errors.

For example:

calculate_turbine_cost_use_case(repo, get_tax, 10)
"""
{
      "type": "Success",
      "errors": [],
      "data": {
            "total": 16625.0
      }
}
"""

In the error case.

calculate_turbine_cost_use_case(repo, get_tax, "foo")
"""
{
      "type": "ParametersError",
      "errors": [
            {
                  "field": "number_of_turbines",
                  "message": "Number of turbines is required"
            }
      ],
      "data": {}
}
"""

To restrict the error types, we will create a class with the error types.

# mi_proyecto/core/use_cases/turbine/calculate_turbine_installation.py
from mi_proyecto.core.entities.constants import INSTALLATION_PRICE


class ResponseTypes:
    SUCCESS = "Success"  # The process ended correctly
    PARAMETERS_ERROR = "ParametersError"  # Missing or invalid parameters
    RESOURCE_ERROR = "ResourceError"  # The process ended correctly but the resource is not available (DB, file, etc)
    SYSTEM_ERROR = "SystemError"  # The process ended with an error. Python error


def calculate_turbine_cost_use_case(repo, get_tax, number_of_turbines: int) -> dict:
    # Check if the number of turbines is present and valid
    if not isinstance(number_of_turbines, int) or number_of_turbines <= 0:
        return {
            "type": ResponseTypes.PARAMETERS_ERROR,
            "errors": [
                {
                    "field": "number_of_turbines",
                    "message": "Number of turbines is required",
                }
            ],
            "data": {},
        }
    # Logic
    try:
        tax = get_tax()
        salaries = repo.fetch_all("workers", ["salary"])
    except Exception as error:
        return {
            "type": ResponseTypes.SYSTEM_ERROR,
            "errors": [{"field": "system", "message": str(error)}],
            "data": {},
        }
    if not salaries:
        return {
            "type": ResponseTypes.RESOURCE_ERROR,
            "errors": [{"field": "salaries", "message": "Salaries not found"}],
            "data": {},
        }
    average_salary = sum(row["salary"] for row in salaries) / len(salaries)
    total = number_of_turbines * INSTALLATION_PRICE
    total += total * tax
    total += average_salary
    return {"type": ResponseTypes.SUCCESS, "errors": [], "data": {"total": total}}

If the input is wrong, or we cannot get the data we need from the database, we will return a dictionary with the error type and the errors. And if any Python exception occurs (the tax API does not respond, the database is corrupted), we catch it and turn it into a SystemError structure: the exception never leaves the use case. Otherwise we will return a dictionary with the Success type and the data.

If we implement it in an external interface such as an API, we will return a status code depending on each error type.

• 200 for Success.
• 400 for ParametersError.
• 500 for SystemError.
• 503 for ResourceError.

We could create a decorator for that purpose. Before returning the result, it will check the type returned by the use case and modify the status code before responding.

If we were working with FastAPI, a simple implementation would be the following.

from functools import wraps

from fastapi import FastAPI, Response, status

app = FastAPI()


class ResponseTypes:
    # The process ended correctly. HTTP 200
    SUCCESS = "Success"
    # Missing or invalid parameters. HTTP 400
    PARAMETERS_ERROR = "ParametersError"
    # The process ended with an error. Python error. HTTP 500
    SYSTEM_ERROR = "SystemError"
    # The process ended correctly but the resource is not available (DB, file, etc). HTTP 503
    RESOURCE_ERROR = "ResourceError"


def correct_http_code(func):
    """
    Adjust the HTTP code based on the response type
    """

    @wraps(func)
    async def wrapper(*args, **kwargs):
        response = kwargs.get("response")
        output = await func(*args, **kwargs)
        status_type = output.get("type")
        response.status_code = status.HTTP_200_OK
        if status_type == ResponseTypes.PARAMETERS_ERROR:
            response.status_code = status.HTTP_400_BAD_REQUEST
        elif status_type == ResponseTypes.RESOURCE_ERROR:
            response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE
        elif status_type == ResponseTypes.SYSTEM_ERROR:
            response.status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
        return output

    return wrapper


def calculate_turbine_cost_use_case(number_of_turbines: int) -> dict:
    pass


@app.post("/api/calculate-turbine-cost")
@correct_http_code
async def calculate_turbine_cost(response: Response, payload: dict | None = None):
    return calculate_turbine_cost_use_case((payload or {}).get("number_of_turbines"))

There is a problem when sending or receiving data. Since we use JSON to communicate, we will receive and send Camel Case, but in Python we use Snake Case. We can configure FastAPI to do the conversion automatically.

Type and structure validation

We continue digging into the possible errors. The data we may receive in the use cases can be wild, with inappropriate structures and types. To reduce the complexity we can use a widely known library in the Python ecosystem called pydantic. It is a library for validating types. It would be a good idea to automate the boring task with a decorator in charge of returning the errors in the format we are using in the architecture.

from functools import wraps

from pydantic import ValidationError


class ResponseTypes:
    SUCCESS = "Success"  # The process ended correctly
    PARAMETERS_ERROR = "ParametersError"  # Missing or invalid parameters
    RESOURCE_ERROR = "ResourceError"  # The process ended correctly but the resource is not available (DB, file, etc)
    SYSTEM_ERROR = "SystemError"  # The process ended with an error. Python error


def check_params(Model):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            params = kwargs.get("params", None)
            if params is not None:
                try:
                    Model.model_validate(params, strict=True)
                except ValidationError as e:
                    errors = []
                    for error in e.errors():
                        errors.append(
                            {
                                "field": error["loc"][0],
                                "message": error["msg"],
                            }
                        )
                    return {
                        "type": ResponseTypes.PARAMETERS_ERROR,
                        "errors": errors,
                        "data": {},
                    }
            return func(*args, **kwargs)

        return wrapper

    return decorator

We define the Pydantic model, the structure we are looking to receive. As an example, we will receive a user's information.

from pydantic import BaseModel


class UserInfoModel(BaseModel):
    id: int
    name: str
    is_active: bool
    weight: float
    favorites: list[int]

To use it, we just add the decorator to the use case along with the previous model.

@check_params(UserInfoModel)
def set_user_info(params):
    return {
        "type": ResponseTypes.SUCCESS,
        "errors": [],
        "data": {},
    }

We can now receive data.

inputExample = {
    "id": 1,
    "name": "John Doe",
    "is_active": True,
    "weight": 75.5,
    "favorites": [1, 2, 3],
}

set_user_info(params=inputExample)
# {'type': 'Success', 'errors': [], 'data': {}}

If the data is not correct, we will return a dictionary with the error type and the errors.

inputExample = {
    "id": False,
    "name": 23,
    "is_active": True,
    "weight": 75.5,
    "favorites": [1, 2, 3],
}

set_user_info(params=inputExample)
# {'type': 'ParametersError', 'errors': [{'field': 'id', 'message': 'Input should be a valid integer'}, {'field': 'name', 'message': 'Input should be a valid string'}], 'data': {}}

It is important that when using set_user_info, we pass the params parameter so that the decorator can validate the data.

Now our use cases are protected from incorrect data and the validation process is automated.

Testing

What should we test? We actually answered this question at the beginning, and I am sure you see it more clearly after reading the previous example. We must test the use cases, not the rest of the layers (the API in this case). FastAPI is just an interface that receives data and sends it to the use cases; it does not care about the structure of the received dictionary (JSON), whether it is correct or not. Validating is the responsibility of the use cases, in addition to giving us the list of errors if there are any. The only thing we could check is whether the API returns the correct status code.

Let's see how this advantage materializes with pytest. Thanks to dependency inversion we need neither a real database nor a real tax API: we create fake implementations in a couple of lines and inject them.

# tests/use_cases/test_calculate_turbine_installation.py
from mi_proyecto.core.use_cases.turbine.calculate_turbine_installation import (
    ResponseTypes,
    calculate_turbine_cost_use_case,
)


class FakeRepo:
    def fetch_all(self, table, columns=None):
        return [{"salary": 1000}, {"salary": 2000}]


def fake_get_tax() -> float:
    return 0.21


def test_calculate_turbine_cost_returns_the_total():
    # Given
    repo = FakeRepo()
    number_of_turbines = 10

    # When
    result = calculate_turbine_cost_use_case(repo, fake_get_tax, number_of_turbines)

    # Then
    assert result["type"] == ResponseTypes.SUCCESS
    # 10 * 1250 = 12500, plus 21% tax = 15125, plus 1500 of average salary
    assert result["data"]["total"] == 16625.0


def test_calculate_turbine_cost_requires_a_valid_number():
    # Given
    repo = FakeRepo()
    number_of_turbines = "foo"

    # When
    result = calculate_turbine_cost_use_case(repo, fake_get_tax, number_of_turbines)

    # Then
    assert result["type"] == ResponseTypes.PARAMETERS_ERROR
    assert result["errors"][0]["field"] == "number_of_turbines"

Notice that the test knows nothing about Flask, FastAPI, SQLite or requests. It only knows the use case, its injected dependencies and the dictionary it returns.

Final notes

Only the business logic has been modified, not the external interfaces. Therefore the APIs and the CLI remain the same, without changes. That is the magic of clean architecture.

I hope this article has helped you understand how to implement clean architecture and where to start implementing it in Python. Even though the examples are simple, you will be able to apply it to bigger projects. My advice is not to stop here. Keep learning by reading books specialized in design patterns and software architecture, and practice a lot to apply the concepts.