Arquitectura hexagonal: qué es y cómo aplicarla en tu proyecto

D
DanisCh
• 17 min de lectura
Arquitectura hexagonal: qué es y cómo aplicarla en tu proyecto
Java

La arquitectura hexagonal, también conocida como Ports and Adapters, es un patrón de diseño de software propuesto por Alistair Cockburn en 2005. Su objetivo es crear aplicaciones donde la lógica de negocio esté completamente aislada de los detalles de infraestructura: la base de datos, el framework web, los servicios externos, la interfaz de usuario. El resultado es código más fácil de testear, más fácil de mantener y más resistente a los cambios tecnológicos.

En esta guía vas a entender el problema que resuelve, cómo funciona, y cómo aplicarla en un proyecto real con ejemplos en Python y Node.js/TypeScript.

El problema que resuelve

En una aplicación típica sin arquitectura definida, la lógica de negocio está entrelazada con los detalles de infraestructura. Un servicio que gestiona pedidos puede hacer llamadas directas a la base de datos, llamar a APIs externas de pago, enviar emails y todo ello dentro de la misma función. Esto crea varios problemas:

  • Imposible testear sin infraestructura real: para testear la lógica de negocio necesitas una base de datos, una cuenta de Stripe real, un servidor de email. Los tests son lentos, frágiles y difíciles de configurar.
  • Difícil de cambiar tecnología: si quieres cambiar de PostgreSQL a MongoDB, o de Stripe a PayPal, tienes que modificar el código de negocio, que no debería saber nada de esos detalles.
  • Difícil de entender: la lógica de negocio está oculta entre código de infraestructura. Es difícil entender qué hace el sistema sin leer implementaciones de bajo nivel.
# ❌ Código sin arquitectura: lógica de negocio entrelazada con infraestructura

import psycopg2
import requests
import smtplib

def procesar_pedido(datos_pedido):
    # Lógica de negocio mezclada con acceso a BD
    conn = psycopg2.connect("postgresql://...")
    cur = conn.cursor()

    cur.execute("SELECT stock FROM productos WHERE id = %s", [datos_pedido['producto_id']])
    stock = cur.fetchone()[0]

    if stock < datos_pedido['cantidad']:
        raise Exception("Stock insuficiente")   # ¿esto es lógica de negocio o error de BD?

    # Lógica mezclada con llamada a API de pagos
    respuesta = requests.post("https://api.stripe.com/v1/charges", {
        "amount": datos_pedido['total'],
        "source": datos_pedido['token_tarjeta']
    })
    if respuesta.status_code != 200:
        raise Exception("Pago fallido")

    cur.execute("INSERT INTO pedidos (...) VALUES (...)", [...])
    cur.execute("UPDATE productos SET stock = stock - %s WHERE id = %s",
                [datos_pedido['cantidad'], datos_pedido['producto_id']])
    conn.commit()

    # Email mezclado con el resto
    with smtplib.SMTP("smtp.gmail.com") as smtp:
        smtp.sendmail("no-reply@tienda.com", datos_pedido['email'], "Pedido confirmado")

# Para testear esto necesitas: BD real, cuenta Stripe, servidor SMTP
# Un test de esta función tarda segundos y puede fallar por razones ajenas al negocio

La idea central: el hexágono

La arquitectura hexagonal divide la aplicación en tres partes:

  • El dominio: el núcleo de la aplicación. Contiene la lógica de negocio pura, las entidades del dominio y las reglas que hacen que el negocio funcione. No depende de nada externo. No sabe qué base de datos se usa, qué framework web sirve las peticiones, ni qué servicio de email envía los mensajes.
  • Los puertos: interfaces (abstracciones) que define el dominio para comunicarse con el exterior. Hay dos tipos: puertos de entrada (cómo el mundo exterior llama al dominio) y puertos de salida (cómo el dominio pide cosas al mundo exterior).
  • Los adaptadores: implementaciones concretas de los puertos. El adaptador de base de datos implementa el puerto de repositorio. El adaptador HTTP implementa el puerto de entrada web. El adaptador de Stripe implementa el puerto de pagos.
