Skip to Content
Course content

289: Connecting to Redis with redis-py

Click on the "Edit" button in the top corner of the screen to edit your slide content.

So, you've probably realized by now that while Python's dictionaries are great, they disappear the moment your script stops running. When we need a data store that's incredibly fast, lives outside our application process, but doesn't require the overhead of a full relational database, Redis is usually my first choice. It's basically a giant, networked dictionary with some very clever superpowers.

To get us started, we'll need the redis-py library. I'm going to build a simple page-view tracker for a blog. It's a classic use case because updating a count in a SQL database every time a page loads is a great way to kill your performance. Redis handles this with ease.

Getting the connection established

First, we need to create a client. I usually keep my Redis connection in a dedicated utility function or a singleton, but for this example, we'll keep it inline. Assuming you have Redis running on your local machine at the default port, here is how we hook into it.

import redis

# I'm creating the Redis client here. 
# By default, it connects to localhost:6379
r = redis.Redis(host='localhost', port=6379)

# Let's test the connection with a simple ping
if r.ping():
    print("Connected to Redis!")

Incrementing views atomically

The magic of Redis for this project is the incr command. In a standard database, you'd have to read the value, add one, and save it back—which creates a race condition if two people visit the page at the exact same millisecond. Redis does this "atomically," meaning it happens in one unbreakable step.

# Imagine this function runs every time someone hits the 'About Me' page
def track_page_view(page_id):
    key = f"views:{page_id}"
    new_count = r.incr(key)
    return new_count

print(f"Current views for home page: {track_page_view('home')}")
print(f"Current views for home page: {track_page_view('home')}")

The "Byte-String" trap

Now, I'll do something I've done a dozen times in my early projects. Let's say I want to retrieve the current count without incrementing it using get. Watch what happens:

# Trying to fetch the value
current_views = r.get("views:home")
print(f"The count is: {current_views}") 
# Output: The count is: b'2'

See that b'2'? That's a byte string. By default, redis-py returns data as bytes because Redis itself is binary-safe. It doesn't know or care if you're storing a string, an integer, or a serialized image. But in Python, having to call .decode('utf-8') on every single result is a massive pain in the neck.

The fix is simple: when you initialize the client, tell it to decode the responses for you. I'm going to restart the client with that flag.

# The corrected connection
r = redis.Redis(host='localhost', port=6379, decode_responses=True)

current_views = r.get("views:home")
print(f"The count is: {current_views}") 
# Output: The count is: 2

Adding a "Trending" window with expiration

A total view count is fine, but it's more interesting to know how many people visited in the last hour. This is where Redis expire comes in. We can create a key for the current hour and tell Redis to delete it automatically after 3600 seconds.

import time

def track_trending_view(page_id):
    # Create a key based on the current hour (e.g., views:home:2023102714)
    hour_timestamp = time.strftime("%Y%m%d%H")
    key = f"trending:{page_id}:{hour_timestamp}"
    
    r.incr(key)
    # Set the key to expire in 1 hour if it's new
    r.expire(key, 3600)
    
    return r.get(key)

print(f"Hourly views: {track_trending_view('home')}")



📋 Practical Task

Build a Temporary User Session Store

Your task is to create a small script that simulates a user session manager using redis-py. This is a common way to handle "logged-in" states without hitting a primary database on every request.

  • Initialize a Redis client with decode_responses=True.
  • Write a function set_session(user_id, session_token) that stores the token under a key like session:user_123.
  • The session should be temporary; set the key to expire in 300 seconds (5 minutes).
  • Write a function get_session(user_id) that retrieves the token. If the token has expired or doesn't exist, it should return the string "Session Expired".
  • Test your logic by setting a session, retrieving it, and then manually deleting the key using r.delete() to simulate an expiration before retrieving it again.
Rating
0 0

There are no comments for now.

to be the first to leave a comment.