Composable Architecture: Building Flexible Enterprise Systems with MACH and API-First Design

# products.graphql — owned by catalog service
type Product @key(fields: "id") {
    id: ID!
    name: String!
    price: Float!
    currency: String!
    category: String
    inStock: Boolean
}

extend type Query {
    products(category: String): [Product!]!
    product(id: ID!): Product
}

# cart.graphql — cart service extends Product with cart-specific data
type CartItem @key(fields: "id") {
    id: ID!
    productId: ID!
    quantity: Int!
}

extend type Product @key(fields: "id") {
    id: ID! @external
    cartQuantity: Int @requires(fields: "id")
}

extend type Query {
    cart(userId: ID!): [CartItem!]!
}

# router.yaml — Apollo Federation Router
supergraph:
  listen: 0.0.0.0:4000
  cors:
    origins:
      - https://www.example.com
      - https://admin.example.com

rhai:
  scripts: ./rhai-scripts
  main: main.rhai

headers:
  all:
    request:
      - propagate:
          named: "authorization"

# Rate limiting per subgraph
subgraphs:
  products:
    routing_url: http://catalog-service:4001/graphql
  cart:
    routing_url: http://cart-service:4002/graphql
  checkout:
    routing_url: http://checkout-service:4003/graphql

# Frontend calls the federation router with one query
mutation Checkout {
    checkout {
        createOrder(input: { paymentMethod: "card" }) {
            orderId
            total
            items {
                product { name price }
                quantity
            }
            status
        }
    }
}

# catalog-service.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: catalog-service
  labels:
    app: catalog
    service: composable-commerce
spec:
  replicas: 2
  selector:
    matchLabels:
      app: catalog
  template:
    metadata:
      labels:
        app: catalog
    spec:
      containers:
      - name: catalog
        image: registry.example.com/catalog:v1.24.0
        ports:
        - containerPort: 4001
        env:
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: catalog-db
              key: url
        resources:
          requests:
            memory: "256Mi"
            cpu: "250m"
          limits:
            memory: "512Mi"
            cpu: "500m"
        livenessProbe:
          httpGet:
            path: /health
            port: 4001
---
apiVersion: v1
kind: Service
metadata:
  name: catalog-service
spec:
  selector:
    app: catalog
  ports:
  - port: 4001
    targetPort: 4001

# .github/workflows/deploy-service.yml
name: Deploy Composable Service

on:
  push:
    paths:
      - 'services/catalog/**'

jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Build and push Docker image
        run: |
          docker build -t registry.example.com/catalog:${{ github.sha }} ./services/catalog
          docker push registry.example.com/catalog:${{ github.sha }}

      - name: Deploy to Kubernetes
        run: |
          kubectl set image deployment/catalog-service \
            catalog=registry.example.com/catalog:${{ github.sha }}
          kubectl rollout status deployment/catalog-service

# Event schema for inter-service communication
from pydantic import BaseModel
from datetime import datetime
from enum import Enum
from typing import Any, Dict

class EventType(str, Enum):
    ORDER_CREATED = "order.created"
    ORDER_CANCELLED = "order.cancelled"
    INVENTORY_UPDATED = "inventory.updated"
    PAYMENT_PROCESSED = "payment.processed"
    SHIPMENT_CREATED = "shipment.created"

class DomainEvent(BaseModel):
    event_id: str
    event_type: EventType
    source_service: str
    timestamp: datetime
    data: Dict[str, Any]
    correlation_id: str  # Track across services

# Cart service publishes event
async def on_order_created(order_id: str, items: list):
    event = DomainEvent(
        event_id=str(uuid.uuid4()),
        event_type=EventType.ORDER_CREATED,
        source_service="cart",
        timestamp=datetime.utcnow(),
        data={"order_id": order_id, "items": items},
        correlation_id=order_id
    )
    await event_bus.publish("commerce.orders", event.model_dump_json())

# Inventory service subscribes
async def handle_order_created(message: str):
    event = DomainEvent.model_validate_json(message)
    for item in event.data["items"]:
        await inventory_service.reserve(item["product_id"], item["quantity"])
    await event_bus.publish("commerce.inventory", DomainEvent(
        event_id=str(uuid.uuid4()),
        event_type=EventType.INVENTORY_UPDATED,
        source_service="inventory",
        timestamp=datetime.utcnow(),
        data={"correlation_id": event.correlation_id},
        correlation_id=event.correlation_id
    ).model_dump_json())

# Istio Ambient mesh configuration for composable services
apiVersion: v1
kind: ServiceAccount
metadata:
  name: catalog-service
---
apiVersion: security.istio.io/v1beta1
kind: PeerAuthentication
metadata:
  name: catalog-mtls
spec:
  selector:
    matchLabels:
      app: catalog
  mtls:
    mode: STRICT  # All traffic encrypted
---
apiVersion: telemetry.istio.io/v1
kind: Telemetry
metadata:
  name: catalog-tracing
spec:
  selector:
    matchLabels:
      app: catalog
  tracing:
    - providers:
        - name: otel-tracer
      randomSamplingPercentage: 1.0
---
# Traffic management between composable services
apiVersion: networking.istio.io/v1
kind: VirtualService
metadata:
  name: catalog-routing
