Key-Value Concept

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

# Conceptual key-value store operations
 
class KeyValueStore:
    """
    The fundamental key-value store abstraction.
    All key-value databases implement these core operations.
    """
    
    def put(self, key: str, value: bytes) -> None:
        """
        Store a value associated with a key.
        
        Args:
            key: Unique identifier for the data (typically a string)
            value: The data to store (opaque bytes from the store's perspective)
        
        Semantics:
            - If key exists, the old value is replaced
            - The operation is atomic
            - Durability guarantees depend on configuration
        """
        pass
    
    def get(self, key: str) -> bytes | None:
        """
        Retrieve the value associated with a key.
        
        Args:
            key: The key to look up
            
        Returns:
            The value if key exists, None otherwise
            
        Semantics:
            - Read is atomic
            - May return stale data in eventually consistent systems
        """
        pass
    
    def delete(self, key: str) -> bool:
        """
        Remove a key-value pair from the store.
        
        Args:
            key: The key to delete
            
        Returns:
            True if key existed and was deleted, False otherwise
            
        Semantics:
            - Operation is atomic
            - Deletion may be logical (tombstone) or physical
        """
        pass
 
# Example usage
store = KeyValueStore()
store.put("user:1001", b'{"name": "Alice", "email": "alice@example.com"}')
user = store.get("user:1001")  # Returns the JSON bytes
store.delete("user:1001")       # User data is removed

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

# Key design patterns for key-value stores
 
class KeyDesigner:
    """
    Demonstrates common key design patterns used in production systems.
    """
    
    @staticmethod
    def user_profile_key(user_id: int) -> str:
        """Simple entity key"""
        return f"user:{user_id}:profile"
    
    @staticmethod
    def user_session_key(user_id: int, session_id: str) -> str:
        """Compound key for user sessions"""
        return f"session:{user_id}:{session_id}"
    
    @staticmethod
    def rate_limit_key(ip_address: str, endpoint: str, window: str) -> str:
        """Rate limiting key with temporal window"""
        return f"ratelimit:{ip_address}:{endpoint}:{window}"
    
    @staticmethod
    def cache_key(entity_type: str, entity_id: str, version: int = None) -> str:
        """Cache key with optional versioning for cache busting"""
        base = f"cache:{entity_type}:{entity_id}"
        return f"{base}:v{version}" if version else base
    
    @staticmethod
    def leaderboard_key(game_id: str, period: str) -> str:
        """Leaderboard key with time period"""
        return f"leaderboard:{game_id}:{period}"
    
    @staticmethod
    def distributed_lock_key(resource: str) -> str:
        """Distributed lock key"""
        return f"lock:{resource}"
    
    @staticmethod
    def counter_key(metric: str, date: str) -> str:
        """Counter with date partitioning"""
        return f"counter:{metric}:{date}"
 
 
# Example: E-commerce application key design
class ECommerceKeys:
    """
    Key design for an e-commerce platform demonstrating
    how to support multiple access patterns.
    """
    
    # Product access patterns
    @staticmethod
    def product_by_id(product_id: str) -> str:
        return f"product:{product_id}"
    
    @staticmethod
    def product_by_sku(sku: str) -> str:
        # Secondary "index" - stores the product_id
        return f"product:sku:{sku}"
    
    @staticmethod
    def product_inventory(product_id: str, warehouse_id: str) -> str:
        return f"inventory:{warehouse_id}:{product_id}"
    
    # Shopping cart
    @staticmethod
    def cart(user_id: str) -> str:
        return f"cart:{user_id}"
    
    # Order access patterns
    @staticmethod
    def order_by_id(order_id: str) -> str:
        return f"order:{order_id}"
    
    @staticmethod
    def orders_by_user(user_id: str) -> str:
        # Returns a list/set of order IDs
        return f"user:{user_id}:orders"

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105

import json
import pickle
import msgpack
import struct
from dataclasses import dataclass
from typing import Any
 
@dataclass
class User:
    id: int
    name: str
    email: str
    preferences: dict
 
