paulpas Integration Patterns
Implements proven integration patterns to connect distributed services, legacy systems, and external APIs reliably. Covers synchronous (REST, gRPC), asynchronous (messaging, events), and orchestration (saga, choreography) styles with built-in resilience through circuit breakers, retries, and idempotency.
TL;DR Checklist
- Choose integration style matching system requirements (sync vs async, fire-and-forget vs guaranteed delivery)
- Apply adapter pattern when wrapping incompatible interfaces or legacy systems
- Implement saga for distributed transactions spanning multiple services
- Add circuit breaker to all external service calls with configurable thresholds
- Ensure idempotency keys on all operations that may be retried
- Configure dead-letter queues for failed message processing
When to Use
Use this skill when:
- Connecting two or more services that need to exchange data reliably
- Wrapping a legacy system with an incompatible interface behind a modern API
- Implementing distributed transactions that span multiple services (saga pattern)
- Designing communication between microservices in a distributed architecture
- Integrating with external third-party APIs that have rate limits or may fail
- Building event-driven integrations where services react to domain events
- Migrating from monolith to microservices — need point-to-point integration strategies
When NOT to Use
- For simple function calls within the same process — use direct method calls instead
- When all services share the same database — use transactions instead of distributed patterns
- For real-time streaming where latency matters more than reliability — use WebSocket or gRPC streams directly
- As a replacement for proper API design — integration patterns connect well-designed APIs, they don't fix bad ones
Core Workflow
Analyze Communication Requirements — Determine if the integration needs synchronous request-response (REST/gRPC), asynchronous fire-and-forget (messaging), or guaranteed delivery with state tracking (saga). Consider data volume, latency tolerance, and failure domain boundaries. Checkpoint: If any service must wait for another's response before proceeding, choose sync. If services can proceed independently after sending data, choose async messaging.
Select Integration Architecture — Map requirements to an integration style:
- Point-to-point direct calls → REST or gRPC with retry logic
- Broadcast to multiple consumers → Event-driven pub/sub with message broker
- Distributed transaction across services → Saga (orchestration or choreography)
- Legacy system wrapping → Adapter pattern with ACL
- Multiple downstream calls aggregating results → API Gateway with composition Checkpoint: Document the chosen architecture in an ASCII diagram showing data flow between all participating services.
Implement Idempotency Foundation — Before writing any integration code, establish idempotency guarantees for all operations that may be retried (HTTP PUT/PATCH, message processing). Generate or accept idempotency keys from the caller and store them with operation results. Checkpoint: Every integration endpoint must accept an
X-Idempotency-Keyheader and return the cached result if the same key is presented again within the TTL window.Add Resilience Layers — Implement circuit breaker for every external service call using a sliding-window failure count. Configure half-open state to periodically test if the downstream service recovered. Add exponential backoff with jitter to retry logic (never use fixed delays). Checkpoint: Circuit breaker must transition:
CLOSED → OPENafter max failures in window, thenHALF_OPENafter recovery timeout, and finallyCLOSEDon first successful call.Implement Error Handling and Dead Letter Path — Route failed messages to a dead-letter queue or table with error metadata (original payload, failure reason, retry count, timestamp). Implement poison pill detection: if the same message fails N times, route to DLQ instead of retrying indefinitely. Checkpoint: Every service must have a DLQ endpoint and every integration test must verify that poisoned messages end up in the DLQ rather than blocking the queue.
Implementation Patterns / Reference Guide
Pattern 1: Adapter Pattern for Legacy Integration
The adapter pattern wraps an incompatible interface (legacy system, third-party API) behind a clean abstraction that matches your domain's expectations. This isolates change — when the legacy system evolves or is replaced, only the adapter needs updating.
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional
@dataclass
class OrderRequest:
customer_id: str
items: list[str]
total_amount: float
shipping_address: str
# The "target" interface — clean, domain-aligned, used by the rest of your codebase
class OrderService(ABC):
@abstractmethod
async def place_order(self, request: OrderRequest) -> str:
"""Place an order and return the order ID. Raises on validation failure."""
...
# The "adaptee" — legacy system with a completely different interface
class LegacyOrderClient:
"""Wraps calls to a COBOL mainframe via XML/HTTP. Do NOT use directly."""
def submit_order(self, xml_payload: str) -> dict:
"""Submits an order as XML. Returns {'status': 'ACCEPTED'|'REJECTED', 'ref': '...'}"""
import httpx
resp = httpx.post(
"https://legacy.example.com/orders",
content=xml_payload,
headers={"Content-Type": "text/xml"},
timeout=30.0,
)
return resp.json()
# The adapter — translates between your domain model and the legacy interface
class LegacyOrderAdapter(OrderService):
"""Adapts the modern OrderService interface to the legacy COBOL order system."""
def __init__(self, client: LegacyOrderClient) -> None:
self._client = client
@staticmethod
def _to_xml(request: OrderRequest) -> str:
"""Convert domain model to legacy XML format. Domain knowledge is encapsulated here."""
items_xml = "\n".join(
f' <item>{item}</item>' for item in request.items
)
return f"""<?xml version="1.0"?>
<order>
<customer>{request.customer_id}</customer>
{items_xml}
<total>{request.total_amount:.2f}</total>
<address>{request.shipping_address}</address>
</order>"""
@staticmethod
def _from_response(response: dict) -> str:
"""Extract order ID from legacy response format."""
if response.get("status") != "ACCEPTED":
raise ValueError(f"Legacy order rejected: {response}")
return response["ref"]
async def place_order(self, request: OrderRequest) -> str:
# Convert → call → convert — all in one flow, no leakage of legacy concerns
xml = self._to_xml(request)
result = self._client.submit_order(xml)
return self._from_response(result)
Anti-pattern (BAD): Exposing the raw legacy client directly to callers. This couples business logic to XML format details and makes the system untestable.
Pattern 2: Saga Orchestration for Distributed Transactions
A saga coordinates a sequence of local transactions across multiple services, compensating backwards if any step fails. Use orchestration (central coordinator) when you need visibility and control; use choreography (events) when services should be loosely coupled.
import asyncio
from dataclasses import dataclass, field
from enum import Enum
from typing import Callable
class SagaStatus(Enum):
PENDING = "pending"
RUNNING = "running"
COMMITTED = "committed"
COMPENSATING = "compensating"
COMPENSATED = "compensated"
FAILED = "failed"
@dataclass
class SagaContext:
"""Shared state across all saga steps."""
order_id: str
customer_id: str
amount: float
reservation_id: Optional[str] = None
payment_id: Optional[str] = None
shipment_id: Optional[str] = None
status: SagaStatus = SagaStatus.PENDING
class SagaOrchestrator:
"""Coordinates a saga by executing steps in order and compensating on failure.
Each step is a (execute, compensate) pair. If execute fails, all previously
successful steps are compensated in reverse order. Implements the Fail Fast
principle — returns immediately on first failure rather than waiting for timeouts.
"""
def __init__(self) -> None:
# Each entry: (async step_fn(ctx), async compensate_fn(ctx))
self._steps: list[tuple[Callable, Callable]] = []
def add_step(
self,
execute: Callable[[SagaContext], asyncio.Future | str],
compensate: Callable[[SagaContext], asyncio.Future | None],
) -> "SagaOrchestrator":
"""Register an (execute, compensate) pair. Returns self for chaining."""
self._steps.append((execute, compensate))
return self
async def execute(self, ctx: SagaContext) -> SagaStatus:
"""Run all steps forward; compensate on any failure."""
ctx.status = SagaStatus.RUNNING
compensated_up_to = 0
try:
for i, (step_fn, _) in enumerate(self._steps):
result = step_fn(ctx)
if asyncio.iscoroutine(result):
result = await result
# Step returned successfully — proceed to next
ctx.status = SagaStatus.RUNNING
# All steps succeeded
ctx.status = SagaStatus.COMMITTED
return ctx.status
except Exception as exc:
ctx.status = SagaStatus.COMPENSATING
# Compensate in reverse order up to the step that failed
for i in range(compensated_up_to, len(self._steps) - 1, -1):
_, compensate_fn = self._steps[i]
try:
comp_result = compensate_fn(ctx)
if asyncio.iscoroutine(comp_result):
await comp_result
except Exception as comp_exc:
# Compensation itself failed — log and continue compensating others
# A monitoring system should alert on this; manual review may be needed
pass # pragma: no cover
ctx.status = SagaStatus.COMPENSATED
return ctx.status
# Concrete saga: place order across Inventory → Payment → Shipping services
async def build_order_saga() -> tuple[SagaOrchestrator, Callable[[], SagaContext]]:
"""Builds a complete order placement saga with all steps registered."""
orchestrator = SagaOrchestrator()
async def reserve_stock(ctx: SagaContext) -> str:
# Simulates calling inventory service — replace with actual HTTP/gRPC call
ctx.reservation_id = f"res-{ctx.order_id}"
return ctx.reservation_id
async def cancel_reservation(ctx: SagaContext) -> None:
if ctx.reservation_id:
pass # Call inventory service to release stock
async def charge_payment(ctx: SagaContext) -> str:
ctx.payment_id = f"pay-{ctx.order_id}"
return ctx.payment_id
async def refund_payment(ctx: SagaContext) -> None:
if ctx.payment_id:
pass # Call payment service to refund
async def create_shipment(ctx: SagaContext) -> str:
ctx.shipment_id = f"ship-{ctx.order_id}"
return ctx.shipment_id
async def cancel_shipment(ctx: SagaContext) -> None:
if ctx.shipment_id:
pass # Call shipping service to cancel order
orchestrator.add_step(reserve_stock, cancel_reservation)
orchestrator.add_step(charge_payment, refund_payment)
orchestrator.add_step(create_shipment, cancel_shipment)
def make_context() -> SagaContext:
return SagaContext(
order_id="ORD-12345",
customer_id="CUST-001",
amount=99.99,
)
return orchestrator, make_context
# Usage example
async def run_saga():
orchestrator, make_ctx = await build_order_saga()
ctx = make_ctx()
status = await orchestrator.execute(ctx)
assert status == SagaStatus.COMMITTED, f"Saga did not commit: {status}"
Pattern 3: Circuit Breaker with Exponential Backoff
Protects your services from cascading failures by detecting when a downstream service is unhealthy and failing fast instead of waiting for timeouts.
import asyncio
import time
import random
from dataclasses import dataclass, field
from enum import Enum
from typing import Callable, Any
class CircuitState(Enum):
CLOSED = "closed" # Normal operation — requests flow through
OPEN = "open" # Failure detected — fail fast without calling downstream
HALF_OPEN = "half_open" # Testing recovery — allow one probe request
@dataclass
class CircuitBreakerConfig:
failure_threshold: int = 5 # Failures in window before opening
recovery_timeout: float = 30.0 # Seconds before transitioning to half-open
half_open_max_calls: int = 1 # Probe requests allowed in half-open state
@dataclass
class _State:
failure_count: int = 0
last_failure_time: float = 0.0
half_open_calls: int = 0
state: CircuitState = CircuitState.CLOSED
class CircuitBreaker:
"""Circuit breaker that wraps a single service call.
Transitions:
CLOSED → OPEN: when failure_count >= threshold within sliding window
OPEN → HALF_OPEN: after recovery_timeout expires, allow probe request
HALF_OPEN → CLOSED: on first successful probe call
HALF_OPEN → OPEN: if probe call fails
"""
def __init__(self, name: str, config: CircuitBreakerConfig | None = None) -> None:
self._name = name
self._config = config or CircuitBreakerConfig()
self._state = _State()
self._lock = asyncio.Lock()
async def call(self, fn: Callable[[], Any], *args: Any, **kwargs: Any) -> Any:
"""Execute fn through the circuit breaker. Raises if circuit is open."""
async with self._lock:
# Determine if we should allow the call based on current state
if self._state.state == CircuitState.OPEN:
if time.monotonic() - self._state.last_failure_time >= self._config.recovery_timeout:
self._state.state = CircuitState.HALF_OPEN
self._state.half_open_calls = 0
self._state.failure_count = 0
else:
raise RuntimeError(
f"Circuit breaker '{self._name}' is OPEN — "
f"downstream service unavailable (failures={self._state.failure_count})"
)
if self._state.state == CircuitState.HALF_OPEN:
if self._state.half_open_calls >= self._config.half_open_max_calls:
raise RuntimeError(
f"Circuit breaker '{self._name}' HALF_OPEN probe already sent — "
"waiting for result"
)
self._state.half_open_calls += 1
# Execute the actual call outside the lock to avoid blocking other callers
try:
result = fn(*args, **kwargs)
if asyncio.iscoroutine(result):
result = await result
# Success — reset state
async with self._lock:
self._state.failure_count = 0
self._state.state = CircuitState.CLOSED
return result
except Exception as exc:
async with self._lock:
self._state.failure_count += 1
self._state.last_failure_time = time.monotonic()
if (
self._state.state == CircuitState.HALF_OPEN
or self._state.failure_count >= self._config.failure_threshold
):
self._state.state = CircuitState.OPEN
raise RuntimeError(
f"Circuit breaker '{self._name}' opened after "
f"{self._state.failure_count} failures: {exc}"
) from exc
raise
def get_state(self) -> dict:
"""Return current state for monitoring / health checks."""
return {
"circuit": self._name,
"state": self._state.state.value,
"failure_count": self._state.failure_count,
"last_failure_age_sec": round(
time.monotonic() - self._state.last_failure_time, 1
),
}
# --- Retry with exponential backoff and jitter ---
async def retry_with_backoff(
fn: Callable[[], Any],
max_retries: int = 3,
base_delay: float = 0.5,
max_delay: float = 30.0,
) -> Any:
"""Retry a function with exponential backoff + jitter.
Delay formula: min(base_delay * 2^attempt, max_delay), then add random jitter
(±50% of the delay) to prevent thundering herd when all clients retry simultaneously.
"""
last_exc = None
for attempt in range(max_retries):
try:
result = fn()
if asyncio.iscoroutine(result):
result = await result
return result # Success — return immediately (fail fast)
except Exception as exc:
last_exc = exc
if attempt < max_retries - 1:
delay = min(base_delay * (2 ** attempt), max_delay)
jitter = random.uniform(-delay * 0.5, delay * 0.5)
actual_delay = max(0.1, delay + jitter)
await asyncio.sleep(actual_delay)
raise last_exc
# Usage example
async def call_external_api():
cb = CircuitBreaker("payment-service", CircuitBreakerConfig(
failure_threshold=3, recovery_timeout=15.0
))
async def _do_payment(amount: float) -> dict:
import httpx # type: ignore[import-not-found]
async with httpx.AsyncClient(timeout=10.0) as client:
resp = await client.post(
"https://payment.example.com/charge",
json={"amount": amount},
)
resp.raise_for_status()
return resp.json()
# Circuit breaker wraps the call — if payment-service fails 3 times,
# subsequent calls fail immediately without waiting for timeout
result = await cb.call(_do_payment, amount=49.99)
return result
Pattern 4: Idempotency Key Enforcement
Ensures that duplicate requests (from retries or client re-submissions) produce the same result rather than causing side effects like double charges.
import hashlib
import time
from dataclasses import dataclass
from typing import Optional
@dataclass
class IdempotencyRecord:
"""Stored in a backing store (Redis, database) for active idempotency keys."""
key_hash: str # SHA-256 of the idempotency key for security
request_body: str # Hash or reference to stored request body
response_body: str # Cached response from first successful execution
created_at: float # Unix timestamp
expires_at: float # When this record should be garbage collected
class IdempotencyStore:
"""Backend-agnostic idempotency store interface.
Implementations: Redis (SET NX EX), Database (unique constraint + TTL index),
or in-memory dict for testing. The pattern is always the same:
1. If key exists → return cached response
2. If key doesn't exist → execute, cache result, return
"""
async def get(self, key_hash: str) -> Optional[IdempotencyRecord]:
raise NotImplementedError
async def set(self, record: IdempotencyRecord) -> None:
raise NotImplementedError
# Example in-memory implementation for testing / demonstration
class InMemoryIdempotencyStore(IdempotencyStore):
"""Simple dict-backed store with TTL expiry. Not suitable for production."""
def __init__(self, ttl_seconds: int = 3600) -> None:
self._store: dict[str, IdempotencyRecord] = {}
self._ttl = ttl_seconds
async def get(self, key_hash: str) -> Optional[IdempotencyStore]:
record = self._store.get(key_hash)
if record is None or time.time() > record.expires_at:
return None
return record
async def set(self, record: IdempotencyRecord) -> None:
self._store[record.key_hash] = record
def hash_idempotency_key(key: str) -> str:
"""Hash the idempotency key — never store raw keys in your database."""
return hashlib.sha256(key.encode()).hexdigest()
# --- Request handler with idempotency check ---
async def handle_order_request(
order_data: dict,
idempotency_key: str | None,
store: IdempotencyStore,
) -> dict:
"""Process an order with full idempotency protection.
If the same key arrives twice (due to client retry), return the original
response without re-processing — preventing duplicate charges or orders.
Raises ValueError if no idempotency key is provided.
"""
if not idempotency_key:
raise ValueError(
"Missing required header: X-Idempotency-Key. "
"Required to ensure request idempotency for retry safety."
)
key_hash = hash_idempotency_key(idempotency_key)
# Fast path: check if we already processed this request
cached = await store.get(key_hash)
if cached is not None:
return {"status": "already_processed", "original_response": cached.response_body}
# Execute the actual business logic
# ... (order creation, payment processing, etc.) ...
result = {"status": "created", "order_id": f"ORD-{hash(key_hash)[:8]}"}
# Cache the response for future duplicate requests
await store.set(IdempotencyRecord(
key_hash=key_hash,
request_body=str(order_data),
response_body=str(result),
created_at=time.time(),
expires_at=time.time() + 3600, # 1 hour TTL
))
return result
Constraints
MUST DO
- Always include an idempotency key on HTTP operations that may be retried (POST with
X-Idempotency-Key, or use PUT for idempotent resources) - Configure circuit breaker on every external service call, never rely on timeouts alone
- Use exponential backoff with random jitter — never fixed delay retries (causes thundering herd)
- Design sagas so that every forward step has a compensating action that can run independently
- Route poisoned messages to dead-letter queues after N failed attempts; never retry indefinitely
- Log all integration failures with correlation ID, service names, and failure reason for distributed tracing
MUST NOT DO
- Call external services synchronously in a chain without circuit breakers (cascading failure risk)
- Use string comparison on idempotency keys — always hash them before storing
- Implement saga compensation as nested try/finally inside the execute step (compensation must be independent and testable)
- Skip idempotency for non-GET HTTP operations — clients will retry, servers must handle duplicates
- Put business logic inside adapters — adapters translate, they don't implement domain behavior
- Use fixed retry delays of less than 1 second or more than 60 seconds
Output Template
When implementing or reviewing an integration pattern, produce:
- Integration Style Selection — Which pattern (REST, gRPC, events, saga) and why, with trade-off analysis
- Architecture Diagram — ASCII art showing services, data flow, and failure boundaries
- Implementation Code — Complete, typed Python functions/classes matching the selected patterns
- Resilience Configuration — Circuit breaker thresholds, retry settings, DLQ routing rules
- Idempotency Strategy — How duplicate requests are detected and handled
Related Skills
| Skill | Purpose |
|---|---|
anti-corruption-layer |
Design ACLs to isolate domain models when integrating with incompatible external systems |
event-driven-patterns |
Deep dive into event sourcing, CQRS, and event stream processing patterns |
rest-api-patterns |
REST API design best practices for the services you're connecting |
microservice-resilience-patterns |
Advanced resilience strategies including bulkheads, rate limiting, and graceful degradation |