Working with Redis in Python: A Practical Guide to redis-py

Redis is the in-memory data store you reach for when the database is too slow and a local variable is too small. From Python, the official client is redis (a.k.a. redis-py); it ships a sync client, an asyncio client, connection pooling, pipelines, and pub/sub out of the box. This guide walks through a real, runnable setup: install, your first connection, the most useful data commands, pipelines, pooling for web apps, async usage, and the gotchas that catch people on day one.

TL;DR

• pip install "redis[hiredis]" gets you the client plus a fast C parser.
• A redis.Redis(decode_responses=True) is a single client instance; pass it around your app.
• The five commands you will type 95% of the time: set, get, hset, hgetall, expire.
• pipeline() batches commands in one round trip and is roughly an order of magnitude faster than one-call-per-command.
• Share a ConnectionPool across requests in a web app; don’t build a new Redis() per request.
• The async client lives at redis.asyncio and uses await plus aclose() instead of close().

Install redis-py and connect

You need a running Redis server. The fastest way to get one for local development is Docker; everything else (Redis Stack, Homebrew, a managed instance) also works.

docker run -d --name redis -p 6379:6379 redis:7-alpine

Now install the Python client. The [hiredis] extra pulls in a compiled C parser that is meaningfully faster than the pure-Python one. If you cannot install compiled packages in your environment, drop the extra and you will still get a working client:

pip install "redis[hiredis]"

The client connects on first use, not at construction. This is the smallest end-to-end example and the one to keep open while you read the rest of the article:

import redis

r = redis.Redis(host="localhost", port=6379, decode_responses=True)
print(r.ping())
# output: True

A few things to notice in that snippet. host and port default to localhost and 6379, so you can usually drop them. decode_responses=True is the most important flag in this guide: without it, r.get("foo") returns b'foo' (bytes), and you will spend your first hour debugging string-versus-bytes errors. With it, you get str. Turn it on unless you have a reason not to.

r.ping() returns True if the server is reachable and replies with PONG. If it raises redis.exceptions.ConnectionError, the client is down, the port is wrong, or a firewall is in the way.

URL-style connection

For anything other than a quick script, store the connection details in a URL and use from_url. The same URL works for the sync client, the async client, and ConnectionPool:

import redis

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

Supported schemes: redis:// (plain TCP), rediss:// (TLS), and unix://path/to/socket.sock (Unix domain socket). You can embed a username and password: redis://user:pass@host:6379/0. Keep URLs out of source control; pull them from an environment variable.

Strings, hashes, and TTL

Most Redis usage boils down to a few data types. Strings are the simplest and the most common; hashes are the next step up when you want to group fields under a single key.

Strings with TTL

set and get are the bread and butter. setex combines a SET with an expiration in one call, which is what you want for session tokens, OTPs, and any cache entry that should disappear on its own:

r.set("user:42:name", "Ada")
print(r.get("user:42:name"))      # Ada
print(r.exists("user:42:name"))   # 1

r.setex("session:abc", 60, "logged-in")
print(r.ttl("session:abc"))       # 60 (or just under)

r.get returns None for a missing key, not an exception. r.ttl returns -1 if the key exists with no expiration, and -2 if the key does not exist. Both are worth knowing because they show up in log output and dashboards.

To read or write many keys in one round trip, use mget and mset. mget returns a list aligned with the keys you passed, and missing keys come back as None in their slot:

r.mset({"k1": "v1", "k2": "v2"})
print(r.mget("k1", "k2", "missing"))
# output: ['v1', 'v2', None]

Atomic counters are the other thing strings are good for. incr and decr are atomic, which makes them safe for rate limiting, sequence numbers, and leaderboards across multiple worker processes:

r.set("page:views", 0)
for _ in range(5):
    r.incr("page:views")
print(r.get("page:views"))   # 5

The gotcha: incr on a key whose value is not an integer raises redis.exceptions.ResponseError. That is a feature, not a bug, but it bites people who use the same key for both counters and free-form values.

Hashes

