# 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
View Examples →

🎨 Media Processing Examples

Image, audio, and video processing examples

  • Image caching system
  • Audio conversion service
  • Video thumbnail generation
  • Media file management
View Examples →

💼 Real-World Cases

Complete implementations of real application scenarios

  • User system
  • Real-time chat room
  • Task queue
  • Caching system
View Examples →

# 🚀 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

# 📱 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

# 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

# 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

# 🎯 Suggested Learning Path

# Beginner Route

  1. Basic Operations → Start with Hello World
  2. Data Types → Understand serialization features
  3. Simple Applications → Implement counters, caches
  4. Publish/Subscribe → Try message passing

# Advanced Route

  1. Batch Operations → Improve performance
  2. Media Processing → Handle images and audio
  3. Error Handling → Build stable systems
  4. 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. 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

# 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

# 📚 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

# 🎯 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