Module 1: Redis Fundamentals

redis-py and the Redis Explorer Mini-project

Overview

You've mastered Redis's 5 data types from redis-cli. Now you move to Python — where Redis is going to live in your real applications. This capsule covers redis-py (the official Python client for Redis), the CLI-equivalent operations now in code, pipelines for efficient batch operations, and it closes with the Redis Explorer mini-project: a Python CLI tool that exercises the 5 data types with real use cases.

Every operation you learned in redis-cli has its equivalent in redis-py. The Python syntax is very similar — r.set("key", "value") in Python is SET key value in the CLI. But there are three important details the CLI doesn't have: how to manage connections, how to handle string vs bytes encoding, and how to use pipelines to cut down network round-trips. Those three points are the difference between Python code that works in development and code that scales in production.

It's module 1's final capsule. By the end, you'll have Redis running, command of the 5 data types from Python, and a mini-project that demonstrates all of it in executable code. Ready to get into caching patterns in module 2.

Installing redis-py

Activate your virtual environment and install it:

cd ~/projects/redis-guide/module-01-fundamentals/redis-explorer
source .venv/bin/activate

pip install redis

Verify:

python -c "import redis; print(redis.__version__)"

Expected output: 5.0.1 or higher. Any 5.x version works perfectly for this guide.

⚠️ IMPORTANT NOTE: The package you install is called redis on pip, but when you import it in Python you use import redis. Do NOT install aioredis — it's been deprecated since 2021 and it was absorbed into redis-py in version 4.2+. When we get to module 4 with async code, we'll use redis.asyncio (a module inside redis-py), NOT the separate aioredis library.

Basic connection from Python

import redis

r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

# Verify the connection
print(r.ping())
# True

Breaking down the parameters:

host='localhost': where Redis is running. If you're using Docker locally, localhost works because we mapped -p 6379:6379
port=6379: Redis's default port
db=0: the database to use (Redis has 16 by default, numbered 0-15)
decode_responses=True: critical. Without this, Redis returns bytes (b'value') instead of strings ("value")

The detail about `decode_responses=True`

Without decode_responses=True:

r = redis.Redis(host='localhost', port=6379)
r.set('greeting', 'hello')
print(r.get('greeting'))
# b'hello'   <-- bytes, not a string

print(type(r.get('greeting')))
# <class 'bytes'>

With decode_responses=True:

r = redis.Redis(host='localhost', port=6379, decode_responses=True)
r.set('greeting', 'hello')
print(r.get('greeting'))
# 'hello'   <-- a string

print(type(r.get('greeting')))
# <class 'str'>

Recommendation: Use decode_responses=True in 95% of cases. Only set it to False when you're storing genuinely binary data (files, pickled objects, images). For normal strings, serialized JSON, or numbers, decoded is what you want.

Connecting with a URL (more portable)

r = redis.from_url("redis://localhost:6379/0", decode_responses=True)

This is the idiomatic way to configure Redis with environment variables:

import os

REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")
r = redis.from_url(REDIS_URL, decode_responses=True)

In development: redis://localhost:6379/0 In production: redis://default:password@redis-prod.internal:6379/0 (with auth) In the cloud: rediss://... (with TLS) or whatever connection string your provider gives you (Upstash, Redis Cloud)

Operations from Python

Every CLI operation has a Python equivalent. The simple rule: the CLI command in lowercase is the Python method.

Strings

import redis
r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# SET / GET
r.set('greeting', 'hello')
print(r.get('greeting'))
# 'hello'

# SET with a TTL
r.set('cache:item', 'temporary', ex=60)  # ex = expire in seconds
print(r.ttl('cache:item'))
# 58 (or a bit less)

# INCR / DECR
r.set('counter', 0)
r.incr('counter')
r.incr('counter')
r.incrby('counter', 10)
print(r.get('counter'))
# '12'

