Anti-Corruption Layer System Design
Anti-Corruption Layer System Design
Real-World Problem Context
Your company acquires a competitor. Their order management system uses a legacy SOAP API with XML payloads, Roman numeral status codes (status: "III" means "shipped"), and date formats like DD/MM/YYYY. Your system uses REST with JSON, integer enums, and ISO 8601 dates. The CTO says: "integrate both systems — customers should see all their orders in one place."
The naive approach: sprinkle XML parsing and Roman numeral conversion directly into your clean domain services. Six months later, your codebase is littered with if legacy_system: branches, your domain model has fields like legacy_status_roman that nobody new on the team understands, and every change to the legacy integration risks breaking your core order logic.
The Anti-Corruption Layer (ACL) is a boundary that translates between two systems with incompatible models, preventing the legacy system's concepts and quirks from leaking into your domain.
Problem Statement
When integrating with external systems — legacy monoliths, acquired company APIs, third-party services, or partner integrations — the external model rarely matches your domain model. Field names differ, enums have different values, data structures are shaped differently, and business concepts map imperfectly.
The core challenge: how do you integrate with external systems whose data models and APIs differ significantly from yours, without contaminating your domain model with their concepts, quirks, and technical debt?
Potential Solutions
1. Translation Layer Between Domains
Place an ACL at the boundary that translates between the external model and your domain model:
Without ACL: With ACL:
┌──────────────────┐ ┌──────────────────┐
│ Your Domain │ │ Your Domain │
│ │ │ (clean model) │
│ Order { │ │ Order { │
│ status: "III" │ ← leaked! │ status: SHIPPED│ ← clean
│ date: "25/12" │ ← leaked! │ date: ISO 8601│ ← clean
│ xml_payload.. │ ← leaked! │ } │
│ } │ └────────┬─────────┘
└────────┬─────────┘ │
│ ┌────────▼─────────┐
│ │ Anti-Corruption │
│ │ Layer │
│ │ Translates: │
│ │ "III" → SHIPPED │
│ │ "25/12" → ISO │
│ │ XML → JSON │
│ └────────┬─────────┘
│ │
┌────────▼─────────┐ ┌────────▼─────────┐
│ Legacy System │ │ Legacy System │
│ (SOAP/XML) │ │ (SOAP/XML) │
└──────────────────┘ └──────────────────┘
# Your clean domain model — no legacy concepts
from enum import Enum
from datetime import datetime
from dataclasses import dataclass
class OrderStatus(Enum):
PENDING = "pending"
CONFIRMED = "confirmed"
SHIPPED = "shipped"
DELIVERED = "delivered"
CANCELLED = "cancelled"
@dataclass
class Order:
id: str
customer_id: str
status: OrderStatus
total_cents: int # Money in cents, not floats
created_at: datetime # UTC, ISO 8601
items: list["OrderItem"]
# Anti-Corruption Layer: translates legacy → domain
import xml.etree.ElementTree as ET
from datetime import datetime
class LegacyOrderACL:
"""Translates between the legacy SOAP order system and our domain model."""
# Legacy uses Roman numerals for status
STATUS_MAP = {
"I": OrderStatus.PENDING,
"II": OrderStatus.CONFIRMED,
"III": OrderStatus.SHIPPED,
"IV": OrderStatus.DELIVERED,
"V": OrderStatus.CANCELLED,
}
def to_domain(self, legacy_xml: str) -> Order:
"""Convert a legacy XML order to our domain Order."""
root = ET.fromstring(legacy_xml)
return Order(
id=f"legacy-{root.find('OrderNumber').text}",
customer_id=self._map_customer_id(root.find('CustRef').text),
status=self.STATUS_MAP[root.find('StatusCode').text],
total_cents=self._parse_legacy_amount(root.find('TotalAmount').text),
created_at=self._parse_legacy_date(root.find('OrderDate').text),
items=[
self._map_item(item)
for item in root.findall('.//LineItem')
]
)
def to_legacy(self, order: Order) -> str:
"""Convert our domain Order to legacy XML format for writes."""
reverse_status = {v: k for k, v in self.STATUS_MAP.items()}
return f"""<?xml version="1.0"?>
<Order>
<OrderNumber>{order.id.replace('legacy-', '')}</OrderNumber>
<CustRef>{self._reverse_customer_id(order.customer_id)}</CustRef>
<StatusCode>{reverse_status[order.status]}</StatusCode>
<TotalAmount>{order.total_cents / 100:.2f}</TotalAmount>
<OrderDate>{order.created_at.strftime('%d/%m/%Y')}</OrderDate>
</Order>"""
def _parse_legacy_date(self, date_str: str) -> datetime:
"""Legacy uses DD/MM/YYYY, we use ISO 8601."""
return datetime.strptime(date_str, "%d/%m/%Y")
def _parse_legacy_amount(self, amount_str: str) -> int:
"""Legacy uses decimal strings like '99.99', we use cents."""
return int(float(amount_str) * 100)
def _map_customer_id(self, legacy_ref: str) -> str:
"""Legacy uses 'CUST-00123', we use UUIDs.
Look up the mapping or create one."""
return customer_mapping.get_or_create(legacy_ref)
2. ACL as a Service (Facade)
For complex integrations, the ACL can be a standalone service:
┌──────────────┐ ┌────────────────────┐ ┌──────────────────┐
│ Order Service │────▶│ Legacy Order ACL │────▶│ Legacy System │
│ (your domain) │ │ Service │ │ (SOAP/XML) │
│ │ │ │ │ │
│ REST/JSON │ │ - Translates │ │ - WSDL endpoint │
│ domain model │ │ requests/responses│ │ - Roman numerals │
│ │ │ - Retries/circuit │ │ - DD/MM/YYYY │
│ │ │ - Caches mappings │ │ │
└──────────────┘ └────────────────────┘ └──────────────────┘
Your Order Service never sees XML, Roman numerals, or legacy date formats.
The ACL Service exposes a clean REST API that matches your domain model.
# ACL Service: clean REST API that hides the legacy system
from fastapi import FastAPI
app = FastAPI()
acl = LegacyOrderACL()
legacy_client = LegacySOAPClient(wsdl_url="https://legacy.internal/orders?wsdl")
@app.get("/orders/{order_id}")
async def get_order(order_id: str) -> Order:
# Call legacy system
legacy_xml = await legacy_client.get_order(order_id)
# Translate to domain model
return acl.to_domain(legacy_xml)
@app.get("/orders")
async def list_orders(customer_id: str) -> list[Order]:
legacy_ref = customer_mapping.to_legacy(customer_id)
legacy_xml_list = await legacy_client.list_orders(legacy_ref)
return [acl.to_domain(xml) for xml in legacy_xml_list]
@app.post("/orders/{order_id}/ship")
async def ship_order(order_id: str):
# Translate domain action to legacy operation
legacy_order_num = order_id.replace("legacy-", "")
await legacy_client.update_status(legacy_order_num, "III") # Roman numeral
return {"status": "shipped"}
3. ACL for Database Migration (Strangler Fig Support)
During a gradual migration from legacy to new system, the ACL serves both:
Phase 1: Read from legacy, write to both
┌──────────────────┐
│ API Layer │
└────────┬──────────┘
│
┌────────▼──────────┐
│ ACL │
│ Read: legacy DB │──────▶ Legacy DB (source of truth)
│ Write: both DBs │──────▶ New DB (shadow writes)
└───────────────────┘
Phase 2: Read from new, write to both (verify consistency)
┌────────▼──────────┐
│ ACL │
│ Read: new DB │──────▶ New DB (source of truth now)
│ Write: both DBs │──────▶ Legacy DB (kept in sync)
└───────────────────┘
Phase 3: Cut over
┌────────▼──────────┐
│ Domain Service │──────▶ New DB only
│ (ACL removed) │
└───────────────────┘ Legacy DB decommissioned
class MigrationACL:
def __init__(self, legacy_repo, new_repo, phase: str = "phase1"):
self.legacy = legacy_repo
self.new = new_repo
self.phase = phase
self.translator = LegacyOrderACL()
async def get_order(self, order_id: str) -> Order:
if self.phase == "phase1":
legacy_data = await self.legacy.get(order_id)
return self.translator.to_domain(legacy_data)
else:
return await self.new.get(order_id)
async def save_order(self, order: Order):
if self.phase in ("phase1", "phase2"):
# Write to both during migration
legacy_data = self.translator.to_legacy(order)
await self.legacy.save(legacy_data)
await self.new.save(order)
# Verify consistency
await self._verify_consistency(order.id)
else:
await self.new.save(order)
async def _verify_consistency(self, order_id: str):
legacy = self.translator.to_domain(await self.legacy.get(order_id))
new = await self.new.get(order_id)
if legacy != new:
log.error(f"Consistency mismatch for order {order_id}")
metrics.increment("acl.consistency.mismatch")
4. ACL for Third-Party API Integration
Every third-party integration should go through an ACL:
# Without ACL: Stripe concepts leak into your domain
class PaymentService:
async def charge(self, order: Order):
# Stripe-specific concepts everywhere
intent = stripe.PaymentIntent.create(
amount=order.total_cents,
currency="usd",
payment_method=order.stripe_payment_method_id, # ← Stripe leaked
capture_method="manual",
)
order.stripe_payment_intent_id = intent.id # ← Stripe leaked
order.stripe_status = intent.status # ← Stripe leaked
# With ACL: your domain knows about "payments", not "Stripe"
class StripeACL:
"""Translates between our Payment domain and Stripe's API."""
async def create_charge(self, payment: Payment) -> ChargeResult:
intent = stripe.PaymentIntent.create(
amount=payment.amount_cents,
currency=payment.currency.lower(),
payment_method=self._to_stripe_method(payment.method_token),
capture_method="manual",
idempotency_key=payment.idempotency_key,
)
return ChargeResult(
payment_id=payment.id,
provider_reference=intent.id,
status=self._map_status(intent.status),
captured=False,
)
def _map_status(self, stripe_status: str) -> PaymentStatus:
return {
"requires_payment_method": PaymentStatus.FAILED,
"requires_confirmation": PaymentStatus.PENDING,
"requires_action": PaymentStatus.ACTION_REQUIRED,
"processing": PaymentStatus.PROCESSING,
"succeeded": PaymentStatus.SUCCEEDED,
"canceled": PaymentStatus.CANCELLED,
}[stripe_status]
# Domain service uses ACL — no Stripe concepts
class PaymentService:
def __init__(self, payment_gateway: PaymentGateway):
self.gateway = payment_gateway # StripeACL implements PaymentGateway
async def charge(self, order: Order) -> ChargeResult:
payment = Payment(
id=str(uuid4()),
amount_cents=order.total_cents,
currency=order.currency,
method_token=order.payment_token,
)
return await self.gateway.create_charge(payment)
# Switching from Stripe to Adyen? Replace StripeACL with AdyenACL.
# Domain code unchanged.
Trade-offs & Considerations
ACL Placement Isolation Latency Maintenance Deployment
──────────────────────────────────────────────────────────────────────────
In-process module Medium None (in-proc) Coupled Same deploy
Separate service High Network hop Independent Independent
Library/SDK Medium None Versioned Per consumer
API Gateway High Gateway hop Centralized Gateway deploy
When you DON'T need an ACL:
- The external system's model closely matches your domain
- The integration is trivial (e.g., a single field mapping)
- You control both systems and can align their models
- The integration is temporary and will be removed soon
When you DO need an ACL:
- External model uses fundamentally different concepts
- Multiple services need to integrate with the same external system
- You're migrating away from a legacy system gradually
- The external API changes frequently and you want to absorb changes in one place
Best Practices
-
Never expose external model types in your domain — your
Orderclass should not have astripe_payment_intent_idfield. That's Stripe's concept, not yours. Use a genericprovider_reference: str. -
ACL translates in both directions —
to_domain()for reads,to_external()for writes. Both directions need testing. -
Put the ACL on your side of the boundary — the ACL belongs to the consuming team, not the providing team. You control when and how you adapt to external changes.
-
Use interfaces behind the ACL — define a
PaymentGatewayinterface.StripeACLimplements it. Swapping providers means implementing a new ACL, not changing domain code. -
Test the ACL with recorded real responses — capture actual responses from the external system and use them as test fixtures. This catches translation bugs that unit tests with hand-crafted data miss.
-
Monitor ACL translation failures — when the external system changes its response format, your ACL is where it'll break. Alert on translation errors.
Step-by-Step Approach
Step 1: Define your clean domain model
├── What fields and concepts does YOUR system need?
├── Don't look at the external system yet
└── Use your own naming, enums, types
Step 2: Map external model to domain model
├── Document field-by-field mapping
├── Identify concepts that don't translate cleanly
├── Decide how to handle missing/extra fields
└── Handle edge cases (null values, unknown enums)
Step 3: Implement the ACL translator
├── to_domain(): external → your model
├── to_external(): your model → external
├── Fail explicitly on unmapped values (don't silently drop)
└── Log/metric every translation for debugging
Step 4: Wire through the ACL
├── All reads from external system go through ACL
├── All writes to external system go through ACL
├── Domain services receive/return only domain types
└── No external types cross the ACL boundary
Step 5: Test with real data
├── Record actual responses from external system
├── Use as test fixtures for ACL translation tests
├── Include edge cases: empty fields, unknown statuses
└── Regression test when external system changes
Step 6: Plan for ACL removal
└── If migrating away from legacy:
Phase 1: read legacy via ACL, write both
Phase 2: read new DB, write both, verify consistency
Phase 3: remove ACL, decommission legacy
Conclusion
The Anti-Corruption Layer is how you integrate with external systems without letting their models, quirks, and technical debt infect your domain. Whether it's a legacy SOAP API with Roman numeral status codes, a third-party payment provider with vendor-specific concepts, or an acquired company's database with incompatible schemas — the ACL translates at the boundary so your domain code stays clean. The investment pays off immediately: when Stripe changes their API, you update one file (the StripeACL), not 50 files across your codebase. When you migrate away from the legacy system, the ACL is already the seam where you cut. Start by defining your domain model without looking at the external system, then build the translator that bridges the gap.
What did you think?