# Estructura de carpetas de un proyecto con arquitectura hexagonal

mi-app/
├── src/
│   ├── dominio/                    ← núcleo, sin dependencias externas
│   │   ├── entidades/
│   │   │   ├── pedido.py           ← entidad del dominio
│   │   │   └── producto.py
│   │   ├── puertos/
│   │   │   ├── entrada/
│   │   │   │   └── servicio_pedidos.py    ← interfaz del caso de uso
│   │   │   └── salida/
│   │   │       ├── repositorio_pedidos.py ← puerto de persistencia
│   │   │       ├── servicio_pagos.py      ← puerto de pagos
│   │   │       └── servicio_email.py      ← puerto de notificaciones
│   │   └── servicios/
│   │       └── servicio_pedidos_impl.py  ← lógica de negocio pura
│   └── infraestructura/            ← adaptadores, dependen de librerías externas
│       ├── persistencia/
│       │   └── repositorio_pedidos_postgres.py
│       ├── pagos/
│       │   └── adaptador_stripe.py
│       ├── email/
│       │   └── adaptador_smtp.py
│       └── web/
│           └── controlador_pedidos.py   ← adaptador de entrada HTTP
└── tests/
    ├── unitarios/        ← testean el dominio con adaptadores falsos (mocks)
    └── integracion/      ← testean los adaptadores con infraestructura real

Implementación paso a paso en Python

Paso 1: las entidades del dominio

# src/dominio/entidades/pedido.py
# Las entidades son objetos del dominio con lógica de negocio propia
# No importan nada de infraestructura

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum

class EstadoPedido(Enum):
    PENDIENTE    = "pendiente"
    PAGADO       = "pagado"
    ENVIADO      = "enviado"
    ENTREGADO    = "entregado"
    CANCELADO    = "cancelado"

@dataclass
class LineaPedido:
    producto_id: int
    nombre:      str
    cantidad:    int
    precio_unit: float

    @property
    def subtotal(self) -> float:
        return self.cantidad * self.precio_unit

@dataclass
class Pedido:
    id:            int | None
    cliente_email: str
    lineas:        list[LineaPedido] = field(default_factory=list)
    estado:        EstadoPedido     = EstadoPedido.PENDIENTE
    creado_en:     datetime         = field(default_factory=datetime.now)

    @property
    def total(self) -> float:
        return sum(linea.subtotal for linea in self.lineas)

    def confirmar_pago(self) -> None:
        """Regla de negocio: solo se puede confirmar un pedido pendiente."""
        if self.estado != EstadoPedido.PENDIENTE:
            raise ValueError(f"No se puede confirmar un pedido en estado {self.estado.value}")
        self.estado = EstadoPedido.PAGADO

    def cancelar(self) -> None:
        """Regla de negocio: no se puede cancelar un pedido ya enviado."""
        if self.estado in (EstadoPedido.ENVIADO, EstadoPedido.ENTREGADO):
            raise ValueError("No se puede cancelar un pedido ya enviado")
        self.estado = EstadoPedido.CANCELADO

    def puede_enviarse(self) -> bool:
        return self.estado == EstadoPedido.PAGADO and len(self.lineas) > 0

Paso 2: los puertos de salida

# src/dominio/puertos/salida/repositorio_pedidos.py
# Puerto de salida: contrato que el dominio espera de la persistencia
# Es una interfaz abstracta. El dominio define qué necesita, sin saber cómo.

from abc import ABC, abstractmethod
from dominio.entidades.pedido import Pedido

class RepositorioPedidos(ABC):

    @abstractmethod
    def guardar(self, pedido: Pedido) -> Pedido:
        """Guarda un pedido y devuelve el pedido con el id asignado."""
        ...

    @abstractmethod
    def buscar_por_id(self, pedido_id: int) -> Pedido | None:
        ...

    @abstractmethod
    def buscar_por_cliente(self, email: str) -> list[Pedido]:
        ...

 

# src/dominio/puertos/salida/servicio_pagos.py

from abc import ABC, abstractmethod
from dataclasses import dataclass

