Skip to main content
Markdown

Redis Client Applications

The native Redis client supports typed values, pipelines, transactions, streams, and Pub/Sub. Use it to combine Redis access with computation in a standalone Codon application; the Redis server needs no Codon module.

This guide covers the common command families and the differences between batched commands, transactions, and replayable event consumption.

Before you begin: install the all-SDK distribution and run a development Redis server on localhost. The examples write keys; use a database without application data you need to keep.

Read, Compute, and Write​

An ordinary Redis server needs no Codon module for client access. Save the example below as client.codon. It writes two readings, computes their total in Codon, and stores a summary hash. Use a development database: the example writes to reading:a, reading:b, and codon:summary.

client.codon
from exaloop.databases.redis import Redis
import sys

port = int(sys.argv[1]) if len(sys.argv) > 1 else 6379
with Redis(port=port) as client:
    client.mset({"reading:a": 1.5, "reading:b": 2.5})
    readings = client.mget(["reading:a", "reading:b", "reading:missing"], float)
    total = 0.0
    for value in readings:
        if value is not None:
            total += value
    client.hset("codon:summary", {"total": str(total), "source": "Codon"})
    print(client.hgetall("codon:summary", str))

With a prebuilt SDK and a Redis server listening on port 6379:

Terminal
exaloop --with redis run -release client.codon
Use an individual SDK without the CLI
Terminal
codon run -release -plugin /path/to/sdk/redis client.codon

The resulting hash contains a total of 4.0. Requesting float in mget() converts the stored values for the calculation; the missing key becomes None and is skipped.

Connections and Security​

Redis() connects to the local server on port 6379. For a remote deployment, provide ACL credentials from private configuration and enable verified TLS: Redis(host="redis.example.com", ssl=True, ssl_ca_certs="/etc/ssl/redis-ca.pem").

Each concurrent worker needs a dedicated client. A with block closes it deterministically; connection loss raises ConnectionError without retrying writes whose outcome may be uncertain. Server command failures raise ResponseError instead. The connection reference lists all options.

Types and Materialization​

Reads return bytes unless you ask for a type. For example, client.get("count", int) reads an integer and client.hgetall("record", str) reads a hash as text. Missing values use Optional[T], so they are distinct from an empty string or zero.

Choose the Redis data structure that matches your operation:

DataCommon writesCommon reads and materialized results
Strings and binary blobsset, mset, append, setrange, incr, incrbyfloatget -> Optional[T], mget -> List[Optional[T]], getrange -> bytes, strlen
Hasheshset(key, mapping), hdel, hincrbyhget, hmget, hgetall -> Dict[T,T], hexists, hlen
Listslpush, rpush, lset, ltrimlrange -> List[T], lpop/rpop -> Optional[T], llen, blpop -> Optional[Tuple[T,T]]
Setssadd, srem, spopsmembers/sunion/sinter/sdiff -> Set[T], sismember, scard
Sorted setszadd(key, member_to_score), zincrby, zremzrange -> List[T], zrange_withscores -> List[Tuple[T,float]], zscore -> Optional[float], zcard
Bitmaps and bitfieldssetbit, bitop, bitfieldgetbit, bitcount, bitfield -> List[Optional[int]]
HyperLogLogpfadd, pfmergepfcount -> int (approximate cardinality)
Geospatial indexesgeoadd(key, member_to_lon_lat)geopos -> List[Optional[Tuple[float,float]]], geodist -> Optional[float], geosearch -> List[T]
Streams and consumer groupsxadd, xdel, xtrim, xgroup_create, xack, xclaimxrange, xread, xreadgroup, xpending, xlen

Common Redis options keep their familiar names. For example, set() accepts ex for an expiry in seconds and nx to write only if the key does not exist. You can also use client[key] for string values; direct indexing raises KeyError for a missing key.

Large Collections and Bulk Work​

Use scan_iter(match="prefix:*", count=100) to walk a keyspace without a single large KEYS request. Like Redis SCAN, it is not a snapshot and may return duplicates while keys are changing. Split large reads and writes into batches to avoid holding an entire workload in memory.

A nontransactional pipeline sends queued commands together and then reads all replies, avoiding one network round trip per command:

Source
from exaloop.databases.redis import Redis

with Redis() as client:
    with client.pipeline(transaction=False) as batch:
        batch.set("batch:count", "10").incr("batch:count").get("batch:count")
        replies = batch.execute()
        assert replies[1].int() == 11
        assert replies[2].as_type(int) == 11
        for reply in replies:
            reply.close()

Call execute() to send the queued work, then read the replies in command order. The example sets a counter to 10, increments it, and reads back 11. Leaving the block without execute() discards the queue.

Transactions​

Without transaction=False, a pipeline uses MULTI/EXEC, so Redis executes the queued commands together without interleaving other clients' commands. This does not provide SQL-style rollback: a failed command does not undo earlier writes in the transaction.

For a read/modify/write operation, use WATCH to detect a concurrent update. The example retries the whole operation up to three times:

Source
from exaloop.databases.redis import Redis, WatchError

with Redis() as client:
    for attempt in range(3):
        try:
            with client.pipeline() as transaction:
                transaction.watch("balance")
                with transaction.read("GET", "balance") as reply:
                    balance = 0 if reply.is_null else reply.as_type(int)
                transaction.set("balance", balance + 1)
                results = transaction.execute()
                for result in results:
                    result.close()
            break
        except WatchError:
            if attempt == 2:
                raise

Keep the client's connection dedicated to this operation until the transaction finishes. If a watched key changes, WatchError lets the application decide whether to retry.

Streams and Pub/Sub​

Use streams when consumers need stored events and acknowledgements. An entry has an ID and a collection of fields; xread() groups the returned entries by stream:

Source
from exaloop.databases.redis import Redis

with Redis() as client:
    entry = client.xadd("measurements", {"sensor": "A", "value": "12.5"})
    rows = client.xread({"measurements": "0"}, count=100, T=str)
    assert rows[0][0] == "measurements"
    assert client.xlen("measurements") > 0

For consumer groups, create the group with xgroup_create(), read deliveries with xreadgroup(), and acknowledge completed work with xack(). If you use a blocking read, give the socket a longer timeout than the requested wait.

Pub/Sub is a simpler choice for live notifications that do not need replay. It uses its own connection, so publishing does not interrupt the subscriber:

Source
from exaloop.databases.redis import Redis

with Redis() as client:
    with client.pubsub() as subscription:
        subscription.subscribe("events")
        assert subscription.get_message(1.0).__val__().kind == "subscribe"
        client.publish("events", b"updated")
        message = subscription.get_message(1.0)
        assert message is not None
        assert message.channel == b"events" and message.data == b"updated"

The first message confirms the subscription; the next contains the published event. get_message() returns None on timeout. A disconnected Pub/Sub consumer misses events, so choose streams when recovery matters.

Commands Beyond the Convenience API​

Use execute_command("COMMAND", *args) for a command without a convenience method, including commands provided by an installed Redis module. It returns a reply that you can read through typed accessors; see the reply reference.

The client is synchronous and single-node. Cluster routing, Sentinel discovery, and connection pooling are not supported.

Next Steps​