Redis with Python

Use Redis in Python with redis-py. Install, connect, and work with strings, hashes, lists, and more.

Installation

Install redis-py with pip:

BASH

pip install redis

For better performance, install the optional hiredis parser. It replaces the pure-Python response parser with a C extension and is faster under high throughput:

BASH

pip install "redis[hiredis]"

Connecting

Default connection

redis.Redis() connects to 127.0.0.1 on port 6379, database 0, with no password:

PYTHON

import redis

r = redis.Redis()
r.ping()  # True

Connecting with host, port, and password

PYTHON

r = redis.Redis(
    host="10.0.1.50",
    port=6379,
    password="s3cr3t",
    db=0,
    decode_responses=True,
)

Setting decode_responses=True makes the client return strings instead of bytes. This is usually what you want.

Connecting with a URL string

redis.from_url() accepts a Redis URL, which is convenient for reading connection details from an environment variable:

PYTHON

import os
import redis

r = redis.from_url(os.environ["REDIS_URL"], decode_responses=True)
# e.g. REDIS_URL=redis://:s3cr3t@10.0.1.50:6379/0

SSL/TLS connections

For managed Redis services that require TLS, pass ssl=True:

PYTHON

r = redis.Redis(
    host="my-redis.example.com",
    port=6380,
    password="s3cr3t",
    ssl=True,
    decode_responses=True,
)

Or via URL:

PYTHON

r = redis.from_url("rediss://10.0.1.50:6380", decode_responses=True)

The rediss:// scheme (with double s) enables TLS.

Strings

Strings are the most common Redis type. Use them for caching, counters, and session tokens.

PYTHON

# Basic set and get
r.set("product:name:42", "Wireless Keyboard")
r.get("product:name:42")  # "Wireless Keyboard"

# Set with expiry (seconds)
r.set("session:token:abc123", "user_id=88", ex=3600)

# Set with expiry in milliseconds
r.set("rate:user:88", "1", px=60000)

# Set with expiry at a Unix timestamp
import time
r.set("promo:flash_sale", "active", exat=int(time.time()) + 86400)

# NX: only set if the key does not exist
r.set("lock:job:export", "1", ex=30, nx=True)

# XX: only set if the key already exists
r.set("session:token:abc123", "user_id=88", ex=7200, xx=True)

Counters

INCR and DECR increment and decrement integer values atomically:

PYTHON

r.set("stats:page_views:home", 0)
r.incr("stats:page_views:home")        # 1
r.incrby("stats:page_views:home", 10)  # 11
r.decr("stats:page_views:home")        # 10

Bulk operations

MSET and MGET write and read multiple keys in a single round trip:

PYTHON

r.mset({
    "config:max_retries": "3",
    "config:timeout_ms": "5000",
    "config:env": "production",
})

values = r.mget("config:max_retries", "config:timeout_ms", "config:env")
# ["3", "5000", "production"]

Hashes

Hashes store field-value pairs under a single key. They are well suited for representing objects where you need to read or update individual fields.

PYTHON

# Set one or more fields
r.hset("user:profile:55", mapping={
    "name": "Priya Sharma",
    "email": "priya@example.com",
    "plan": "pro",
    "login_count": "0",
})

# Get a single field
r.hget("user:profile:55", "email")  # "priya@example.com"

# Get all fields and values
r.hgetall("user:profile:55")
# {"name": "Priya Sharma", "email": "priya@example.com", ...}

# Check if a field exists
r.hexists("user:profile:55", "phone")  # False

# Delete a field
r.hdel("user:profile:55", "login_count")

hmset is deprecated since Redis 4.0. Use hset with the mapping argument instead.

Lists

Lists are ordered sequences of strings. New elements can be added to either end, making them useful for queues and activity feeds.

PYTHON

# Push to the left (head)
r.lpush("notifications:user:88", "Your order shipped")
r.lpush("notifications:user:88", "Payment received")

# Push to the right (tail)
r.rpush("queue:email_jobs", '{"to": "alice@example.com", "template": "welcome"}')

# Pop from the left
r.lpop("queue:email_jobs")

# Get a range (0 to -1 returns all elements)
r.lrange("notifications:user:88", 0, 4)
# ["Payment received", "Your order shipped"]

# Length
r.llen("notifications:user:88")  # 2

A common pattern for a recent activity feed: push new events with lpush and trim the list to a fixed size:

PYTHON

r.lpush("activity:user:88", "Logged in from 192.168.1.10")
r.ltrim("activity:user:88", 0, 49)  # Keep only the 50 most recent events

Sets

Sets store unique, unordered values. They are useful for tracking memberships, tags, and deduplication.

PYTHON

r.sadd("article:tags:301", "postgresql", "performance", "indexing")

# All members
r.smembers("article:tags:301")  # {"postgresql", "performance", "indexing"}

# Check membership
r.sismember("article:tags:301", "performance")  # True

# Remove a member
r.srem("article:tags:301", "indexing")

Set operations

PYTHON

r.sadd("user:interests:55", "databases", "python", "linux")
r.sadd("user:interests:60", "databases", "go", "linux")

# Union: all interests from either user
r.sunion("user:interests:55", "user:interests:60")
# {"databases", "python", "linux", "go"}

# Intersection: interests both users share
r.sinter("user:interests:55", "user:interests:60")
# {"databases", "linux"}

# Difference: interests user 55 has that user 60 does not
r.sdiff("user:interests:55", "user:interests:60")
# {"python"}

Sorted sets

Sorted sets associate a floating-point score with each member. Members are always returned in score order. They are commonly used for leaderboards, rate limiting windows, and priority queues.

PYTHON

