More about Redis, hands-on in Python

Do in redis-cli:

help
help ping
ping
ping message

Redis is more than just a simple key-value store, it is a data structure server. It can store several more complex data types and manipulate them directly on the server.

Redis data types:

String

The most basic Redis data type. Strings can contain any kind of data, of up to 512 MB in size each.

Use cases:

• caching web pages and such

• counters (INCR, DECR)

• session management

• and other simple key-value lookups

Examples:

SET user:1000:name "Alice"
GET user:1000:name
SET page:views 100
INCR page:views

List

A linked list of strings, ordered by insertion order. Elements can be added to the head (left) or tail (right).

• queues or stacks

• background job queues (using LPUSH and RPOP)

• timeline activities (e.g., latest posts from a user).

LPUSH task:queue "task 3"
LPUSH task:queue "task 2" "task 1"
LRANGE task:queue 0 -1
RPOP task:queue

Hash

A collection of field-value pairs associated with a single key.

• representing objects with named attributes

HSET user:profile:1000 name "Bob" age 30 role "developer"
HGET user:profile:1000 name
HGETALL user:profile:1000

Set

An unordered collection of unique strings. Server-side operations: intersections, unions, differences.

• tagging systems

• tracking unique visitors (IPs or user IDs)

SADD article:1:tags redis database nosql
SADD article:1:tags redis mongo

SMEMBERS article:1:tags
SISMEMBER article:1:tags redis

SADD article:2:tags  redis postgres nosql
SINTER article:1:tags article:2:tags

Sorted Set (ZSet)

Similar to a Set, but every member is associated with a floating-point score. Elements are sorted by their score.

• leaderboards

• priority queues

ZADD leaderboard 1500 Alice 2200 Bob 1800 Charlie
ZRANGE leaderboard 0 -1 WITHSCORES
ZRANK leaderboard Bob
ZINCRBY leaderboard 1500 Bob

Bitmaps & Bitfields

Built on top of the String type, perform bit-level operations: setting, getting, counting bits.

• recording whether a user logged in on a specific day of the year.

GET user:logins:dodo
SETBIT user:logins:dodo 123 1
GETBIT user:logins:dodo 123
BITCOUNT user:logins:dodo

HyperLogLog

A probabilistic data structure used to estimate (with a small margin of error) the number of unique elements in a set, using a small fixed amount of memory.

• counting unique IP addresses or user IDs

GET website:visitors
PFADD website:visitors "192.168.1.1" "10.0.0.5" "172.16.0.2"
PFADD website:visitors "192.168.1.1"
PFCOUNT website:visitors

Streams

An append-only log data structure, such as a message queue.

• event queues, real-time data feeds, messaging systems

XADD sensor:stream * temperature 22.5 humidity 45
XLEN sensor:stream
XRANGE sensor:stream - +
XDEL sensor:stream 1680000000000
XTRIM sensor:stream MINID ~ 1000

Geospatial Indexes

Built on top of Sorted Sets, stores coordinates and efficiently queries coordinates.

• location-based services

GEOADD locations 13.404954 52.520008 "Berlin" 2.352222 48.856613 "Paris"
GEODIST locations Berlin Paris km
GEORADIUS locations 2.352222 48.856613 1000 km WITHDIST

Redis in Python

A client is a connection pool

redis.Redis(...) creates a client with a connection pool. Connections are borrowed for command execution and then returned.

Keep one client per process, not one per command. The pool is thread-safe, so multiple threads can share the same client.

common.pyfrom typing import Any

import redis

_CONN_PARS: Any = {
    "host": "localhost",
    "port": 6379,
    "db": 0,
    "decode_responses": True,
    "socket_timeout": 1.0,
    "socket_connect_timeout": 1.0,
    "max_connections": 40,  # defaults to inf
}


# redis does not provide proper type hints, so we have to use Any


# single client, private connection pool
def connect() -> Any:
    return redis.Redis(**_CONN_PARS)


_connection_pool = redis.ConnectionPool(**_CONN_PARS)


# multiple clients can share one connection pool
def connect_pool() -> Any:
    return redis.Redis(connection_pool=_connection_pool)


# delete all keys in the database
def cleanup(r: Any) -> None:
    if keys := r.keys("*"):
        r.delete(*keys)


