Implementando arquitectura limpia en Python

La Arquitectura Limpia es una variante de la arquitectura hexagonal de Alistair Cockburn. La idea principal es separar la lógica de negocio de la infraestructura. Propuesta por Robert C. Martin en 2012, combina los principios de la arquitectura hexagonal, la arquitectura cebolla y otras variantes, definiendo con más precisión la responsabilidad de cada capa y cómo deben comunicarse entre ellas.

Un proyecto se divide en diferentes capas:

Arquitectura limpia

🟡 Entities: Variables/Constantes/Clases/Objetos del negocio. El núcleo de la aplicación. Por ejemplo, Usuario, Producto, Factura, etc.
🔴 Use Cases: La implementación de las lógicas de negocio. Principalmente funciones. Por ejemplo, la lógica para crear un nuevo usuario, calcular el total de una factura, etc.
🟢 Gateways: Las interfaces que los Casos de Uso necesitan para interactuar con el mundo externo. Por ejemplo, la interfaz de la base de datos, la interfaz del sistema de archivos, etc.
🔵 External Interfaces: Las aplicaciones que interactúan con el mundo exterior.
- Base de datos
- Frameworks
- ORM
- Sistema de archivos
- Dispositivos
- API externas
- UI
  - HTTP (sitio web)
  - CLI (interfaz de línea de comando)
  - API (API REST)

La regla general principal es que las capas interiores no pueden depender de las exteriores: las dependencias siempre apuntan hacia dentro. Cuando una capa interior necesita algo del mundo exterior (una base de datos, una API externa), no lo importa directamente: recibe una interfaz que le proporciona la capa exterior, y solo devuelve estructuras simples (en Python usaremos diccionarios).

⬆️ Las llamadas entran a través de interfaces | ⬇️ Los resultados salen como estructuras simples

Y estas normas nunca las romperemos. Incluso cuando ocurre una excepción, la trataremos y devolveremos una estructura.

Ventajas

Vamos a repasar algunas de las ventajas por las cuales deberíamos implementar la arquitectura limpia en nuestros proyectos.

Mantenimiento y modularidad: Las separaciones aportan facilidad a la hora de realizar cambios sin que afecten a otras partes del código.
Facilita el testeo: Al ser todos los componentes independientes, podemos testearlos de forma aislada. Sabemos qué recibimos y qué debemos devolver, independientemente de la interfaz o tecnología que exista detrás.
Reutilización de código: Podemos reutilizar los casos de uso en diferentes interfaces o entidades en diferentes casos de uso.
Aislamiento de tecnología: Podemos cambiar elementos como frameworks, bases de datos, UIs, etc. sin afectar a la lógica de negocio.
Escalabilidad: Podemos añadir nuevas funcionalidades sin afectar a las ya existentes.
Documentación: Al tener una estructura delimitada, la documentación será más sencilla de redactar.

Todo ello desemboca en un código más robusto, limpio y fácil de mantener.

Desventajas

No todo es color de rosa. La arquitectura limpia también tiene sus inconvenientes:

Incrementa el tiempo inicial de desarrollo: Al tener que definir la estructura y las interfaces, el tiempo de desarrollo inicial puede ser mayor. Se requiere una concienzuda planificación previa.
Curva de aprendizaje: Al principio puede resultar complicado entender cómo se comunican las diferentes capas. El equipo debe estar formado y conocer la arquitectura.
Sobrecarga de abstracción: La creación de muchas interfaces y capas puede resultar en una sobrecarga de abstracción. Hay que tener cuidado, no es necesario crear una interfaz para cada función, solo para las que interactúan con el mundo exterior.
Fragmentación de ficheros: Al tener diferentes capas, cada una en su carpeta, puede resultar en un árbol de ficheros considerable.

Sin embargo, con la práctica y la formación, estas desventajas se pueden mitigar.

Ejemplo sencillo de implementación en Python

Veamos un ejemplo de una funcionalidad que calcula el precio de instalación de una turbina. Necesitaremos, del usuario, el número de turbinas a instalar. El precio por cada instalación de turbina será una constante.

La estructura de carpetas será la siguiente.

mi_proyecto: La carpeta principal. El nombre del proyecto.
- core: Lógica de negocio.
  - entities
    - constants.py
  - use_cases
    - turbine
      - calculate_turbine_installation.py
- infra: Infraestructura o interfaces externas.
  - cli
    - click
      - src
  - api
    - flask
      - src
    - fastapi
      - src
  - http
    - django

Como hemos adelantado, el precio de la instalación será una constante. Crearemos un archivo constants.py en la carpeta entities.

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

En entities también viven las clases y objetos del negocio. En este artículo nos basta con la constante, pero si el proyecto creciera, la turbina tendría su propia entidad.

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


@dataclass
class Turbine:
    model: str
    power_kw: float

El caso de uso estará en la carpeta use_cases.

# 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}

La primera interfaz externa será una API REST. En el ejemplo usaremos Flask por simplicidad.