# MSET / MGET
r.mset({'key1': 'a', 'key2': 'b', 'key3': 'c'})
print(r.mget(['key1', 'key2', 'key3']))
# ['a', 'b', 'c']

# SETNX (a simple lock)
acquired = r.set('lock:42', 'worker-A', nx=True, ex=30)
if acquired:
    print("Lock acquired")
else:
    print("The lock was already taken")

# DEL
r.delete('counter', 'key1', 'key2', 'key3')

# EXISTS
print(r.exists('greeting'))
# 1 (it exists)

print(r.exists('nonexistent'))
# 0 (it doesn't exist)

Notice:

r.set("key", "value", ex=60) for SET with a TTL (in the CLI: SET key value EX 60)
r.set("key", "value", nx=True) for SETNX (in the CLI: SET key value NX)
r.delete() instead of r.del() (because del is a reserved word in Python)
r.mget() returns a list directly (in the CLI: a numbered list)

Hashes

# HSET
r.hset('user:1', 'name', 'Alex')
r.hset('user:1', mapping={'email': 'alex@example.com', 'age': 30})

# HGET
print(r.hget('user:1', 'name'))
# 'Alex'

# HGETALL returns a Python dict (this is very idiomatic)
print(r.hgetall('user:1'))
# {'name': 'Alex', 'email': 'alex@example.com', 'age': '30'}

# HMGET (several fields)
print(r.hmget('user:1', ['name', 'email']))
# ['Alex', 'alex@example.com']

# HINCRBY
r.hset('stats:user:1', 'logins', 0)
r.hincrby('stats:user:1', 'logins', 1)
r.hincrby('stats:user:1', 'page_views', 50)
print(r.hgetall('stats:user:1'))
# {'logins': '1', 'page_views': '50'}

# HEXISTS
print(r.hexists('user:1', 'email'))
# True

# HDEL
r.hdel('user:1', 'age')
print(r.hgetall('user:1'))
# {'name': 'Alex', 'email': 'alex@example.com'}

# HKEYS / HVALS / HLEN
print(r.hkeys('user:1'))
# ['name', 'email']

print(r.hvals('user:1'))
# ['Alex', 'alex@example.com']

print(r.hlen('user:1'))
# 2

Notice:

r.hset("key", "field", "value") for an individual field
r.hset("key", mapping={...}) for several fields in one call
r.hgetall() returns a Python dict directly (not an alternating list)

Lists

# RPUSH / LPUSH (they accept multiple values)
r.rpush('queue', 'task-1', 'task-2', 'task-3')

# LPOP / RPOP
print(r.lpop('queue'))
# 'task-1'

# LRANGE
print(r.lrange('queue', 0, -1))
# ['task-2', 'task-3']

# LLEN
print(r.llen('queue'))
# 2

# LINDEX
print(r.lindex('queue', 0))
# 'task-2'

# LREM (syntax: lrem(key, count, value))
r.rpush('items', 'a', 'b', 'a', 'c', 'a')
r.lrem('items', 0, 'a')  # remove ALL the "a"s
print(r.lrange('items', 0, -1))
# ['b', 'c']

# LTRIM
r.rpush('log', *[f'event-{i}' for i in range(10)])
r.ltrim('log', -5, -1)  # keep only the last 5
print(r.lrange('log', 0, -1))
# ['event-5', 'event-6', 'event-7', 'event-8', 'event-9']

# BLPOP (blocking pop with a timeout)
result = r.blpop(['work_queue'], timeout=5)
if result:
    queue_name, value = result
    print(f"Received from {queue_name}: {value}")
else:
    print("Timeout, no messages arrived")

Notice:

r.rpush(key, *values): the * unpacking lets you pass a list or individual args
r.lrem(key, count, value): a positive count works from the beginning, a negative one from the end, 0 means all of them
r.blpop([list_keys], timeout=N): you pass a list of keys (BLPOP can wait on several queues)

Sets

# SADD
r.sadd('tags', 'python', 'redis', 'fastapi')
r.sadd('tags', 'python')  # a duplicate, it isn't added
print(r.scard('tags'))
# 3