# printing support
def Title(title: str) -> None:
    print("\n===", title, "===\n")


def Line(tag: str, val: Any) -> None:
    print(f"{tag:16} -> {val}", flush=True)

Testing the connection

connect.pyimport redis
from common import Line, Title, connect, connect_pool

Title("Testing the connection")

# single client, private connection pool
r = connect()

Line("PING", r.ping())

info = r.info("server")

# mode can be: "standalone", "sentinel", "cluster"
Line("server", f"version: {info['redis_version']}, mode: {info['redis_mode']}")

# common pool
r = connect_pool()

Line("PING shared", r.ping())

# error handling
dead = redis.Redis(host="localhost", port=6399)
try:
    dead.ping()  # type: ignore
except redis.ConnectionError as exc:
    Line("dead server", type(exc).__name__)

The main types

The return types in Python:

• get -> str or None; mget -> list

• hgetall -> dict; hget -> str or None

• lrange -> list; smembers -> set

• zrange(..., withscores=True) -> list[tuple[member, float]]

• incr, incrby, hincrby -> int

• set(..., nx=True) -> True if written, None if not

• …

redis_types.pyfrom typing import Any

from common import Line, Title, cleanup, connect

r: Any = connect()
cleanup(r)

Title("Strings")

# keys
name1 = "user:1:name"
name2 = "user:2:name"
name3 = "user:3:name"

r.set(name1, "Alice")
alice = r.get(name1)
Line("get name1", f"{alice}: {type(alice).__name__}")

r.mset({name2: "Bob", name3: "Cara"})

Line("mget", r.mget(name1, name2, name3))
Line("append", f"{r.append(name1, ' Smith')} | {r.get(name1)}")
Line("getrange", r.getrange(name1, 0, 4))

likes = "post:42:likes"

incrBy = r.incrby(likes, 2)
Line("incrby", f"{incrBy}: {type(incrBy).__name__}")

Title("Hashes")

user1 = "user:1"

r.hset(
    user1,
    mapping={"name": "Alice", "role": "member", "age": 30},
)

Line("hgetall", r.hgetall(user1))
Line("hget role", r.hget(user1, "role"))
Line("hincrby age", r.hincrby(user1, "age", 1))
Line("hkeys", r.hkeys(user1))

Title("Lists")

feed1 = "feed:alice"

r.rpush(feed1, "post:7", "post:9")
r.lpush(feed1, "post:3")

Line("lrange", r.lrange(feed1, 0, -1))
Line("lpop", r.lpop(feed1))
Line("rpop", r.rpop(feed1))

r.rpush(feed1, "post:11", "post:12")

Line("lrange", r.lrange(feed1, 0, -1))

r.ltrim(feed1, 0, 1)

Line("after ltrim 0..1", r.lrange(feed1, 0, -1))

Title("Sets - membership and set algebra")

follows_alice = "follows:alice"
follows_bob = "follows:bob"

r.sadd(follows_alice, "bob", "carol")
r.sadd(follows_bob, "alice", "carol")
Line("follows alice", sorted(r.smembers(follows_alice)))
Line("scard", r.scard(follows_alice))
Line("bob follows alice", r.sismember(follows_alice, "bob"))
Line("sinter", sorted(r.sinter(follows_alice, follows_bob)))
Line("sunion", sorted(r.sunion(follows_alice, follows_bob)))
Line("sdiff", sorted(r.sdiff(follows_alice, follows_bob)))

Title("Sorted sets - a leaderboard")

leaderboard = "leaderboard"

r.zadd(leaderboard, {"post:42": 2, "post:7": 5, "post:9": 1})
r.zincrby(leaderboard, 4, "post:42")
Line("zrange", r.zrange(leaderboard, 0, -1, withscores=True))
Line("zrevrange", r.zrevrange(leaderboard, 0, -1, withscores=True))
Line("zscore post:42", r.zscore(leaderboard, "post:42"))
Line("zrank post:7", r.zrank(leaderboard, "post:7"))

Title("Three specialised types, one line each")

visitors = "hll:visitors"

r.pfadd(visitors, "alice", "bob", "alice")
print("HyperLogLog pfcount", r.pfcount(visitors), "(approximate)")

logins = "bitmap:logins"

r.setbit(logins, 3, 1)
r.setbit(logins, 5, 1)
Line("bitmap bitcount", r.bitcount(logins))

