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.

Un proyecto se divide en diferentes capas:

• 🟡 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 interactuan 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 exteriores no pueden acceder directamente a las interiores. La comunicación se realiza a través de interfaces y estructuras. Las capas externas se comunican con las internas a través de interfaces, y las internas solo devuelven estructuras simples (en Python usaremos diccionarios).

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 que recibimos y que 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 como 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
      • constantes.py
    • use_cases
      • turbina
        • calcular_turbina_instalación.py
  • infra: Infraestructura o interfaces externas.
    • cli
      • click
        • src
    • api
      • fastapi
        • src
    • http
      • django
        • src

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

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 UI será una interfaz gráfica en HTML. 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 http.

mi_proyecto/infra/http/flask/

Usamos el siguiente código. El input lo obtendremos a través de un parámetro presente en la URL.

# mi_proyecto/infra/http/flask/src/app.py
from flask import Flask, request, jsonify
from 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()

Recuerde: la idea principal es mantener la lógica de negocio separado 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.

Crearemos una carpeta en infra con el nombre api, y dentro de ella otra con el nombre fastapi.

mi_proyecto/infra/api/fastapi/

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

from fastapi import FastAPI
from 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):
    result = calculate_turbine_cost_use_case(number_of_turbines)
    return {
        "total": result
    }

Ahora tenemos 2 interfaces diferentes para la lógica de negocio. 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 agregando 2 variables de interfaz externa.

• 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.

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 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/puertas de enlace/

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'})
    return response.json()['tax']

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

# mi_proyecto/infra/database/SQLiteRepo.py
import sqlite3

class SQLiteRepo():
    def __init__(self, db_path: str):
        self.connection = sqlite3.connect(db_path)

    def get(self, table: str, column: str) -> float:
        if table == "workers" and column == "salary":
            return self.connection.execute('SELECT salary FROM workers').fetchall()

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
from mi_proyecto.infra.gateways.tax import get_tax
from mi_proyecto.infra.database import SQLiteRepo

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

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 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(10)
"""
{
      "type": "Success",
      "errors": [],
      "data": {
            "total": 1000
      }
}
"""

En el caso de error.

calculate_turbine_cost_use_case("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
from mi_proyecto.infra.gateways.tax import get_tax
from mi_proyecto.infra.database import SQLiteRepo

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(number_of_turbines: int) -> dict:
    # Check if the number of turbines is present
    if not number_of_turbines or not isinstance(number_of_turbines, int):
        return {
            "type": ResponseTypes.PARAMETERS_ERROR,
            "errors": [
                {
                    "field": "number_of_turbines",
                    "message": "Number of turbines is required"
                }
            ],
            "data": {}
        }
    # Logic
    tax = get_tax()
    salaries = SQLiteRepo().get("workers", "salary")
    if not salaries:
        return {
            "type": ResponseTypes.RESOURCE_ERROR,
            "errors": [
                {
                    "field": "salaries",
                    "message": "Salaries not found"
                }
            ],
            "data": {}
        }
    average_salary = sum([salary for salary in salaries]) / len(salaries)
    total = number_of_turbines * INSTALLATION_PRICE
    total += total * tax
    total += average_salary
    return {
        "type": ResponseTypes.SUCCESS,
        "errors": [],
        "data": {
            "total": 1000
        }
    }

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. 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 estuviemos trabajando en FastAPI, una sencilla implementación sería la siguiente.

from fastapi import FastAPI, Response, status
from functools import wraps

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, input: dict | None = None):
    return calculate_turbine_cost_use_case(input)

Testeo

¿Qué debemos testear? En realidad esta pregunta la hemos respondido al principio, y estoy seguro 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.

Apuntes finales

Solo se ha modificado la lógica de negocio, no las UIs. Por lo tanto la web y API se mantiene 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 donde 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, arquitectura de software y mucha práctica para aplicar los conceptos.