Database per Service

Introduction

One of the most important principles in microservices is that each service owns its data. From building distributed systems, I've learned that sharing databases between services creates tight coupling that defeats the purpose of microservices. When multiple services access the same tables, schema changes become coordination nightmares.

This article covers data ownership, polyglot persistence, and strategies for maintaining data isolation.

Why Database per Service?

Trade-offs

Aspect	Shared Database	Database per Service
Coupling	Tight	Loose
Schema Changes	Coordinate all services	Independent
Data Consistency	ACID transactions	Eventual consistency
Queries	Easy joins	Cross-service calls
Scaling	Limited options	Independent scaling
Technology	One database	Polyglot persistence

Data Ownership Principles

Bounded Context Data

# Each service owns its domain data
# user_service/models.py
from sqlalchemy import Column, String, DateTime
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()


class User(Base):
    """User entity - owned by User Service."""
    __tablename__ = "users"
    
    id = Column(String, primary_key=True)
    email = Column(String, unique=True, nullable=False)
    name = Column(String, nullable=False)
    password_hash = Column(String, nullable=False)
    created_at = Column(DateTime, nullable=False)
    
    # User service owns these fields
    # Other services should NOT access this table directly


# order_service/models.py
class Order(Base):
    """Order entity - owned by Order Service."""
    __tablename__ = "orders"
    
    id = Column(String, primary_key=True)
    user_id = Column(String, nullable=False)  # Reference only, not FK
    status = Column(String, nullable=False)
    total_amount = Column(Float, nullable=False)
    created_at = Column(DateTime, nullable=False)
    
    # Note: No foreign key to users table
    # Order service stores user_id but doesn't access user data


class OrderItem(Base):
    """Order item - owned by Order Service."""
    __tablename__ = "order_items"
    
    id = Column(String, primary_key=True)
    order_id = Column(String, ForeignKey("orders.id"), nullable=False)
    product_id = Column(String, nullable=False)  # Reference only
    product_name = Column(String, nullable=False)  # Denormalized snapshot
    quantity = Column(Integer, nullable=False)
    unit_price = Column(Float, nullable=False)

Data Duplication is Acceptable

# Order service stores product snapshots
# This is intentional - orders preserve historical data

class OrderItem(Base):
    __tablename__ = "order_items"
    
    id = Column(String, primary_key=True)
    order_id = Column(String, ForeignKey("orders.id"))
    
    # Reference to product service
    product_id = Column(String, nullable=False)
    
    # Snapshot at time of order - intentional duplication
    product_name = Column(String, nullable=False)
    product_sku = Column(String, nullable=False)
    unit_price = Column(Float, nullable=False)  # Price at order time
    
    quantity = Column(Integer, nullable=False)


class OrderService:
    async def create_order(
        self, 
        user_id: str, 
        items: list[dict],
    ) -> Order:
        """Create order with product snapshots."""
        order = Order(
            id=str(uuid.uuid4()),
            user_id=user_id,
            status="pending",
            total_amount=0,
            created_at=datetime.utcnow(),
        )
        
        total = 0
        for item in items:
            # Get current product info from Product Service
            product = await self.product_client.get_product(item["product_id"])
            
            order_item = OrderItem(
                id=str(uuid.uuid4()),
                order_id=order.id,
                product_id=product.id,
                # Store snapshot - won't change if product updates
                product_name=product.name,
                product_sku=product.sku,
                unit_price=product.price,
                quantity=item["quantity"],
            )
            
            total += product.price * item["quantity"]
            order.items.append(order_item)
        
        order.total_amount = total
        await self.repository.save(order)
        
        return order

Polyglot Persistence

Choosing the Right Database

Service	Database	Reason
User Service	PostgreSQL	ACID transactions, complex queries
Order Service	PostgreSQL	Transactional integrity
Product Catalog	Elasticsearch	Full-text search, faceted navigation
Session/Cart	Redis	High speed, TTL support
Analytics	ClickHouse/TimescaleDB	Time-series, aggregations
Documents	MongoDB	Flexible schema

Implementation Examples

# user_service/database.py - PostgreSQL
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession


class PostgresDatabase:
    def __init__(self, url: str):
        self.engine = create_async_engine(url)
    
    async def get_session(self) -> AsyncSession:
        async with AsyncSession(self.engine) as session:
            yield session


# product_service/database.py - Elasticsearch
from elasticsearch import AsyncElasticsearch