# SMEMBERS returns a Python set
print(r.smembers('tags'))
# {'fastapi', 'python', 'redis'}

# SISMEMBER
print(r.sismember('tags', 'python'))
# True

print(r.sismember('tags', 'javascript'))
# False

# SREM
r.srem('tags', 'redis')
print(r.smembers('tags'))
# {'fastapi', 'python'}

# Operations between sets
r.sadd('users:online:web', 'alice', 'bob', 'charlie')
r.sadd('users:online:mobile', 'bob', 'charlie', 'diana')

# Intersection
print(r.sinter('users:online:web', 'users:online:mobile'))
# {'bob', 'charlie'}

# Union
print(r.sunion('users:online:web', 'users:online:mobile'))
# {'alice', 'bob', 'charlie', 'diana'}

# Difference
print(r.sdiff('users:online:web', 'users:online:mobile'))
# {'alice'}

# SRANDMEMBER / SPOP
print(r.srandmember('tags'))
# some random one

print(r.srandmember('tags', 2))
# 2 random ones

Notice:

r.smembers() returns a Python set (not a list) — useful for membership operations
The operations between sets (sinter, sunion, sdiff) also return a set

Sorted Sets

# ZADD takes a dict {member: score}
r.zadd('leaderboard', {
    'alice': 1500,
    'bob': 2300,
    'charlie': 1800,
    'diana': 950,
    'eve': 2100
})

# ZRANGE / ZREVRANGE
# ZRANGE is ascending, ZREVRANGE is descending
print(r.zrange('leaderboard', 0, 2))
# ['diana', 'alice', 'charlie']  (the 3 with the lowest scores)

print(r.zrevrange('leaderboard', 0, 2))
# ['bob', 'eve', 'charlie']  (the 3 top scores)

# With scores
print(r.zrevrange('leaderboard', 0, 2, withscores=True))
# [('bob', 2300.0), ('eve', 2100.0), ('charlie', 1800.0)]

# ZSCORE
print(r.zscore('leaderboard', 'alice'))
# 1500.0

# ZRANK / ZREVRANK
print(r.zrevrank('leaderboard', 'alice'))
# 3 (fourth position from the top, 0-indexed)

print(r.zrevrank('leaderboard', 'bob'))
# 0 (first position = the winner)

# ZINCRBY
r.zincrby('leaderboard', 200, 'charlie')
print(r.zscore('leaderboard', 'charlie'))
# 2000.0

# ZRANGEBYSCORE (the key to sliding windows)
import time

now = time.time()
r.zadd('rate:user42', {f'req-{i}': now + i for i in range(5)})

# Requests in the window of the last 60 sec
window_start = now - 60
in_window = r.zrangebyscore('rate:user42', window_start, '+inf')
print(in_window)
# ['req-0', 'req-1', 'req-2', 'req-3', 'req-4']

# How many are there
count = r.zcard('rate:user42')
print(count)
# 5

# Count within a specific range
count_in_range = r.zcount('rate:user42', window_start, '+inf')
print(count_in_range)
# 5

# ZREMRANGEBYSCORE (cleaning out the old ones)
r.zremrangebyscore('rate:user42', 0, now - 60)

Notice:

r.zadd(key, {member: score}) takes a dict (more Pythonic than passing the score and member separately)
r.zrevrange(key, start, end, withscores=True) returns a list of tuples [(member, score), ...]
r.zrangebyscore(key, min, max) with '-inf' and '+inf' as strings

Pipelines: batch operations

Every command you send to Redis costs a network round-trip (~0.1 ms on localhost, ~1-5 ms in the cloud). If you need to run 100 commands, that's 100 round-trips. Pipelines let you send 100 commands in a single round-trip.

Without a pipeline (slow)

import time

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

start = time.time()
for i in range(100):
    r.set(f'key:{i}', f'value:{i}')
