Database Domain Specialist

Quick Reference (30 seconds)

Enterprise Database Expertise - Comprehensive database patterns and implementations covering PostgreSQL, MongoDB, Redis, and advanced data management for scalable modern applications.

Core Capabilities:

• PostgreSQL: Advanced relational patterns, optimization, and scaling
• MongoDB: Document modeling, aggregation, and NoSQL performance tuning
• Redis: In-memory caching, real-time analytics, and distributed systems
• Multi-Database: Hybrid architectures and data integration patterns
• Performance: Query optimization, indexing strategies, and scaling
• Operations: Connection management, migrations, and monitoring

When to Use:

• Designing database schemas and data models
• Implementing caching strategies and performance optimization
• Building scalable data architectures
• Working with multi-database systems
• Optimizing database queries and performance

Implementation Guide (5 minutes)

Quick Start Workflow

Database Stack Initialization:

from moai_domain_database import DatabaseManager

# Initialize multi-database stack
db_manager = DatabaseManager()

# Configure PostgreSQL for relational data
postgresql = db_manager.setup_postgresql(
 connection_string="postgresql://...",
 connection_pool_size=20,
 enable_query_logging=True
)

# Configure MongoDB for document storage
mongodb = db_manager.setup_mongodb(
 connection_string="mongodb://...",
 database_name="app_data",
 enable_sharding=True
)

# Configure Redis for caching and real-time features
redis = db_manager.setup_redis(
 connection_string="redis://...",
 max_connections=50,
 enable_clustering=True
)

# Use unified database interface
user_data = db_manager.get_user_with_profile(user_id)
analytics = db_manager.get_user_analytics(user_id, time_range="30d")

Single Database Operations:

# PostgreSQL schema migration
moai db:migrate --database postgresql --migration-file schema_v2.sql

# MongoDB aggregation pipeline
moai db:aggregate --collection users --pipeline analytics_pipeline.json

# Redis cache warming
moai db:cache:warm --pattern "user:*" --ttl 3600

Core Components

1. PostgreSQL (modules/postgresql.md)

• Advanced schema design and constraints
• Complex query optimization and indexing
• Window functions and CTEs
• Partitioning and materialized views
• Connection pooling and performance tuning

2. MongoDB (modules/mongodb.md)

• Document modeling and schema design
• Aggregation pipelines for analytics
• Indexing strategies and performance
• Sharding and scaling patterns
• Data consistency and validation

3. Redis (modules/redis.md)

• Multi-layer caching strategies
• Real-time analytics and counting
• Distributed locking and coordination
• Pub/sub messaging and streams
• Advanced data structures (HyperLogLog, Geo)

Advanced Patterns (10+ minutes)

Multi-Database Architecture

Polyglot Persistence Pattern:

class DataRouter:
 def __init__(self):
 self.postgresql = PostgreSQLConnection()
 self.mongodb = MongoDBConnection()
 self.redis = RedisConnection()

 def get_user_profile(self, user_id):
 # Get structured user data from PostgreSQL
 user = self.postgresql.get_user(user_id)

 # Get flexible profile data from MongoDB
 profile = self.mongodb.get_user_profile(user_id)

 # Get real-time status from Redis
 status = self.redis.get_user_status(user_id)

 return self.merge_user_data(user, profile, status)

 def update_user_data(self, user_id, data):
 # Route different data types to appropriate databases
 if 'structured_data' in data:
 self.postgresql.update_user(user_id, data['structured_data'])

 if 'profile_data' in data:
 self.mongodb.update_user_profile(user_id, data['profile_data'])

 if 'real_time_data' in data:
 self.redis.set_user_status(user_id, data['real_time_data'])

 # Invalidate cache across databases
 self.invalidate_user_cache(user_id)

Data Synchronization:

class DataSyncManager:
 def sync_user_data(self, user_id):
 # Sync from PostgreSQL to MongoDB for search
 pg_user = self.postgresql.get_user(user_id)
 search_document = self.create_search_document(pg_user)
 self.mongodb.upsert_user_search(user_id, search_document)

 # Update cache in Redis
 cache_data = self.create_cache_document(pg_user)
 self.redis.set_user_cache(user_id, cache_data, ttl=3600)

