Storage Examples

This section demonstrates True Storage’s core storage capabilities, including hot (in-memory), cold (disk-based), and mixed (tiered) storage strategies.

Storage Types Overview

True Storage provides three main storage types:

  1. Hot Storage: In-memory storage with LRU caching

  2. Cold Storage: Disk-based storage with compression

  3. Mixed Storage: Tiered storage combining hot and cold strategies

Hot Storage

Hot storage provides fast, in-memory data access with LRU (Least Recently Used) caching:

Hot Storage Operations
  1"""
  2Hot Storage Demo
  3
  4This demo shows operations with the True Storage hot storage system,
  5which provides in-memory caching with LRU eviction and expiration for high-performance access.
  6"""
  7
  8import logging
  9import time
 10from true_storage.storage.hot import HotStorage
 11
 12def measure_time(func):
 13    """Decorator to measure execution time."""
 14    def wrapper(*args, **kwargs):
 15        start = time.perf_counter()
 16        result = func(*args, **kwargs)
 17        end = time.perf_counter()
 18        print(f"Operation took: {(end - start)*1000:.2f}ms")
 19        return result
 20    return wrapper
 21
 22def main():
 23    # Configure logging
 24    logging.basicConfig(
 25        level=logging.INFO,
 26        format='%(asctime)s - %(levelname)s - %(message)s'
 27    )
 28
 29    print("🚀 Initializing hot storage...")
 30    # Initialize with a small max size and short expiration time for demonstration
 31    storage = HotStorage(max_size=5, expiration_time=10)
 32
 33    # Performance demonstration
 34    print("\n⚡ Performance comparison...")
 35
 36    @measure_time
 37    def store_data():
 38        """Store sample data."""
 39        for i in range(10):
 40            key = f"key_{i}"
 41            value = {"data": f"value_{i}", "size": i * 100}
 42            storage.store(key, value)
 43            print(f"Stored {key}")
 44
 45    @measure_time
 46    def retrieve_data_first_time():
 47        """First retrieval."""
 48        results = []
 49        for i in range(10):
 50            key = f"key_{i}"
 51            value = storage.retrieve(key)
 52            results.append(value)
 53            print(f"Retrieved {key}: {value}")
 54        return results
 55
 56    print("\n💾 Storing sample data...")
 57    store_data()
 58
 59    print("\n📊 Current storage stats:")
 60    stats = storage.get_stats()
 61    print(f"Storage size: {stats['size']} items")
 62    print(f"Max size: {stats['max_size']} items")
 63    print(f"Hit ratio: {stats['hit_ratio']:.2%}")
 64    print(f"Average response time: {stats['avg_response_time']:.2f}ms")
 65
 66    print("\n🔍 Retrieving data...")
 67    results = retrieve_data_first_time()
 68
 69    # Demonstrate LRU eviction
 70    print("\n🔄 Demonstrating LRU eviction...")
 71    print("Adding more items than max_size to trigger eviction...")
 72    for i in range(10, 15):
 73        key = f"key_{i}"
 74        value = {"data": f"value_{i}", "size": i * 100}
 75        storage.store(key, value)
 76        print(f"Stored {key}")
 77
 78    print("\nChecking which items remain in storage:")
 79    remaining_keys = storage.get_keys()
 80    print(f"Remaining keys: {remaining_keys}")
 81    
 82    # Demonstrate expiration
 83    print("\n⏰ Demonstrating expiration...")
 84    print("Waiting for items to expire...")
 85    time.sleep(11)  # Wait longer than expiration_time
 86    
 87    print("\nTrying to retrieve expired items:")
 88    for key in remaining_keys:
 89        value = storage.retrieve(key)
 90        print(f"Retrieving {key}: {'Found' if value else 'Expired'}")
 91
 92    # Final statistics
 93    print("\n📊 Final storage stats:")
 94    stats = storage.get_stats()
 95    print(f"Storage size: {stats['size']} items")
 96    print(f"Hit ratio: {stats['hit_ratio']:.2%}")
 97    print(f"Total operations: {stats['total_operations']}")
 98    print(f"Average response time: {stats['avg_response_time']:.2f}ms")
 99
100if __name__ == "__main__":
101    main()
Key features demonstrated:

  • In-memory caching with configurable size

  • LRU eviction for memory management

  • Thread-safe operations

  • Performance metrics and statistics

  • Optional item expiration