@dataclass
class ResultadoPago:
    exitoso:         bool
    id_transaccion:  str | None
    mensaje_error:   str | None = None

class ServicioPagos(ABC):

    @abstractmethod
    def cobrar(self, importe: float, token_pago: str) -> ResultadoPago:
        """Intenta cobrar un importe con el token de pago del cliente."""
        ...

    @abstractmethod
    def reembolsar(self, id_transaccion: str) -> bool:
        ...

 

# src/dominio/puertos/salida/servicio_notificaciones.py

from abc import ABC, abstractmethod
from dominio.entidades.pedido import Pedido

class ServicioNotificaciones(ABC):

    @abstractmethod
    def enviar_confirmacion_pedido(self, pedido: Pedido) -> None:
        ...

    @abstractmethod
    def enviar_notificacion_envio(self, pedido: Pedido, num_seguimiento: str) -> None:
        ...

Paso 3: el servicio de dominio (la lógica de negocio)

# src/dominio/servicios/servicio_pedidos.py
# Lógica de negocio pura. Solo depende de puertos (abstracciones).
# No sabe nada de PostgreSQL, Stripe, SMTP ni de ningún framework.

from dominio.entidades.pedido import Pedido, EstadoPedido
from dominio.puertos.salida.repositorio_pedidos import RepositorioPedidos
from dominio.puertos.salida.servicio_pagos import ServicioPagos
from dominio.puertos.salida.servicio_notificaciones import ServicioNotificaciones

class ServicioPedidos:

    def __init__(
        self,
        repositorio:    RepositorioPedidos,
        pagos:          ServicioPagos,
        notificaciones: ServicioNotificaciones
    ):
        # Inyección de dependencias: el servicio recibe las implementaciones concretas
        # pero solo las conoce como abstracciones (puertos)
        self._repositorio    = repositorio
        self._pagos          = pagos
        self._notificaciones = notificaciones

    def crear_pedido(self, pedido: Pedido, token_pago: str) -> Pedido:
        """
        Caso de uso: crear y pagar un pedido.
        Lógica de negocio pura, sin detalles de infraestructura.
        """
        # Regla de negocio: el pedido debe tener al menos una línea
        if not pedido.lineas:
            raise ValueError("Un pedido debe tener al menos un producto")

        # Regla de negocio: el total mínimo es 1€
        if pedido.total < 1.0:
            raise ValueError("El importe mínimo del pedido es 1€")

        # Intentar cobrar
        resultado = self._pagos.cobrar(pedido.total, token_pago)
        if not resultado.exitoso:
            raise ValueError(f"El pago no pudo procesarse: {resultado.mensaje_error}")

        # Confirmar el pago en la entidad (la entidad valida su propio estado)
        pedido.confirmar_pago()

        # Persistir
        pedido_guardado = self._repositorio.guardar(pedido)

        # Notificar (si falla, no revertimos el pago: lo gestionamos con reintentos)
        try:
            self._notificaciones.enviar_confirmacion_pedido(pedido_guardado)
        except Exception:
            pass   # en producción: loggear y encolar para reintento

        return pedido_guardado

    def cancelar_pedido(self, pedido_id: int) -> Pedido:
        pedido = self._repositorio.buscar_por_id(pedido_id)
        if not pedido:
            raise ValueError(f"Pedido {pedido_id} no encontrado")

        pedido.cancelar()   # la entidad valida si el estado permite cancelación
        return self._repositorio.guardar(pedido)

    def obtener_pedidos_cliente(self, email: str) -> list[Pedido]:
        return self._repositorio.buscar_por_cliente(email)

 

Paso 4: los adaptadores de infraestructura

# src/infraestructura/persistencia/repositorio_pedidos_postgres.py
# Adaptador de salida: implementa el puerto con PostgreSQL

import psycopg2
from dominio.entidades.pedido import Pedido, EstadoPedido, LineaPedido
from dominio.puertos.salida.repositorio_pedidos import RepositorioPedidos