class ElasticsearchDatabase:
    def __init__(self, hosts: list[str]):
        self.client = AsyncElasticsearch(hosts)
    
    async def search_products(
        self, 
        query: str,
        filters: dict = None,
    ) -> list[dict]:
        body = {
            "query": {
                "bool": {
                    "must": [
                        {"multi_match": {
                            "query": query,
                            "fields": ["name^2", "description", "category"],
                        }}
                    ],
                    "filter": self._build_filters(filters) if filters else [],
                }
            },
        }
        
        response = await self.client.search(index="products", body=body)
        return [hit["_source"] for hit in response["hits"]["hits"]]


# cart_service/database.py - Redis
import redis.asyncio as redis


class RedisCartStore:
    def __init__(self, url: str):
        self.client = redis.from_url(url)
    
    async def get_cart(self, user_id: str) -> dict:
        cart = await self.client.hgetall(f"cart:{user_id}")
        return {k.decode(): json.loads(v) for k, v in cart.items()}
    
    async def add_item(
        self, 
        user_id: str, 
        product_id: str, 
        item: dict,
    ):
        await self.client.hset(
            f"cart:{user_id}",
            product_id,
            json.dumps(item),
        )
        # Cart expires after 7 days
        await self.client.expire(f"cart:{user_id}", 604800)
    
    async def remove_item(self, user_id: str, product_id: str):
        await self.client.hdel(f"cart:{user_id}", product_id)

Querying Across Services

API Composition

# When you need data from multiple services
# Use API composition at the application layer

class OrderQueryService:
    """Query service that composes data from multiple services."""
    
    def __init__(
        self,
        order_repo,
        user_client: UserServiceClient,
        product_client: ProductServiceClient,
    ):
        self.order_repo = order_repo
        self.user_client = user_client
        self.product_client = product_client
    
    async def get_order_details(self, order_id: str) -> dict:
        """Get order with enriched data."""
        # Get order from our database
        order = await self.order_repo.get(order_id)
        
        # Enrich with user info from User Service
        user = await self.user_client.get_user(order.user_id)
        
        # Enrich items with current product info (optional)
        enriched_items = []
        for item in order.items:
            try:
                product = await self.product_client.get_product(item.product_id)
                enriched_items.append({
                    **item.__dict__,
                    "current_price": product.price,
                    "in_stock": product.stock > 0,
                })
            except ProductNotFoundError:
                enriched_items.append({
                    **item.__dict__,
                    "current_price": None,
                    "in_stock": False,
                    "discontinued": True,
                })
        
        return {
            "order": order.__dict__,
            "customer": {
                "id": user.id,
                "name": user.name,
                "email": user.email,
            },
            "items": enriched_items,
        }


# Parallel fetching for efficiency
async def get_orders_with_users(self, order_ids: list[str]) -> list[dict]:
    """Get multiple orders with parallel user fetches."""
    orders = await asyncio.gather(*[
        self.order_repo.get(oid) for oid in order_ids
    ])
    
    # Get unique user IDs
    user_ids = list(set(o.user_id for o in orders))
    
    # Batch fetch users
    users = await self.user_client.get_users_batch(user_ids)
    user_map = {u.id: u for u in users}
    
    return [
        {
            "order": order.__dict__,
            "customer": user_map.get(order.user_id).__dict__,
        }
        for order in orders
    ]

CQRS for Complex Queries

# Maintain read-optimized views for complex queries

# Write side - Order Service
class OrderCommandHandler:
    async def create_order(self, command: CreateOrderCommand):
        order = Order.create(command)
        await self.repository.save(order)
        
        # Publish event
        await self.event_publisher.publish(
            OrderCreatedEvent(order_id=order.id, data=order.__dict__)
        )


# Read side - Query Service
class OrderReadModel:
    """Denormalized read model for order queries."""
    __tablename__ = "order_read_models"
    
    id = Column(String, primary_key=True)
    order_id = Column(String, nullable=False)
    user_id = Column(String, nullable=False)
    user_name = Column(String)  # Denormalized
    user_email = Column(String)  # Denormalized
    status = Column(String)
    total_amount = Column(Float)
    item_count = Column(Integer)
    product_names = Column(ARRAY(String))  # Denormalized
    created_at = Column(DateTime)
    updated_at = Column(DateTime)


class OrderEventHandler:
    """Update read model from events."""
    
    async def handle_order_created(self, event: dict):
        """Populate read model on order creation."""
        order_data = event["data"]
        
        # Fetch user data to denormalize
        user = await self.user_client.get_user(order_data["user_id"])
        
        read_model = OrderReadModel(
            id=str(uuid.uuid4()),
            order_id=order_data["id"],
            user_id=order_data["user_id"],
            user_name=user.name,
            user_email=user.email,
            status=order_data["status"],
            total_amount=order_data["total_amount"],
            item_count=len(order_data["items"]),
            product_names=[i["product_name"] for i in order_data["items"]],
            created_at=order_data["created_at"],
            updated_at=datetime.utcnow(),
        )
        
        await self.read_repo.save(read_model)
    
    async def handle_user_updated(self, event: dict):
        """Update denormalized user data."""
        user_data = event["data"]
        
        # Update all order read models for this user
        await self.read_repo.update_many(
            filter={"user_id": user_data["id"]},
            update={
                "user_name": user_data["name"],
                "user_email": user_data["email"],
                "updated_at": datetime.utcnow(),
            },
        )