Usage Tips:

  • Ideal for frequently accessed data

  • Monitor memory usage with get_stats()

  • Set appropriate max_size for your use case

  • Use set_ttl() for time-sensitive data

Cold Storage

Cold storage provides persistent, disk-based storage with compression:

Cold Storage Operations
  1"""
  2Cold Storage Demo
  3
  4This demo shows operations with the True Storage cold storage system,
  5which provides compressed, long-term storage capabilities with metadata tracking.
  6"""
  7
  8import logging
  9import json
 10import tempfile
 11import time
 12from datetime import datetime
 13from true_storage.storage.cold import ColdStorage
 14
 15def format_size(size_bytes):
 16    """Format size in bytes to human readable format."""
 17    for unit in ['B', 'KB', 'MB', 'GB']:
 18        if size_bytes < 1024:
 19            return f"{size_bytes:.2f} {unit}"
 20        size_bytes /= 1024
 21    return f"{size_bytes:.2f} TB"
 22
 23def main():
 24    # Configure logging
 25    logging.basicConfig(
 26        level=logging.INFO,
 27        format='%(asctime)s - %(levelname)s - %(message)s'
 28    )
 29
 30    # Create a temporary directory for cold storage
 31    with tempfile.TemporaryDirectory() as temp_dir:
 32        print("Initializing cold storage...")
 33        storage = ColdStorage(
 34            storage_directory=temp_dir,
 35            compression_level=6  # Medium compression
 36        )
 37
 38        print("\nStoring different types of data...")
 39
 40        # 1. Store large text data
 41        print("\nStoring large text data:")
 42        large_text = "This is a sample text that will be compressed." * 1000
 43        storage.store("text_data", large_text)
 44        text_meta = storage.get_item_metadata("text_data")
 45        print(f"Stored text data (compressed size: {format_size(text_meta['size'])})")
 46
 47        # 2. Store structured data
 48        print("\nStoring structured data:")
 49        structured_data = {
 50            "project": "True Storage Demo",
 51            "created_at": datetime.now().isoformat(),
 52            "settings": {
 53                "compression": True,
 54                "backup_enabled": True,
 55                "retention_days": 30
 56            },
 57            "data_points": [
 58                {"id": i, "value": f"point_{i}"} 
 59                for i in range(100)
 60            ]
 61        }
 62        storage.store("config_data", structured_data)
 63        config_meta = storage.get_item_metadata("config_data")
 64        print(f"Stored configuration data (compressed size: {format_size(config_meta['size'])})")
 65
 66        # 3. Store binary data
 67        print("\nStoring binary data:")
 68        binary_data = bytes([i % 256 for i in range(10000)])
 69        storage.store("binary_data", binary_data)
 70        binary_meta = storage.get_item_metadata("binary_data")
 71        print(f"Stored binary data (compressed size: {format_size(binary_meta['size'])})")
 72
 73        # Display storage statistics
 74        print("\nCurrent storage statistics:")
 75        stats = storage.get_stats()
 76        print(f"Total items: {stats['total_items']}")
 77        print(f"Total size: {format_size(stats['total_size'])}")
 78        print(f"Average item size: {format_size(stats['avg_item_size'])}")
 79        print(f"Average response time: {stats['avg_response_time']:.2f}ms")
 80
 81        # Demonstrate data retrieval
 82        print("\nRetrieving and verifying data...")
 83        
 84        # Verify text data
 85        retrieved_text = storage.retrieve("text_data")
 86        print(f"Retrieved text data matches: {retrieved_text == large_text}")
 87        
 88        # Verify structured data
 89        retrieved_config = storage.retrieve("config_data")
 90        print(f"Retrieved config data matches: {retrieved_config == structured_data}")
 91        
 92        # Verify binary data
 93        retrieved_binary = storage.retrieve("binary_data")
 94        print(f"Retrieved binary data matches: {retrieved_binary == binary_data}")
 95
 96        # Show item metadata
 97        print("\nItem metadata:")
 98        for key in storage.get_keys():
 99            meta = storage.get_item_metadata(key)