spec:
  hosts:
    - catalog-service
  http:
    - match:
        - headers:
            x-deployment-preview:
              exact: "true"
      route:
        - destination:
            host: catalog-service-preview
      timeout: 5s
    - route:
        - destination:
            host: catalog-service
          weight: 99
        - destination:
            host: catalog-service-preview
          weight: 1
      retries:
        attempts: 3
        perTryTimeout: 2s
      timeout: 10s

import asyncio
from circuitbreaker import circuit
from typing import TypeVar, Callable, Awaitable

T = TypeVar("T")

class ComposableServiceClient:
    """Client with built-in resilience patterns."""

    def __init__(self, service_name: str, base_url: str, timeout: float = 5.0):
        self.service_name = service_name
        self.base_url = base_url
        self.timeout = timeout
        self.fallback_cache = {}

    @circuit(failure_threshold=5, recovery_timeout=30)
    async def call(self, endpoint: str, **kwargs) -> dict:
        """Circuit-breaker protected service call."""
        async with asyncio.timeout(self.timeout):
            async with aiohttp.ClientSession() as session:
                async with session.get(
                    f"{self.base_url}{endpoint}",
                    headers={"X-Correlation-Id": kwargs.get("correlation_id", "")},
                    **kwargs
                ) as response:
                    if response.status >= 500:
                        raise ServiceError(f"{self.service_name} returned {response.status}")
                    return await response.json()

    async def call_with_fallback(
        self, endpoint: str, fallback_key: str, **kwargs
    ) -> dict:
        """Call service with cached fallback on failure."""
        try:
            result = await self.call(endpoint, **kwargs)
            self.fallback_cache[fallback_key] = result  # Update cache on success
            return result
        except (ServiceError, asyncio.TimeoutError) as e:
            # Return cached data if available
            if fallback_key in self.fallback_cache:
                return {
                    "data": self.fallback_cache[fallback_key],
                    "source": "cache",
                    "error": str(e)
                }
            raise

# Usage
catalog_client = ComposableServiceClient("catalog", "http://catalog-service:4001")

async def get_product_details(product_id: str) -> dict:
    return await catalog_client.call_with_fallback(
        f"/products/{product_id}",
        fallback_key=f"product_{product_id}"
    )

import aiocache
from aiocache import cached
from aiocache.serializers import JsonSerializer

class CatalogServiceCache:
    """Multi-layer caching for composable catalog service."""

    @cached(
        ttl=300,  # 5 minutes
        key_builder=lambda fn, *args, **kwargs: f"product:{args[0]}",
        serializer=JsonSerializer(),
        namespace="catalog"
    )
    async def get_product(self, product_id: str) -> dict:
        """Cache product data with Redis backend."""
        return await self.db.fetch_product(product_id)

    @cached(
        ttl=60,  # 1 minute for frequently changing inventory
        key_builder=lambda fn, *args: f"inventory:{args[0]}",
        serializer=JsonSerializer(),
        namespace="inventory"
    )
    async def get_inventory(self, product_id: str) -> dict:
        return await self.inventory_client.check_stock(product_id)

    # Invalidate cache when product updates
    async def update_product(self, product_id: str, data: dict):
        await self.db.update_product(product_id, data)
        await aiocache.Cache().delete(f"catalog:product:{product_id}")

# GraphQL DataLoader prevents N+1 queries
from graphql import GraphQLResolveInfo
from aiodataloader import DataLoader

class ProductLoader(DataLoader):
    async def batch_load(self, keys: list) -> list:
        products = await self.db.fetch_products_by_ids(keys)
        product_map = {p["id"]: p for p in products}
        return [product_map.get(k) for k in keys]

product_loader = ProductLoader()

# In resolver
async def resolve_product(parent, info: GraphQLResolveInfo, id: str):
    return await product_loader.load(id)

import pact

# Consumer-side test (frontend team)
@pytest.fixture
def consumer():
    return pact.Consumer("web-frontend")

@pytest.fixture
def provider():
    return pact.Provider("catalog-service")

def test_get_product(consumer, provider):
    (consumer
     .upon_receiving("a request for product 123")
     .with_request("GET", "/products/123")
     .will_respond_with(200, body={
         "id": "123",
         "name": "Widget",
         "price": 29.99
     }))

    with provider:
        result = await catalog_client.get_product("123")
        assert result["name"] == "Widget"

import pytest
from testcontainers.postgres import PostgresContainer
from testcontainers.redis import RedisContainer

@pytest.mark.asyncio
async def test_catalog_integration():
    with PostgresContainer("postgres:16") as pg:
        with RedisContainer("redis:7") as redis:
            # Wire up real dependencies
            db_url = pg.get_connection_url()
            redis_url = redis.get_connection_url()

            # Start the catalog service with real DB and cache
            app = create_catalog_service(db_url=db_url, redis_url=redis_url)

            # Run tests against real service
            result = await app.get_product("123")
            assert result is not None

# API Gateway strangler fig routing
routes = {
    "/api/checkout": {"service": "checkout-service", "weight": 10},  # Start with 10%
    "/api/catalog": {"service": "catalog-service", "weight": 50},    # Half migrated
    "/api/orders": {"service": "order-service", "weight": 100},      # Fully migrated
    "/api/*": {"service": "monolith", "weight": 100}                 # Everything else
}