Implementando arquitectura limpia en Python | Programador Web Valencia

Implementando arquitectura limpia en Python

8 minutos

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:

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

猬嗭笍 Interfaces 猬囷笍 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 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.

Esta obra est谩 bajo una Licencia Creative Commons Atribuci贸n-NoComercial-SinDerivadas 4.0 Internacional.

Atribuci贸n/Reconocimiento-NoComercial-SinDerivados 4.0 Internacional

驴Me ayudas?

Comprame un caf茅
Pulsa sobre la imagen

No te sientas obligado a realizar una donaci贸n, pero cada aportaci贸n mantiene el sitio en activo logrando que contin煤e existiendo y sea accesible para otras personas. Adem谩s me motiva a crear nuevo contenido.

Comentarios

{{ comments.length }} comentarios

Nuevo comentario

Nueva replica  {{ formatEllipsisAuthor(replyComment.author) }}

Acepto la pol铆tica de Protecci贸n de Datos.

Escribe el primer comentario

Tal vez tambi茅n te interese...