class ValueSerializer:
    """
    Demonstrates various serialization strategies for key-value store values.
    The application is responsible for choosing the right serialization format.
    """
    
    # JSON: Human-readable, widely supported, larger size
    @staticmethod
    def to_json(obj: Any) -> bytes:
        return json.dumps(obj).encode('utf-8')
    
    @staticmethod
    def from_json(data: bytes) -> Any:
        return json.loads(data.decode('utf-8'))
    
    # MessagePack: Compact binary, fast, good for structured data
    @staticmethod
    def to_msgpack(obj: Any) -> bytes:
        return msgpack.packb(obj)
    
    @staticmethod
    def from_msgpack(data: bytes) -> Any:
        return msgpack.unpackb(data, raw=False)
    
    # Pickle: Python-specific, preserves custom objects
    # WARNING: Never unpickle untrusted data (security risk)
    @staticmethod
    def to_pickle(obj: Any) -> bytes:
        return pickle.dumps(obj)
    
    @staticmethod
    def from_pickle(data: bytes) -> Any:
        return pickle.loads(data)
    
    # Custom binary: Maximum efficiency for known structures
    @staticmethod
    def to_binary_user(user: User) -> bytes:
        """
        Custom binary format for User objects.
        Format: [id:8B][name_len:2B][name][email_len:2B][email][prefs_json]
        """
        name_bytes = user.name.encode('utf-8')
        email_bytes = user.email.encode('utf-8')
        prefs_bytes = json.dumps(user.preferences).encode('utf-8')
        
        return struct.pack(
            f'>QH{len(name_bytes)}sH{len(email_bytes)}s',
            user.id,
            len(name_bytes), name_bytes,
            len(email_bytes), email_bytes
        ) + prefs_bytes
    
    @staticmethod
    def from_binary_user(data: bytes) -> User:
        """Deserialize binary user data."""
        offset = 0
        user_id = struct.unpack_from('>Q', data, offset)[0]
        offset += 8
        
        name_len = struct.unpack_from('>H', data, offset)[0]
        offset += 2
        name = data[offset:offset + name_len].decode('utf-8')
        offset += name_len
        
        email_len = struct.unpack_from('>H', data, offset)[0]
        offset += 2
        email = data[offset:offset + email_len].decode('utf-8')
        offset += email_len
        
        preferences = json.loads(data[offset:].decode('utf-8'))
        
        return User(id=user_id, name=name, email=email, preferences=preferences)
 
 
# Size comparison for the same User object
user = User(
    id=1001,
    name="Alice Johnson",
    email="alice.johnson@example.com",
    preferences={"theme": "dark", "notifications": True}
)
 
json_size = len(ValueSerializer.to_json(user.__dict__))
msgpack_size = len(ValueSerializer.to_msgpack(user.__dict__))
pickle_size = len(ValueSerializer.to_pickle(user))
 
# Typical results:
# JSON:     ~120 bytes (human readable)
# MsgPack:  ~85 bytes  (30% smaller)
# Pickle:   ~180 bytes (larger but preserves Python types)
# Binary:   ~75 bytes  (smallest, but most rigid)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

# Redis durability configuration examples
 
# Option 1: No persistence (pure cache)
# - Fastest
# - All data lost on restart
# - Use for: Pure caching, ephemeral data
"""
save ""
appendonly no
"""
 
# Option 2: RDB snapshots only
# - Point-in-time snapshots every N seconds
# - Can lose up to N seconds of data
# - Use for: Data that can be reconstructed
"""
save 900 1      # Snapshot if 1+ keys changed in 900 seconds
save 300 10     # Snapshot if 10+ keys changed in 300 seconds
save 60 10000   # Snapshot if 10000+ keys changed in 60 seconds
appendonly no
"""
 
# Option 3: AOF (Append-Only File)
# - Every write logged to disk
# - Configurable fsync frequency
# - Use for: Important data with durability needs
"""
save ""
appendonly yes
appendfsync everysec    # fsync every second (good balance)
# appendfsync always    # fsync every write (slowest, safest)
# appendfsync no        # OS decides when to fsync (fastest, risky)
"""
 
# Option 4: RDB + AOF (Belt and suspenders)
# - Combines point-in-time snapshots with continuous logging
# - Best durability guarantees
# - Use for: Critical data
"""
save 900 1
save 300 10
save 60 10000
appendonly yes
appendfsync everysec
"""
 
