application · temario
5.5tema 5 de 8

Puertos Inbound y Outbound

Los puertos son contratos (ABC/Protocol) que protegen el dominio de la infraestructura y viceversa.

En la arquitectura hexagonal, un puerto es una interfaz que define cómo el núcleo de la aplicación (dominio + aplicación) se comunica con el mundo exterior. Los puertos hacen posible que el dominio no sepa nada de HTTP, bases de datos ni colas de mensajes, mientras que los adapters implementan esos contratos con tecnología concreta.

Los puertos inbound (driving ports) definen cómo el mundo exterior puede iniciar una acción en la aplicación. Son las interfaces que los adapters de entrada (FastAPI controller, CLI, test) usan para hablar con la capa de aplicación. En Python con CQRS, suele ser suficiente con el contrato implícito del handler: 'acepta este Command/Query y devuelve este tipo'.

Los puertos outbound (driven ports) definen cómo la aplicación habla con sistemas externos: bases de datos, servicios de email, colas de mensajes, APIs de terceros. Son implementados por los adapters de infraestructura. El dominio y la aplicación solo conocen la abstracción (ABC/Protocol), nunca la implementación concreta.

En Python, los puertos outbound se expresan como clases abstractas (ABC) o como Protocols (duck typing estructural). ABC es más explícito y detecta incumplimientos en tiempo de clase; Protocol permite implementaciones sin herencia, lo que es más flexible para adapters externos que no puedes modificar.

La regla de oro: la flecha de dependencia siempre apunta hacia adentro. El dominio no importa infraestructura. La infraestructura importa dominio/aplicación para implementar los puertos. Los adapters inbound conocen la capa de aplicación; los adapters outbound son conocidos por la capa de aplicación solo a través del puerto.

structure.txt
# Puertos en la arquitectura hexagonal

# ── PUERTOS OUTBOUND (driven) ──────────────────────────────────────────────
# Viven en domain/ (repositorios) o en application/ports/outbound/

# application/ports/outbound/<service>_port.py
from abc import ABC, abstractmethod
from typing import Protocol, runtime_checkable

# Opción A: ABC (herencia explícita, error en tiempo de definición)
class EmailNotificationPort(ABC):
    """Puerto outbound: el dominio/aplicación notifica; infra envía el email."""
    @abstractmethod
    def notify(self, recipient: str, subject: str, body: str) -> None: ...

# Opción B: Protocol (duck typing estructural, sin herencia requerida)
@runtime_checkable
class PaymentGatewayPort(Protocol):
    """Puerto outbound: el dominio/aplicación carga; infra procesa el pago."""
    def charge(self, amount: float, currency: str, token: str) -> str: ...

# Puerto de repositorio (outbound) — en domain/repositories/
class <Aggregate>Repository(ABC):
    @abstractmethod
    def get_by_id(self, id: <Aggregate>Id) -> <Aggregate> | None: ...
    @abstractmethod
    def save(self, aggregate: <Aggregate>) -> None: ...

# ── PUERTOS INBOUND (driving) ──────────────────────────────────────────────
# En CQRS el contrato inbound es implícito: la firma del handler.
# Para bus de comandos explícito:

# application/ports/inbound/command_bus.py
class CommandBus(ABC):
    """Puerto inbound: adapter HTTP/CLI envía commands aquí."""
    @abstractmethod
    def dispatch(self, command: object) -> None: ...

# application/ports/inbound/query_bus.py
from typing import TypeVar
R = TypeVar('R')

class QueryBus(ABC):
    @abstractmethod
    def ask(self, query: object) -> object: ...

# ── REGLA DE DEPENDENCIA (visualización) ────────────────────────────────────
# [FastAPI Adapter] ──uses──► [CommandBus Port] ──implemented by──►
# [InMemoryCommandBus] ──dispatches──► [Handler] ──uses──►
# [Repository Port] ◄──implemented by── [SqlRepository Adapter]
#
# Dirección: Infra → Application → Domain (nunca al revés)
project/
ports· Ports root (application layer)
inbound· Driving ports (optional in CQRS)
outbound· Driven ports (application → infra)

Debugging lab

Detecta y corrige el error o la violación de diseño.

0/5 tests passing0%
  1. 5.5.5.1

    # application/ports/outbound/smtp_port.py from smtplib import SMTP class SmtpPort(ABC): def __init__(self, smtp: SMTP): ...

  2. 5.5.5.2

    # domain/model/order.py from sqlalchemy.orm import Session class Order: def save_to_db(self, session: Session): ...

  3. 5.5.5.3

    class SendEmailHandler: def __init__(self): import smtplib self._smtp = smtplib.SMTP('smtp.example.com') def handle(self, cmd): self._smtp.sendmail(...)

  4. 5.5.5.4

    class OrderRepository(ABC): @abstractmethod def get_by_id(self, id: str) -> dict: ... # devuelve dict con todos los campos de la tabla

  5. 5.5.5.5

    # infrastructure/adapters/order_sql_repository.py from abc import ABC class OrderSqlRepository(ABC): # el adapter hereda de ABC también