Performance Optimization

Query Performance Analysis:

# PostgreSQL query optimization
def analyze_query_performance(query):
 explain_result = postgresql.execute(f"EXPLAIN (ANALYZE, BUFFERS) {query}")
 return QueryAnalyzer(explain_result).get_optimization_suggestions()

# MongoDB aggregation optimization
def optimize_aggregation_pipeline(pipeline):
 optimizer = AggregationOptimizer()
 return optimizer.optimize_pipeline(pipeline)

# Redis performance monitoring
def monitor_redis_performance():
 metrics = redis.info()
 return PerformanceAnalyzer(metrics).get_recommendations()

Scaling Strategies:

# Read replicas for PostgreSQL
read_replicas = postgresql.setup_read_replicas([
 "postgresql://replica1...",
 "postgresql://replica2..."
])

# Sharding for MongoDB
mongodb.setup_sharding(
 shard_key="user_id",
 num_shards=4
)

# Redis clustering
redis.setup_cluster([
 "redis://node1:7000",
 "redis://node2:7000",
 "redis://node3:7000"
])

Works Well With

Complementary Skills:

• moai-domain-backend - API integration and business logic
• moai-foundation-core - Database migration and schema management
• moai-workflow-project - Database project setup and configuration
• moai-platform-baas - BaaS database integration patterns

Technology Integration:

• ORMs and ODMs (SQLAlchemy, Mongoose, TypeORM)
• Connection pooling (PgBouncer, connection pools)
• Migration tools (Alembic, Flyway)
• Monitoring (pg_stat_statements, MongoDB Atlas)
• Cache invalidation and synchronization

Usage Examples

Database Operations

# PostgreSQL advanced queries
users = postgresql.query(
 "SELECT * FROM users WHERE created_at > %s ORDER BY activity_score DESC LIMIT 100",
 [datetime.now() - timedelta(days=30)]
)

# MongoDB analytics
analytics = mongodb.aggregate('events', [
 {"$match": {"timestamp": {"$gte": start_date}}},
 {"$group": {"_id": "$type", "count": {"$sum": 1}}},
 {"$sort": {"count": -1}}
])

# Redis caching operations
async def get_user_data(user_id):
 cache_key = f"user:{user_id}"
 data = await redis.get(cache_key)

 if not data:
 data = fetch_from_database(user_id)
 await redis.setex(cache_key, 3600, json.dumps(data))

 return json.loads(data)

Multi-Database Transactions

async def create_user_with_profile(user_data, profile_data):
 try:
 # Start transaction across databases
 async with transaction_manager():
 # Create user in PostgreSQL
 user_id = await postgresql.insert_user(user_data)

 # Create profile in MongoDB
 await mongodb.insert_user_profile(user_id, profile_data)

 # Set initial cache in Redis
 await redis.set_user_cache(user_id, {
 "id": user_id,
 "status": "active",
 "created_at": datetime.now().isoformat()
 })

 return user_id

 except Exception as e:
 # Automatic rollback across databases
 logger.error(f"User creation failed: {e}")
 raise

Technology Stack

Relational Database:

• PostgreSQL 14+ (primary)
• MySQL 8.0+ (alternative)
• Connection pooling (PgBouncer, SQLAlchemy)

NoSQL Database:

• MongoDB 6.0+ (primary)
• Document modeling and validation
• Aggregation framework
• Sharding and replication

In-Memory Database:

• Redis 7.0+ (primary)
• Redis Stack for advanced features
• Clustering and high availability
• Advanced data structures

Supporting Tools:

• Migration tools (Alembic, Flyway)
• Monitoring (Prometheus, Grafana)
• ORMs/ODMs (SQLAlchemy, Mongoose)
• Connection management

Performance Features:

• Query optimization and analysis
• Index management and strategies
• Caching layers and invalidation
• Load balancing and failover

For detailed implementation patterns and database-specific optimizations, see the modules/ directory.