# Add members with scores
r.zadd("leaderboard:week", {
    "alice": 9850,
    "bob": 12400,
    "carol": 7200,
})

# Range from lowest to highest score (index-based)
r.zrange("leaderboard:week", 0, -1, withscores=True)
# [("carol", 7200.0), ("alice", 9850.0), ("bob", 12400.0)]

# Top 3 by highest score
r.zrange("leaderboard:week", 0, 2, rev=True, withscores=True)
# [("bob", 12400.0), ("alice", 9850.0), ("carol", 7200.0)]

# Members with scores between 8000 and 15000
r.zrangebyscore("leaderboard:week", 8000, 15000)
# ["alice", "bob"]

# Rank of a member (0-indexed, ascending)
r.zrank("leaderboard:week", "alice")  # 1

# Remove a member
r.zrem("leaderboard:week", "carol")

Key expiry and TTL

PYTHON

# Set a TTL in seconds on an existing key
r.expire("session:token:abc123", 3600)

# Check remaining TTL (in seconds)
r.ttl("session:token:abc123")  # e.g. 3598

# TTL in milliseconds
r.pttl("session:token:abc123")  # e.g. 3597842

# Remove the expiry (make the key persistent)
r.persist("session:token:abc123")

# TTL returns -1 if no expiry, -2 if the key does not exist
r.ttl("nonexistent:key")  # -2

Pipelining

Each Redis command normally requires a round trip to the server. Pipelining batches multiple commands into a single network call, which cuts latency significantly when you need to run many commands together.

PYTHON

import redis

r = redis.Redis(decode_responses=True)

# Without pipelining: 1000 round trips
for i in range(1000):
    r.set(f"counter:{i}", 0)

# With pipelining: 1 round trip
with r.pipeline() as pipe:
    for i in range(1000):
        pipe.set(f"counter:{i}", 0)
    pipe.execute()

The context manager calls execute() for you when the block exits. Commands inside the pipeline are queued locally and sent to Redis in one batch.

You can also read results back from a pipeline:

PYTHON

with r.pipeline() as pipe:
    pipe.get("product:name:42")
    pipe.get("product:name:43")
    pipe.get("product:name:44")
    results = pipe.execute()
# results = ["Wireless Keyboard", "USB-C Hub", None]

For atomic operations, use pipe.watch() to implement optimistic locking with MULTI/EXEC. For simpler atomicity requirements, Lua scripts via r.eval() are often cleaner.

Connection pooling

By default, each redis.Redis() instance creates its own connection. In a web application handling concurrent requests, you want to reuse connections instead of opening a new one per request.

ConnectionPool manages a pool of connections shared across threads:

PYTHON

import redis

pool = redis.ConnectionPool(
    host="10.0.1.50",
    port=6379,
    password="s3cr3t",
    max_connections=20,
    decode_responses=True,
)

# Pass the pool to Redis clients
r = redis.Redis(connection_pool=pool)

Create the pool once at application startup and share the r instance (or create new Redis instances from the same pool). The pool is thread-safe.

For URL-based configuration:

PYTHON

pool = redis.ConnectionPool.from_url(
    "redis://:s3cr3t@10.0.1.50:6379/0",
    max_connections=20,
    decode_responses=True,
)
r = redis.Redis(connection_pool=pool)

Error handling

The most common exceptions come from the redis.exceptions module:

PYTHON

import redis
from redis.exceptions import ConnectionError, TimeoutError, ResponseError

r = redis.Redis(
    host="10.0.1.50",
    socket_connect_timeout=2,
    socket_timeout=2,
    decode_responses=True,
)

try:
    r.set("product:name:42", "Wireless Keyboard")
except ConnectionError:
    # Server unreachable or connection dropped
    print("Could not connect to Redis")
except TimeoutError:
    # Command took too long
    print("Redis command timed out")
except ResponseError as e:
    # Wrong type, invalid arguments, etc.
    print(f"Redis error: {e}")

ResponseError covers cases like calling a list command on a string key, or exceeding memory limits. Set socket_connect_timeout and socket_timeout on the connection to avoid blocking indefinitely when Redis is slow or unreachable.

Async support

redis-py includes a built-in async client for use with asyncio. The API mirrors the synchronous client:

PYTHON

import asyncio
import redis.asyncio as aioredis

async def main():
    r = aioredis.Redis(host="localhost", decode_responses=True)
    await r.set("session:token:abc123", "user_id=88", ex=3600)
    value = await r.get("session:token:abc123")
    print(value)  # "user_id=88"
    await r.aclose()

asyncio.run(main())

The async client supports the same commands, pipelining, connection pooling, and pub/sub as the sync client. Use it in FastAPI, Starlette, or any other asyncio-based application.

Quick reference

Command	Description
`r.set(key, value, ex=n)`	Set a string with optional TTL in seconds
`r.get(key)`	Get a string value
`r.mset({k: v, ...})`	Set multiple keys at once
`r.mget(k1, k2, ...)`	Get multiple keys at once
`r.incr(key)`	Increment an integer value
`r.hset(key, mapping={...})`	Set hash fields
`r.hgetall(key)`	Get all hash fields and values
`r.lpush(key, val)`	Prepend to a list
`r.rpush(key, val)`	Append to a list
`r.lrange(key, 0, -1)`	Get all list elements
`r.sadd(key, *members)`	Add to a set
`r.smembers(key)`	Get all set members
`r.zadd(key, {member: score})`	Add to a sorted set
`r.zrange(key, 0, -1, rev=True)`	Get sorted set members by score
`r.expire(key, seconds)`	Set a TTL on an existing key
`r.ttl(key)`	Get remaining TTL in seconds
`r.pipeline()`	Open a pipeline context
`redis.ConnectionPool(...)`	Create a shared connection pool