class RepositorioPedidosPostgres(RepositorioPedidos):

    def __init__(self, conexion_string: str):
        self._conn = psycopg2.connect(conexion_string)

    def guardar(self, pedido: Pedido) -> Pedido:
        cur = self._conn.cursor()
        if pedido.id is None:
            cur.execute(
                "INSERT INTO pedidos (cliente_email, estado, total, creado_en) "
                "VALUES (%s, %s, %s, %s) RETURNING id",
                [pedido.cliente_email, pedido.estado.value, pedido.total, pedido.creado_en]
            )
            pedido.id = cur.fetchone()[0]
        else:
            cur.execute(
                "UPDATE pedidos SET estado = %s WHERE id = %s",
                [pedido.estado.value, pedido.id]
            )
        self._conn.commit()
        return pedido

    def buscar_por_id(self, pedido_id: int) -> Pedido | None:
        cur = self._conn.cursor()
        cur.execute("SELECT * FROM pedidos WHERE id = %s", [pedido_id])
        fila = cur.fetchone()
        if not fila:
            return None
        return self._mapear_a_pedido(fila)

    def buscar_por_cliente(self, email: str) -> list[Pedido]:
        cur = self._conn.cursor()
        cur.execute("SELECT * FROM pedidos WHERE cliente_email = %s", [email])
        return [self._mapear_a_pedido(fila) for fila in cur.fetchall()]

    def _mapear_a_pedido(self, fila) -> Pedido:
        return Pedido(
            id=fila[0],
            cliente_email=fila[1],
            estado=EstadoPedido(fila[2]),
        )
# src/infraestructura/pagos/adaptador_stripe.py
# Adaptador de salida: implementa el puerto de pagos con Stripe

import stripe
from dominio.puertos.salida.servicio_pagos import ServicioPagos, ResultadoPago

class AdaptadorStripe(ServicioPagos):

    def __init__(self, api_key: str):
        stripe.api_key = api_key

    def cobrar(self, importe: float, token_pago: str) -> ResultadoPago:
        try:
            cargo = stripe.Charge.create(
                amount=int(importe * 100),   # Stripe trabaja en céntimos
                currency="eur",
                source=token_pago,
            )
            return ResultadoPago(exitoso=True, id_transaccion=cargo.id)
        except stripe.error.CardError as e:
            return ResultadoPago(exitoso=False, id_transaccion=None, mensaje_error=str(e))

    def reembolsar(self, id_transaccion: str) -> bool:
        try:
            stripe.Refund.create(charge=id_transaccion)
            return True
        except stripe.error.StripeError:
            return False

 

# src/infraestructura/web/controlador_pedidos.py
# Adaptador de entrada: convierte peticiones HTTP en llamadas al dominio

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from dominio.entidades.pedido import Pedido, LineaPedido
from dominio.servicios.servicio_pedidos import ServicioPedidos

app = FastAPI()

class LineaRequest(BaseModel):
    producto_id: int
    nombre:      str
    cantidad:    int
    precio_unit: float

class PedidoRequest(BaseModel):
    cliente_email: str
    lineas:        list[LineaRequest]
    token_pago:    str

# El controlador recibe el servicio de dominio ya configurado
servicio: ServicioPedidos = None   # se inyecta al iniciar la app

@app.post("/pedidos", status_code=201)
def crear_pedido(request: PedidoRequest):
    try:
        pedido = Pedido(
            id=None,
            cliente_email=request.cliente_email,
            lineas=[
                LineaPedido(
                    producto_id=l.producto_id,
                    nombre=l.nombre,
                    cantidad=l.cantidad,
                    precio_unit=l.precio_unit
                )
                for l in request.lineas
            ]
        )
        resultado = servicio.crear_pedido(pedido, request.token_pago)
        return {"id": resultado.id, "total": resultado.total, "estado": resultado.estado.value}
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

@app.delete("/pedidos/{pedido_id}")
def cancelar_pedido(pedido_id: int):
    try:
        resultado = servicio.cancelar_pedido(pedido_id)
        return {"id": resultado.id, "estado": resultado.estado.value}
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

 

Paso 5: el punto de entrada — ensamblar todo

# src/main.py
# El punto de entrada conecta el dominio con los adaptadores concretos

