smart-redis-storage

• Simple CRUD Operations — Easy-to-use methods for Create, Read, Update, and Delete operations
• JSON Serialization — Automatic serialization and deserialization of Python objects
• Hash-based Storage — Organized data storage using Redis hashes with unique keys
• TTL Support — Configurable expiration time for stored data
• Thread-safe — Built on Redis-py, suitable for multi-threaded applications
• Flexible Data Structure — Store multiple key-value pairs under a single unique identifier

pip install smart-redis-storage

Requirements:

• Python 3.6+
• Redis server
• redis-py library

from smart_redis_storage.redis_storage import RedisStorageManager

# Initialize with default Redis connection (localhost:6379)
redis_storage = RedisStorageManager()

# Store user data
user_id = 123
redis_storage.set_data(
    uniq_key=user_id,
    key='profile',
    value={'name': 'John', 'age': 30, 'email': 'john@example.com'}
)

# Retrieve user data
profile = redis_storage.get_data(uniq_key=user_id, key='profile')
print(profile)  # {'name': 'John', 'age': 30, 'email': 'john@example.com'}

# Store with expiration (30 seconds)
redis_storage.set_data(
    uniq_key=user_id,
    key='session_data',
    value={'token': 'abc123', 'last_login': '2026-01-01'},
    expiration=30
)

Custom Redis Connection:

# Connect to custom Redis instance
redis_storage = RedisStorageManager(
    host='redis-server.com',
    port=6379,
    db=1
)

Multiple Data Operations:

# Store multiple pieces of data under same unique key
user_id = 456
redis_storage.set_data(user_id, 'preferences', {'theme': 'dark', 'language': 'en'})
redis_storage.set_data(user_id, 'cart', {'items': [1, 2, 3], 'total': 150.0})

# Get all data for a user
all_user_data = redis_storage.get_all_data(user_id)
print(all_user_data)
# {'preferences': {'theme': 'dark', 'language': 'en'}, 'cart': {'items': [1, 2, 3], 'total': 150.0}}

# Update specific data
redis_storage.update_data(user_id, 'preferences', {'theme': 'light', 'language': 'en'})

# Pop data (read and remove)
cart_data = redis_storage.pop_data(user_id, 'cart')
print(cart_data)  # {'items': [1, 2, 3], 'total': 150.0}

# Check remaining TTL
ttl = redis_storage.get_ttl(user_id)
print(f"TTL: {ttl} seconds")

Data Management:

# Delete specific key
redis_storage.delete_key(user_id, 'preferences')

# Delete all data for a unique key
redis_storage.delete_all_hash(user_id)

# Check if data exists
data = redis_storage.get_data(user_id, 'nonexistent_key')
print(data)  # None

RedisStorageManager Class

Constructor: RedisStorageManager(host='localhost', port=6379, db=0)

Session Management:

# Store user session
def store_user_session(user_id, session_data):
    redis_storage.set_data(
        uniq_key=user_id,
        key='session',
        value=session_data,
        expiration=3600  # 1 hour
    )

# Retrieve and validate session
def get_valid_session(user_id):
    session = redis_storage.get_data(user_id, 'session')
    if session and session.get('valid_until') > time.time():
        return session
    return None

Cache Management:

# Cache expensive computation results
def get_cached_data(user_id, data_key, compute_function):
    cached = redis_storage.get_data(user_id, data_key)
    if cached is not None:
        return cached

    # Compute and cache if not found
    result = compute_function()
    redis_storage.set_data(user_id, data_key, result, expiration=300)
    return result

Multi-tenant Data Isolation:

# Different tenants can use the same storage with different unique keys
tenant_a_data = redis_storage.get_all_data('tenant_a')
tenant_b_data = redis_storage.get_all_data('tenant_b')

The library uses Redis hashes with the following pattern:

• Key format: uniq:{unique_identifier}
• Field: Your specified key
• Value: JSON-serialized data

Example:

uniq:123 (Hash)
  ├── profile → '{"name": "John", "age": 30}'
  ├── session → '{"token": "abc123", "last_login": "2026-01-01"}'
  └── preferences → '{"theme": "dark", "language": "en"}'

The library handles Redis connection errors and JSON serialization/deserialization automatically. In case of connection issues, Redis-py will raise appropriate exceptions that you can catch in your application.

try:
    redis_storage.set_data(user_id, 'key', {'data': 'value'})
except redis.ConnectionError as e:
    print(f"Redis connection error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

• Uses Redis hashes for efficient storage of multiple related data
• JSON serialization adds overhead but provides flexibility
• Consider data size and expiration policies for optimal performance
• Batch operations can be implemented by storing multiple keys under same unique identifier

By using this software, you agree to the full disclaimer terms.

Software provided "AS IS" without warranty. You assume all risks.

Full legal disclaimer: See DISCLAIMER.md

License: BSD 3-Clause License