Usage

This guide covers all aspects of using Nicolas Cache in your projects.

Basic Operations

Initialization

Create a cache instance with your preferred backend:

from nicolas.cache import Cache

# In-memory cache
cache = Cache(backend="memory")

# Redis cache
cache = Cache(
    backend="redis",
    host="localhost",
    port=6379,
    db=0,
    password=None,
    prefix="myapp:"  # Optional key prefix
)

# Redis Sentinel cache
cache = Cache(
    backend="redis-sentinel",
    sentinels=[("localhost", 26379)],
    service_name="mymaster"
)

Setting Values

Store any Python object that can be pickled:

# Simple values
cache.set("name", "Alice")
cache.set("count", 42)
cache.set("active", True)

# Complex objects
cache.set("user", {
    "id": 1,
    "name": "Alice",
    "roles": ["admin", "user"]
})

# With tags
cache.set("session:123", session_data, tags=["sessions", "user:1"])

# With TTL (Redis backends only)
cache.set("token", "abc123", ttl=3600)  # Expires in 1 hour

Getting Values

Retrieve stored values:

# Get a value
name = cache.get("name")

# Returns None if key doesn't exist
missing = cache.get("non-existent")
assert missing is None

# Get with default value
value = cache.get("key") or "default"

# Get all values
all_data = cache.getall()
for key, value in all_data.items():
    print(f"{key}: {value}")

Checking Existence

if cache.exists("user:123"):
    user = cache.get("user:123")
else:
    user = fetch_from_database(123)
    cache.set("user:123", user)

Deleting Values

# Delete a single key
deleted = cache.delete("old_key")
print(f"Deleted: {deleted}")  # True if existed, False otherwise

# Delete multiple keys using tags
count = cache.delete_by_tag("temporary")
print(f"Deleted {count} entries")

Advanced Tagging

Tags provide powerful cache management capabilities.

Tag Strategies

# Hierarchical tags
cache.set("product:123", product_data, tags=[
    "products",
    "category:electronics",
    "brand:apple",
    "store:main"
])

# User-specific tags
cache.set(f"cart:{user_id}", cart_data, tags=[
    "carts",
    f"user:{user_id}",
    "active_sessions"
])

# Time-based tags
from datetime import date
today = date.today().isoformat()
cache.set("report", data, tags=[
    "reports",
    f"date:{today}",
    f"month:{today[:7]}"
])

Bulk Operations

# Cache multiple related items
for product in products:
    cache.set(
        f"product:{product['id']}",
        product,
        tags=["products", f"category:{product['category']}"]
    )

# Get all products in a category
electronics = cache.get_by_tag("category:electronics")

# Invalidate a category
cache.delete_by_tag("category:electronics")

Cache Invalidation

def update_user(user_id, new_data):
    # Update database
    save_to_database(user_id, new_data)

    # Invalidate all user-related cache
    cache.delete_by_tag(f"user:{user_id}")

def clear_old_sessions():
    # Clear all session data
    cache.delete_by_tag("sessions")

Pattern Examples

Caching Database Queries

def get_user(user_id):
    cache_key = f"user:{user_id}"

    # Try cache first
    user = cache.get(cache_key)
    if user is not None:
        return user

    # Fetch from database
    user = db.query("SELECT * FROM users WHERE id = ?", user_id)

    # Cache with appropriate tags
    if user:
        cache.set(cache_key, user, tags=[
            "users",
            f"org:{user['org_id']}",
            f"role:{user['role']}"
        ])

    return user

API Response Caching

import hashlib
import json

def cached_api_call(endpoint, params):
    # Create cache key from endpoint and params
    cache_key = hashlib.md5(
        f"{endpoint}:{json.dumps(params, sort_keys=True)}".encode()
    ).hexdigest()

    # Check cache
    response = cache.get(cache_key)
    if response is not None:
        return response

    # Make API call
    response = requests.get(endpoint, params=params).json()

    # Cache with TTL
    cache.set(cache_key, response,
              tags=["api_responses", f"endpoint:{endpoint}"],
              ttl=300)  # 5 minutes

    return response

Session Management

class SessionManager:
    def __init__(self, cache):
        self.cache = cache

    def create_session(self, user_id, data):
        session_id = generate_session_id()
        session_key = f"session:{session_id}"

        session_data = {
            "user_id": user_id,
            "created_at": time.time(),
            **data
        }

        self.cache.set(
            session_key,
            session_data,
            tags=["sessions", f"user:{user_id}"],
            ttl=3600  # 1 hour
        )

        return session_id

    def get_session(self, session_id):
        return self.cache.get(f"session:{session_id}")

    def destroy_session(self, session_id):
        return self.cache.delete(f"session:{session_id}")

    def destroy_user_sessions(self, user_id):
        return self.cache.delete_by_tag(f"user:{user_id}")

Performance Optimization

Batch Operations

# Instead of multiple individual sets
for item in items:
    cache.set(f"item:{item['id']}", item)  # Multiple round trips

# Consider grouping related data
cache.set("items:batch:1", items[:100], tags=["items"])
cache.set("items:batch:2", items[100:200], tags=["items"])

Cache Warming

def warm_cache():
    """Pre-populate cache with frequently accessed data."""
    # Load popular products
    popular_products = db.query("SELECT * FROM products WHERE popular = 1")
    for product in popular_products:
        cache.set(
            f"product:{product['id']}",
            product,
            tags=["products", "popular"]
        )

    # Load configuration
    config = load_configuration()
    cache.set("config", config, tags=["system"])

Cache Aside Pattern

class CacheAsideRepository:
    def __init__(self, cache, db):
        self.cache = cache
        self.db = db

    def get(self, key):
        # Check cache
        value = self.cache.get(key)
        if value is not None:
            return value

        # Load from database
        value = self.db.get(key)
        if value:
            # Update cache
            self.cache.set(key, value)

        return value

    def update(self, key, value):
        # Update database
        self.db.update(key, value)
        # Invalidate cache
        self.cache.delete(key)

    def delete(self, key):
        # Delete from database
        self.db.delete(key)
        # Remove from cache
        self.cache.delete(key)

Error Handling

from nicolas.cache import Cache
import logging

logger = logging.getLogger(__name__)

class ResilientCache:
    """Cache wrapper with fallback behavior."""

    def __init__(self, primary_backend="redis", fallback_backend="memory"):
        try:
            self.cache = Cache(backend=primary_backend)
            self.fallback = None
        except Exception as e:
            logger.warning(f"Primary cache failed: {e}, using fallback")
            self.cache = Cache(backend=fallback_backend)
            self.fallback = True

    def get(self, key, default=None):
        try:
            value = self.cache.get(key)
            return value if value is not None else default
        except Exception as e:
            logger.error(f"Cache get failed: {e}")
            return default

    def set(self, key, value, **kwargs):
        try:
            self.cache.set(key, value, **kwargs)
            return True
        except Exception as e:
            logger.error(f"Cache set failed: {e}")
            return False