100            created = datetime.fromtimestamp(meta['created_at'])
101            last_accessed = datetime.fromtimestamp(meta['last_accessed'])
102            print(f"\n{key}:")
103            print(f"  Size: {format_size(meta['size'])}")
104            print(f"  Created: {created}")
105            print(f"  Last accessed: {last_accessed}")
106
107        # Demonstrate deletion
108        print("\nDemonstrating deletion...")
109        storage.delete("binary_data")
110        print("Deleted binary data")
111        print(f"Remaining items: {storage.get_keys()}")
112
113        # Final statistics
114        print("\nFinal storage statistics:")
115        stats = storage.get_stats()
116        print(f"Total items: {stats['total_items']}")
117        print(f"Total size: {format_size(stats['total_size'])}")
118        print(f"Hit ratio: {stats['hit_ratio']:.2%}")
119        print(f"Total operations: {stats['total_operations']}")
120
121if __name__ == "__main__":
122    main()
Key features demonstrated:

  • File-based persistent storage

  • Automatic data compression

  • Metadata tracking and statistics

  • Atomic file operations

  • Directory management

Best Practices:

  • Use for large, infrequently accessed data

  • Enable compression for large datasets

  • Monitor disk usage with get_stats()

  • Implement proper error handling

  • Use context managers for safe operations

Mixed Storage

Mixed storage combines hot and cold storage for optimal performance:

Mixed Storage Operations
  1"""
  2Mixed Storage Demo
  3
  4This demo shows operations with the True Storage mixed storage system,
  5which combines hot and cold storage for optimal performance and storage efficiency.
  6"""
  7
  8import logging
  9import time
 10import tempfile
 11from datetime import datetime
 12from true_storage.storage.mixed import MixedStorage
 13
 14def format_size(size_bytes):
 15    """Format size in bytes to human readable format."""
 16    for unit in ['B', 'KB', 'MB', 'GB']:
 17        if size_bytes < 1024:
 18            return f"{size_bytes:.2f} {unit}"
 19        size_bytes /= 1024
 20    return f"{size_bytes:.2f} TB"
 21
 22def measure_time(func):
 23    """Decorator to measure execution time."""
 24    def wrapper(*args, **kwargs):
 25        start = time.perf_counter()
 26        result = func(*args, **kwargs)
 27        end = time.perf_counter()
 28        print(f"Operation took: {(end - start)*1000:.2f}ms")
 29        return result
 30    return wrapper
 31
 32def main():
 33    # Configure logging
 34    logging.basicConfig(
 35        level=logging.INFO,
 36        format='%(asctime)s - %(levelname)s - %(message)s'
 37    )
 38
 39    # Create temporary directory for cold storage
 40    with tempfile.TemporaryDirectory() as temp_dir:
 41        print("Initializing mixed storage...")
 42        storage = MixedStorage(
 43            storage_directory=temp_dir,
 44            max_size=5,  # Small hot storage size for demonstration
 45            expiration_time=10,  # Short expiration for demonstration
 46            compression_level=6
 47        )
 48
 49        print("\nDemonstrating tiered storage...")
 50
 51        # 1. Store frequently accessed data
 52        print("\nStoring frequently accessed data:")
 53        frequent_data = {
 54            "type": "frequent",
 55            "timestamp": datetime.now().isoformat(),
 56            "data": "This is frequently accessed data" * 100
 57        }
 58        storage.store("frequent_data", frequent_data)
 59
 60        # 2. Store rarely accessed data
 61        print("\nStoring rarely accessed data:")
 62        rare_data = {
 63            "type": "rare",
 64            "timestamp": datetime.now().isoformat(),
 65            "data": "This is rarely accessed data" * 100
 66        }
 67        storage.store("rare_data", rare_data)
 68
 69        # Display initial statistics
 70        print("\nInitial storage statistics:")
 71        stats = storage.get_stats()
 72        print("\nHot Storage:")
 73        hot_stats = stats["hot_storage"]
 74        print(f"Size: {hot_stats['size']} items")
 75        print(f"Max size: {hot_stats['max_size']} items")
 76        print(f"Hit ratio: {hot_stats['hit_ratio']:.2%}")
 77        print(f"Average response time: {hot_stats['avg_response_time']:.2f}ms")
 78
 79        print("\nCold Storage:")
 80        cold_stats = stats["cold_storage"]
 81        print(f"Total items: {cold_stats['total_items']}")
 82        print(f"Total size: {format_size(cold_stats['total_size'])}")
 83        print(f"Average item size: {format_size(cold_stats['avg_item_size'])}")
 84
 85        # Demonstrate access patterns
 86        print("\nSimulating access patterns...")
 87
 88        @measure_time
 89        def access_frequent_data():
 90            """Access frequent data multiple times."""
 91            for _ in range(5):
 92                value = storage.retrieve("frequent_data")
 93                print(f"Retrieved frequent data (type: {value['type']})")
 94                time.sleep(0.1)  # Simulate processing
 95
 96        @measure_time
 97        def access_rare_data():
 98            """Access rare data once."""
 99            value = storage.retrieve("rare_data")
