Anti-Corruption Layer (ACL)
Traducir el modelo externo para no contaminar el modelo propio.
Cuando un Bounded Context necesita integrar con un sistema externo —otro contexto, un servicio de terceros, una API legada— existe el riesgo de que el modelo externo contamine el dominio propio. Si el código de dominio empieza a manejar tipos, vocabulario y estructuras del sistema externo, el lenguaje ubicuo del contexto propio se erosiona. La Anti-Corruption Layer (ACL) es la barrera que previene esa contaminación: es un adapter cuya misión es traducir el modelo externo al lenguaje del contexto propio.
La ACL es un adapter de infraestructura que realiza traducción semántica, no solo técnica. Un adapter HTTP simple cambia el protocolo (serializa/deserializa JSON); la ACL además cambia el vocabulario, los tipos y las estructuras. Si el sistema externo habla de `user_account` con campos `account_balance` y `txn_history`, y el contexto propio habla de `<Wallet>` con `<Balance>` y `<TransactionLog>`, la ACL hace esa traducción de nombre y de concepto. El resultado es que el dominio propio nunca ve ni sabe del modelo externo.
La diferencia con un simple adapter HTTP es precisamente esa capa semántica. Un adapter HTTP puede vivir en infraestructura y solo serializar/deserializar; es lo que haces en una relación Conformist o Customer/Supplier sin conflicto de modelos. La ACL se justifica cuando el modelo externo tiene un vocabulario, una lógica de estados o una estructura que, si se introduce en el dominio propio, lo degradaría. La ACL absorbe ese vocabulario y lo transforma en los tipos del lenguaje ubicuo propio antes de que el dominio los vea.
No siempre necesitas una ACL. En una relación Conformist, el contexto downstream adopta el modelo del upstream sin resistencia: no hay traducción, solo consumo. En un Shared Kernel, ambos contextos comparten una parte del modelo acordado entre los equipos. En Open Host Service, el upstream expone un protocolo bien definido y estable que el downstream consume directamente. La ACL se justifica cuando hay conflicto real entre los modelos, cuando el sistema externo es inestable o legado, o cuando el equipo quiere proteger el lenguaje ubicuo ante cambios externos.
En Python, la ACL se implementa con una clase que contiene un método `translate` (o `from_external`) que recibe el DTO externo —típicamente un dataclass o un dict deserializado— y retorna el tipo del dominio propio. El cliente HTTP o SDK que llama al sistema externo es una clase separada: hace la llamada de red y retorna el DTO externo crudo. La ACL solo traduce; no hace llamadas de red. Esta separación permite testear la ACL de forma unitaria sin mockear HTTP.
# infrastructure/acl/<external_context>_client.py — HTTP/SDK Client
# Makes network calls. Returns raw external DTOs only. No domain types.
import httpx
from dataclasses import dataclass
@dataclass
class <ExternalContextDTO>:
external_id: str
external_<field>: str
external_<status>: str # external system's vocabulary
class <ExternalContext>Client:
def __init__(self, base_url: str) -> None:
self._base_url = base_url
def fetch_<resource>(self, external_id: str) -> <ExternalContextDTO>:
response = httpx.get(f'{self._base_url}/<resource>/{external_id}')
response.raise_for_status()
data = response.json()
return <ExternalContextDTO>(
external_id=data['id'],
external_<field>=data['<external_field_name>'],
external_<status>=data['<external_status_name>'],
)
# infrastructure/acl/<external_context>_acl.py — Anti-Corruption Layer
# Translates external DTO to own ubiquitous language. No HTTP calls.
from ...<own_domain>.model.<value_object> import <ValueObject>
from .<external_context>_client import <ExternalContextDTO>
class <ExternalContext>ACL:
def translate(self, dto: <ExternalContextDTO>) -> <ValueObject>:
# Semantic translation: external vocabulary → own ubiquitous language
return <ValueObject>(
<own_field>=dto.external_<field>,
<own_status>=self._map_status(dto.external_<status>),
)
@staticmethod
def _map_status(external_status: str) -> <OwnStatus>:
mapping = {
'ext_pending': <OwnStatus>.PENDING,
'ext_active': <OwnStatus>.ACTIVE,
'ext_closed': <OwnStatus>.INACTIVE,
}
return mapping[external_status]
# If external model changes, only this class needs updating.
# Domain model stays untouched.Debugging lab
Detecta y corrige el error o la violación de diseño.
- 6.5.5.1
# domain/model/<aggregate>.py from infrastructure.acl.<external_context>_client import <ExternalContextDTO> class <Aggregate>: def sync_with_external(self, dto: <ExternalContextDTO>) -> None: # Domain using external DTO directly self._<field> = dto.external_<field> self._<status> = dto.external_<status>
- 6.5.5.2
# infrastructure/acl/<external_context>_acl.py class <ExternalContext>ACL: def translate(self, dto: <ExternalContextDTO>) -> <ValueObject>: if dto.external_<status> not in ('<ext_active>', '<ext_pending>'): raise <DomainError>('Invalid external status for this context') # apply business rule during translation result = <ValueObject>(<own_field>=dto.external_<field>) result._<computed_field> = self._compute_business_rule(dto) return result
- 6.5.5.3
# infrastructure/acl/<external_context>_acl.py from sqlalchemy.orm import Session from ..persistence.models import <AggregateORM> class <ExternalContext>ACL: def __init__(self, session: Session) -> None: self._session = session def translate(self, dto: <ExternalContextDTO>) -> <AggregateORM>: row = self._session.get(<AggregateORM>, dto.external_id) row.<field> = dto.external_<field> return row # returns ORM model
- 6.5.5.4
# infrastructure/api/controllers.py import httpx async def sync_<resource>_controller( resource_id: UUID, handler: Sync<Resource>Handler, ) -> <Resource>Response: # Controller calling external API directly and mapping to domain resp = httpx.get(f'https://external.service/<resource>/{resource_id}') data = resp.json() domain_obj = <ValueObject>(<own_field>=data['external_<field>']) command = Sync<Resource>Command(resource_id=resource_id, value=domain_obj) handler.handle(command)
- 6.5.5.5
# infrastructure/acl/<external_context>_acl.py class <ExternalContext>ACL: def translate(self, dto: <ExternalContextDTO>) -> <ValueObject>: try: return <ValueObject>(<own_field>=dto.external_<field>) except Exception: raise HTTPException(status_code=502, detail='External service error')