# Now queries are simple and fast
class OrderQueryService:
    async def search_orders(
        self,
        user_email: str = None,
        product_name: str = None,
        status: str = None,
    ) -> list[OrderReadModel]:
        """Search orders using read model."""
        query = self.session.query(OrderReadModel)
        
        if user_email:
            query = query.filter(OrderReadModel.user_email.ilike(f"%{user_email}%"))
        if product_name:
            query = query.filter(OrderReadModel.product_names.any(product_name))
        if status:
            query = query.filter(OrderReadModel.status == status)
        
        return await query.all()

Schema Migration Strategies

Independent Migrations

# Each service manages its own migrations
# order_service/alembic/versions/001_initial.py

from alembic import op
import sqlalchemy as sa


def upgrade():
    op.create_table(
        "orders",
        sa.Column("id", sa.String(), primary_key=True),
        sa.Column("user_id", sa.String(), nullable=False),
        sa.Column("status", sa.String(), nullable=False),
        sa.Column("total_amount", sa.Float(), nullable=False),
        sa.Column("created_at", sa.DateTime(), nullable=False),
    )
    
    op.create_index("ix_orders_user_id", "orders", ["user_id"])


def downgrade():
    op.drop_table("orders")

Backward Compatible Changes

# Adding new columns - backward compatible
def upgrade():
    # Add column with default (safe)
    op.add_column(
        "orders",
        sa.Column("shipping_address", sa.JSON(), nullable=True),
    )


# Removing columns - two-phase approach
# Phase 1: Stop using the column in code
# Phase 2: Remove column after all instances updated

def upgrade_phase2():
    op.drop_column("orders", "legacy_field")


# Renaming columns - expand and contract pattern
# Phase 1: Add new column, write to both
def upgrade_phase1():
    op.add_column("orders", sa.Column("customer_id", sa.String()))
    # Application writes to both user_id and customer_id


# Phase 2: Migrate data, read from new column
def upgrade_phase2():
    op.execute("UPDATE orders SET customer_id = user_id WHERE customer_id IS NULL")


# Phase 3: Remove old column after all readers updated
def upgrade_phase3():
    op.drop_column("orders", "user_id")

Data Synchronization Patterns

Event-Driven Sync

class ProductCatalogSync:
    """Sync product data to local cache from events."""
    
    def __init__(self, cache: redis.Redis):
        self.cache = cache
    
    async def handle_product_created(self, event: dict):
        """Cache product on creation."""
        product = event["data"]
        await self.cache.hset(
            f"products:{product['id']}",
            mapping={
                "name": product["name"],
                "price": str(product["price"]),
                "sku": product["sku"],
            },
        )
    
    async def handle_product_updated(self, event: dict):
        """Update cached product."""
        product = event["data"]
        await self.cache.hset(
            f"products:{product['id']}",
            mapping={
                "name": product["name"],
                "price": str(product["price"]),
                "sku": product["sku"],
            },
        )
    
    async def handle_product_deleted(self, event: dict):
        """Remove product from cache."""
        await self.cache.delete(f"products:{event['aggregate_id']}")
    
    async def get_product(self, product_id: str) -> dict | None:
        """Get product from cache."""
        data = await self.cache.hgetall(f"products:{product_id}")
        if not data:
            return None
        return {
            k.decode(): v.decode() for k, v in data.items()
        }

Change Data Capture (CDC)

# Using Debezium-style CDC for database sync
# This captures changes directly from database logs

from dataclasses import dataclass
from enum import Enum


class ChangeType(Enum):
    INSERT = "insert"
    UPDATE = "update"
    DELETE = "delete"


@dataclass
class ChangeEvent:
    table: str
    change_type: ChangeType
    before: dict | None
    after: dict | None
    timestamp: datetime


class CDCHandler:
    """Handle Change Data Capture events."""
    
    def __init__(self, sync_targets: dict):
        self.sync_targets = sync_targets
    
    async def handle_change(self, event: ChangeEvent):
        """Route change to appropriate handler."""
        handler = self.sync_targets.get(event.table)
        if handler:
            await handler.sync(event)