elapsed = time.time() - start
print(f"Without a pipeline: {elapsed*1000:.1f}ms")
# Without a pipeline: 25.3ms (on localhost; in the cloud it's 100-500ms)

With a pipeline (fast)

start = time.time()
pipe = r.pipeline()
for i in range(100):
    pipe.set(f'key:{i}', f'value:{i}')
pipe.execute()
elapsed = time.time() - start
print(f"With a pipeline: {elapsed*1000:.1f}ms")
# With a pipeline: 2.1ms (10x faster)

The difference is dramatic in production, where network latency is higher.

A pipeline with results

pipe = r.pipeline()
pipe.set('counter', 0)
pipe.incr('counter')
pipe.incr('counter')
pipe.get('counter')

results = pipe.execute()
print(results)
# [True, 1, 2, '2']
# Each result corresponds to each command, in order

A real use case: loading 1000 products into the cache

products = db.query(Product).all()  # 1000 products

pipe = r.pipeline()
for p in products:
    pipe.hset(f'cache:product:{p.id}', mapping={
        'name': p.name,
        'price': str(p.price),
        'stock': str(p.stock)
    })
    pipe.expire(f'cache:product:{p.id}', 300)
pipe.execute()

Without a pipeline: 2,000 commands × 1 ms = 2 seconds With a pipeline: 1 round-trip = ~5 ms

Rule: If you're going to do more than 5 Redis operations back to back, consider using a pipeline.

Pipelines with transactions (MULTI/EXEC)

By default, redis-py's pipelines also work as transactions (they run all the commands atomically):

# Transfer points from one user to another (atomically)
def transfer_points(from_user, to_user, points):
    pipe = r.pipeline()
    pipe.zincrby('balances', -points, from_user)
    pipe.zincrby('balances', points, to_user)
    results = pipe.execute()
    return results

If you want pipelining without a transaction (faster but without atomicity):

pipe = r.pipeline(transaction=False)
# ... operations that don't need atomicity
pipe.execute()

For this guide, sticking with the default (with a transaction) is fine.

Connection pooling: an early note

When you create redis.Redis(...), redis-py automatically creates a connection pool internally. It shares connections between calls, reusing them efficiently.

# redis-py creates a pool implicitly
r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# 1000 operations from different "places" in your code
for _ in range(1000):
    r.set('key', 'value')  # it reuses the pool's connections internally

In FastAPI with many concurrent workers, the pool handles the connections automatically. It's efficient without you doing anything special.

⚠️ NOTE: In module 4 we'll cover explicit async pools for FastAPI with redis.asyncio.ConnectionPool. For now, the implicit pool is enough.

Mini-project: Redis Explorer

It's time to consolidate everything in code. You're going to build a CLI tool called Redis Explorer that exercises the 5 data types with realistic use cases.

Specifications

Redis Explorer is a Python script that offers a menu with 6 options:

=== Redis Explorer ===
1. Page counter (string + INCR)
2. User profile cache (hash)
3. Capped activity log (list)
4. Tag system (set)
5. Leaderboard (sorted set)
6. Server stats (INFO)
0. Exit

Select an option:

Each option runs a series of Redis operations and shows the results. The user can run the options in any order and they all modify the same local Redis.

Project structure

redis-explorer/
├── .venv/
├── explorer.py
└── requirements.txt

requirements.txt:

redis>=7.4

Implementation: `explorer.py`

"""
Redis Explorer — a CLI tool for experimenting with Redis's 5 data types.
Mini-project for Module 1 of the Redis & Caching Strategies Guide.
"""
import redis
import time
import json
from datetime import datetime


# A global connection (in production, you'd use dependency injection)
r = redis.Redis(host='localhost', port=6379, decode_responses=True)


