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.