Before we begin, we need to have a Redis server running and the necessary Python library installed.
Get the Redis Server with Docker
Docker is a straightforward way to get a Redis instance running locally.
# Pull the official Redis image
docker pull redis
# Run the container, mapping the port to your machine
docker run --name my-redis -p 6379:6379 -d redis
pip install redis
2. Connecting to Redis
import redisimport time# NOTE: The standard Docker command uses port 6379.redis_client = redis.StrictRedis( host='localhost', port=6379, decode_responses=True,)try: redis_client.ping()print("Connected to Redis server successfully.")except redis.ConnectionErroras e:print(f"Failed to connect: {e}")
Connected to Redis server successfully.
3. Basic Operations: Strings
# Setting and Getting values:redis_client.set('example_key', 'hello ajulabs!')value = redis_client.get('example_key')print(f"Value for 'example_key': {value}")# Set with expiration (seconds)redis_client.setex('temp_key', 5, 'This will expire in 5 seconds')print(f"Temp Key (before expiration): {redis_client.get('temp_key')}")# Wait for the key to expiretime.sleep(6)print(f"Temp Key (after expiration): {redis_client.get('temp_key')}") # Output: None
Value for 'example_key': hello ajulabs!
Temp Key (before expiration): This will expire in 5 seconds
Temp Key (after expiration): None
4. Working with Lists
# Clean up the key to ensure a fresh startredis_client.delete('ajulabs_team')# Add members to the end of the listredis_client.rpush('ajulabs_team', 'diogo', 'tiago', 'esdras', 'gabriel')# Retrieve all members from the listteam_members = redis_client.lrange('ajulabs_team', 0, -1)print(f"Team members: {team_members}")# You can use other data types to store metadataredis_client.set('members_count', len(team_members))print(f"Number of team members: {redis_client.get('members_count')}")
Team members: ['diogo', 'tiago', 'esdras', 'gabriel']
Number of team members: 4
5. Working with Hashes
# Create a profile for 'esdras'redis_client.hset('user:esdras', mapping={'name': 'Esdras','role': 'GOAT','tasks_completed': 1000})# And another for 'andressa'redis_client.hset('user:andressa', mapping={'name': 'Andressa','role': 'Front-end Developer','tasks_completed': 8})# HGETALL retrieves all fields and values in a hashesdras_profile = redis_client.hgetall('user:esdras')print(f"Esdras's Profile: {esdras_profile}")# HINCRBY increments a numeric value in a hashredis_client.hincrby('user:esdras', 'tasks_completed', 1)esdras_tasks = redis_client.hget('user:esdras', 'tasks_completed')print(f"Esdras's updated tasks: {esdras_tasks}")
# Clean up previous keys before startingredis_client.delete('project:alpha', 'project:beta')# SADD adds members to a set. Let's assign teams to projects.# Project Alpha teamredis_client.sadd('project:alpha', 'diogo', 'tassio', 'andressa', 'julio')# Project Beta teamredis_client.sadd('project:beta', 'esdras', 'gabriel', 'andressa', 'cristiano')# SINTER finds the intersection of sets (members on BOTH projects)both_projects = redis_client.sinter('project:alpha', 'project:beta')print(f"Team members on both projects: {both_projects}") # Output: {'andressa'}
Team members on both projects: {'andressa'}
7. Working with Sorted Sets (ZSETs)
leaderboard_key ='leaderboard:loc_weekly'redis_client.delete(leaderboard_key)# ZADD adds members to the sorted set with a scoreredis_client.zadd(leaderboard_key, {'gabriel': 2000,'leticia': 950,'tiago': 1500,'lucas': 800,'esdras': 1999})# ZINCRBY increments the score of a memberredis_client.zincrby(leaderboard_key, 250, 'leticia')# ZREVRANGE retrieves a range of members, ordered from highest to lowest scoretop_3 = redis_client.zrevrange(leaderboard_key, 0, 2, withscores=True)print("🏆 Weekly Leaderboard (Top 3):")for i, (member, score) inenumerate(top_3):print(f"{i+1}. {member} - {int(score)} lines")# ZSCORE gets the score for a specific membertiago_score = redis_client.zscore(leaderboard_key, 'tiago')print(f"\nTiago's current score: {int(tiago_score)}")
# Create a pipelinepipe = redis_client.pipeline()# Assume a hash for 'daniel' exists or will be createdredis_client.hset('user:daniel', 'tasks_completed', 0)# Stage the commands. They are not executed yet.pipe.hincrby('user:daniel', 'tasks_completed', 1) pipe.lpush('activity_log', 'User daniel completed a task.')# Execute all commands in the pipelineresults = pipe.execute()print(f"Pipeline execution results: {results}")print(f"Daniel's new task count: {redis_client.hget('user:daniel', 'tasks_completed')}")print(f"Latest activity log: {redis_client.lrange('activity_log', 0, 0)}")# Clean up logsredis_client.delete('activity_log')
Pipeline execution results: [1, 1]
Daniel's new task count: 1
Latest activity log: ['User daniel completed a task.']