import os
from dominio.servicios.servicio_pedidos import ServicioPedidos
from infraestructura.persistencia.repositorio_pedidos_postgres import RepositorioPedidosPostgres
from infraestructura.pagos.adaptador_stripe import AdaptadorStripe
from infraestructura.email.adaptador_smtp import AdaptadorSMTP
from infraestructura.web import controlador_pedidos

# Construir los adaptadores de infraestructura
repositorio    = RepositorioPedidosPostgres(os.environ["DATABASE_URL"])
pagos          = AdaptadorStripe(os.environ["STRIPE_API_KEY"])
notificaciones = AdaptadorSMTP(os.environ["SMTP_HOST"])

# Ensamblar el servicio de dominio inyectando los adaptadores
# El dominio no sabe que son Postgres, Stripe o SMTP: los ve como puertos
servicio = ServicioPedidos(
    repositorio=repositorio,
    pagos=pagos,
    notificaciones=notificaciones
)

# Inyectar el servicio en el controlador HTTP
controlador_pedidos.servicio = servicio

 

La gran ventaja: tests rápidos y sin infraestructura

Con la arquitectura hexagonal puedes testear toda la lógica de negocio usando implementaciones falsas (stubs/mocks) de los puertos, sin necesidad de base de datos, Stripe ni SMTP.

# tests/unitarios/test_servicio_pedidos.py

import pytest
from dominio.entidades.pedido import Pedido, LineaPedido, EstadoPedido
from dominio.servicios.servicio_pedidos import ServicioPedidos
from dominio.puertos.salida.repositorio_pedidos import RepositorioPedidos
from dominio.puertos.salida.servicio_pagos import ServicioPagos, ResultadoPago
from dominio.puertos.salida.servicio_notificaciones import ServicioNotificaciones


# ─── Implementaciones falsas (stubs) ──────────────────────────────────────────

class RepositorioEnMemoria(RepositorioPedidos):
    """Implementación falsa que guarda los pedidos en un diccionario."""

    def __init__(self):
        self._pedidos = {}
        self._siguiente_id = 1

    def guardar(self, pedido: Pedido) -> Pedido:
        if pedido.id is None:
            pedido.id = self._siguiente_id
            self._siguiente_id += 1
        self._pedidos[pedido.id] = pedido
        return pedido

    def buscar_por_id(self, pedido_id: int) -> Pedido | None:
        return self._pedidos.get(pedido_id)

    def buscar_por_cliente(self, email: str) -> list[Pedido]:
        return [p for p in self._pedidos.values() if p.cliente_email == email]


class PagosSimulados(ServicioPagos):
    """Simulador de pagos configurable para los tests."""

    def __init__(self, exitoso: bool = True):
        self.exitoso = exitoso
        self.cobros_realizados = []

    def cobrar(self, importe: float, token_pago: str) -> ResultadoPago:
        self.cobros_realizados.append({"importe": importe, "token": token_pago})
        if self.exitoso:
            return ResultadoPago(exitoso=True, id_transaccion="txn_test_123")
        return ResultadoPago(exitoso=False, id_transaccion=None, mensaje_error="Tarjeta rechazada")

    def reembolsar(self, id_transaccion: str) -> bool:
        return True


class NotificacionesFalsas(ServicioNotificaciones):
    """Captura las notificaciones enviadas para verificarlas en los tests."""

    def __init__(self):
        self.confirmaciones_enviadas = []

    def enviar_confirmacion_pedido(self, pedido: Pedido) -> None:
        self.confirmaciones_enviadas.append(pedido)

    def enviar_notificacion_envio(self, pedido: Pedido, num_seguimiento: str) -> None:
        pass


# ─── Fixtures ─────────────────────────────────────────────────────────────────

@pytest.fixture
def pedido_valido():
    return Pedido(
        id=None,
        cliente_email="ana@mail.com",
        lineas=[
            LineaPedido(producto_id=1, nombre="Teclado", cantidad=1, precio_unit=59.99),
            LineaPedido(producto_id=2, nombre="Ratón",   cantidad=2, precio_unit=29.99),
        ]
    )