def page_counter():
    """Demonstrates strings with atomic INCR for counting visits."""
    print("\n--- Page counter (string + INCR) ---")

    pages = ['home', 'products', 'about', 'contact']
    print(f"Simulating 10 random visits among: {pages}")

    import random
    for _ in range(10):
        page = random.choice(pages)
        new_count = r.incr(f'pageviews:{page}')
        print(f"  Visit to /{page} → total: {new_count}")

    print("\nFinal statistics:")
    for page in pages:
        count = r.get(f'pageviews:{page}') or 0
        print(f"  /{page}: {count} visits")


def user_profile_cache():
    """Demonstrates hashes for caching objects with editable fields."""
    print("\n--- User profile cache (hash) ---")

    user_id = 42
    key = f'user:{user_id}'

    # Create the profile
    r.hset(key, mapping={
        'name': 'Alex',
        'email': 'alex@example.com',
        'age': '30',
        'country': 'Argentina',
        'last_login': datetime.now().isoformat()
    })
    r.expire(key, 600)  # 10 min TTL

    print(f"Profile created for user:{user_id}:")
    profile = r.hgetall(key)
    for field, value in profile.items():
        print(f"  {field}: {value}")

    # A granular update: only last_login
    print(f"\nUpdating only last_login...")
    r.hset(key, 'last_login', datetime.now().isoformat())

    # Increment the login counter
    r.hincrby(key, 'login_count', 1)

    print(f"Profile updated:")
    for field, value in r.hgetall(key).items():
        print(f"  {field}: {value}")


def activity_log():
    """Demonstrates lists with LTRIM for a capped activity feed."""
    print("\n--- Capped activity log (list) ---")

    user_id = 42
    key = f'activity:user:{user_id}'

    # Clean up so we start fresh
    r.delete(key)

    # Generate 15 events
    events = [
        'login',
        'view_product:1',
        'view_product:5',
        'add_to_cart:5',
        'view_product:8',
        'remove_from_cart:5',
        'add_to_cart:8',
        'checkout_started',
        'payment_completed',
        'order_confirmed',
        'logout',
        'login',
        'view_orders',
        'view_order_details:42',
        'logout'
    ]

    for event in events:
        r.rpush(key, json.dumps({
            'event': event,
            'timestamp': datetime.now().isoformat()
        }))

    print(f"Total events in the log: {r.llen(key)}")

    # A capped log: keep only the last 10
    r.ltrim(key, -10, -1)
    print(f"After LTRIM (-10, -1): {r.llen(key)} events")

    print("\nThe last 10 events:")
    for raw_event in r.lrange(key, 0, -1):
        event_data = json.loads(raw_event)
        print(f"  [{event_data['timestamp'][11:19]}] {event_data['event']}")


def tag_system():
    """Demonstrates sets for a tag system with intersection."""
    print("\n--- Tag system (set) ---")

    # Products with their tags
    products_tags = {
        100: ['laptop', 'apple', 'macbook', 'm3'],
        101: ['laptop', 'lenovo', 'gaming', 'nvidia'],
        102: ['laptop', 'apple', 'macbook', 'air'],
        103: ['phone', 'apple', 'iphone'],
        104: ['phone', 'samsung', 'android'],
        105: ['laptop', 'gaming', 'nvidia', 'rgb'],
    }

    # Clear the previous indexes
    for key in r.keys('tag:*:products'):
        r.delete(key)
    for key in r.keys('product:*:tags'):
        r.delete(key)

    # Index it: tag → products, product → tags
    for product_id, tags in products_tags.items():
        for tag in tags:
            r.sadd(f'tag:{tag}:products', product_id)
        r.sadd(f'product:{product_id}:tags', *tags)

    # Searches with set operations
    print("Products with the 'laptop' tag:")
    laptops = r.smembers('tag:laptop:products')
    print(f"  {sorted(laptops)}")

    print("\nProducts with the 'apple' tag:")
    apples = r.smembers('tag:apple:products')
    print(f"  {sorted(apples)}")

    print("\nIntersection — laptop AND apple:")
    laptops_apple = r.sinter('tag:laptop:products', 'tag:apple:products')
    print(f"  {sorted(laptops_apple)}")

    print("\nLaptops that are NOT apple (gaming, lenovo, etc.):")
    laptops_not_apple = r.sdiff('tag:laptop:products', 'tag:apple:products')
    print(f"  {sorted(laptops_not_apple)}")

    print("\nUnion — laptop OR phone:")
    devices = r.sunion('tag:laptop:products', 'tag:phone:products')
    print(f"  {sorted(devices)}")

    print("\nTags of product 100:")
    tags_100 = r.smembers('product:100:tags')
    print(f"  {sorted(tags_100)}")