100            print(f"Retrieved rare data (type: {value['type']})")
101            time.sleep(0.1)  # Simulate processing
102
103        print("\nAccessing frequent data multiple times:")
104        access_frequent_data()
105
106        print("\nAccessing rare data once:")
107        access_rare_data()
108
109        # Store more data to demonstrate hot storage eviction
110        print("\nStoring additional data to demonstrate hot storage eviction...")
111        for i in range(5):
112            data = {
113                "type": f"additional_{i}",
114                "timestamp": datetime.now().isoformat(),
115                "data": f"Additional data {i}" * 50
116            }
117            storage.store(f"additional_{i}", data)
118            print(f"Stored additional_{i}")
119
120        # Optimize hot storage
121        print("\nOptimizing hot storage...")
122        storage.optimize_hot_storage()
123
124        # Show keys in both storages
125        print("\nKeys in storage after optimization:")
126        keys = storage.get_keys()
127        print(f"Total keys: {len(keys)}")
128        print(f"Keys: {keys}")
129
130        # Warm up hot storage with specific keys
131        print("\nWarming up hot storage with frequent data...")
132        storage.warm_up_hot_storage(["frequent_data"])
133
134        # Final statistics
135        print("\nFinal storage statistics:")
136        stats = storage.get_stats()
137        
138        print("\nHot Storage:")
139        hot_stats = stats["hot_storage"]
140        print(f"Size: {hot_stats['size']} items")
141        print(f"Max size: {hot_stats['max_size']} items")
142        print(f"Hit ratio: {hot_stats['hit_ratio']:.2%}")
143        print(f"Average response time: {hot_stats['avg_response_time']:.2f}ms")
144
145        print("\nCold Storage:")
146        cold_stats = stats["cold_storage"]
147        print(f"Total items: {cold_stats['total_items']}")
148        print(f"Total size: {format_size(cold_stats['total_size'])}")
149        print(f"Average item size: {format_size(cold_stats['avg_item_size'])}")
150
151        print("\nCombined Statistics:")
152        combined_stats = stats["combined"]
153        print(f"Total operations: {combined_stats['total_operations']}")
154        print(f"Hit ratio: {combined_stats['hit_ratio']:.2%}")
155        print(f"Average response time: {combined_stats['avg_response_time']:.2f}ms")
156
157if __name__ == "__main__":
158    main()
Key features demonstrated:

  • Automatic tiered storage management

  • Smart data migration based on access patterns

  • Combined statistics and monitoring

  • Optimized read/write operations

  • Configurable storage ratios

Performance Tips:

  • Tune hot/cold ratio based on workload

  • Monitor access patterns with get_stats()

  • Use appropriate storage backends

  • Implement proper error handling

  • Consider data locality

Advanced Usage

Here are some advanced storage patterns:

from true_storage.storage import MixedStorage
from true_storage.compression import ZstdCompression

# Create a mixed storage with custom configuration
storage = MixedStorage(
    hot_size=1000,            # Hot storage size
    cold_path="/data/cache",  # Cold storage path
    compression=ZstdCompression(level=3),  # Custom compression
    hot_ratio=0.2,           # Keep 20% in hot storage
    stats_enabled=True       # Enable detailed statistics
)

# Store with metadata
storage.store(
    "key1",
    "value1",
    metadata={
        "type": "user_data",
        "priority": "high",
        "ttl": 3600  # 1 hour
    }
)

# Bulk operations
with storage.batch() as batch:
    batch.store("key2", "value2")
    batch.store("key3", "value3")

# Get storage statistics
stats = storage.get_stats()
print(f"Hot storage hit ratio: {stats.hot_hit_ratio}%")
print(f"Cold storage size: {stats.cold_size_bytes} bytes")

Error Handling

Proper error handling is crucial for robust storage operations:

from true_storage.exceptions import StorageError, StorageFullError

try:
    storage.store("key", "value")
except StorageFullError:
    # Handle storage capacity issues
    storage.evict_least_used(count=10)
    storage.store("key", "value")
except StorageError as e:
    # Handle other storage errors
    logger.error(f"Storage error: {e}")
    raise

See Also