# Performance impact of durability settings:
# 
# | Setting              | Writes/sec | Data Loss Risk    |
# |---------------------|------------|-------------------|
# | No persistence      | 500,000+   | Total on crash    |
# | RDB only (5 min)    | 400,000+   | Up to 5 minutes   |
# | AOF (everysec)      | 100,000+   | Up to 1 second    |
# | AOF (always)        | 10,000     | Minimal           |

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148

import time
from typing import Optional, List, Tuple
 
class ExtendedKeyValueStore:
    """
    Extended operations beyond basic GET/PUT/DELETE.
    These operations enable sophisticated use cases.
    """
    
    def set_with_ttl(
        self, 
        key: str, 
        value: bytes, 
        ttl_seconds: int
    ) -> None:
        """
        Store a value that automatically expires after ttl_seconds.
        
        Use cases:
        - Session tokens that expire
        - Cache entries with freshness limits
        - Rate limiting windows
        """
        pass
    
    def get_with_ttl(self, key: str) -> Tuple[Optional[bytes], int]:
        """
        Get value and remaining TTL.
        Returns (value, remaining_seconds) or (None, -1) if not found.
        """
        pass
    
    def compare_and_swap(
        self, 
        key: str, 
        expected: bytes, 
        new_value: bytes
    ) -> bool:
        """
        Atomic compare-and-swap operation.
        
        Only updates if current value equals expected value.
        Returns True if swap succeeded, False otherwise.
        
        Use cases:
        - Optimistic locking
        - Distributed coordination
        - Safe concurrent updates
        """
        pass
    
    def increment(self, key: str, delta: int = 1) -> int:
        """
        Atomically increment a numeric value.
        
        Creates key with value 0 if it doesn't exist.
        Returns the new value after increment.
        
        Use cases:
        - Page view counters
        - Rate limiting
        - Generating unique IDs
        """
        pass
    
    def multi_get(self, keys: List[str]) -> dict[str, Optional[bytes]]:
        """
        Retrieve multiple keys in a single operation.
        
        Returns a dict mapping each key to its value (or None).
        Much more efficient than individual GETs.
        
        Use cases:
        - Loading user profile + preferences + settings
        - Fetching multiple cache entries
        - Bulk data retrieval
        """
        pass
    
    def multi_set(self, pairs: dict[str, bytes]) -> None:
        """
        Store multiple key-value pairs atomically.
        
        Either all keys are set or none are (transactional).
        
        Use cases:
        - Initializing related data atomically
        - Bulk cache population
        - Transactional updates across keys
        """
        pass
    
    def set_if_not_exists(self, key: str, value: bytes) -> bool:
        """
        Only set the key if it doesn't already exist.
        
        Returns True if key was set, False if it already existed.
        
        Use cases:
        - Distributed locks (acquire only if free)
        - Idempotent initialization
        - First-write-wins semantics
        """
        pass
 
 
# Practical example: Rate Limiter using extended operations
class RateLimiter:
    """
    Token bucket rate limiter implemented with key-value store.
    Demonstrates practical use of INCR and TTL operations.
    """
    
    def __init__(
        self, 
        store: ExtendedKeyValueStore,
        max_requests: int = 100,
        window_seconds: int = 60
    ):
        self.store = store
        self.max_requests = max_requests
        self.window_seconds = window_seconds
    
    def is_allowed(self, client_id: str) -> Tuple[bool, int]:
        """
        Check if request is allowed and update counter.
        
        Returns:
            (allowed: bool, remaining: int)
        """
        key = f"ratelimit:{client_id}:{self._current_window()}"
        
        # Atomic increment with TTL
        # If key doesn't exist, creates with value 1 and TTL
        current_count = self.store.increment(key)
        
        # Set TTL only on first request of window
        if current_count == 1:
            self.store.set_with_ttl(key, str(current_count).encode(), self.window_seconds)
        
        remaining = max(0, self.max_requests - current_count)
        allowed = current_count <= self.max_requests
        
        return allowed, remaining
    
    def _current_window(self) -> str:
        """Get current time window identifier."""
        return str(int(time.time()) // self.window_seconds)