def leaderboard():
    """Demonstrates sorted sets for rankings with a score."""
    print("\n--- Leaderboard (sorted set) ---")

    key = 'leaderboard:weekly'
    r.delete(key)

    # Add the scores
    scores = {
        'alice': 1500,
        'bob': 2300,
        'charlie': 1800,
        'diana': 950,
        'eve': 2100,
        'frank': 1200,
        'grace': 2700,
        'henry': 800,
    }
    r.zadd(key, scores)

    print(f"Total players: {r.zcard(key)}")

    # Top 5
    print("\n--- Top 5 ---")
    top5 = r.zrevrange(key, 0, 4, withscores=True)
    for i, (player, score) in enumerate(top5, start=1):
        print(f"  {i}. {player}: {int(score)} pts")

    # A specific position
    target = 'charlie'
    rank = r.zrevrank(key, target)
    score = r.zscore(key, target)
    print(f"\n{target} is in position #{rank + 1} with {int(score)} pts")

    # Add points to a player
    print(f"\nCharlie earns 500 extra points...")
    new_score = r.zincrby(key, 500, 'charlie')
    new_rank = r.zrevrank(key, 'charlie')
    print(f"Charlie now has {int(new_score)} pts, position #{new_rank + 1}")

    # How many players have a score >= 2000
    print(f"\nPlayers with 2000+ pts: {r.zcount(key, 2000, '+inf')}")

    # The updated top 5
    print("\n--- Top 5 (updated) ---")
    top5_new = r.zrevrange(key, 0, 4, withscores=True)
    for i, (player, score) in enumerate(top5_new, start=1):
        print(f"  {i}. {player}: {int(score)} pts")


def server_info():
    """Demonstrates INFO for inspecting the server."""
    print("\n--- Server stats (INFO) ---")

    print("General information:")
    info = r.info('server')
    print(f"  Redis version: {info.get('redis_version')}")
    print(f"  Uptime: {info.get('uptime_in_seconds')} seconds")

    print("\nMemory:")
    mem = r.info('memory')
    print(f"  Used: {mem.get('used_memory_human')}")
    print(f"  Peak: {mem.get('used_memory_peak_human')}")

    print("\nConnected clients:")
    clients = r.info('clients')
    print(f"  Connected: {clients.get('connected_clients')}")

    print("\nUsage stats:")
    stats = r.info('stats')
    hits = stats.get('keyspace_hits', 0)
    misses = stats.get('keyspace_misses', 0)
    total = hits + misses
    hit_rate = (hits / total * 100) if total > 0 else 0
    print(f"  Total commands processed: {stats.get('total_commands_processed')}")
    print(f"  Keyspace hits: {hits}")
    print(f"  Keyspace misses: {misses}")
    print(f"  Hit rate: {hit_rate:.1f}%")

    print(f"\nTotal keys right now: {r.dbsize()}")


