Nosql Databases

NoSQL Databases

Production-grade NoSQL database patterns with MongoDB, Redis, Cassandra, and DynamoDB.

Quick Start

# MongoDB with PyMongo
from pymongo import MongoClient
from pymongo.errors import DuplicateKeyError
from datetime import datetime

client = MongoClient("mongodb://localhost:27017/")
db = client.analytics
events = db.events

# Create index for query performance
events.create_index([("user_id", 1), ("timestamp", -1)])
events.create_index([("event_type", 1)])

# Insert with retry pattern
def insert_event(event: dict, retries: int = 3):
    event["_id"] = f"{event['user_id']}_{event['timestamp'].isoformat()}"
    event["created_at"] = datetime.utcnow()

    for attempt in range(retries):
        try:
            events.insert_one(event)
            return True
        except DuplicateKeyError:
            return False  # Already exists
        except Exception as e:
            if attempt == retries - 1:
                raise
    return False

# Aggregation pipeline
pipeline = [
    {"$match": {"event_type": "purchase", "timestamp": {"$gte": datetime(2024, 1, 1)}}},
    {"$group": {"_id": "$user_id", "total_purchases": {"$sum": "$amount"}, "count": {"$sum": 1}}},
    {"$sort": {"total_purchases": -1}},
    {"$limit": 100}
]
top_customers = list(events.aggregate(pipeline))

Core Concepts

1. Redis for Caching & Real-time

import redis
import json
from datetime import timedelta

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# Cache pattern with TTL
def get_user_profile(user_id: str) -> dict:
    cache_key = f"user:{user_id}:profile"

    # Try cache first
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)

    # Cache miss - fetch from DB
    profile = fetch_from_database(user_id)

    # Set with 1 hour TTL
    r.setex(cache_key, timedelta(hours=1), json.dumps(profile))
    return profile

# Rate limiting
def check_rate_limit(user_id: str, limit: int = 100, window: int = 60) -> bool:
    key = f"rate:{user_id}:{int(time.time()) // window}"
    current = r.incr(key)

    if current == 1:
        r.expire(key, window)

    return current <= limit

# Real-time leaderboard with sorted sets
def update_leaderboard(user_id: str, score: float):
    r.zadd("leaderboard:daily", {user_id: score})

def get_top_users(n: int = 10) -> list:
    return r.zrevrange("leaderboard:daily", 0, n-1, withscores=True)

# Pub/Sub for event streaming
def publish_event(channel: str, event: dict):
    r.publish(channel, json.dumps(event))

def subscribe_events(channel: str):
    pubsub = r.pubsub()
    pubsub.subscribe(channel)
    for message in pubsub.listen():
        if message['type'] == 'message':
            yield json.loads(message['data'])

2. DynamoDB Patterns

import boto3
from boto3.dynamodb.conditions import Key, Attr
from decimal import Decimal

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Events')

# Single table design pattern
def put_event(event: dict):
    item = {
        'PK': f"USER#{event['user_id']}",
        'SK': f"EVENT#{event['timestamp']}#{event['event_id']}",
        'GSI1PK': f"TYPE#{event['event_type']}",
        'GSI1SK': f"DATE#{event['timestamp'][:10]}",
        'data': event
    }
    table.put_item(Item=item)

# Query by user
def get_user_events(user_id: str, limit: int = 100):
    response = table.query(
        KeyConditionExpression=Key('PK').eq(f"USER#{user_id}") & Key('SK').begins_with("EVENT#"),
        ScanIndexForward=False,
        Limit=limit
    )
    return response['Items']

# Query by event type (using GSI)
def get_events_by_type(event_type: str, date: str):
    response = table.query(
        IndexName='GSI1',
        KeyConditionExpression=Key('GSI1PK').eq(f"TYPE#{event_type}") & Key('GSI1SK').eq(f"DATE#{date}")
    )
    return response['Items']

# Batch write with exponential backoff
def batch_write_events(events: list):
    with table.batch_writer() as batch:
        for event in events:
            batch.put_item(Item=event)

3. Cassandra for Time Series

from cassandra.cluster import Cluster
from cassandra.query import BatchStatement, SimpleStatement
from datetime import datetime

cluster = Cluster(['node1', 'node2', 'node3'])
session = cluster.connect('analytics')

# Create table with time-based partitioning
session.execute("""
    CREATE TABLE IF NOT EXISTS events_by_day (
        date date,
        user_id uuid,
        event_time timestamp,
        event_type text,
        data text,
        PRIMARY KEY ((date), event_time, user_id)
    ) WITH CLUSTERING ORDER BY (event_time DESC)
""")