@pytest.fixture
def servicio_con_pago_exitoso():
    return ServicioPedidos(
        repositorio=RepositorioEnMemoria(),
        pagos=NotificacionesFalsas() or PagosSimulados(exitoso=True),
        notificaciones=NotificacionesFalsas()
    )


# ─── Tests ────────────────────────────────────────────────────────────────────

def test_crear_pedido_con_pago_exitoso(pedido_valido):
    pagos = PagosSimulados(exitoso=True)
    notificaciones = NotificacionesFalsas()
    servicio = ServicioPedidos(
        repositorio=RepositorioEnMemoria(),
        pagos=pagos,
        notificaciones=notificaciones
    )

    resultado = servicio.crear_pedido(pedido_valido, "tok_test")

    assert resultado.id is not None
    assert resultado.estado == EstadoPedido.PAGADO
    assert resultado.total == pytest.approx(119.97)
    assert len(pagos.cobros_realizados) == 1
    assert pagos.cobros_realizados[0]["importe"] == pytest.approx(119.97)
    assert len(notificaciones.confirmaciones_enviadas) == 1

def test_crear_pedido_falla_si_pago_rechazado(pedido_valido):
    servicio = ServicioPedidos(
        repositorio=RepositorioEnMemoria(),
        pagos=PagosSimulados(exitoso=False),
        notificaciones=NotificacionesFalsas()
    )

    with pytest.raises(ValueError, match="El pago no pudo procesarse"):
        servicio.crear_pedido(pedido_valido, "tok_invalido")

def test_no_se_puede_crear_pedido_sin_lineas():
    servicio = ServicioPedidos(
        repositorio=RepositorioEnMemoria(),
        pagos=PagosSimulados(exitoso=True),
        notificaciones=NotificacionesFalsas()
    )
    pedido_vacio = Pedido(id=None, cliente_email="ana@mail.com", lineas=[])

    with pytest.raises(ValueError, match="al menos un producto"):
        servicio.crear_pedido(pedido_vacio, "tok_test")

def test_cancelar_pedido_pagado(pedido_valido):
    repositorio = RepositorioEnMemoria()
    servicio = ServicioPedidos(
        repositorio=repositorio,
        pagos=PagosSimulados(exitoso=True),
        notificaciones=NotificacionesFalsas()
    )

    pedido_creado = servicio.crear_pedido(pedido_valido, "tok_test")
    pedido_cancelado = servicio.cancelar_pedido(pedido_creado.id)

    assert pedido_cancelado.estado == EstadoPedido.CANCELADO

# Estos tests no abren ninguna conexión de red, BD ni servicio externo.
# Corren en milisegundos y son completamente deterministas.

 

Equivalente en TypeScript con Express

// src/dominio/puertos/salida/repositorio-pedidos.ts
// Puerto de salida: interfaz que el dominio define

import { Pedido } from '../entidades/pedido';

export interface RepositorioPedidos {
  guardar(pedido: Pedido): Promise;
  buscarPorId(id: number): Promise;
  buscarPorCliente(email: string): Promise;
}

export interface ServicioPagos {
  cobrar(importe: number, tokenPago: string): Promise;
  reembolsar(idTransaccion: string): Promise;
}

export interface ResultadoPago {
  exitoso:        boolean;
  idTransaccion?: string;
  mensajeError?:  string;
}
// src/dominio/servicios/servicio-pedidos.ts

import { Pedido, EstadoPedido } from '../entidades/pedido';
import { RepositorioPedidos, ServicioPagos } from '../puertos/salida/repositorio-pedidos';

export class ServicioPedidos {
  constructor(
    private readonly repositorio:    RepositorioPedidos,
    private readonly pagos:          ServicioPagos,
  ) {}

  async crearPedido(pedido: Pedido, tokenPago: string): Promise {
    if (pedido.lineas.length === 0) {
      throw new Error('Un pedido debe tener al menos un producto');
    }

    const resultado = await this.pagos.cobrar(pedido.total, tokenPago);
    if (!resultado.exitoso) {
      throw new Error(`El pago no pudo procesarse: ${resultado.mensajeError}`);
    }

    pedido.confirmarPago();
    return this.repositorio.guardar(pedido);
  }
}
// src/infraestructura/persistencia/repositorio-pedidos-prisma.ts
// Adaptador de salida: implementa el puerto usando Prisma