Podemos tener diferentes implementaciones en diferentes frameworks. Por tanto, cada repositorio (no confundir con un repositorio de versionado como Git) tendrá su propia carpeta. En esta situación, tendremos una carpeta para Flask en la carpeta api.

mi_proyecto/infra/api/flask/

Usamos el siguiente código. El input lo obtendremos a través de un parámetro presente en la 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()

Recuerda: la idea principal es mantener la lógica de negocio separada de las interfaces externas.

Ahora aparece una nueva necesidad: crear una API para interactuar con una aplicación móvil. Nos piden que usemos FastAPI para ello.

Dentro de la carpeta api de infra crearemos otra con el nombre 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)

Después nos piden una página web que muestre el resultado en HTML a un humano. Usaremos Django. Al tratarse de un sitio web, su carpeta estará en 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),
]

Por último, ¿por qué no un cliente de terminal? Usaremos Click. Al tratarse de una CLI, su carpeta estará en 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()

Visto en un diagrama, las cuatro interfaces convergen en el mismo punto:

flowchart LR
    subgraph infra["infra (interfaces externas)"]
        F["API con Flask"]
        FA["API con FastAPI"]
        D["Web con Django"]
        C["CLI con Click"]
    end
    subgraph core["core (lógica de negocio)"]
        U["calculate_turbine_cost_use_case"]
    end
    F --> U
    FA --> U
    D --> U
    C --> U

Ahora tenemos 4 interfaces diferentes para la lógica de negocio: una API con Flask, otra API con FastAPI, una web en HTML con Django y una CLI con Click. Podemos cambiar la interfaz de usuario sin tocar el corazón de la aplicación. Impresionante, ¿no?

Otro ejemplo de implementación en Python con diversas interfaces externas

Veamos el ejemplo anterior con una nueva necesidad. El precio ya no saldrá solo de una constante. Ahora el cálculo necesita 2 datos que viven en el mundo exterior.

tax (impuesto). Lo obtendremos de una API pública. Usaremos España como ejemplo.
Average salary o salario medio de todos nuestros trabajadores. Lo obtendremos de una base de datos.

Esto nos genera un problema. El caso de uso necesita hablar con una API y con una base de datos, pero recuerda la regla: las capas interiores no pueden depender de las exteriores. ¿Cómo lo solucionamos? Con inversión de dependencias. El caso de uso recibirá esas piezas desde fuera, como interfaces, sin saber qué tecnología hay detrás.

Reestructuraremos el caso de uso. Ahora usaremos el número de turbinas, el porcentaje de impuesto y el salario medio de los trabajadores.

# 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}

Crearemos una nueva carpeta en la carpeta infra con el nombre gateways.

mi_proyecto/infra/gateways/

Creemos la interfaz para la API de impuestos. Usaremos la biblioteca requests para hacer la solicitud y haremos una petición a una API ficticia que devolverá un impuesto filtrado por país.

# 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"]

El siguiente paso es crear la interfaz para la base de datos. Usaremos SQLite para el ejemplo.

# 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()

Dos detalles del repositorio: activamos sqlite3.Row para que cada fila se comporte como un diccionario, y el valor de key se pasa como parámetro de la consulta, nunca interpolado en el SQL, para evitar inyecciones.

Ahora modificaremos el caso de uso para usar las 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}

En el caso de uso hemos añadido dos nuevos parámetros: repo, la interfaz de la base de datos, y get_tax, la interfaz de la API de impuestos. Fíjate en que el caso de uso ya no importa nada de infra: recibe sus dependencias desde fuera. Esto se conoce como inversión de dependencias, y es lo que nos permitirá cambiar la base de datos o el proveedor de impuestos sin afectar a la lógica de negocio.

Quizá te preguntes dónde está definida la interfaz. En Python no hace falta declararla formalmente: cualquier objeto con un método fetch_all sirve como repositorio (duck typing). Si prefieres un contrato explícito, puedes definir un Protocol dentro de core y usarlo como tipo del parámetro repo. Las implementaciones de infra lo cumplirán sin necesidad de heredar de nada.

# 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]: ...

Cuando llamemos al caso de uso desde una interfaz, le pasaremos las interfaces de la base de datos y de la API de impuestos.

# 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)

Y así evoluciona el cliente de terminal con Click que creamos en el primer ejemplo: solo cambia lo que inyectamos.

# 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()

No rompemos la regla: las capas interiores siguen sin conocer a las exteriores. El caso de uso recibe sus dependencias como interfaces y devuelve estructuras simples.

Este es el viaje completo de una petición, de fuera hacia dentro y de vuelta:

