0
# Caching System
1
2
The caching system provides caching interface supporting multiple backends including Redis and Memcached. It offers key-value storage for improved application performance with configurable backends and TTL support.
3
4
## Capabilities
5
6
### Cache Handler Interface
7
8
Base interface for caching functionality that defines the contract for cache operations.
9
10
```python { .api }
11
class CacheHandler:
12
"""
13
Cache handler interface for key-value caching operations.
14
15
Provides methods for storing, retrieving, and managing cached
16
data with support for expiration and cache invalidation.
17
"""
18
19
def get(self, key: str) -> Any:
20
"""
21
Get a value from cache by key.
22
23
Args:
24
key: Cache key to retrieve
25
26
Returns:
27
Cached value or None if key not found or expired
28
"""
29
30
def set(self, key: str, value: Any, time: int = None) -> None:
31
"""
32
Set a value in cache with optional expiration.
33
34
Args:
35
key: Cache key to store under
36
value: Value to cache (must be serializable)
37
time: Expiration time in seconds (None for no expiration)
38
"""
39
40
def delete(self, key: str) -> None:
41
"""
42
Delete a key from cache.
43
44
Args:
45
key: Cache key to delete
46
"""
47
48
def purge(self) -> None:
49
"""
50
Clear all cached data.
51
52
Removes all keys and values from the cache.
53
"""
54
```
55
56
## Usage Examples
57
58
### Basic Caching
59
60
```python
61
from cement import App, Controller, ex, init_defaults
62
import time
63
64
CONFIG = init_defaults('myapp')
65
CONFIG['cache.redis'] = {
66
'host': 'localhost',
67
'port': 6379,
68
'db': 0,
69
'password': None
70
}
71
72
class DataController(Controller):
73
class Meta:
74
label = 'data'
75
stacked_on = 'base'
76
stacked_type = 'nested'
77
78
@ex(help='get user data with caching')
79
def user_info(self):
80
"""Get user information with caching."""
81
user_id = 12345
82
cache_key = f'user:{user_id}'
83
84
# Try to get from cache first
85
user_data = self.app.cache.get(cache_key)
86
87
if user_data is None:
88
print('Cache miss - fetching from database')
89
# Simulate database query
90
time.sleep(1)
91
user_data = {
92
'id': user_id,
93
'name': 'John Doe',
94
'email': 'john@example.com',
95
'last_login': '2023-01-15T10:30:00Z'
96
}
97
98
# Cache for 5 minutes
99
self.app.cache.set(cache_key, user_data, time=300)
100
print('Data cached')
101
else:
102
print('Cache hit - returning cached data')
103
104
output = self.app.render(user_data)
105
print(output)
106
107
@ex(help='clear user cache')
108
def clear_cache(self):
109
"""Clear user cache."""
110
user_id = 12345
111
cache_key = f'user:{user_id}'
112
113
self.app.cache.delete(cache_key)
114
print(f'Cleared cache for {cache_key}')
115
116
class BaseController(Controller):
117
class Meta:
118
label = 'base'
119
120
class MyApp(App):
121
class Meta:
122
label = 'myapp'
123
base_controller = 'base'
124
extensions = ['redis']
125
cache_handler = 'redis'
126
config_defaults = CONFIG
127
handlers = [BaseController, DataController]
128
129
with MyApp() as app:
130
app.run()
131
132
# Usage:
133
# myapp data user-info # First call - cache miss
134
# myapp data user-info # Second call - cache hit
135
# myapp data clear-cache # Clear cache
136
```
137
138
### Redis Caching
139
140
```python
141
from cement import App, Controller, ex, init_defaults
142
import json
143
import time
144
145
CONFIG = init_defaults('myapp')
146
CONFIG['cache.redis'] = {
147
'host': 'localhost',
148
'port': 6379,
149
'db': 1, # Use database 1 for this app
150
'password': 'your_redis_password',
151
'socket_timeout': 5,
152
'connection_pool_kwargs': {
153
'max_connections': 50
154
}
155
}
156
157
class ApiController(Controller):
158
class Meta:
159
label = 'api'
160
stacked_on = 'base'
161
stacked_type = 'nested'
162
163
def fetch_api_data(self, endpoint):
164
"""Simulate API data fetching."""
165
print(f'Fetching data from API: {endpoint}')
166
time.sleep(2) # Simulate API delay
167
return {
168
'endpoint': endpoint,
169
'data': ['item1', 'item2', 'item3'],
170
'timestamp': time.time()
171
}
172
173
@ex(
174
help='get API data with Redis caching',
175
arguments=[
176
(['endpoint'], {'help': 'API endpoint to fetch'})
177
]
178
)
179
def fetch(self):
180
"""Fetch API data with Redis caching."""
181
endpoint = self.app.pargs.endpoint
182
cache_key = f'api:data:{endpoint}'
183
184
# Check cache first
185
cached_data = self.app.cache.get(cache_key)
186
187
if cached_data:
188
print('Returning cached API data')
189
print(f'Cache key: {cache_key}')
190
data = json.loads(cached_data)
191
else:
192
print('Cache miss - fetching from API')
193
data = self.fetch_api_data(endpoint)
194
195
# Cache for 10 minutes
196
self.app.cache.set(cache_key, json.dumps(data), time=600)
197
print(f'Data cached with key: {cache_key}')
198
199
output = self.app.render(data)
200
print(output)
201
202
@ex(help='show cache statistics')
203
def stats(self):
204
"""Show Redis cache statistics."""
205
# This would typically use Redis commands
206
# For demo purposes, show cached keys
207
print('Cache Statistics:')
208
print('- Cache backend: Redis')
209
print('- Connection: Active')
210
print('- Sample operations demonstrated')
211
212
class BaseController(Controller):
213
class Meta:
214
label = 'base'
215
216
class MyApp(App):
217
class Meta:
218
label = 'myapp'
219
base_controller = 'base'
220
extensions = ['redis']
221
cache_handler = 'redis'
222
config_defaults = CONFIG
223
handlers = [BaseController, ApiController]
224
225
with MyApp() as app:
226
app.run()
227
228
# Usage:
229
# myapp api fetch /users/active
230
# myapp api fetch /products/popular
231
# myapp api stats
232
```
233
234
### Memcached Caching
235
236
```python
237
from cement import App, Controller, ex, init_defaults
238
import pickle
239
240
CONFIG = init_defaults('myapp')
241
CONFIG['cache.memcached'] = {
242
'hosts': ['127.0.0.1:11211'],
243
'binary': True,
244
'behaviors': {
245
'tcp_nodelay': True,
246
'ketama': True
247
}
248
}
249
250
class SessionController(Controller):
251
class Meta:
252
label = 'session'
253
stacked_on = 'base'
254
stacked_type = 'nested'
255
256
@ex(
257
help='create user session',
258
arguments=[
259
(['username'], {'help': 'username for session'})
260
]
261
)
262
def create(self):
263
"""Create user session with Memcached."""
264
username = self.app.pargs.username
265
session_id = f'session:{username}:{int(time.time())}'
266
267
session_data = {
268
'username': username,
269
'created_at': time.time(),
270
'permissions': ['read', 'write'],
271
'preferences': {
272
'theme': 'dark',
273
'language': 'en'
274
}
275
}
276
277
# Store session for 1 hour
278
self.app.cache.set(session_id, session_data, time=3600)
279
280
print(f'Session created: {session_id}')
281
print(f'Session data: {session_data}')
282
283
@ex(
284
help='get user session',
285
arguments=[
286
(['session_id'], {'help': 'session ID to retrieve'})
287
]
288
)
289
def get(self):
290
"""Get user session from Memcached."""
291
session_id = self.app.pargs.session_id
292
293
session_data = self.app.cache.get(session_id)
294
295
if session_data:
296
print(f'Session found: {session_id}')
297
output = self.app.render(session_data)
298
print(output)
299
else:
300
print(f'Session not found or expired: {session_id}')
301
302
@ex(
303
help='delete user session',
304
arguments=[
305
(['session_id'], {'help': 'session ID to delete'})
306
]
307
)
308
def delete(self):
309
"""Delete user session from Memcached."""
310
session_id = self.app.pargs.session_id
311
312
self.app.cache.delete(session_id)
313
print(f'Session deleted: {session_id}')
314
315
class BaseController(Controller):
316
class Meta:
317
label = 'base'
318
319
class MyApp(App):
320
class Meta:
321
label = 'myapp'
322
base_controller = 'base'
323
extensions = ['memcached']
324
cache_handler = 'memcached'
325
config_defaults = CONFIG
326
handlers = [BaseController, SessionController]
327
328
with MyApp() as app:
329
app.run()
330
331
# Usage:
332
# myapp session create johndoe
333
# myapp session get session:johndoe:1642248000
334
# myapp session delete session:johndoe:1642248000
335
```
336
337
### Cache Wrapper Decorator
338
339
```python
340
from cement import App, Controller, ex
341
import functools
342
import hashlib
343
import json
344
import time
345
346
def cached(timeout=300, key_prefix=''):
347
"""Decorator to cache function results."""
348
def decorator(func):
349
@functools.wraps(func)
350
def wrapper(self, *args, **kwargs):
351
# Generate cache key from function name and arguments
352
key_data = {
353
'function': func.__name__,
354
'args': args,
355
'kwargs': kwargs
356
}
357
key_string = json.dumps(key_data, sort_keys=True)
358
key_hash = hashlib.md5(key_string.encode()).hexdigest()
359
cache_key = f'{key_prefix}{func.__name__}:{key_hash}'
360
361
# Try to get from cache
362
result = self.app.cache.get(cache_key)
363
364
if result is None:
365
print(f'Cache miss for {func.__name__}')
366
result = func(self, *args, **kwargs)
367
self.app.cache.set(cache_key, result, time=timeout)
368
print(f'Result cached with key: {cache_key}')
369
else:
370
print(f'Cache hit for {func.__name__}')
371
372
return result
373
return wrapper
374
return decorator
375
376
class ComputeController(Controller):
377
class Meta:
378
label = 'compute'
379
stacked_on = 'base'
380
stacked_type = 'nested'
381
382
@cached(timeout=600, key_prefix='compute:')
383
def expensive_calculation(self, n):
384
"""Expensive calculation that benefits from caching."""
385
print(f'Performing expensive calculation for n={n}')
386
# Simulate expensive computation
387
time.sleep(3)
388
result = sum(i * i for i in range(n))
389
return result
390
391
@ex(
392
help='perform cached calculation',
393
arguments=[
394
(['number'], {'type': int, 'help': 'number for calculation'})
395
]
396
)
397
def calculate(self):
398
"""Perform calculation with caching."""
399
n = self.app.pargs.number
400
401
start_time = time.time()
402
result = self.expensive_calculation(n)
403
end_time = time.time()
404
405
print(f'Result: {result}')
406
print(f'Execution time: {end_time - start_time:.2f} seconds')
407
408
@ex(help='clear computation cache')
409
def clear(self):
410
"""Clear all cached computations."""
411
self.app.cache.purge()
412
print('Computation cache cleared')
413
414
class BaseController(Controller):
415
class Meta:
416
label = 'base'
417
418
class MyApp(App):
419
class Meta:
420
label = 'myapp'
421
base_controller = 'base'
422
extensions = ['redis']
423
cache_handler = 'redis'
424
handlers = [BaseController, ComputeController]
425
426
with MyApp() as app:
427
app.run()
428
429
# Usage:
430
# myapp compute calculate 1000 # First call - slow
431
# myapp compute calculate 1000 # Second call - fast (cached)
432
# myapp compute clear # Clear cache
433
```
434
435
### Cache Configuration and Management
436
437
```python
438
from cement import App, Controller, ex, init_defaults
439
440
CONFIG = init_defaults('myapp')
441
CONFIG['cache.redis'] = {
442
'host': 'localhost',
443
'port': 6379,
444
'db': 0,
445
'default_timeout': 300, # 5 minutes default
446
'key_prefix': 'myapp:'
447
}
448
449
class CacheController(Controller):
450
class Meta:
451
label = 'cache'
452
stacked_on = 'base'
453
stacked_type = 'nested'
454
455
@ex(
456
help='set cache value',
457
arguments=[
458
(['key'], {'help': 'cache key'}),
459
(['value'], {'help': 'cache value'}),
460
(['--ttl'], {'type': int, 'help': 'time to live in seconds'})
461
]
462
)
463
def set(self):
464
"""Set a cache value."""
465
key = self.app.pargs.key
466
value = self.app.pargs.value
467
ttl = self.app.pargs.ttl
468
469
self.app.cache.set(key, value, time=ttl)
470
471
ttl_msg = f' (TTL: {ttl}s)' if ttl else ' (no expiration)'
472
print(f'Set cache key "{key}" = "{value}"{ttl_msg}')
473
474
@ex(
475
help='get cache value',
476
arguments=[
477
(['key'], {'help': 'cache key to retrieve'})
478
]
479
)
480
def get(self):
481
"""Get a cache value."""
482
key = self.app.pargs.key
483
value = self.app.cache.get(key)
484
485
if value is not None:
486
print(f'Cache key "{key}" = "{value}"')
487
else:
488
print(f'Cache key "{key}" not found or expired')
489
490
@ex(
491
help='delete cache key',
492
arguments=[
493
(['key'], {'help': 'cache key to delete'})
494
]
495
)
496
def delete(self):
497
"""Delete a cache key."""
498
key = self.app.pargs.key
499
self.app.cache.delete(key)
500
print(f'Deleted cache key "{key}"')
501
502
@ex(help='clear all cache')
503
def clear(self):
504
"""Clear all cached data."""
505
self.app.cache.purge()
506
print('All cache data cleared')
507
508
@ex(help='show cache info')
509
def info(self):
510
"""Show cache configuration information."""
511
cache_config = self.app.config.get_section_dict('cache.redis')
512
513
print('Cache Configuration:')
514
print(f' Backend: Redis')
515
print(f' Host: {cache_config.get("host", "localhost")}')
516
print(f' Port: {cache_config.get("port", 6379)}')
517
print(f' Database: {cache_config.get("db", 0)}')
518
print(f' Default TTL: {cache_config.get("default_timeout", 300)}s')
519
print(f' Key Prefix: {cache_config.get("key_prefix", "")}')
520
521
class BaseController(Controller):
522
class Meta:
523
label = 'base'
524
525
class MyApp(App):
526
class Meta:
527
label = 'myapp'
528
base_controller = 'base'
529
extensions = ['redis']
530
cache_handler = 'redis'
531
config_defaults = CONFIG
532
handlers = [BaseController, CacheController]
533
534
with MyApp() as app:
535
app.run()
536
537
# Usage:
538
# myapp cache set user:123 "John Doe" --ttl 60
539
# myapp cache get user:123
540
# myapp cache delete user:123
541
# myapp cache clear
542
# myapp cache info
543
```