def main():
    """The Redis Explorer's main menu."""
    print("=" * 50)
    print("  REDIS EXPLORER — Module 1 mini-project")
    print("=" * 50)

    # Verify the connection
    try:
        r.ping()
    except redis.ConnectionError:
        print("\n❌ Couldn't connect to Redis at localhost:6379")
        print("   Make sure Redis is running:")
        print("   docker run -d --name redis-dev -p 6379:6379 redis:7")
        return

    actions = {
        '1': ('Page counter (string + INCR)', page_counter),
        '2': ('User profile cache (hash)', user_profile_cache),
        '3': ('Capped activity log (list)', activity_log),
        '4': ('Tag system (set)', tag_system),
        '5': ('Leaderboard (sorted set)', leaderboard),
        '6': ('Server stats (INFO)', server_info),
    }

    while True:
        print("\n--- Menu ---")
        for key, (name, _) in actions.items():
            print(f"  {key}. {name}")
        print("  0. Exit")

        choice = input("\nSelect an option: ").strip()

        if choice == '0':
            print("See you later!")
            break
        elif choice in actions:
            _, func = actions[choice]
            try:
                func()
            except Exception as e:
                print(f"❌ Error: {e}")
        else:
            print("Invalid option")


if __name__ == '__main__':
    main()

Running it

# Make sure Redis is running
docker ps | grep redis

# Activate the venv if it isn't active
source .venv/bin/activate

# Run the explorer
python explorer.py

Expected output:

==================================================
  REDIS EXPLORER — Module 1 mini-project
==================================================

--- Menu ---
  1. Page counter (string + INCR)
  2. User profile cache (hash)
  3. Capped activity log (list)
  4. Tag system (set)
  5. Leaderboard (sorted set)
  6. Server stats (INFO)
  0. Exit

Select an option: 1

--- Page counter (string + INCR) ---
Simulating 10 random visits among: ['home', 'products', 'about', 'contact']
  Visit to /products → total: 1
  Visit to /home → total: 1
  ...

Verifying with redis-cli

After running several of the Explorer's options, check from redis-cli that the data is there:

docker exec -it redis-dev redis-cli

127.0.0.1:6379> KEYS *
1) "pageviews:home"
2) "user:42"
3) "activity:user:42"
4) "tag:laptop:products"
5) "leaderboard:weekly"
... etc

127.0.0.1:6379> GET pageviews:home
"3"

127.0.0.1:6379> HGETALL user:42
1) "name"
2) "Alex"
...

127.0.0.1:6379> ZREVRANGE leaderboard:weekly 0 2 WITHSCORES
1) "grace"
2) "2700"
3) "bob"
4) "2300"
5) "charlie"
6) "2300"   # with the 500 extra points we added

The mini-project's success criteria

The script runs without errors
All 6 options work correctly
Each option uses the data type appropriate for its use case
You verify with redis-cli that the data really is in Redis
You understand why each use case chose that specific data type

If all 5 are ✅, you've completed module 1.

Troubleshooting

Problem 1: `ImportError: No module named 'redis'`

Cause: You didn't install redis-py in your active venv.

Solution:

# Check that the venv is active
which python  # it should point to .venv/bin/python

# If it isn't active:
source .venv/bin/activate

# Install it
pip install redis

Problem 2: `redis.exceptions.ConnectionError: Error 111 connecting to localhost:6379`

Cause: Redis isn't running, or it isn't on the host/port you expect.

Solution:

docker ps | grep redis
# If it doesn't show up:
docker start redis-dev
# Or create it:
docker run -d --name redis-dev -p 6379:6379 redis:7

Problem 3: Values come back as bytes instead of strings

Cause: You forgot decode_responses=True.

Solution:

# Wrong:
r = redis.Redis(host='localhost', port=6379)
r.get('key')  # b'value'  <-- bytes

# Right:
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
r.get('key')  # 'value'  <-- str

Problem 4: `HSET` doesn't accept multiple fields in older versions

Cause: redis-py < 4.0 doesn't support r.hset(key, mapping={...}).

Solution: Upgrade redis-py:

pip install --upgrade redis

For modern versions (5.x), mapping={...} works perfectly.

Problem 5: The pipeline isn't faster on localhost

Cause: On localhost, network latency is <1 ms, so the speedup isn't noticeable.

Solution: The pipeline's benefit shows up in production where latency is 5-50 ms. On localhost the difference is marginal but real. For real benchmarks, use Redis in the cloud or add artificial latency:

import time
# Without a pipeline in the cloud (30 ms latency):
# 100 ops × 30ms = 3 seconds

# With a pipeline:
# 1 round-trip × 30ms = 30ms
# = 100x faster in production

Problem 6: `r.zadd` with the old syntax

Cause: redis-py 3.x used r.zadd(key, score, member). Modern versions (4+) use a dict.

Solution:

# The current syntax (4.x+):
r.zadd('leaderboard', {'alice': 1500, 'bob': 2300})

# If you see tutorials with the old syntax:
# r.zadd('leaderboard', 1500, 'alice')   # does NOT work in 5.x

Module 1 summary

You've closed module 1. Now you can:

✅ Install Redis with Docker in 30 seconds
✅ Operate redis-cli fluently (PING, SET/GET, EXPIRE, INFO, MONITOR, etc.)
✅ Command the 5 data types with judgment:
- Strings for atomic counters, simple caching, locks
- Hashes for objects with editable fields, sessions
- Lists for queues, event logs, job queues
- Sets for unique collections, tags, tracking
- Sorted sets for rankings, leaderboards, sliding window rate limiting
✅ Connect redis-py from Python (decode_responses=True)
✅ Use pipelines for efficient batch operations
✅ Build a mini-project that exercises the 5 data types with real use cases

What's coming in module 2: Caching patterns. You'll take everything you learned here and apply it strategically — when to use cache-aside vs write-through vs write-behind, how to design a TTL, how to prevent cache stampede. It's the heart of the guide.

Additional resources

redis-py Documentation — The official Python client, complete reference
redis-py Connection Patterns — Synchronous and async connections, pools
redis-py Examples — Official examples in the repo
redis.asyncio API — A module 4 preview (the async client)
Redis University: RU102 (Redis for Java/Python Developers) — A free official course focused on Python
Real Python: Using Redis with Python — A complementary tutorial with use cases

What's next?

You've completed Module 1: Redis Fundamentals. In Module 2: Caching Patterns & TTL you get into strategy: when to cache, which pattern to use, how much TTL, and how to invalidate when the data changes. It's the most important module in the guide — where Redis goes from a tool to a strategy.

Before moving on, make sure you:

Have Redis Explorer running without errors
Have tried all 6 menu options
Verified with redis-cli that each option left data in Redis
Can explain why each use case chose that specific data type

If all 4 are ✅, you're ready. On to module 2.

← PreviousLists, Sets, and Sorted Sets Next →Introduction to Module 2: Caching Patterns and TTL

redis-py and the Redis Explorer Mini-project

Overview

Installing redis-py

Basic connection from Python

The detail about decode_responses=True

Connecting with a URL (more portable)

Operations from Python

Strings

Hashes

Lists

Sets

Sorted Sets

Pipelines: batch operations

Without a pipeline (slow)

With a pipeline (fast)

A pipeline with results

A real use case: loading 1000 products into the cache

Pipelines with transactions (MULTI/EXEC)

Connection pooling: an early note

Mini-project: Redis Explorer

Specifications

Project structure

Implementation: explorer.py

Running it

Verifying with redis-cli

The mini-project's success criteria

Troubleshooting

Problem 1: ImportError: No module named 'redis'

Problem 2: redis.exceptions.ConnectionError: Error 111 connecting to localhost:6379

Problem 3: Values come back as bytes instead of strings

Problem 4: HSET doesn't accept multiple fields in older versions

Problem 5: The pipeline isn't faster on localhost

Problem 6: r.zadd with the old syntax

Module 1 summary

Additional resources

What's next?

The detail about `decode_responses=True`

Implementation: `explorer.py`

Problem 1: `ImportError: No module named 'redis'`

Problem 2: `redis.exceptions.ConnectionError: Error 111 connecting to localhost:6379`

Problem 4: `HSET` doesn't accept multiple fields in older versions

Problem 6: `r.zadd` with the old syntax