A hash is a flat dictionary stored under one key. Use them when you would otherwise build keys like user:42:name, user:42:email, user:42:age. The mapping= keyword is the cleanest way to write the whole hash at once:

r.hset(
    "user:42",
    mapping={"name": "Ada", "email": "ada@example.com", "age": 30},
)
print(r.hgetall("user:42"))
# output: {'name': 'Ada', 'email': 'ada@example.com', 'age': '30'}

Two real gotchas in that output. First, every field comes back as a string, even numbers. r.hget("user:42", "age") returns "30", not 30. Coerce explicitly with int(...) or store the value as JSON. Second, this works only because decode_responses=True is set. Without it, both keys and values are bytes, and the dict looks like {b'name': b'Ada', ...}.

For typed wrappers around hgetall, the dataclasses guide is a useful follow-up.

Pipelines for batched commands

Every command in Redis is one round trip. If you do 100 INCRs, the network is the bottleneck. A pipeline buffers commands client-side and ships them in one batch, which is roughly an order of magnitude faster for bulk work. The redis-py docs quote 100,000 INCRs at about 21.7 seconds without a pipeline and about 2.4 seconds with one.

Pipelines are chainable. The pattern reads almost like a script:

pipe = r.pipeline(transaction=True)
pipe.set("counter", 0)
for _ in range(3):
    pipe.incr("counter")
pipe.get("counter")
results = pipe.execute()
print(results)
# output: [True, 1, 2, 3, '3']

Two things to know. transaction=True (the default) wraps the buffered commands in Redis MULTI/EXEC, so the batch runs atomically: either every command succeeds or none of them do. Pass transaction=False for pure batching without atomicity, which is faster but lets other clients interleave. And pipe.execute() returns a list with the per-command results in the same order you queued them, so you can destructure or index into it directly.

Pipelines are not thread-safe. Create one per request or per worker; do not share a single pipeline across threads.

Connection pools for web apps

A redis.Redis() is cheap to construct, but every connection it opens is a real TCP socket plus a server-side file descriptor. In a long-running web app, the right pattern is one shared ConnectionPool and a Redis() instance that points at it:

import redis

POOL = redis.ConnectionPool.from_url(
    "redis://localhost:6379/0",
    max_connections=20,
    decode_responses=True,
)

def get_redis() -> redis.Redis:
    return redis.Redis(connection_pool=POOL)

A few defaults worth knowing. max_connections=None means “unlimited”, which is fine on a workstation and a footgun on a server with hundreds of workers. health_check_interval=0 (the default) disables the periodic PING that recycles dead connections; set it to 30 seconds or so once you deploy. The pool is thread-safe; pipelines and pub/sub objects built on top of it are not.

For production URLs, front Redis with a load balancer or Sentinel. The URL form redis://sentinel1:26379;sentinel2:26379 plus sentinel_kwargs={"master_name": "mymaster"} is documented in the redis-py Sentinel examples.

Async Redis with redis.asyncio

The async client lives in the same package and mirrors the sync API, with the obvious differences: every method is a coroutine, you use aclose() instead of close(), and the factory redis.asyncio.from_url is itself a coroutine that returns a connected client:

import asyncio
import redis.asyncio as redis

async def main() -> None:
    client = redis.Redis(decode_responses=True)
    await client.set("hello", "world")
    print(await client.get("hello"))   # world
    await client.aclose()

asyncio.run(main())

Use this in asyncio web frameworks (FastAPI, Starlette, aiohttp) and any code that already runs on an event loop. The sync client is the right choice for requests-style code, flask, django, and most scripts; mixing the two within a single function is almost always the wrong call.

Pub/sub works the same shape: open a channel with async with client.pubsub() as pubsub:, subscribe, and consume messages with await pubsub.get_message(...). The aclose() / close() distinction is the most common mistake when porting sync code: using the wrong one leaks the connection back to the pool.

Common gotchas

A few things go wrong the first time and are easy to fix once you know they exist. Most of these come from real outages, not hypothetical edge cases:

• decode_responses=False returns bytes. New users copy a tutorial, get b'foo', and panic. Turn decode_responses=True on at the client level, not per-call.

• Numeric fields in hashes are strings. r.hget("user", "age") returns "30", not 30. Coerce explicitly or store the value as JSON with the stdlib json module.

• keys is O(N) and blocks the server. Use r.scan_iter("user:*") to iterate in production-sized keyspaces. keys is fine in development, never in production.

  # Bad: blocks the server on a large keyspace
  users = r.keys("user:*")
  
  # Good: cursor-based, O(1) per call
  for key in r.scan_iter("user:*", count=500):
      print(key)

• Pipelines and pub/sub are not thread-safe. The connection pool is. Don’t share a pipeline or pub/sub across threads, and don’t share them across coroutines either.

• aclose() vs close(). Sync client uses r.close(). Async client uses await client.aclose(). Using the wrong one leaks the connection.

• Slow networks and blocking commands. socket_timeout=5 (the default) will trip you on large BLPOP waits or slow WAN links. Bump it explicitly for blocking commands, and use a higher socket_timeout for any client that talks to a remote Redis.

• set(..., nx=True) for “set if not exists”. This is the standard primitive for distributed locks. Don’t roll your own with GET + SET; you will lose the race.

• Store Python objects as JSON, not pickle. r.set("user:42", pickle.dumps(user)) works, but it is a remote-code-execution hole if the data ever comes from an untrusted source. Stick to json.dumps / json.loads for any value that crosses a trust boundary.

• Separate broker and result backend DBs in Celery. If you also use Redis as a Celery broker, keep the broker and your application keys on different database numbers. They have different access patterns and different lifecycle requirements; sharing a DB invites cross-talk.

Conclusion

The shape of a Redis-backed Python service is small: a connection pool, a Redis client with decode_responses=True, the half-dozen commands that match your data shape, and pipelines wherever you do bulk work. The operational pitfalls (bytes versus strings, keys versus scan_iter, pool sharing, aclose versus close) are all avoidable once you know they exist.

From here, the natural next step is wiring Redis into a real workload. The task queues with Celery and Redis guide shows Redis as a Celery broker and result backend, which is the most common production use. The shelve module is a useful contrast if you only need local key-value persistence with no server. And the stdlib json module is the right tool for serializing dicts before you set them.

Frequently asked questions

Should I use decode_responses=True or leave it off?

For most Python code, yes. The default False returns bytes, so r.get("foo") gives you b"foo", and you will hit string-versus-bytes bugs the first time you compare it to a Python string. Turn decode_responses=True on at the client level and stay in str land.

Is the sync client faster than the async client?

Not in any meaningful way for typical workloads; both clients are bindings to the same RESP protocol. Pick sync for flask, django, scripts, and any code that does not already have an event loop. Pick redis.asyncio for FastAPI, aiohttp, and other asyncio code; the async advantage only shows up when you are already inside an event loop with hundreds of concurrent awaits.

When does a connection pool actually matter?

Always in a web app. Every Redis() instance with no connection_pool= argument opens a fresh socket on first use, and a server with hundreds of worker threads can exhaust the server-side file descriptor limit quickly. One ConnectionPool.from_url(...) shared across the app gives you bounded concurrency (max_connections=20) and dead-connection recycling via health_check_interval.

How do I safely run a blocking command like BLPOP?

Bump socket_timeout on the client and pass timeout= to the blocking call. r.blpop("queue", timeout=10) is the right form: the call returns None if nothing arrives in 10 seconds instead of hanging, and you can fall back to logging, retrying, or yielding to other work. The default 5-second socket timeout will trip on slow networks before your business logic can finish.

See also

• Task Queues with Celery and Redis: Redis as a Celery broker and result backend.
• shelve module: a stdlib key-value store for local-only persistence.
• json module: serialize dicts and lists before storing them in Redis.
• Dataclasses: typed wrappers around hgetall that coerce fields back to int, float, or datetime.