Database Design

Database Design

What I Do

I provide comprehensive guidance on database design principles, schema modeling, normalization, indexing strategies, and architectural decisions. I help choose between relational and NoSQL approaches, design for scalability, and implement data integrity patterns.

When to Use Me

• Starting new database projects
• Schema migration and refactoring
• Performance optimization
• Choosing database technology
• Normalization/denormalization decisions
• Microservices data architecture
• Data migration strategies
• Audit and compliance requirements

Core Concepts

• Normalization: Reducing data redundancy (1NF, 2NF, 3NF, BCNF)
• Denormalization: Intentionally adding redundancy for performance
• Primary Keys: Unique entity identifiers (surrogate vs natural)
• Foreign Keys: Referential integrity constraints
• Indexes: Performance optimization structures
• Cardinality: Uniqueness of column values
• Selectivity: Index efficiency measurement
• Partitioning: Horizontal data distribution
• Sharding: Cross-node data distribution
• ACID: Transaction guarantees
• CAP Theorem: Consistency vs Availability trade-offs
• CQRS: Command Query Responsibility Segregation

Code Examples

Relational Schema Design (SQLAlchemy)

from sqlalchemy import Column, Integer, String, ForeignKey, DateTime, Boolean, Index
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    email = Column(String(255), unique=True, nullable=False, index=True)
    username = Column(String(50), unique=True, nullable=False = Column(String()
    password_hash255), nullable=False)
    is_active = Column(Boolean, default=True, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
    updated_at = Column(DateTime, default=datetime.utcnow,
    
    __table_args__ = (
        Index('idx_users_active_email', 'email', 'is_active'),
    )

class Post(Base):
    __tablename__ = 'posts'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    user_id = Column(Integer, ForeignKey('users.id', nullable=False)
    title = Column(String(255), nullable=False)
    slug = Column(String(255), unique=True, nullable=False)
    content = Column(String, nullable=True)
    published_at = Column(DateTime, nullable=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow,
    
    author = relationship('User', back_populates='posts')
    
    __table_args__ = (
        Index('idx_posts_user_published', 'user_id', 'published_at'),
        ForeignKeyConstraint(['user_id'], ['users.id'],
    )

User.posts = relationship('Post', order_by='Post.created_at.desc()', back_populates='author')

NoSQL Document Schema (MongoDB)

from datetime import datetime
from typing import Optional, List, Dict, Any

def create_product_schema() -> dict:
    return {
        'bsonType': 'object',
        'required': ['name', 'sku', 'price', 'inventory'],
        'properties': {
            'name': {
                'bsonType': 'string',
                'description': 'Product name'
            },
            'sku': {
                'bsonType': 'string',
                'description': 'Stock keeping unit'
            },
            'price': {
                'bsonType': 'number',
                'minimum': 0,
                'description': 'Product price'
            },
            'inventory': {
                'bsonType': 'object',
                'required': ['quantity', 'warehouse'],
                'properties': {
                    'quantity': {'bsonType': 'int', 'minimum': 0},
                    'warehouse': {'bsonType': 'string'},
                    'reserved': {'bsonType': 'int', 'minimum': 0}
                }
            },
            'categories': {
                'bsonType': 'array',
                'items': {'bsonType': 'string'}
            },
            'attributes': {
                'bsonType': 'object',
                'additionalProperties': True
            },
            'created_at': {'bsonType': 'date'},
            'updated_at': {'bsonType': 'date'}
        }
    }

Audit Trail Pattern

from datetime import datetime
from typing import Optional, Dict, Any

class AuditLogMixin:
    def to_audit_record(self, action: str, user_id: Optional[int] = None) -> Dict[str, Any]:
        return {
            'entity_type': self.__class__.__name__,
            'entity_id': getattr(self, 'id', None),
            'action': action,
            'changes': self._get_changes(),
            'user_id': user_id,
            'timestamp': datetime.utcnow().isoformat()
        }
    
    def _get_changes(self) -> Dict[str, Any]:
        return {}

class AuditableMixin:
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._original_values = {}
    
    def __setattr__(self, key: str, value: Any) -> None:
        if hasattr(self, 'id') and key not in ('_original_values',):
            if key not in ('created_at', 'updated_at'):
                if key not in self._original_values:
                    self._original_values[key] = getattr(self, key, None)
        super().__setattr__(key, value)

Soft Delete Pattern

from datetime import datetime
from typing import TypeVar, Generic
from sqlalchemy import Column, DateTime, Boolean
from sqlalchemy.orm import declared_attr

T = TypeVar('T')

class SoftDeleteMixin:
    deleted_at = Column(DateTime, nullable=True)
    is_deleted = Column(Boolean, default=False, nullable=False, index=True)
    
    @declared_attr
    def __mapper_args__(cls):
        return {
            'with_polymorphic': '*',
            'always_refresh': True
        }

class SoftDeleteRepository(Generic[T]):
    def query(self):
        return self.session.query(self.model).filter(
            self.model.is_deleted == False
        )
    
    def delete(self, entity: T) -> None:
        entity.deleted_at = datetime.utcnow()
        entity.is_deleted = True
        self.session.add(entity)
    
    def hard_delete(self, entity: T) -> None:
        self.session.delete(entity)
    
    def find_deleted(self, limit: int = 100) -> list:
        return self.session.query(self.model).filter(
            self.model.is_deleted == True
        ).limit(limit).all()

Polymorphic Associations

from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint
from sqlalchemy.orm import relationship

class Comment(Base):
    __tablename__ = 'comments'
    
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.id'), nullable=False)
    content = Column(String, nullable=False)
    commentable_type = Column(String(50), nullable=False)
    commentable_id = Column(Integer, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    __mapper_args__ = {
        'polymorphic_identity': 'comment',
        'polymorphic_on': commentable_type
    }

class CommentableMixin:
    @declared_attr
    def comments(cls):
        return relationship(
            'Comment',
            primaryjoin=f'and_(Comment.commentable_id=={cls.__name__}.id, Comment.commentable_type=="{cls.__name__}")',
            backref='commentable',
            lazy='dynamic'
        )

class Post(CommentableMixin, Base):
    __tablename__ = 'posts'
    
    id = Column(Integer, primary_key=True)
    title = Column(String(255), nullable=False)
    content = Column(String)
    
    def add_comment(self, user_id: int, content: str) -> Comment:
        return Comment(
            user_id=user_id,
            content=content,
            commentable_type='Post',
            commentable_id=self.id
        )

Best Practices

1. Start Simple: Design for current needs, not hypothetical future scale
2. Choose the Right Tool: Match database type to data access patterns
3. Normalize First: Start with normalized design, denormalize based on performance needs
4. Index Strategically: Index columns used in WHERE, JOIN, ORDER BY clauses
5. Use Appropriate Data Types: Choose smallest type that fits your data
6. Plan for Growth: Consider partitioning and sharding from the start
7. Document Your Schema: Maintain ER diagrams and data dictionaries
8. Implement Soft Deletes: Never truly delete critical business data
9. Version Your Migrations: Use migration tools (Alembic, Flyway)
10. Test Performance Early: Profile queries with realistic data volumes
11. Use Constraints: Enforce data integrity at the database level
12. Separate Concerns: Use different databases for different access patterns

Common Patterns

Event Sourcing:

class EventStore:
    def append(self, aggregate_id: str, events: list) -> None:
        for event in events:
            self.session.execute(
                """
                INSERT INTO events (aggregate_id, event_type, data, created_at)
                VALUES (%s, %s, %s, %s)
                """,
                (aggregate_id, event['type'], json.dumps(event['data']), datetime.utcnow())
            )
    
    def get_events(self, aggregate_id: str) -> list:
        return self.session.execute(
            "SELECT * FROM events WHERE aggregate_id = ? ORDER BY created_at",
            (aggregate_id,)
        ).fetchall()

Saga Pattern for Distributed Transactions:

class OrderSaga:
    def execute(self, order_id: int) -> bool:
        try:
            self.create_order(order_id)
            self.reserve_inventory(order_id)
            self.process_payment(order_id)
            self.confirm_order(order_id)
            return True
        except SagaException as e:
            self.rollback(order_id, e.step)
            return False
    
    def rollback(self, order_id: int, failed_step: str) -> None:
        if failed_step == 'inventory':
            self.release_inventory(order_id)
        if failed_step == 'payment':
            self.refund_payment(order_id)
        self.cancel_order(order_id)

Data Vault Modeling:

# Hub (business keys)
class CustomerHub:
    customer_id = Column(String, primary_key=True)
    load_date = Column(DateTime, primary_key=True)
    record_source = Column(String)

# Satellite (contextual data)
class CustomerSatellite:
    customer_id = Column(String, ForeignKey('hub_customer.customer_id'), primary_key=True)
    load_date = Column(DateTime, primary_key=True)
    hash_diff = Column(String)
    name = Column(String)
    email = Column(String)
    address = Column(String)

# Link (relationships)
class OrderLink:
    order_id = Column(String, primary_key=True)
    load_date = Column(DateTime, primary_key=True)
    customer_id = Column(String, ForeignKey('hub_customer.customer_id'))
    product_id = Column(String, ForeignKey('hub_product.product_id'))