# Example Code
Welcome to the example code section! Here we've collected various practical Redis Toolkit usage examples, from basic to advanced, to help you get started quickly.
# 📂 Example Categories
🎯 Basic Examples
Simple examples suitable for beginners
- Hello World
- Basic operations
- Data type handling
- Simple publish/subscribe
🎨 Media Processing Examples
Image, audio, and video processing examples
- Image caching system
- Audio conversion service
- Video thumbnail generation
- Media file management
💼 Real-World Cases
Complete implementations of real application scenarios
- User system
- Real-time chat room
- Task queue
- Caching system
# 🚀 Quick Start Examples
# Hello Redis Toolkit
from redis_toolkit import RedisToolkit
# Initialize
toolkit = RedisToolkit()
# Store data
toolkit.setter("greeting", "Hello, Redis Toolkit!")
# Retrieve data
message = toolkit.getter("greeting")
print(message) # Hello, Redis Toolkit!
# Store complex data
user = {
"id": 1,
"name": "Alice",
"scores": [95, 87, 92]
}
toolkit.setter("user:1", user)
# Automatic deserialization
retrieved_user = toolkit.getter("user:1")
print(f"{retrieved_user['name']}'s scores: {retrieved_user['scores']}")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# 📱 Useful Small Examples
# Counter
class Counter:
def __init__(self, name):
self.toolkit = RedisToolkit()
self.key = f"counter:{name}"
def increment(self):
current = self.toolkit.getter(self.key) or 0
self.toolkit.setter(self.key, current + 1)
return current + 1
def get(self):
return self.toolkit.getter(self.key) or 0
def reset(self):
self.toolkit.setter(self.key, 0)
# Usage example
page_views = Counter("page_views")
page_views.increment()
print(f"Page views: {page_views.get()}")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# Simple Cache
import time
def cached(key, ttl=300):
"""Cache decorator"""
def decorator(func):
def wrapper(*args, **kwargs):
toolkit = RedisToolkit()
# Try to get from cache
cached_result = toolkit.getter(key)
if cached_result is not None:
return cached_result
# Execute function and cache result
result = func(*args, **kwargs)
toolkit.setter(key, result, ex=ttl)
return result
return wrapper
return decorator
# Usage example
@cached("expensive_calculation", ttl=600)
def calculate_something():
print("Performing expensive calculation...")
time.sleep(2) # Simulate time-consuming operation
return {"result": 42}
# First call executes the calculation
result1 = calculate_something()
# Subsequent calls return cached result
result2 = calculate_something()
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
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
# Leaderboard
class Leaderboard:
def __init__(self, name):
self.toolkit = RedisToolkit()
self.key = f"leaderboard:{name}"
def add_score(self, player, score):
"""Add or update player score"""
self.toolkit.client.zadd(self.key, {player: score})
def get_top(self, count=10):
"""Get top N players"""
top_players = self.toolkit.client.zrevrange(
self.key, 0, count-1, withscores=True
)
return [(p.decode(), int(s)) for p, s in top_players]
def get_rank(self, player):
"""Get player rank"""
rank = self.toolkit.client.zrevrank(self.key, player)
return rank + 1 if rank is not None else None
# Usage example
game_scores = Leaderboard("game_2024")
# Add scores
game_scores.add_score("Alice", 1500)
game_scores.add_score("Bob", 1200)
game_scores.add_score("Charlie", 1800)
# Get leaderboard
top_3 = game_scores.get_top(3)
for i, (player, score) in enumerate(top_3, 1):
print(f"{i}. {player}: {score} points")
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
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
# 🎯 Suggested Learning Path
# Beginner Route
- Basic Operations → Start with Hello World
- Data Types → Understand serialization features
- Simple Applications → Implement counters, caches
- Publish/Subscribe → Try message passing
# Advanced Route
- Batch Operations → Improve performance
- Media Processing → Handle images and audio
- Error Handling → Build stable systems
- Real Projects → Complete application development
# 💡 Example Usage Tips
# 1. Copy and Modify
All examples can be directly copied and modified according to your needs:
# Copy example code
# Modify configuration parameters
# Adjust business logic
# Add error handling
1
2
3
4
2
3
4
# 2. Combine Usage
Different examples can be combined:
# Combine cache + counter
class CachedCounter(Counter):
def increment(self):
result = super().increment()
# Clear related cache
self.toolkit.client.delete("stats:cache")
return result
1
2
3
4
5
6
7
2
3
4
5
6
7
# 3. Performance Testing
Conduct performance tests based on examples:
import time
def benchmark(func, iterations=1000):
start = time.time()
for _ in range(iterations):
func()
elapsed = time.time() - start
print(f"Executed {iterations} times in: {elapsed:.3f} seconds")
print(f"Average per execution: {elapsed/iterations*1000:.2f} ms")
# Test example
benchmark(lambda: toolkit.setter("test", "value"))
1
2
3
4
5
6
7
8
9
10
11
12
2
3
4
5
6
7
8
9
10
11
12
# 📚 Complete Example Projects
# Mini Blog System
A complete example combining multiple features:
class MiniBlog:
def __init__(self):
self.toolkit = RedisToolkit()
def create_post(self, author, title, content):
"""Create a post"""
post_id = self.toolkit.client.incr("post:id")
post = {
"id": post_id,
"author": author,
"title": title,
"content": content,
"created_at": time.time(),
"views": 0
}
# Save post
self.toolkit.setter(f"post:{post_id}", post)
# Add to recent posts list
self.toolkit.client.lpush("posts:recent", post_id)
# Update author's post list
self.toolkit.client.sadd(f"author:{author}:posts", post_id)
return post_id
def get_post(self, post_id):
"""Get post and increment view count"""
key = f"post:{post_id}"
post = self.toolkit.getter(key)
if post:
# Increment view count
post["views"] += 1
self.toolkit.setter(key, post)
return post
def get_recent_posts(self, count=10):
"""Get recent posts"""
post_ids = self.toolkit.client.lrange("posts:recent", 0, count-1)
posts = []
for pid in post_ids:
post = self.toolkit.getter(f"post:{int(pid)}")
if post:
posts.append(post)
return posts
# Usage example
blog = MiniBlog()
# Create post
post_id = blog.create_post(
author="Alice",
title="My Experience with Redis Toolkit",
content="This is an amazing tool..."
)
# Read post
post = blog.get_post(post_id)
print(f"'{post['title']}' has been viewed {post['views']} times")
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
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
# 🔗 Related Resources
- GitHub Examples Repository (opens new window)
- API Documentation - Detailed API reference
- Best Practices - Development recommendations
# 🎯 Next Steps
Choose the example category you're interested in to start exploring:
Tips
- All examples are tested and can be run directly
- It's recommended to read the code and understand the logic before running
- If you encounter issues, check Troubleshooting