import { PrismaClient } from '@prisma/client';
import { Pedido } from '../../dominio/entidades/pedido';
import { RepositorioPedidos } from '../../dominio/puertos/salida/repositorio-pedidos';

export class RepositorioPedidosPrisma implements RepositorioPedidos {
  constructor(private readonly prisma: PrismaClient) {}

  async guardar(pedido: Pedido): Promise {
    if (pedido.id === null) {
      const creado = await this.prisma.pedido.create({
        data: {
          clienteEmail: pedido.clienteEmail,
          estado:       pedido.estado,
          total:        pedido.total,
        },
      });
      pedido.id = creado.id;
    } else {
      await this.prisma.pedido.update({
        where: { id: pedido.id },
        data:  { estado: pedido.estado },
      });
    }
    return pedido;
  }

  async buscarPorId(id: number): Promise {
    const fila = await this.prisma.pedido.findUnique({ where: { id } });
    return fila ? this.mapearAPedido(fila) : null;
  }

  async buscarPorCliente(email: string): Promise {
    const filas = await this.prisma.pedido.findMany({
      where: { clienteEmail: email }
    });
    return filas.map(this.mapearAPedido);
  }

  private mapearAPedido(fila: any): Pedido {
    return new Pedido(fila.id, fila.clienteEmail, fila.estado);
  }
}

Cuándo aplicar arquitectura hexagonal

Tiene sentido cuando:

  • La lógica de negocio es compleja y necesitas testearla de forma aislada, rápida y fiable.
  • El proyecto va a vivir mucho tiempo y la tecnología probablemente cambiará (nuevo ORM, nuevo proveedor de pagos, nueva API de email).
  • El equipo es lo suficientemente grande como para que diferentes personas trabajen en el dominio y en la infraestructura de forma independiente.
  • Necesitas soportar múltiples adaptadores de entrada o salida: una API REST y una CLI, o PostgreSQL y MongoDB dependiendo del entorno.

Puede ser excesiva cuando:

  • El proyecto es pequeño, de corta vida o es un prototipo. La complejidad añadida de separar puertos y adaptadores no se amortiza.
  • La lógica de negocio es casi inexistente (un CRUD simple). Separar dominio e infraestructura cuando no hay lógica real que aislar es burocracia sin valor.
  • El equipo no está familiarizado con el patrón. Una arquitectura que el equipo no entiende bien produce más problemas de los que resuelve.

Resumen

  • La arquitectura hexagonal separa la lógica de negocio (dominio) de los detalles de infraestructura (base de datos, APIs externas, framework web) mediante puertos (interfaces) y adaptadores (implementaciones concretas).
  • El dominio no depende de nada externo. Todo lo que necesita lo pide a través de puertos, que son interfaces abstractas que él mismo define.
  • Los adaptadores implementan los puertos. Puedes cambiar de PostgreSQL a MongoDB, de Stripe a PayPal, de SMTP a SendGrid, sin tocar una sola línea del dominio.
  • La principal ventaja es la testeabilidad: puedes testear toda la lógica de negocio con implementaciones en memoria, sin infraestructura real. Los tests corren en milisegundos y son completamente deterministas.
  • La inyección de dependencias es el mecanismo que conecta el dominio con los adaptadores. El dominio recibe los adaptadores concretos en tiempo de construcción (en el punto de entrada de la aplicación), pero los ve como interfaces.
  • Aplícala cuando la lógica de negocio es compleja y el proyecto es de larga duración. Para proyectos simples o prototipos, el coste de la estructura adicional no se justifica.

¿Te ha gustado esta entrada?

Compártela con tus compañeros para que también sigan aprendiendo.

Comunidad y Comentarios

0 COMENTARIOS

No hay comentarios todavía. Sé el primero en compartir tu opinión.

Escribe tu opinión
Respondiendo a