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

• 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

• 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

• 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.