# Insert with prepared statement
insert_stmt = session.prepare("""
    INSERT INTO events_by_day (date, user_id, event_time, event_type, data)
    VALUES (?, ?, ?, ?, ?)
""")

def insert_event(event: dict):
    session.execute(insert_stmt, [
        event['timestamp'].date(),
        event['user_id'],
        event['timestamp'],
        event['event_type'],
        json.dumps(event['data'])
    ])

# Query by date range
def get_events_for_date(date: datetime.date):
    rows = session.execute(
        "SELECT * FROM events_by_day WHERE date = %s",
        [date]
    )
    return list(rows)

Tools & Technologies

Troubleshooting Guide

Best Practices

# ✅ DO: Design for access patterns (NoSQL)
# Primary key = partition key + sort key

# ✅ DO: Use connection pooling
pool = redis.ConnectionPool(max_connections=20)
r = redis.Redis(connection_pool=pool)

# ✅ DO: Set TTLs on cache data
r.setex(key, ttl_seconds, value)

# ✅ DO: Handle eventual consistency
# Read-your-writes with consistent reads where needed

# ❌ DON'T: Use NoSQL for complex joins
# ❌ DON'T: Store unbounded data in single document
# ❌ DON'T: Ignore partition sizing

Resources

• MongoDB University
• Redis University
• DynamoDB Guide

---

Skill Certification Checklist:

• [ ] Can design document schemas for MongoDB
• [ ] Can implement caching patterns with Redis
• [ ] Can model time series data in Cassandra
• [ ] Can use DynamoDB single-table design
• [ ] Can choose appropriate consistency levels

Nosql Databases Guide

Overview

This guide provides comprehensive documentation for the nosql-databases skill in the custom-plugin-data-engineer plugin.

Category: Database

Quick Start

Prerequisites

• Familiarity with database concepts
• Development environment set up
• Plugin installed and configured

Basic Usage

# Invoke the skill
claude "nosql-databases - [your task description]"

# Example
claude "nosql-databases - analyze the current implementation"

Core Concepts

Key Principles

1. Consistency - Follow established patterns 2. Clarity - Write readable, maintainable code 3. Quality - Validate before deployment

Best Practices

• Always validate input data
• Handle edge cases explicitly
• Document your decisions
• Write tests for critical paths

Common Tasks

Task 1: Basic Implementation

# Example implementation pattern
def implement_nosql_databases(input_data):
    """
    Implement nosql-databases functionality.

    Args:
        input_data: Input to process

    Returns:
        Processed result
    """
    # Validate input
    if not input_data:
        raise ValueError("Input required")

    # Process
    result = process(input_data)

    # Return
    return result

Task 2: Advanced Usage

For advanced scenarios, consider:

• Configuration customization via assets/config.yaml
• Validation using scripts/validate.py
• Integration with other skills

Troubleshooting

Common Issues

Related Resources

• SKILL.md - Skill specification
• config.yaml - Configuration options
• validate.py - Validation script

---

Last updated: 2025-12-30

Nosql Databases Patterns

Design Patterns

Pattern 1: Input Validation

Always validate input before processing:

def validate_input(data):
    if data is None:
        raise ValueError("Data cannot be None")
    if not isinstance(data, dict):
        raise TypeError("Data must be a dictionary")
    return True

Pattern 2: Error Handling

Use consistent error handling:

try:
    result = risky_operation()
except SpecificError as e:
    logger.error(f"Operation failed: {e}")
    handle_error(e)
except Exception as e:
    logger.exception("Unexpected error")
    raise

Pattern 3: Configuration Loading

Load and validate configuration:

import yaml

def load_config(config_path):
    with open(config_path) as f:
        config = yaml.safe_load(f)
    validate_config(config)
    return config

Anti-Patterns to Avoid

❌ Don't: Swallow Exceptions

# BAD
try:
    do_something()
except:
    pass

✅ Do: Handle Explicitly

# GOOD
try:
    do_something()
except SpecificError as e:
    logger.warning(f"Expected error: {e}")
    return default_value

Category-Specific Patterns: Database

Recommended Approach

1. Start with the simplest implementation 2. Add complexity only when needed 3. Test each addition 4. Document decisions

Common Integration Points

• Configuration: assets/config.yaml
• Validation: scripts/validate.py
• Documentation: references/GUIDE.md

---

Pattern library for nosql-databases skill