class ProductSyncHandler:
    """Sync product changes to search index."""
    
    def __init__(self, elasticsearch: Elasticsearch):
        self.es = elasticsearch
    
    async def sync(self, event: ChangeEvent):
        if event.change_type == ChangeType.DELETE:
            await self.es.delete(
                index="products",
                id=event.before["id"],
            )
        else:
            await self.es.index(
                index="products",
                id=event.after["id"],
                body=event.after,
            )

Handling Referential Integrity

class OrderService:
    """Handle cross-service references gracefully."""
    
    async def create_order(
        self, 
        user_id: str, 
        items: list[dict],
    ) -> Order:
        """Create order with validated references."""
        # Validate user exists
        user = await self.user_client.get_user(user_id)
        if not user:
            raise ValidationError(f"User {user_id} not found")
        
        # Validate products exist and have stock
        for item in items:
            product = await self.product_client.get_product(item["product_id"])
            if not product:
                raise ValidationError(f"Product {item['product_id']} not found")
            if product.stock < item["quantity"]:
                raise ValidationError(f"Insufficient stock for {product.name}")
        
        # Create order (references validated at this point)
        order = Order(
            id=str(uuid.uuid4()),
            user_id=user_id,
            items=[...],
        )
        
        await self.repository.save(order)
        return order
    
    async def get_order(self, order_id: str) -> dict:
        """Get order handling missing references."""
        order = await self.repository.get(order_id)
        
        # User might have been deleted
        try:
            user = await self.user_client.get_user(order.user_id)
            user_info = {"id": user.id, "name": user.name}
        except UserNotFoundError:
            user_info = {"id": order.user_id, "name": "[Deleted User]"}
        
        return {
            "order": order.__dict__,
            "customer": user_info,
        }


class SoftDeleteHandler:
    """Handle soft deletes across services."""
    
    async def handle_user_deleted(self, event: dict):
        """Handle user deletion - anonymize orders."""
        user_id = event["aggregate_id"]
        
        # Don't delete orders, anonymize them
        await self.order_repo.update_many(
            filter={"user_id": user_id},
            update={
                "customer_name": "[Deleted User]",
                "customer_email": "[redacted]",
            },
        )

Practical Exercise

Exercise: Implement Database per Service

# inventory_service/models.py
from sqlalchemy import Column, String, Integer, DateTime
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()


class InventoryItem(Base):
    """Inventory owned by Inventory Service."""
    __tablename__ = "inventory"
    
    id = Column(String, primary_key=True)
    product_id = Column(String, nullable=False, unique=True)
    warehouse_id = Column(String, nullable=False)
    quantity = Column(Integer, nullable=False, default=0)
    reserved = Column(Integer, nullable=False, default=0)
    updated_at = Column(DateTime, nullable=False)
    
    @property
    def available(self) -> int:
        return self.quantity - self.reserved


class InventoryRepository:
    async def reserve(
        self, 
        product_id: str, 
        quantity: int, 
        order_id: str,
    ) -> bool:
        """Reserve inventory for order."""
        async with self.session.begin():
            item = await self.session.execute(
                select(InventoryItem)
                .where(InventoryItem.product_id == product_id)
                .with_for_update()
            )
            item = item.scalar_one_or_none()
            
            if not item or item.available < quantity:
                return False
            
            item.reserved += quantity
            item.updated_at = datetime.utcnow()
            
            # Store reservation
            reservation = Reservation(
                id=str(uuid.uuid4()),
                product_id=product_id,
                order_id=order_id,
                quantity=quantity,
            )
            self.session.add(reservation)
        
        return True
    
    async def commit_reservation(self, order_id: str):
        """Convert reservation to actual deduction."""
        async with self.session.begin():
            reservations = await self.get_reservations(order_id)
            
            for res in reservations:
                item = await self.get_item(res.product_id)
                item.quantity -= res.quantity
                item.reserved -= res.quantity
                item.updated_at = datetime.utcnow()
            
            await self.delete_reservations(order_id)
    
    async def release_reservation(self, order_id: str):
        """Release reservation (order cancelled)."""
        async with self.session.begin():
            reservations = await self.get_reservations(order_id)
            
            for res in reservations:
                item = await self.get_item(res.product_id)
                item.reserved -= res.quantity
                item.updated_at = datetime.utcnow()
            
            await self.delete_reservations(order_id)

Key Takeaways

1. Own your data - Each service has exclusive access to its database

2. No cross-database joins - Use API composition or CQRS

3. Duplicate when needed - Store snapshots for historical accuracy

4. Polyglot persistence - Choose the right database for each use case

5. Events for sync - Keep data synchronized through events

What's Next?

With separate databases, we need patterns for distributed transactions. In Article 8: Distributed Data Patterns, we'll explore Sagas, event sourcing, and CQRS.

---

This article is part of the Microservice Architecture 101 series.