events = "stream:events"

r.xadd(events, {"user": "alice", "action": "login"})
r.xadd(events, {"user": "bob", "action": "purchase"})

Line("stream xlen", r.xlen(events))
Line("stream xrange", r.xrange(events, min="-", max="+"))
Line("stream newest first", r.xrevrange(events, count=1))
Line("stream read", r.xread({events: "0-0"}, count=10))

Title("TYPEs")

for name in (
    name1,
    likes,
    user1,
    feed1,
    follows_alice,
    follows_bob,
    leaderboard,
    visitors,
    events,
):
    print(f"TYPE {name:<16} -> {r.type(name)}")

Expiration, TTL, and conditional expiry

A TTL is attached to the whole key, not to a field or a list element. TTL returns seconds, PTTL milliseconds, and two sentinel values:

• -1 - the key exists and has no expiration.

• -2 - the key does not exist (already expired, or never created).

pythonr.set("feed:alice", html, ex=60)          # seconds
r.set("feed:alice", html, px=60_000)      # milliseconds
r.set("feed:alice", html, ex=60, nx=True) # only if absent
r.set("feed:alice", html, keepttl=True)   # overwrite, keep the old TTL

EXPIRE can be conditional, which is useful when several workers refresh the same key:

• expire(key, 30, nx=True) - only if no TTL is set.

• expire(key, 30, xx=True) - only if a TTL already exists.

• expire(key, 30, gt=True) - only if the new TTL is greater.

• expire(key, 30, lt=True) - only if the new TTL is smaller.

expiry.pyimport time

from common import Line, Title, cleanup, connect

r = connect()
cleanup(r)

Title("A key with a deadline")

feed = "cache:feed:alice"
r.set(feed, "<html>cached feed for Alice</html>", ex=2)

Line("GET", r.get(feed))
Line("TTL, PTTL", f"{r.ttl(feed)} s, {r.pttl(feed)} ms")

time.sleep(2.1)

# -1 = no expiry, -2 = key does not exist
Line("TTL after 2.1 s", r.ttl(feed))

Title("Changing a deadline without rewriting the value")

session = "session:alice"
r.set(session, "token-123")

Line("fresh key TTL", r.ttl(session))
Line("EXPIRE 30", f"{r.expire(session, 30)} | TTL {r.ttl(session)}")
Line("EXPIRE 10 GT", f"{r.expire(session, 10, gt=True)} | TTL {r.ttl(session)}")
Line("EXPIRE 60 GT", f"{r.expire(session, 60, gt=True)} | TTL {r.ttl(session)}")
Line("PERSIST", f"{r.persist(session)} | TTL {r.ttl(session)}")

Title("GETEX: read and refresh in one round trip")

r.set(feed, "<html>v2</html>", ex=30)
Line("GETEX EX 60", f"{r.getex(feed, ex=60)} | TTL {r.ttl(feed)}")

Title("SET ... KEEPTTL keeps the old deadline")

r.set(feed, "<html>v3</html>", keepttl=True)
Line("GET", f"{r.get(feed)} | TTL {r.ttl(feed)}")

Atomicity: commands are atomic

Redis executes commands one at a time. Each single command is atomic. A sequence of commands sent by a client is not:

pythonvalue = int(r.get("counter"))
r.set("counter", value + 1)

Two threads can be interleaved as:

thread 1              thread 2
GET counter  -> 0
                      GET counter  -> 0
SET counter 1
                      SET counter 1

final value: 1, expected 2

INCR is one command:

pythonr.incr("counter")          # read, add, write - inside the server

Pipelining: one round trip to the server

A normal command makes a round trip: send, wait, receive. A pipeline sends a whole batch of commands at once, then reads the replies.

pythonwith r.pipeline(transaction=False) as pipe:
    for _ in range(1000):
        pipe.incr("counter")
    replies = pipe.execute()

• transaction=False makes a plain batch.

• transaction=True makes the batch atomic by wrapping it in a MULTI/EXEC transaction (no interleaving).

Caveats:

• MULTI/EXEC has no rollback, e.g. when a runtime error occurs in the middle of the transaction.

• MULTI/EXEC can be extended with WATCH for optimistic concurrency control.

pipeline.pyimport time