sequenceDiagram
    autonumber
    participant U as Usuario
    participant F as Flask (interfaz externa)
    participant C as Caso de uso (core)
    participant T as Gateway de impuestos (infra)
    participant R as Repositorio SQLite (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: filas con los salarios
    C-->>F: {"total": 16625.0}
    F-->>U: JSON con el total

Fíjate en que las flechas de llamada siempre apuntan hacia dentro y que el caso de uso solo habla con repo y get_tax: no sabe que detrás hay SQLite y una API pública, ni le importa. Y lo que sale de vuelta es siempre una estructura simple, un diccionario.

Ahora se da el caso de que debemos cambiar el repositorio por un archivo JSON en disco. Solo tendremos que cambiar la implementación de la interfaz.

# 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())

El caso de uso no cambia en absoluto. Solo cambiamos la implementación que inyectamos desde la interfaz externa.

# 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  # Nuevo
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"))  # Cambio
    result = calculate_turbine_cost_use_case(repo, get_tax, number_of_turbines)
    return jsonify(result)

Solo hemos modificado la implementación de la base de datos, sin tocar ni una línea de la lógica de negocio.

Gestión de errores

Hemos estado trabajando con casos controlados, donde recibimos datos perfectamente estructurados. La realidad es más sucia.

En arquitectura limpia no podemos devolver excepciones de Python, además de que romperíamos la regla de solo devolver diccionarios. Para ello modificaremos ligeramente la estructura de retorno.

Si todo ha ido bien, devolveremos un diccionario con la clave type con el valor Success. Si ha habido un error, devolveremos un diccionario con la clave type con el valor Error y un diccionario con los errores.

Por ejemplo:

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

En el caso de error.

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

Para restringir los tipos de errores, crearemos una clase con los tipos de errores.

# 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}}

Si el input es incorrecto, o no podemos obtener de la base de datos el dato que necesitamos, devolveremos un diccionario con el tipo de error y los errores. Y si ocurre cualquier excepción de Python (la API de impuestos no responde, la base de datos está corrupta), la capturamos y la convertimos en una estructura SystemError: la excepción nunca sale del caso de uso. En caso contrario devolveremos un diccionario con el tipo Success y los datos.

En el caso de implementarlo en una interfaz externa como una API, devolveremos un código de estado dependiendo de cada tipo de error.

200 para Success.
400 para ParametersError.
500 para SystemError.
503 para ResourceError.

Podríamos crear un decorador para tal fin. Antes de devolver el resultado, revisará el type que nos devuelve el caso de uso y modificará el código de estado antes de responder.

Si estuviéramos trabajando en FastAPI, una sencilla implementación sería la siguiente.

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"))

Hay un problema a la hora de enviar o recibir datos. Al usar JSON para comunicarnos, recibiremos y enviaremos Camel Case pero en Python usamos Snake Case. Podemos configurar FastAPI para que haga la conversión automáticamente.

Validación de tipos y estructuras

Continuamos profundizando en los posibles errores. Los datos que podemos recibir en los casos de uso pueden ser salvajes, con estructuras y tipos inapropiados. Para reducir la complejidad podemos usar una librería ampliamente conocida en el ecosistema de Python llamada pydantic. Es una biblioteca para validar tipos. Sería buena idea automatizar la aburrida tarea con un decorador encargado de devolver los errores con el formato que estamos utilizando en la arquitectura.

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

Definimos el modelo de Pydantic, la estructura que estamos buscando recibir. Como ejemplo, recibiremos la información de un usuario.

from pydantic import BaseModel


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

Para usarlo, tan solo incorporaremos el decorador en el caso de uso y el modelo anterior.

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

Ya podemos recibir datos.

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': {}}

Si los datos no son correctos, devolveremos un diccionario con el tipo de error y los errores.

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': {}}

Es importante que al usar set_user_info, indiquemos el parámetro params para que el decorador pueda validar los datos.

Ahora nuestros casos de uso están protegidos de datos incorrectos y automatizado el proceso de validación.

Testeo

¿Qué debemos testear? En realidad esta pregunta la hemos respondido al principio, y estoy seguro de que lo ves con más claridad al leer el ejemplo anterior. Debemos testear los casos de uso, no el resto de capas (la API en este caso). FastAPI es solo una interfaz que recibe datos y los envía a los casos de uso, no le importa la estructura del diccionario recibido (JSON), si son correctos o no. Eso es responsabilidad de los casos de uso validar, además de darnos el listado de errores si los hubiera. Lo único que podríamos comprobar es si la API devuelve el código de estado correcto.

Veamos cómo se materializa esta ventaja con pytest. Gracias a la inversión de dependencias no necesitamos ni base de datos ni API de impuestos reales: creamos implementaciones falsas en un par de líneas y las inyectamos.

# 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"

Fíjate en que el test no sabe nada de Flask, FastAPI, SQLite o requests. Solo conoce el caso de uso, sus dependencias inyectadas y el diccionario que devuelve.

Apuntes finales

Solo se ha modificado la lógica de negocio, no las interfaces externas. Por lo tanto las APIs y el CLI se mantienen igual, sin cambios. Es la magia de la arquitectura limpia.

Espero que este artículo te haya ayudado a entender cómo implementar la arquitectura limpia y por dónde empezar para implementar en Python. Aunque los ejemplos sean sencillos, podrás aplicarlo en proyectos más grandes. Mi consejo es que no te quedes aquí. Sigue formándote leyendo libros especializados en patrones de diseño y arquitectura de software, y practicando mucho para aplicar los conceptos.