import redis
from common import Line, Title, cleanup, connect

N = 1000

r = connect()
cleanup(r)

counter = "counter"

Title(f"{N} INCRs, one round trip each")

started = time.perf_counter()
for _ in range(N):
    r.incr(counter)
one_by_one = time.perf_counter() - started
Line("one by one", f"{one_by_one * 1000:.0f} ms")

Title(f"{N} INCRs, one pipeline")

r.delete(counter)
started = time.perf_counter()
with r.pipeline(transaction=False) as pipe:
    for _ in range(N):
        pipe.incr(counter)
    results = pipe.execute()

pipelined = time.perf_counter() - started
Line("pipelined", f"{pipelined * 1000:.0f} ms")
Line("last reply", f"{results[-1]} | key {r.get(counter)}")
Line("speed-up", f"{one_by_one / pipelined:.1f}x")

Title("A pipeline is a batch, not automatically a transaction")

with r.pipeline(transaction=False) as pipe:
    pipe.set("a", "1")
    pipe.incr("a")
    pipe.get("a")
    Line("transaction=False", pipe.execute())  # no MULTI/EXEC

with r.pipeline(transaction=True) as pipe:
    pipe.set("b", "1")
    pipe.incr("b")
    pipe.get("b")
    Line("transaction=True ", pipe.execute())  # wrapped in MULTI/EXEC

Title("MULTI/EXEC has no rollback")

with r.pipeline(transaction=True) as pipe:
    pipe.set("text", "not a number")
    pipe.incr("text")  # fails at runtime
    pipe.set("after", "still runs")
    try:
        pipe.execute()
    except redis.ResponseError as exc:
        Line("runtime error", exc)

Line("text ", r.get("text"))
Line("after", r.get("after"))

Title("Read the replies by position")

with r.pipeline(transaction=False) as pipe:
    pipe.set("user:1:name", "Alice")
    pipe.get("user:1:name")
    pipe.strlen("user:1:name")
    set_ok, name, length = pipe.execute()

Line("replies", f"set {set_ok} | get {name} | strlen {length}")

Race conditions and optimistic locking

atomicity.pyimport threading
from typing import Any

import redis
from common import Title, cleanup, connect

N = 20

r = connect()
r2 = connect()  # a second client = a second connection

cleanup(r)

counter = "counter"

Title("A read-modify-write race, forced to lose updates")

r.set(counter, 0)

start = threading.Barrier(N)
all_read = threading.Barrier(N)


def slow_increment():
    start.wait()
    value = int(r.get(counter))  # read
    all_read.wait()  # every thread has read counter
    r.set(counter, value + 1)  # write


threads = [threading.Thread(target=slow_increment) for _ in range(N)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(f"{N} threads, expected {N}, got {r.get(counter)}")

Title("The same race with INCR: one command, no window")

r.set(counter, 0)
start = threading.Barrier(N)


def atomic_increment():
    start.wait()
    r.incr(counter)


threads = [threading.Thread(target=atomic_increment) for _ in range(N)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(f"{N} threads, expected {N}, got {r.get(counter)}")

Title("WATCH/MULTI: abort a transaction if the value moved")

balance = "balance"
r.set(balance, 100)

with r.pipeline() as pipe:
    pipe.watch(balance)
    current = int(pipe.get(balance))
    print("read balance ->", current)

    r2.set(balance, 50)  # a competing writer sneaks in
    print("competitor set balance -> 50")

    pipe.multi()
    pipe.set(balance, current - 30)
    try:
        pipe.execute()
    except redis.WatchError:
        print("EXEC refused: the watched key changed")

print("balance is still ->", r.get(balance))

Title("The same transfer as a retry loop (optimistic locking)")

def debit(pipe: Any, amount: int, attempts: int = 5):
    for attempt in range(1, attempts + 1):
        try:
            pipe.watch(balance)
            current = int(pipe.get(balance))
            if current < amount:  # overdraft protection
                pipe.unwatch()
                print(f"  attempt {attempt}: insufficient funds")
                return False

            pipe.multi()
            pipe.decrby(balance, amount)
            pipe.execute()
            return True

        except redis.WatchError:
            print(f"  attempt {attempt}: lost the race, retrying")

    return False


print("debit 30 ->", debit(r.pipeline(), 30), "| balance ->", r.get(balance))