# Technical Documentation: redis/redis-py
> ℹ️ **Provenance:** Multi-Tier Ingestion · [CodeWiki Reference](https://codewiki.google/github.com/redis/redis-py)
### Overview (`README.md`)
# redis-py
The Python interface to the Redis key-value store.
[](https://github.com/redis/redis-py/actions/workflows/integration.yaml)
[](https://redis.readthedocs.io/en/stable/)
[](https://github.com/redis/redis-py/blob/master/LICENSE)
[](https://pypi.org/project/redis/)
[](https://github.com/redis/redis-py/releases)
[](https://codecov.io/gh/redis/redis-py)
[Installation](#installation) | [Usage](#usage) | [Advanced Topics](#advanced-topics) | [Contributing](https://github.com/redis/redis-py/blob/master/CONTRIBUTING.md)
---------------------------------------------
**Note:** redis-py 5.0 is the last version of redis-py that supports Python 3.7, as it has reached [end of life](https://devguide.python.org/versions/). redis-py 5.1 supports Python 3.8+.
**Note:** redis-py 6.1.0 is the last version of redis-py that supports Python 3.8, as it has reached [end of life](https://devguide.python.org/versions/). redis-py 6.2.0 supports Python 3.9+.
---------------------------------------------
## How do I Redis?
[Learn for free at Redis University](https://redis.io/learn/university)
[Try the Redis Cloud](https://redis.io/try-free/)
[Dive in developer tutorials](https://redis.io/learn)
[Join the Redis community](https://redis.io/community/)
[Work at Redis](https://redis.io/careers/)
## Installation
Start a redis via docker (for Redis versions >= 8.0):
``` bash
docker run -p 6379:6379 -it redis:latest
```
Start a redis via docker (for Redis versions < 8.0):
``` bash
docker run -p 6379:6379 -it redis/redis-stack:latest
```
To install redis-py, simply:
``` bash
$ pip install redis
```
For faster performance, install redis with hiredis support, this provides a compiled response parser, and *for most cases* requires zero code changes.
By default, if hiredis >= 1.0 is available, redis-py will attempt to use it for response parsing.
``` bash
$ pip install "redis[hiredis]"
```
Looking for a high-level library to handle object mapping? See [redis-om-python](https://github.com/redis/redis-om-python)!
## Supported Redis Versions
The most recent version of this library supports Redis version [7.2](https://github.com/redis/redis/blob/7.2/00-RELEASENOTES), [7.4](https://github.com/redis/redis/blob/7.4/00-RELEASENOTES), [8.0](https://github.com/redis/redis/blob/8.0/00-RELEASENOTES), [8.2](https://github.com/redis/redis/blob/8.2/00-RELEASENOTES), [8.4](https://github.com/redis/redis/blob/8.4/00-RELEASENOTES), [8.6](https://github.com/redis/redis/blob/8.6/00-RELEASENOTES) and [8.8](https://github.com/redis/redis/blob/8.8/00-RELEASENOTES).
The table below highlights version compatibility of the most-recent library versions and redis versions.
| Library version | Supported redis versions |
|-----------------|-------------------|
| 3.5.3 | <= 6.2 Family of releases |
| >= 4.5.0 | Version 5.0 to 7.0 |
| >= 5.0.0 | Version 5.0 to 7.4 |
| >= 6.0.0 | Version 7.2 to current |
## Usage
### Basic Example
``` python
>>> import redis
>>> r = redis.Redis(host='localhost', port=6379, db=0)
>>> r.set('foo', 'bar')
True
>>> r.get('foo')
b'bar'
```
The above code connects to localhost on port 6379, sets a value in Redis, and retrieves it. All responses are returned as bytes in Python, to receive decoded strings, set *decode_responses=True*. For this, and more connection options, see [these examples](https://redis.readthedocs.io/en/stable/examples.html).
#### RESP3 Support
redis-py supports RESP3 starting with version 5.0. Starting with redis-py 8.0,
clients use RESP3 on the wire by default while preserving legacy
RESP2-compatible Python response shapes for existing applications.
Set *protocol=3* explicitly when your application should receive
RESP3-specific response shapes or when you want the wire protocol choice to be
visible in code. To force RESP2 on the wire, set *protocol=2*. To opt in to
protocol-independent response shapes, set *legacy_responses=False*.
For new projects, we recommend opting out of legacy response compatibility by
setting *legacy_responses=False*. This makes redis-py return unified Python
response shapes for affected commands whether the connection uses RESP2 or
RESP3. Existing applications can keep the default legacy-compatible behavior
while they migrate response handling at their own pace.
``` python
>>> import redis
>>> r = redis.Redis(host='localhost', port=6379, db=0, legacy_responses=False)
```
See the [unified responses migration guide](https://redis.readthedocs.io/en/stable/unified_responses.html)
for activation instructions and command-by-command response differences. For
RESP3-specific behavior, see the [RESP3 features guide](https://redis.readthedocs.io/en/stable/resp3_features.html).
### Connection Pools
By default, redis-py uses a connection pool to manage connections. Each instance of a Redis class receives its own connection pool. You can however define your own [redis.ConnectionPool](https://redis.readthedocs.io/en/stable/connections.html#connection-pools).
``` python
>>> pool = redis.ConnectionPool(host='localhost', port=6379, db=0)
>>> r = redis.Redis(connection_pool=pool)
```
Alternatively, you might want to look at [Async connections](https://redis.readthedocs.io/en/stable/examples/asyncio_examples.html), or [Cluster connections](https://redis.readthedocs.io/en/stable/connections.html#cluster-client), or even [Async Cluster connections](https://redis.readthedocs.io/en/stable/connections.html#async-cluster-client).
### Redis Commands
There is built-in support for all of the [out-of-the-box Redis commands](https://redis.io/commands). They are exposed using the raw Redis command names (`HSET`, `HGETALL`, etc.) except where a word (i.e. del) is reserved by the language. The complete set of commands can be found [here](https://github.com/redis/redis-py/tree/master/redis/commands), or [the documentation](https://redis.readthedocs.io/en/stable/commands.html).
## Advanced Topics
The [official Redis command documentation](https://redis.io/commands)
does a great job of explaining each command in detail. redis-py attempts
to adhere to the official command syntax. There are a few exceptions:
- **MULTI/EXEC**: These are implemented as part of the Pipeline class.
The pipeline is wrapped with the MULTI and EXEC statements by
default when it is executed, which can be disabled by specifying
transaction=False. See more about Pipelines below.
- **SUBSCRIBE/LISTEN**: Similar to pipelines, PubSub is implemented as
a separate class as it places the underlying connection in a state
where it can\'t execute non-pubsub commands. Calling the pubsub
method from the Redis client will return a PubSub instance where you
can subscribe to channels and listen for messages. You can only call
PUBLISH from the Redis client (see [this comment on issue
#151](https://github.com/redis/redis-py/issues/151#issuecomment-1545015)
for details).
For more details, please see the documentation on [advanced topics page](https://redis.readthedocs.io/en/stable/advanced_features.html).
### Pipelines
The following is a basic example of a [Redis pipeline](https://redis.io/docs/manual/pipelining/), a method to optimize round-trip calls, by batching Redis commands, and receiving their results as a list.
``` python
>>> pipe = r.pipeline()
>>> pipe.set('foo', 5)
>>> pipe.set('bar', 18.5)
>>> pipe.set('blee', "hello world!")
>>> pipe.execute()
[True, True, True]
```
### PubSub
The following example shows how to utilize [Redis Pub/Sub](https://redis.io/docs/manual/pubsub/) to subscribe to specific channels.
``` python
>>> r = redis.Redis(...)
>>> p = r.pubsub()
>>> p.subscribe('my-first-channel', 'my-second-channel', ...)
>>> p.get_message()
{'pattern': None, 'type': 'subscribe', 'channel': b'my-second-channel', 'data': 1}
```
### Redis’ search and query capabilities default dialect
Release 6.0.0 introduces a client-side default dialect for Redis’ search and query capabilities.
By default, the client now overrides the server-side dialect with version 2, automatically appending *DIALECT 2* to commands like *FT.AGGREGATE* and *FT.SEARCH*.
**Important**: Be aware that the query dialect may impact the results returned. If needed, you can revert to a different dialect version by configuring the client accordingly.
``` python
>>> from redis.commands.search.field import TextField
>>> from redis.commands.search.query import Query
>>> from redis.commands.search.index_definition import IndexDefinition
>>> import redis
>>> r = redis.Redis(host='localhost', port=6379, db=0)
>>> r.ft().create_index(
>>> (TextField("name"), TextField("lastname")),
>>> definition=IndexDefinition(prefix=["test:"]),
>>> )
>>> r.hset("test:1", "name", "James")
>>> r.hset("test:1", "lastname", "Brown")
>>> # Query with default DIALECT 2
>>> query = "@name: James Brown"
>>> q = Query(query)
>>> res = r.ft().search(q)
>>> # Query with explicit DIALECT 1
>>> query = "@name: James Brown"
>>> q = Query(query).dialect(1)
>>> res = r.ft().search(q)
```
You can find further details in the [query dialect documentation](https://redis.io/docs/latest/develop/interact/search-and-query/advanced-concepts/dialects/).
### Multi-database client (Active-Active)
The multi-database client allows your application to connect to multiple Redis databases, which are typically replicas of each other. It is designed to work with Redis Software and Redis Cloud Active-Active setups. The client continuously monitors database health, detects failures, and automatically fails over to the next healthy database using a configurable strategy. When the original database becomes healthy again, the client can automatically switch back to it.
This is useful when:
1. You have more than one Redis deployment. This might include two independent Redis servers or two or more Redis databases replicated across multiple [active-active Redis Enterprise](https://redis.io/docs/latest/operate/rs/databases/active-active/) clusters.
2. You want your application to connect to one deployment at a time and to fail over to the next available deployment if the first deployment becomes unavailable.
For the complete failover configuration options and examples, see the [Multi-database client docs](https://redis.readthedocs.io/en/latest/geographic_failover.html).
### Bulk hash ingestion (HIMPORT)
Redis 8.10 adds the `HIMPORT` command family for loading many hashes that share
the same set of field names: register the field names once with `himport_prepare`,
then create each hash by sending only its values. Keys written this way are regular
hashes — every hash command works on them.
``` python
r.himport_prepare("users", ["name", "email", "age"])
r.himport_set("user:1", "users", ["alice", "alice@example.com", "25"])
r.himport_set("user:2", "users", ["bob", "bob@example.com", "30"])
r.himport_discard("users") # => 1
```
Values pair positionally with the prepared fields. Hash enumeration order
(`HGETALL`, `HKEYS`) is not guaranteed to match the prepare order.
**Fieldsets are connection state.** A prepared fieldset lives in the server-side
session of the physical connection that prepared it: it is invisible to other
connections and destroyed by a disconnect or `RESET`. redis-py handles this for you —
`himport_prepare` records the fieldset in a client-level registry, and the `PREPARE`
is applied lazily on whatever pooled connection serves each `himport_set` (and
re-applied automatically after a reconnect, `RESET`, or Sentinel/cluster failover).
You declare each fieldset once per client with `himport_prepare`; there is no
constructor argument for it.
For the highest ingestion throughput, send the `PREPARE` and its `SET`s in one
pipeline — a single batch always executes on one connection:
``` python
with r.pipeline(transaction=False) as pipe:
pipe.himport_prepare("users", ["name", "email", "age"])
for uid, row in rows:
pipe.himport_set(f"user:{uid}", "users", row)
pipe.execute()
```
The automatic re-prepare applies to direct calls only, not to commands inside
`pipeline`/`transaction` blocks: a batched `himport_set` relies on the single
pre-flight `PREPARE` in that batch.
With `RedisCluster`, `himport_prepare` / `himport_discard` / `himport_discard_all`
update the client's shared, cluster-wide registry and return immediately — like the
standalone API, they perform no server I/O of their own. The server-side `PREPARE`
(and, after a discard, `DISCARD`) is applied lazily on each node's connection the
next time it serves an `himport_set`, and re-applied after reconnects or failover;
`himport_set` itself routes by the key's hash slot. A discard is therefore not
removed from every server session at once: each connection drops the fieldset on its
next `himport_set` (or on disconnect). With Sentinel, call
`himport_prepare` on the long-lived client returned by `master_for(...)`; the fieldset
survives failover automatically. HIMPORT is not supported on the multi-database
(Active-Active) client.
The async client mirrors this exactly:
``` python
await r.himport_prepare("users", ["name", "email", "age"])
await r.himport_set("user:1", "users", ["alice", "alice@example.com", "25"])
```
---------------------------------------------
### Author
redis-py is developed and maintained by [Redis Inc](https://redis.io). It can be found [here](
https://github.com/redis/redis-py), or downloaded from [pypi](https://pypi.org/project/redis/).
Special thanks to:
- Andy McCurdy () the original author of redis-py.
- Ludovico Magnocavallo, author of the original Python Redis client,
from which some of the socket code is still used.
- Alexander Solovyov for ideas on the generic response callback
system.
- Paul Hubbard for initial packaging support.
[](https://redis.io)
---
### Overview (`benchmarks/README.md`)
# Benchmarks
This directory contains local benchmark scripts for redis-py. They are not part
of the normal CI test matrix.
## OpenTelemetry Benchmarks
`otel_benchmark.py` measures redis-py operation throughput and latency with
different OpenTelemetry configurations:
- `baseline`: redis-py from a git tag, without the OpenTelemetry code path.
- `otel_disabled`: current working tree with observability not initialized.
- `otel_noop`: observability initialized with a no-op meter provider.
- `otel_inmemory`: observability initialized with an in-memory metric reader.
- `otel_enabled_http`: metrics exported over OTLP/HTTP.
- `otel_enabled_grpc`: metrics exported over OTLP/gRPC.
Run the commands below from the repository root.
The regular development requirements install the OTLP/HTTP exporter only. Install
the gRPC exporter separately when you want to run the `otel_enabled_grpc`
scenario:
```shell
python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r dev_requirements.txt
python -m pip install -r benchmarks/requirements.txt
```
Start Redis before running any scenario. The OpenTelemetry example stack also
starts an OTLP collector on ports `4317` and `4318`, which is useful for the HTTP
and gRPC exporter scenarios:
```shell
docker compose -f docs/examples/opentelemetry/docker-compose.yml up -d
```
Choose one way to run the benchmark and use the matching examples below.
### Run as a module
If you prefer to run the benchmark with `python -m`, use these commands.
Run one scenario at a time:
```shell
python -m benchmarks.otel_benchmark --scenario otel_disabled
python -m benchmarks.otel_benchmark --scenario otel_noop
python -m benchmarks.otel_benchmark --scenario otel_inmemory
python -m benchmarks.otel_benchmark --scenario otel_enabled_http
python -m benchmarks.otel_benchmark --scenario otel_enabled_grpc
```
For a shorter smoke run while checking setup, lower the duration and iteration
count:
```shell
python -m benchmarks.otel_benchmark --scenario otel_enabled_grpc --duration 10 --warmup 2 --iterations 1
```
Run the baseline scenario with a redis-py release tag:
```shell
python -m benchmarks.otel_benchmark --scenario baseline --baseline-tag v5.2.1
```
Use `--async` to run the async client variant:
```shell
python -m benchmarks.otel_benchmark --scenario otel_enabled_http --async
```
By default the benchmark sends exporter traffic to `localhost`. Set
`OTEL_COLLECTOR_HOST` when the collector is on another host:
```shell
OTEL_COLLECTOR_HOST=collector.example.com python -m benchmarks.otel_benchmark --scenario otel_enabled_grpc
```
### Run the file directly
If you prefer to run the Python file directly, use these commands instead.
Run one scenario at a time:
```shell
python benchmarks/otel_benchmark.py --scenario otel_disabled
python benchmarks/otel_benchmark.py --scenario otel_noop
python benchmarks/otel_benchmark.py --scenario otel_inmemory
python benchmarks/otel_benchmark.py --scenario otel_enabled_http
python benchmarks/otel_benchmark.py --scenario otel_enabled_grpc
```
For a shorter smoke run while checking setup, lower the duration and iteration
count:
```shell
python benchmarks/otel_benchmark.py --scenario otel_enabled_grpc --duration 10 --warmup 2 --iterations 1
```
Run the baseline scenario with a redis-py release tag:
```shell
python benchmarks/otel_benchmark.py --scenario baseline --baseline-tag v5.2.1
```
Use `--async` to run the async client variant:
```shell
python benchmarks/otel_benchmark.py --scenario otel_enabled_http --async
```
By default the benchmark sends exporter traffic to `localhost`. Set
`OTEL_COLLECTOR_HOST` when the collector is on another host:
```shell
OTEL_COLLECTOR_HOST=collector.example.com python benchmarks/otel_benchmark.py --scenario otel_enabled_grpc
```
Stop the example stack when finished:
```shell
docker compose -f docs/examples/opentelemetry/docker-compose.yml down
```
---
### Overview (`doctests/README.md`)
# Command examples for redis.io
## How to add an example
Create regular python file in the current folder with meaningful name. It makes sense prefix example files with
command category (e.g. string, set, list, hash, etc) to make navigation in the folder easier. Files ending in *.py*
are automatically run by the test suite.
### Special markup
See https://github.com/redis-stack/redis-stack-website#readme for more details.
## How to test examples
Examples are standalone python scripts, committed to the *doctests* directory. These scripts assume that the
```doctests/requirements.txt``` and ```dev_requirements.txt``` from this repository have been installed, as per below.
```bash
pip install -r dev_requirements.txt
pip uninstall -y redis # uninstall Redis package installed via redis-entraid
pip install -r doctests/requirements.txt
```
Note - the CI process, runs linters against the examples. Assuming
the requirements above have been installed you can run ```ruff check yourfile.py``` and ```ruff format yourfile.py```
locally to validate the linting, prior to CI.
Just include necessary assertions in the example file and run
```bash
sh doctests/run_examples.sh
```
to test all examples in the current folder.
---
### Overview (`docs/examples/README.md`)
# Examples
Examples of redis-py usage go here. They're being linked to the [generated documentation](https://redis.readthedocs.org).
---
### Overview (`docs/examples/opentelemetry/README.md`)
# Example for redis-py OpenTelemetry instrumentation
This example demonstrates how to monitor Redis using [OpenTelemetry](https://opentelemetry.io/) and
[Uptrace](https://github.com/uptrace/uptrace). It requires Docker to start Redis Server and Uptrace.
See
[Monitoring redis-py performance with OpenTelemetry](https://redis.readthedocs.io/en/latest/opentelemetry.html)
for details.
**Step 1**. Download the example using Git:
```shell
git clone https://github.com/redis/redis-py.git
cd example/opentelemetry
```
**Step 2**. Optionally, create a virtualenv:
```shell
python3 -m venv .venv
source .venv/bin/active
```
**Step 3**. Install dependencies:
```shell
pip install -e .
```
**Step 4**. Start the services using Docker and make sure Uptrace is running:
```shell
docker-compose up -d
docker-compose logs uptrace
```
**Step 5**. Run the Redis client example and follow the link from the CLI to view the trace:
```shell
python3 main.py
trace: http://localhost:14318/traces/ee029d8782242c8ed38b16d961093b35
```
You can also open Uptrace UI at [http://localhost:14318](http://localhost:14318) to view available
spans, logs, and metrics.
---
### Advanced Features (`docs/advanced_features.rst`)
Advanced Features
=================
A note about threading
----------------------
Redis client instances can safely be shared between threads. Internally,
connection instances are only retrieved from the connection pool during
command execution, and returned to the pool directly after. Command
execution never modifies state on the client instance.
However, there is one caveat: the Redis SELECT command. The SELECT
command allows you to switch the database currently in use by the
connection. That database remains selected until another is selected or
until the connection is closed. This creates an issue in that
connections could be returned to the pool that are connected to a
different database.
As a result, redis-py does not implement the SELECT command on client
instances. If you use multiple Redis databases within the same
application, you should create a separate client instance (and possibly
a separate connection pool) for each database.
It is not safe to pass PubSub or Pipeline objects between threads.
Pipelines
---------
Default pipelines
~~~~~~~~~~~~~~~~~
Pipelines are a subclass of the base Redis class that provide support
for buffering multiple commands to the server in a single request. They
can be used to dramatically increase the performance of groups of
commands by reducing the number of back-and-forth TCP packets between
the client and server.
Pipelines are quite simple to use:
.. code:: python
>>> r = redis.Redis(...)
>>> r.set('bing', 'baz')
>>> # Use the pipeline() method to create a pipeline instance
>>> pipe = r.pipeline()
>>> # The following SET commands are buffered
>>> pipe.set('foo', 'bar')
>>> pipe.get('bing')
>>> # the EXECUTE call sends all buffered commands to the server, returning
>>> # a list of responses, one for each command.
>>> pipe.execute()
[True, b'baz']
For ease of use, all commands being buffered into the pipeline return
the pipeline object itself. Therefore calls can be chained like:
.. code:: python
>>> pipe.set('foo', 'bar').sadd('faz', 'baz').incr('auto_number').execute()
[True, True, 6]
In addition, pipelines can also ensure the buffered commands are
executed atomically as a group. This happens by default. If you want to
disable the atomic nature of a pipeline but still want to buffer
commands, you can turn off transactions.
.. code:: python
>>> pipe = r.pipeline(transaction=False)
A common issue occurs when requiring atomic transactions but needing to
retrieve values in Redis prior for use within the transaction. For
instance, let's assume that the INCR command didn't exist and we need to
build an atomic version of INCR in Python.
The completely naive implementation could GET the value, increment it in
Python, and SET the new value back. However, this is not atomic because
multiple clients could be doing this at the same time, each getting the
same value from GET.
Enter the WATCH command. WATCH provides the ability to monitor one or
more keys prior to starting a transaction. If any of those keys change
prior the execution of that transaction, the entire transaction will be
canceled and a WatchError will be raised. To implement our own
client-side INCR command, we could do something like this:
.. code:: python
>>> with r.pipeline() as pipe:
... while True:
... try:
... # put a WATCH on the key that holds our sequence value
... pipe.watch('OUR-SEQUENCE-KEY')
... # after WATCHing, the pipeline is put into immediate execution
... # mode until we tell it to start buffering commands again.
... # this allows us to get the current value of our sequence
... current_value = pipe.get('OUR-SEQUENCE-KEY')
... next_value = int(current_value) + 1
... # now we can put the pipeline back into buffered mode with MULTI
... pipe.multi()
... pipe.set('OUR-SEQUENCE-KEY', next_value)
... # and finally, execute the pipeline (the set command)
... pipe.execute()
... # if a WatchError wasn't raised during execution, everything
... # we just did happened atomically.
... break
... except WatchError:
... # another client must have changed 'OUR-SEQUENCE-KEY' between
... # the time we started WATCHing it and the pipeline's execution.
... # our best bet is to just retry.
... continue
Note that, because the Pipeline must bind to a single connection for the
duration of a WATCH, care must be taken to ensure that the connection is
returned to the connection pool by calling the reset() method. If the
Pipeline is used as a context manager (as in the example above) reset()
will be called automatically. Of course you can do this the manual way
by explicitly calling reset():
.. code:: python
>>> pipe = r.pipeline()
>>> while True:
... try:
... pipe.watch('OUR-SEQUENCE-KEY')
... ...
... pipe.execute()
... break
... except WatchError:
... continue
... finally:
... pipe.reset()
A convenience method named "transaction" exists for handling all the
boilerplate of handling and retrying watch errors. It takes a callable
that should expect a single parameter, a pipeline object, and any number
of keys to be WATCHed. Our client-side INCR command above can be written
like this, which is much easier to read:
.. code:: python
>>> def client_side_incr(pipe):
... current_value = pipe.get('OUR-SEQUENCE-KEY')
... next_value = int(current_value) + 1
... pipe.multi()
... pipe.set('OUR-SEQUENCE-KEY', next_value)
>>>
>>> r.transaction(client_side_incr, 'OUR-SEQUENCE-KEY')
[True]
Be sure to call pipe.multi() in the callable passed to Redis.transaction
prior to any write commands.
Pipelines in clusters
~~~~~~~~~~~~~~~~~~~~~
ClusterPipeline is a subclass of RedisCluster that provides support for
Redis pipelines in cluster mode. When calling the execute() command, all
the commands are grouped by the node on which they will be executed, and
are then executed by the respective nodes in parallel. The pipeline
instance will wait for all the nodes to respond before returning the
result to the caller. Command responses are returned as a list sorted in
the same order in which they were sent. Pipelines can be used to
dramatically increase the throughput of Redis Cluster by significantly
reducing the number of network round trips between the client and
the server.
.. code:: python
>>> rc = RedisCluster()
>>> with rc.pipeline() as pipe:
... pipe.set('foo', 'value1')
... pipe.set('bar', 'value2')
... pipe.get('foo')
... pipe.get('bar')
... print(pipe.execute())
[True, True, b'value1', b'value2']
... pipe.set('foo1', 'bar1').get('foo1').execute()
[True, b'bar1']
Please note:
- RedisCluster pipelines currently only support key-based commands.
- The pipeline gets its ‘load_balancing_strategy’ value from the
cluster’s parameter. Thus, if read from replications is enabled in
the cluster instance, the pipeline will also direct read commands to
replicas.
Transactions in clusters
~~~~~~~~~~~~~~~~~~~~~~~~
Transactions are supported in cluster-mode with one caveat: all keys of
all commands issued on a transaction pipeline must reside on the
same slot. This is similar to the limitation of multikey commands in
cluster. The reason behind this is that the Redis engine does not offer
a mechanism to block or exchange key data across nodes on the fly. A
client may add some logic to abstract engine limitations when running
on a cluster, such as the pipeline behavior explained on the previous
block, but there is no simple way that a client can enforce atomicity
across nodes on a distributed system.
The compromise of limiting the transaction pipeline to same-slot keys
is exactly that: a compromise. While this behavior is different from
non-transactional cluster pipelines, it simplifies migration of clients
from standalone to cluster under some circumstances. Note that application
code that issues multi/exec commands on a standalone client without
embedding them within a pipeline would eventually get ‘AttributeError’.
With this approach, if the application uses ‘client.pipeline(transaction=True)’,
then switching the client with a cluster-aware instance would simplify
code changes (to some extent). This may be true for application code that
makes use of hash keys, since its transactions may already be
mapping all commands to the same slot.
An alternative is some kind of two-step commit solution, where a slot
validation is run before the actual commands are run. This could work
with controlled node maintenance but does not cover single node failures.
Given the cluster limitations for transactions, by default pipeline isn't in
transactional mode. To enable transactional context set:
.. code:: python
>>> p = rc.pipeline(transaction=True)
After entering the transactional context you can add commands to a transactional
context, by one of the following ways:
.. code:: python
>>> p = rc.pipeline(transaction=True) # Chaining commands
>>> p.set("key", "value")
>>> p.get("key")
>>> response = p.execute()
Or
.. code:: python
>>> with rc.pipeline(transaction=True) as pipe: # Using context manager
... pipe.set("key", "value")
... pipe.get("key")
... response = pipe.execute()
As you see there's no need to explicitly send `MULTI/EXEC` commands to control context start/end
`ClusterPipeline` will take care of it.
To ensure that different keys will be mapped to a same hash slot on the server side
prepend your keys with the same hash tag, the technique that allows you to control
keys distribution.
More information `here `_
.. code:: python
>>> with rc.pipeline(transaction=True) as pipe:
... pipe.set("{tag}foo", "bar")
... pipe.set("{tag}bar", "foo")
... pipe.get("{tag}foo")
... pipe.get("{tag}bar")
... response = pipe.execute()
CAS Transactions
~~~~~~~~~~~~~~~~~~~~~~~~
If you want to apply optimistic locking for certain keys, you have to execute
`WATCH` command in transactional context. `WATCH` command follows the same limitations
as any other multi key command - all keys should be mapped to the same hash slot.
However, the difference between CAS transaction and normal one is that you have to
explicitly call MULTI command to indicate the start of transactional context, WATCH
command itself and any subsequent commands before MULTI will be immediately executed
on the server side so you can apply optimistic locking and get necessary data before
transaction execution.
.. code:: python
>>> with rc.pipeline(transaction=True) as pipe:
... pipe.watch("mykey") # Apply locking by immediately executing command
... val = pipe.get("mykey") # Immediately retrieves value
... val = val + 1 # Increment value
... pipe.multi() # Starting transaction context
... pipe.set("mykey", val) # Command will be pipelined
... response = pipe.execute() # Returns OK or None if key was modified in the meantime
Publish / Subscribe
-------------------
redis-py includes a PubSub object that subscribes to channels and
listens for new messages. Creating a PubSub object is easy.
.. code:: python
>>> r = redis.Redis(...)
>>> p = r.pubsub()
Once a PubSub instance is created, channels and patterns can be
subscribed to.
.. code:: python
>>> p.subscribe('my-first-channel', 'my-second-channel', ...)
>>> p.psubscribe('my-*', ...)
The PubSub instance is now subscribed to those channels/patterns. The
subscription confirmations can be seen by reading messages from the
PubSub instance.
.. code:: python
>>> p.get_message()
{'pattern': None, 'type': 'subscribe', 'channel': b'my-second-channel', 'data': 1}
>>> p.get_message()
{'pattern': None, 'type': 'subscribe', 'channel': b'my-first-channel', 'data': 2}
>>> p.get_message()
{'pattern': None, 'type': 'psubscribe', 'channel': b'my-*', 'data': 3}
Every message read from a PubSub instance will be a dictionary with the
following keys.
- **type**: One of the following: 'subscribe', 'unsubscribe',
'psubscribe', 'punsubscribe', 'message', 'pmessage'
- **channel**: The channel [un]subscribed to or the channel a message
was published to
- **pattern**: The pattern that matched a published message's channel.
Will be None in all cases except for 'pmessage' types.
- **data**: The message data. With [un]subscribe messages, this value
will be the number of channels and patterns the connection is
currently subscribed to. With [p]message messages, this value will be
the actual published message.
Let's send a message now.
.. code:: python
# the publish method returns the number matching channel and pattern
# subscriptions. 'my-first-channel' matches both the 'my-first-channel'
# subscription and the 'my-*' pattern subscription, so this message will
# be delivered to 2 channels/patterns
>>> r.publish('my-first-channel', 'some data')
2
>>> p.get_message()
{'channel': b'my-first-channel', 'data': b'some data', 'pattern': None, 'type': 'message'}
>>> p.get_message()
{'channel': b'my-first-channel', 'data': b'some data', 'pattern': b'my-*', 'type': 'pmessage'}
Unsubscribing works just like subscribing. If no arguments are passed to
[p]unsubscribe, all channels or patterns will be unsubscribed from.
.. code:: python
>>> p.unsubscribe()
>>> p.punsubscribe('my-*')
>>> p.get_message()
{'channel': b'my-second-channel', 'data': 2, 'pattern': None, 'type': 'unsubscribe'}
>>> p.get_message()
{'channel': b'my-first-channel', 'data': 1, 'pattern': None, 'type': 'unsubscribe'}
>>> p.get_message()
{'channel': b'my-*', 'data': 0, 'pattern': None, 'type': 'punsubscribe'}
redis-py also allows you to register callback functions to handle
published messages. Message handlers take a single argument, the
message, which is a dictionary just like the examples above. To
subscribe to a channel or pattern with a message handler, pass the
channel or pattern name as a keyword argument with its value being the
callback function. For channel or pattern names that cannot be used as
Python keyword arguments, including arbitrary binary names, pass a
``redis.Subscription`` object as a positional argument instead.
Subscription forms can be mixed in the same call. Assuming
``my_handler`` is a callback, a plain channel name can be supplied
alongside a ``Subscription`` object, and ``Subscription`` objects can
also be used together with the existing keyword-argument handler style.
.. code:: python
>>> p.subscribe('logs', redis.Subscription(b'\x80\x81', my_handler))
>>> p.subscribe(redis.Subscription(b'\x82\x83', my_handler), **{'my-channel': my_handler})
When a message is read on a channel or pattern with a message handler,
the message dictionary is created and passed to the message handler. In
this case, a None value is returned from get_message() since the message
was already handled.
.. code:: python
>>> def my_handler(message):
... print('MY HANDLER: ', message['data'])
>>> p.subscribe(**{'my-channel': my_handler})
>>> p.subscribe(redis.Subscription(b'\x80\x81', my_handler))
# read the subscribe confirmation message
>>> p.get_message()
{'pattern': None, 'type': 'subscribe', 'channel': b'my-channel', 'data': 1}
>>> r.publish('my-channel', 'awesome data')
1
# for the message handler to work, we need tell the instance to read data.
# this can be done in several ways (read more below). we'll just use
# the familiar get_message() function for now
>>> message = p.get_message()
MY HANDLER: awesome data
# note here that the my_handler callback printed the string above.
# `message` is None because the message was handled by our handler.
>>> print(message)
None
If your application is not interested in the (sometimes noisy)
subscribe/unsubscribe confirmation messages, you can ignore them by
passing ignore_subscribe_messages=True to r.pubsub(). This will cause
all subscribe/unsubscribe messages to be read, but they won't bubble up
to your application.
.. code:: python
>>> p = r.pubsub(ignore_subscribe_messages=True)
>>> p.subscribe('my-channel')
>>> p.get_message() # hides the subscribe message and returns None
>>> r.publish('my-channel', 'my data')
1
>>> p.get_message()
{'channel': b'my-channel', 'data': b'my data', 'pattern': None, 'type': 'message'}
There are three different strategies for reading messages.
The examples above have been using pubsub.get_message(). Behind the
scenes, get_message() uses the system's 'select' module to quickly poll
the connection's socket. If there's data available to be read,
get_message() will read it, format the message and return it or pass it
to a message handler. If there's no data to be read, get_message() will
immediately return None. This makes it trivial to integrate into an
existing event loop inside your application.
.. code:: python
>>> while True:
>>> message = p.get_message()
>>> if message:
>>> # do something with the message
>>> time.sleep(0.001) # be nice to the system :)
Older versions of redis-py only read messages with pubsub.listen().
listen() is a generator that blocks until a message is available. If
your application doesn't need to do anything else but receive and act on
messages received from redis, listen() is an easy way to get up an
running.
.. code:: python
>>> for message in p.listen():
... # do something with the message
The third option runs an event loop in a separate thread.
pubsub.run_in_thread() creates a new thread and starts the event loop.
The thread object is returned to the caller of run_in_thread(). The
caller can use the thread.stop() method to shut down the event loop and
thread. Behind the scenes, this is simply a wrapper around get_message()
that runs in a separate thread, essentially creating a tiny non-blocking
event loop for you. run_in_thread() takes an optional sleep_time
argument. If specified, the event loop will call time.sleep() with the
value in each iteration of the loop.
Note: Since we're running in a separate thread, there's no way to handle
messages that aren't automatically handled with registered message
handlers. Therefore, redis-py prevents you from calling run_in_thread()
if you're subscribed to patterns or channels that don't have message
handlers attached.
.. code:: python
>>> p.subscribe(**{'my-channel': my_handler})
>>> thread = p.run_in_thread(sleep_time=0.001)
# the event loop is now running in the background processing messages
# when it's time to shut it down...
>>> thread.stop()
run_in_thread also supports an optional exception handler, which lets
you catch exceptions that occur within the worker thread and handle them
appropriately. The exception handler will take as arguments the
exception itself, the pubsub object, and the worker thread returned by
run_in_thread.
.. code:: python
>>> p.subscribe(**{'my-channel': my_handler})
>>> def exception_handler(ex, pubsub, thread):
>>> print(ex)
>>> thread.stop()
>>> thread = p.run_in_thread(exception_handler=exception_handler)
A PubSub object adheres to the same encoding semantics as the client
instance it was created from. Any channel or pattern that's unicode will
be encoded using the encoding specified on the client before being sent
to Redis. If the client's decode_responses flag is set the False (the
default), the 'channel', 'pattern' and 'data' values in message
dictionaries will be byte strings (str on Python 2, bytes on Python 3).
If the client's decode_responses is True, then the 'channel', 'pattern'
and 'data' values will be automatically decoded to unicode strings using
the client's encoding.
PubSub objects remember what channels and patterns they are subscribed
to. In the event of a disconnection such as a network error or timeout,
the PubSub object will re-subscribe to all prior channels and patterns
when reconnecting. Messages that were published while the client was
disconnected cannot be delivered. When you're finished with a PubSub
object, call its .close() method to shutdown the connection.
.. code:: python
>>> p = r.pubsub()
>>> ...
>>> p.close()
The PUBSUB set of subcommands CHANNELS, NUMSUB and NUMPAT are also
supported:
.. code:: python
>>> r.pubsub_channels()
[b'foo', b'bar']
>>> r.pubsub_numsub('foo', 'bar')
[(b'foo', 9001), (b'bar', 42)]
>>> r.pubsub_numsub('baz')
[(b'baz', 0)]
>>> r.pubsub_numpat()
1204
Sharded pubsub
~~~~~~~~~~~~~~
`Sharded pubsub `_ is a feature introduced with Redis 7.0, and fully supported by redis-py as of 5.0. It helps scale the usage of pub/sub in cluster mode, by having the cluster shard messages to nodes that own a slot for a shard channel. Here, the cluster ensures the published shard messages are forwarded to the appropriate nodes. Clients subscribe to a channel by connecting to either the master responsible for the slot, or any of its replicas.
This makes use of the `SSUBSCRIBE `_ and `SPUBLISH `_ commands within Redis.
The following, is a simplified example:
.. code:: python
>>> from redis.cluster import RedisCluster, ClusterNode
>>> r = RedisCluster(startup_nodes=[ClusterNode('localhost', 6379), ClusterNode('localhost', 6380)])
>>> p = r.pubsub()
>>> p.ssubscribe('foo')
>>> # assume someone sends a message along the channel via a publish
>>> message = p.get_sharded_message()
Similarly, the same process can be used to acquire sharded pubsub messages, that have already been sent to a specific node, by passing the node to get_sharded_message:
.. code:: python
>>> from redis.cluster import RedisCluster, ClusterNode
>>> first_node = ClusterNode['localhost', 6379]
>>> second_node = ClusterNode['localhost', 6380]
>>> r = RedisCluster(startup_nodes=[first_node, second_node])
>>> p = r.pubsub()
>>> p.ssubscribe('foo')
>>> # assume someone sends a message along the channel via a publish
>>> message = p.get_sharded_message(target_node=second_node)
The client keeps shard subscriptions attached to the node that currently owns
their slot, so a slot migration or a lost node needs no handling in application
code. When a poll on one node fails, or a node replies ``MOVED`` because it no
longer owns the slot, the client re-routes the affected shard channels to their
new owner and keeps polling the remaining nodes.
**Important**: this affects what ``get_sharded_message()`` raises.
- A connection failure on a single node is no longer surfaced while other nodes
can still deliver. ``get_sharded_message()`` returns the messages those nodes
hold, and raises only when every node polled in the pass failed.
- A ``MOVED`` reply is handled internally and no longer raised. The channels the
former owner no longer serves are re-subscribed on the new owner.
- Code that caught an exception from ``get_sharded_message()`` only to tear the
subscription down and re-subscribe is no longer needed.
- ``get_sharded_message(target_node=...)`` is unchanged: a caller that names a
single node still gets its connection errors raised.
Monitor
~~~~~~~
redis-py includes a Monitor object that streams every command processed
by the Redis server. Use listen() on the Monitor object to block until a
command is received.
.. code:: python
>>> r = redis.Redis(...)
>>> with r.monitor() as m:
>>> for command in m.listen():
>>> print(command)
---
### Backoff (`docs/backoff.rst`)
.. _backoff-label:
Backoff
#############
.. automodule:: redis.backoff
:members:
---
### Clustering (`docs/clustering.rst`)
Clustering
==========
redis-py now supports cluster mode and provides a client for `Redis
Cluster `__.
The cluster client is based on Grokzen’s
`redis-py-cluster `__, has
added bug fixes, and now supersedes that library. Support for these
changes is thanks to his contributions.
To learn more about Redis Cluster, see `Redis Cluster
specifications `__.
`Creating clusters <#creating-clusters>`__ \| `Specifying Target
Nodes <#specifying-target-nodes>`__ \| `Multi-key
Commands <#multi-key-commands>`__ \| `Known PubSub
Limitations <#known-pubsub-limitations>`__
Connecting to cluster
---------------------
Connecting redis-py to a Redis Cluster instance(s) requires at a minimum
a single node for cluster discovery. There are multiple ways in which a
cluster instance can be created:
- Using ‘host’ and ‘port’ arguments:
.. code:: python
>>> from redis.cluster import RedisCluster as Redis
>>> rc = Redis(host='localhost', port=6379)
>>> print(rc.get_nodes())
[[host=127.0.0.1,port=6379,name=127.0.0.1:6379,server_type=primary,redis_connection=Redis>>], [host=127.0.0.1,port=6378,name=127.0.0.1:6378,server_type=primary,redis_connection=Redis>>], [host=127.0.0.1,port=6377,name=127.0.0.1:6377,server_type=replica,redis_connection=Redis>>]]
- Using the Redis URL specification:
.. code:: python
>>> from redis.cluster import RedisCluster as Redis
>>> rc = Redis.from_url("redis://localhost:6379/0")
- Directly, via the ClusterNode class:
.. code:: python
>>> from redis.cluster import RedisCluster as Redis
>>> from redis.cluster import ClusterNode
>>> nodes = [ClusterNode('localhost', 6379), ClusterNode('localhost', 6378)]
>>> rc = Redis(startup_nodes=nodes)
When a RedisCluster instance is being created it first attempts to
establish a connection to one of the provided startup nodes. If none of
the startup nodes are reachable, a ‘RedisClusterException’ will be
thrown. After a connection to the one of the cluster’s nodes is
established, the RedisCluster instance will be initialized with 3
caches: a slots cache which maps each of the 16384 slots to the node/s
handling them, a nodes cache that contains ClusterNode objects (name,
host, port, redis connection) for all of the cluster’s nodes, and a
commands cache contains all the server supported commands that were
retrieved using the Redis ‘COMMAND’ output. See *RedisCluster specific
options* below for more.
RedisCluster instance can be directly used to execute Redis commands.
When a command is being executed through the cluster instance, the
target node(s) will be internally determined. When using a key-based
command, the target node will be the node that holds the key’s slot.
Cluster management commands and other commands that are not key-based
have a parameter called ‘target_nodes’ where you can specify which nodes
to execute the command on. In the absence of target_nodes, the command
will be executed on the default cluster node. As part of cluster
instance initialization, the cluster’s default node is randomly selected
from the cluster’s primaries, and will be updated upon reinitialization.
Using r.get_default_node(), you can get the cluster’s default node, or
you can change it using the ‘set_default_node’ method.
The ‘target_nodes’ parameter is explained in the following section,
‘Specifying Target Nodes’.
.. code:: python
>>> # target-nodes: the node that holds 'foo1's key slot
>>> rc.set('foo1', 'bar1')
>>> # target-nodes: the node that holds 'foo2's key slot
>>> rc.set('foo2', 'bar2')
>>> # target-nodes: the node that holds 'foo1's key slot
>>> print(rc.get('foo1'))
b'bar'
>>> # target-node: default-node
>>> print(rc.keys())
[b'foo1']
>>> # target-node: default-node
>>> rc.ping()
Specifying Target Nodes
-----------------------
As mentioned above, all non key-based RedisCluster commands accept the
kwarg parameter ‘target_nodes’ that specifies the node/nodes that the
command should be executed on. The best practice is to specify target
nodes using RedisCluster class’s node flags: PRIMARIES, REPLICAS,
ALL_NODES, RANDOM. When a nodes flag is passed along with a command, it
will be internally resolved to the relevant node/s. If the nodes
topology of the cluster changes during the execution of a command, the
client will be able to resolve the nodes flag again with the new
topology and attempt to retry executing the command.
.. code:: python
>>> from redis.cluster import RedisCluster as Redis
>>> # run cluster-meet command on all of the cluster's nodes
>>> rc.cluster_meet('127.0.0.1', 6379, target_nodes=Redis.ALL_NODES)
>>> # ping all replicas
>>> rc.ping(target_nodes=Redis.REPLICAS)
>>> # ping a random node
>>> rc.ping(target_nodes=Redis.RANDOM)
>>> # get the keys from all cluster nodes
>>> rc.keys(target_nodes=Redis.ALL_NODES)
[b'foo1', b'foo2']
>>> # execute bgsave in all primaries
>>> rc.bgsave(Redis.PRIMARIES)
You could also pass ClusterNodes directly if you want to execute a
command on a specific node / node group that isn’t addressed by the
nodes flag. However, if the command execution fails due to cluster
topology changes, a retry attempt will not be made, since the passed
target node/s may no longer be valid, and the relevant cluster or
connection error will be returned.
.. code:: python
>>> node = rc.get_node('localhost', 6379)
>>> # Get the keys only for that specific node
>>> rc.keys(target_nodes=node)
>>> # get Redis info from a subset of primaries
>>> subset_primaries = [node for node in rc.get_primaries() if node.port > 6378]
>>> rc.info(target_nodes=subset_primaries)
In addition, the RedisCluster instance can query the Redis instance of a
specific node and execute commands on that node directly. The Redis
client, however, does not handle cluster failures and retries.
.. code:: python
>>> cluster_node = rc.get_node(host='localhost', port=6379)
>>> print(cluster_node)
[host=127.0.0.1,port=6379,name=127.0.0.1:6379,server_type=primary,redis_connection=Redis>>]
>>> r = cluster_node.redis_connection
>>> r.client_list()
[{'id': '276', 'addr': '127.0.0.1:64108', 'fd': '16', 'name': '', 'age': '0', 'idle': '0', 'flags': 'N', 'db': '0', 'sub': '0', 'psub': '0', 'multi': '-1', 'qbuf': '26', 'qbuf-free': '32742', 'argv-mem': '10', 'obl': '0', 'oll': '0', 'omem': '0', 'tot-mem': '54298', 'events': 'r', 'cmd': 'client', 'user': 'default'}]
>>> # Get the keys only for that specific node
>>> r.keys()
[b'foo1']
Multi-key Commands
------------------
Redis supports multi-key commands in Cluster Mode, such as Set type
unions or intersections, mset and mget, as long as the keys all hash to
the same slot. By using RedisCluster client, you can use the known
functions (e.g. mget, mset) to perform an atomic multi-key operation.
However, you must ensure all keys are mapped to the same slot, otherwise
a RedisClusterException will be thrown. Redis Cluster implements a
concept called hash tags that can be used in order to force certain keys
to be stored in the same hash slot, see `Keys hash
tag `__. You can
also use nonatomic for some of the multikey operations, and pass keys
that aren’t mapped to the same slot. The client will then map the keys
to the relevant slots, sending the commands to the slots’ node owners.
Non-atomic operations batch the keys according to their hash value, and
then each batch is sent separately to the slot’s owner.
.. code:: python
# Atomic operations can be used when all keys are mapped to the same slot
>>> rc.mset({'{foo}1': 'bar1', '{foo}2': 'bar2'})
>>> rc.mget('{foo}1', '{foo}2')
[b'bar1', b'bar2']
# Non-atomic multi-key operations splits the keys into different slots
>>> rc.mset_nonatomic({'foo': 'value1', 'bar': 'value2', 'zzz': 'value3')
>>> rc.mget_nonatomic('foo', 'bar', 'zzz')
[b'value1', b'value2', b'value3']
**Cluster PubSub:**
When a ClusterPubSub instance is created without specifying a node, a
single node will be transparently chosen for the pubsub connection on
the first command execution. The node will be determined by: 1. Hashing
the channel name in the request to find its keyslot 2. Selecting a node
that handles the keyslot: If read_from_replicas is set to true or
load_balancing_strategy is provided, a replica can be selected.
For sharded pubsub, the client instead keeps each shard channel on the node
that currently owns its slot and re-routes it when the topology changes; see
the sharded pubsub section of :doc:`advanced_features` for that behavior and
what it means for the errors ``get_sharded_message()`` raises.
Known PubSub Limitations
------------------------
Pattern subscribe and publish do not currently work properly due to key
slots. If we hash a pattern like fo\* we will receive a keyslot for that
string but there are endless possibilities for channel names based on
this pattern - unknowable in advance. This feature is not disabled but
the commands are not currently recommended for use. See
`redis-py-cluster
documentation `__
for more.
.. code:: python
>>> p1 = rc.pubsub()
# p1 connection will be set to the node that holds 'foo' keyslot
>>> p1.subscribe('foo')
# p2 connection will be set to node 'localhost:6379'
>>> p2 = rc.pubsub(rc.get_node('localhost', 6379))
**Read Only Mode**
By default, Redis Cluster always returns MOVE redirection response on
accessing a replica node. You can overcome this limitation and scale
read commands by triggering READONLY mode.
To enable READONLY mode pass read_from_replicas=True or define
a load_balancing_strategy to RedisCluster constructor.
When read_from_replicas is set to true read commands will be assigned between
the primary and its replications in a Round-Robin manner.
With load_balancing_strategy you can define a custom strategy for
assigning read commands to the replicas and primary nodes.
READONLY mode can be set at runtime by calling the readonly() method
with target_nodes=‘replicas’, and read-write access can be restored by
calling the readwrite() method.
.. code:: python
>>> from cluster import RedisCluster as Redis
# Use 'debug' log level to print the node that the command is executed on
>>> rc_readonly = Redis(startup_nodes=startup_nodes,
... read_from_replicas=True)
>>> rc_readonly.set('{foo}1', 'bar1')
>>> for i in range(0, 4):
... # Assigns read command to the slot's hosts in a Round-Robin manner
... rc_readonly.get('{foo}1')
# set command would be directed only to the slot's primary node
>>> rc_readonly.set('{foo}2', 'bar2')
# reset READONLY flag
>>> rc_readonly.readwrite(target_nodes='replicas')
# now the get command would be directed only to the slot's primary node
>>> rc_readonly.get('{foo}1')
---
### Commands (`docs/commands.rst`)
Redis Commands
##############
Core Commands
*************
The following functions can be used to replicate their equivalent `Redis command `_. Generally they can be used as functions on your redis connection. For the simplest example, see below:
Getting and settings data in redis::
import redis
r = redis.Redis(decode_responses=True)
r.set('mykey', 'thevalueofmykey')
r.get('mykey')
.. autoclass:: redis.commands.core.CoreCommands
:inherited-members:
Sentinel Commands
*****************
.. autoclass:: redis.commands.sentinel.SentinelCommands
:inherited-members:
Redis Cluster Commands
**********************
The following `Redis commands `_ are available within a `Redis Cluster `_. Generally they can be used as functions on your redis connection.
.. autoclass:: redis.commands.cluster.RedisClusterCommands
:inherited-members:
---
### Connections (`docs/connections.rst`)
Connecting to Redis
###################
Generic Client
**************
This is the client used to connect directly to a standard Redis node.
.. autoclass:: redis.Redis
:members:
Sentinel Client
***************
Redis `Sentinel `_ provides high availability for Redis. There are commands that can only be executed against a Redis node running in sentinel mode. Connecting to those nodes, and executing commands against them requires a Sentinel connection.
Connection example (assumes Redis exists on the ports listed below):
>>> from redis import Sentinel
>>> sentinel = Sentinel([('localhost', 26379)], socket_timeout=0.1)
>>> sentinel.discover_master('mymaster')
('127.0.0.1', 6379)
>>> sentinel.discover_slaves('mymaster')
[('127.0.0.1', 6380)]
Sentinel
========
.. autoclass:: redis.sentinel.Sentinel
:members:
SentinelConnectionPool
======================
.. autoclass:: redis.sentinel.SentinelConnectionPool
:members:
Cluster Client
**************
This client is used for connecting to a Redis Cluster.
RedisCluster
============
.. autoclass:: redis.cluster.RedisCluster
:members:
ClusterNode
===========
.. autoclass:: redis.cluster.ClusterNode
:members:
Async Client
************
See complete example: `here `__
This client is used for communicating with Redis, asynchronously.
.. autoclass:: redis.asyncio.client.Redis
:members:
Async Cluster Client
********************
RedisCluster (Async)
====================
.. autoclass:: redis.asyncio.cluster.RedisCluster
:members:
:member-order: bysource
ClusterNode (Async)
===================
.. autoclass:: redis.asyncio.cluster.ClusterNode
:members:
:member-order: bysource
ClusterPipeline (Async)
=======================
.. autoclass:: redis.asyncio.cluster.ClusterPipeline
:members: execute_command, execute
:member-order: bysource
Connection
**********
See complete example: `here `__
Connection
==========
.. autoclass:: redis.connection.Connection
:members:
Connection (Async)
==================
.. autoclass:: redis.asyncio.connection.Connection
:members:
Connection Pools
****************
See complete example: `here `__
ConnectionPool
==============
.. autoclass:: redis.connection.ConnectionPool
:members:
ConnectionPool (Async)
======================
.. autoclass:: redis.asyncio.connection.ConnectionPool
:members:
---
### Examples (`docs/examples.rst`)
Examples
########
.. toctree::
:maxdepth: 3
:glob:
examples/connection_examples
examples/ssl_connection_examples
examples/asyncio_examples
examples/search_json_examples
examples/set_and_get_examples
examples/search_vector_similarity_examples
examples/pipeline_examples
examples/timeseries_examples
examples/redis-stream-example
examples/opentelemetry_api_examples
examples/hash_examples
examples/pubsub_examples
---
### Exceptions (`docs/exceptions.rst`)
.. _exceptions-label:
Exceptions
##########
.. automodule:: redis.exceptions
:members:
---
### Geographic Failover (`docs/geographic_failover.rst`)
Client-side geographic failover (Active-Active)
=====================================
The multi-database client allows your application to connect to multiple Redis databases, which are typically replicas of each other.
It is designed to work with Redis Software and Redis Cloud Active-Active setups.
The client continuously monitors database health, detects failures, and automatically fails over to the next healthy database using a configurable strategy.
When the original database becomes healthy again, the client can automatically switch back to it.
Key concepts
------------
- Database and weight:
Each database has a weight indicating its priority. The failover strategy chooses the highest-weight
healthy database as the active one.
- Circuit breaker:
Each database is guarded by a circuit breaker with states CLOSED (healthy), OPEN (unhealthy),
and HALF_OPEN (probing). Health checks toggle these states to avoid hammering a downed database.
- Health checks:
A set of checks determines whether a database is healthy in proactive manner.
By default, an "PING" check runs against the database (all cluster nodes must
pass for a cluster). You can provide your own set of health checks or add an
additional health check on top of the default one. A Redis Enterprise specific
"lag-aware" health check is also available.
- Failure detector:
A detector observes command failures over a moving window (reactive monitoring).
You can specify an exact number of failures and failures rate to have more
fine-grain tuned configuration of triggering fail over based on organic traffic.
You can provide your own set of custom failure detectors or add an additional
detector on top of the default one.
- Failover strategy:
The default strategy is based on statically configured weights. It prefers the highest weighted healthy database.
- Command retry:
Command execution supports retry with backoff. Low-level client retries are disabled and a global retry
setting is applied at the multi-database layer.
- Auto fallback:
If configured with a positive interval, the client periodically attempts to fall back to a higher-weighted
healthy database.
- Events:
The client emits events like "active database changed" and "commands failed". In addition it resubscribes to Pub/Sub channels automatically.
Synchronous usage
-----------------
Minimal example
^^^^^^^^^^^^^^^
.. code-block:: python
from redis.multidb.client import MultiDBClient
from redis.multidb.config import MultiDbConfig, DatabaseConfig
# Two databases. The first has higher weight -> preferred when healthy.
cfg = MultiDbConfig(
databases_config=[
DatabaseConfig(from_url="redis://db-primary:6379/0", weight=1.0),
DatabaseConfig(from_url="redis://db-secondary:6379/0", weight=0.5),
]
)
client = MultiDBClient(cfg)
# First call triggers initialization and health checks.
client.set("key", "value")
print(client.get("key"))
# Pipeline
with client.pipeline() as pipe:
pipe.set("a", 1)
pipe.incrby("a", 2)
values = pipe.execute()
print(values)
# Transaction
def txn(pipe):
current = pipe.get("balance")
current = int(current or 0)
pipe.multi() # mark transaction
pipe.set("balance", current + 100)
client.transaction(txn)
# Pub/Sub usage - will automatically re-subscribe on database switch
pubsub = client.pubsub()
pubsub.subscribe("events")
# In your loop:
message = pubsub.get_message(timeout=1.0)
if message:
print(message)
Asyncio usage
-------------
The asyncio API mirrors the synchronous one and provides async/await semantics.
.. code-block:: python
import asyncio
from redis.asyncio.multidb.client import MultiDBClient
from redis.asyncio.multidb.config import MultiDbConfig, DatabaseConfig
async def main():
cfg = MultiDbConfig(
databases_config=[
DatabaseConfig(from_url="redis://db-primary:6379/0", weight=1.0),
DatabaseConfig(from_url="redis://db-secondary:6379/0", weight=0.5),
]
)
# Context-manager approach for graceful client termination when exits.
# client = MultiDBClient(cfg) could be used instead
async with MultiDBClient(cfg) as client:
await client.set("key", "value")
print(await client.get("key"))
# Pipeline
async with client.pipeline() as pipe:
pipe.set("a", 1)
pipe.incrby("a", 2)
values = await pipe.execute()
print(values)
# Transaction
async def txn(pipe):
current = await pipe.get("balance")
current = int(current or 0)
await pipe.multi()
await pipe.set("balance", current + 100)
await client.transaction(txn)
# Pub/Sub
pubsub = client.pubsub()
await pubsub.subscribe("events")
message = await pubsub.get_message(timeout=1.0)
if message:
print(message)
asyncio.run(main())
MultiDBClient
^^^^^^^^^^^^^
The client exposes the same API as the `Redis` or `RedisCluster` client, making it fully interchangeable and ensuring a simple migration of your application code. Additionally, it supports runtime reconfiguration, allowing you to add features such as health checks, failure detectors, or even new databases without restarting.
Configuration
-------------
MultiDbConfig
^^^^^^^^^^^^^
.. code-block:: python
from redis.multidb.config import (
MultiDbConfig, DatabaseConfig,
DEFAULT_HEALTH_CHECK_INTERVAL, DEFAULT_GRACE_PERIOD
)
from redis.retry import Retry
from redis.backoff import ExponentialWithJitterBackoff
cfg = MultiDbConfig(
databases_config=[
# Construct via URL
DatabaseConfig(
from_url="redis://db-a:6379/0",
weight=1.0,
# Optional: use a custom circuit breaker grace period
grace_period=DEFAULT_GRACE_PERIOD,
# Optional: Redis Enterprise cluster FQDN for REST health checks
health_check_url="https://cluster.example.com",
# Optional: Underlying Redis client related configuration
client_kwargs={"socket_timeout": 5}
),
# Or construct via ConnectionPool
# DatabaseConfig(from_pool=my_pool, weight=1.0),
],
# Global command retry policy (applied at multi-db layer)
command_retry=Retry(
retries=10,
backoff=ExponentialWithJitterBackoff(base=0.01, cap=1),
),
# Health checks
health_check_interval: float = DEFAULT_HEALTH_CHECK_INTERVAL # seconds
health_check_probes: int = DEFAULT_HEALTH_CHECK_PROBES
health_check_delay: float = DEFAULT_HEALTH_CHECK_DELAY # seconds
health_check_timeout: float = DEFAULT_HEALTH_CHECK_TIMEOUT # seconds
health_check_policy: HealthCheckPolicies = DEFAULT_HEALTH_CHECK_POLICY,
# Failure detector
min_num_failures: int = DEFAULT_MIN_NUM_FAILURES
failure_rate_threshold: float = DEFAULT_FAILURE_RATE_THRESHOLD
failures_detection_window: float = DEFAULT_FAILURES_DETECTION_WINDOW # seconds
# Failover behavior
failover_attempts: int = DEFAULT_FAILOVER_ATTEMPTS
failover_delay: float = DEFAULT_FAILOVER_DELAY # seconds
)
Notes:
- Low-level client retries are disabled automatically per database. The multi-database layer handles retries.
- For clusters, health checks validate all nodes.
DatabaseConfig
^^^^^^^^^^^^^^
Each database needs a `DatabaseConfig` that specifies how to connect.
Method 1: Using client_kwargs (most flexible)
~~~~~~~~~~~~~~~~~~~~~
There's an underlying instance of `Redis` or `RedisCluster` client for each database,
so you can pass all the arguments related to them via `client_kwargs` argument:
.. code:: python
database_config = DatabaseConfig(
weight=1.0,
client_kwargs={
'host': 'localhost',
'port': 6379,
'username': "username",
'password': "password",
}
)
Method 2: Using Redis URL
~~~~~~~~~~~~~~~~~~~~~~~~~
.. code:: python
database_config1 = DatabaseConfig(
weight=1.0,
from_url="redis://host1:port1",
client_kwargs={
'username': "username",
'password': "password",
}
)
Method 3: Using Custom Connection Pool
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
.. code:: python
database_config2 = DatabaseConfig(
weight=0.9,
from_pool=connection_pool,
)
**Important**: Don't pass `Retry` objects in `client_kwargs`. `MultiDBClient`
handles all retries at the top level through the `command_retry` configuration.
Health Monitoring
-----------------
The `MultiDBClient` uses two complementary mechanisms to ensure database availability:
- Health Checks (Proactive Monitoring)
- Failure Detection (Reactive Monitoring)
Health Checks (Proactive Monitoring)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
These checks run continuously in the background at configured intervals to proactively
detect database issues. They run in the background with a given interval and
configuration defined in the `MultiDBConfig` class.
To avoid false positives, you can configure amount of health check probes and also
define one of the health check policies to evaluate probes result.
**HealthCheckPolicies.HEALTHY_ALL** - (default) All probes should be successful.
**HealthCheckPolicies.HEALTHY_MAJORITY** - Majority of probes should be successful.
**HealthCheckPolicies.HEALTHY_ANY** - Any of probes should be successful.
The whole health check - all of its probes - runs under `health_check_timeout`
(3 seconds by default), and a single check running out of it marks the database
unhealthy regardless of the failure detector's threshold. See
`Planned maintenance (maintenance notifications)`_ for how this budget behaves
while a database undergoes a planned maintenance.
PingHealthCheck (default)
^^^^^^^^^^^^^^^^^^^^^^^^^
The default health check sends the [PING](https://redis.io/docs/latest/commands/ping/) command
to the database (and to all nodes for clusters).
Lag-Aware Healthcheck (Redis Enterprise Only)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
This is a special type of healthcheck available for Redis (Enterprise) Software
that utilizes a REST API endpoint to obtain information about the synchronization
lag between a given database and all other databases in an Active-Active setup.
To use this healthcheck, first you need to adjust your `DatabaseConfig`
to expose `health_check_url` used by your deployment. By default, your
Cluster FQDN should be used as URL, unless you have some kind of
reverse proxy behind an actual REST API endpoint.
.. code-block:: python
from redis.multidb.client import MultiDBClient
from redis.multidb.config import MultiDbConfig, DatabaseConfig
from redis.multidb.healthcheck import PingHealthCheck, LagAwareHealthCheck
from redis.retry import Retry
from redis.backoff import ExponentialWithJitterBackoff
cfg = MultiDbConfig(
databases_config=[
DatabaseConfig(
from_url="redis://db-primary:6379/0",
weight=1.0,
health_check_url="https://cluster.example.com", # optional for LagAware
),
DatabaseConfig(
from_url="redis://db-secondary:6379/0",
weight=0.5,
health_check_url="https://cluster.example.com",
),
],
# Add custom health check to replace the default
health_checks=[
# Redis Enterprise REST-based lag-aware check
LagAwareHealthCheck(
# Customize REST port, lag tolerance, TLS, etc.
rest_api_port=9443,
lag_aware_tolerance=100, # ms
verify_tls=True,
# auth_basic=("user", "pass"),
# ca_file="/path/ca.pem",
# client_cert_file="/path/cert.pem",
# client_key_file="/path/key.pem",
),
],
)
client = MultiDBClient(cfg)
**Custom Health Checks**
~~~~~~~~~~~~~~~~~~~~~
You can add custom health checks for specific requirements. Please notice that all health checks are executed within
asyncio event loop, so please ensure that `check_health` method is async:
.. code-block:: python
from redis.asyncio.multidb.healthcheck import AbstractHealthCheck, AsyncRedisClientT
class EchoHealthCheck(AbstractHealthCheck):
"""
Health check based on ECHO command.
"""
async def check_health(self, database, hc_client: AsyncRedisClientT) -> bool:
await connection.send_command("ECHO", "healthcheck")
response = await connection.read_response()
return response in (b"healthcheck", "healthcheck")
Failure Detection (Reactive Monitoring)
-----------------
The failure detector monitors command failures and marks a database as unhealthy when its failure rate exceeds a defined threshold within a sliding time window.
Under real traffic conditions, this reactive detection mechanism likely triggers earlier than proactive health checks.
You can extend the set of failure detectors by implementing your own and configuring it through the `MultiDBConfig` class.
By default the failure detector is configured for 1000 failures and a 10% failure rate
threshold within a 2 seconds sliding window. This could be adjusted regarding
your application specifics and traffic pattern.
.. code-block:: python
from redis.multidb.config import MultiDbConfig, DatabaseConfig
from redis.multidb.client import MultiDBClient
cfg = MultiDbConfig(
databases_config=[
DatabaseConfig(from_url="redis://db-a:6379/0", weight=1.0),
DatabaseConfig(from_url="redis://db-b:6379/0", weight=0.5),
],
# Default detector also created from config values
)
client = MultiDBClient(cfg)
# Add an additional detector, optionally limited to specific exception types:
client.add_failure_detector(
CustomFailureDetector()
)
Failover and automatic fallback
--------------------------
Weight-based failover chooses the highest-weighted database with a CLOSED circuit. If no database is
healthy it returns `TemporaryUnavailableException`. This exception indicates that the application should
retry sending requests for a configurable period of time (the configuration (`failover_attempts` * `failover_delay`)
defaults to 120 seconds). If none of the databases became available, then a `NoValidDatabaseException` is thrown.
To enable periodic fallback to a higher-priority healthy database, set `auto_fallback_interval` (seconds):
.. code-block:: python
from redis.multidb.config import MultiDbConfig, DatabaseConfig
cfg = MultiDbConfig(
databases_config=[
DatabaseConfig(from_url="redis://db-primary:6379/0", weight=1.0),
DatabaseConfig(from_url="redis://db-secondary:6379/0", weight=0.5),
],
# Try to fallback to higher-weight healthy database every 30 seconds
auto_fallback_interval=30.0,
)
client = MultiDBClient(cfg)
Planned maintenance (maintenance notifications)
-----------------------------------------------
A planned maintenance on one of the databases - a shard migration, a shard failover or
an endpoint rebind during an upgrade or a rebalancing - is not an outage, and should not
turn into a geo failover: the database stays available, and the operation is resolved
inside its own cluster within seconds. Redis Enterprise announces such operations to the
connected clients through push notifications (`MIGRATING` / `MIGRATED`,
`FAILING_OVER` / `FAILED_OVER`, `MOVING`), and redis-py reacts to them on the
connections of the cluster they concern. What follows describes what the
`MultiDBClient` does with them, how to enable them, and what to expect without them.
Enabling the notifications
^^^^^^^^^^^^^^^^^^^^^^^^^^
`MultiDbConfig` disables maintenance notifications in the underlying clients unless a
`MaintNotificationsConfig` is passed through `client_kwargs`; a database configured
with `from_pool` keeps whatever its pool was built with. The notifications require
RESP3, and the relaxed timeout they apply only means something when the connections
have a finite socket timeout to relax:
.. code-block:: python
from redis.maint_notifications import MaintNotificationsConfig
from redis.multidb.config import MultiDbConfig, DatabaseConfig
maint_config = MaintNotificationsConfig(
enabled=True,
relaxed_timeout=30, # seconds; the socket timeout during a maintenance
)
cfg = MultiDbConfig(
databases_config=[
DatabaseConfig(
from_url="redis://db-primary:6379/0",
weight=1.0,
client_kwargs={
"protocol": 3,
"socket_timeout": 5,
"maint_notifications_config": maint_config,
},
),
DatabaseConfig(
from_url="redis://db-secondary:6379/0",
weight=0.5,
client_kwargs={
"protocol": 3,
"socket_timeout": 5,
"maint_notifications_config": maint_config,
},
),
],
)
`enabled=True` makes the client require the server-side support (the cluster setting
`client_maint_notifications`) and fail the connection otherwise; `enabled="auto"`
attempts to enable the notifications and continues without them if the server rejects
the command.
What the client does during a maintenance
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
With the notifications enabled, a planned maintenance is handled by the underlying
client of the affected database, and the `MultiDBClient` keeps that database active:
- **Connections.** A `MIGRATING` or `FAILING_OVER` notification relaxes the socket
timeout of the connection it arrives on to `relaxed_timeout`, so a command that the
shard switch stalls waits instead of timing out; the matching completion
notification restores the timeout. A `MOVING` notification announces the endpoint
moving to another node: the pool relaxes all of its connections and reconnects them
to the new node, transparently to the commands in flight.
- **Health checks.** The health check client of each database is built from that
database's `client_kwargs`, so it receives the same notifications, and the health
check budget follows the connection timeout they relax: while the database is under
a maintenance, `health_check_timeout` becomes
`health_check_probes * (relaxed_timeout + health_check_delay)` - the time every probe
needs when each may take as long as the relaxed socket timeout - and it returns to
its configured value once the maintenance completes. A stall with no notification
behind it, an unplanned failure, keeps the configured budget and fails the database
over as before. `relaxed_timeout=-1` disables the relaxation of both the connections
and the budget; `relaxed_timeout=None` (blocking sockets) relaxes the budget up to
the notification's time-to-live. The notifications reach the health checks through
the probe client's Redis connections, which only a check that talks to the
database keeps reading - `PingHealthCheck` does, `LagAwareHealthCheck` uses the
REST API alone. Combined with a `PingHealthCheck` (the default), the budget of a
`LagAwareHealthCheck` is relaxed the same way, with each REST request still bounded
by its own `http_timeout`; on its own it keeps the configured budget.
- **Idle connections.** A connection that is idle through the maintenance has the
notifications, and possibly the server's close after an endpoint rebind, waiting in
its buffer. When the pool hands it out again it reads them first. If the connection
is still alive the maintenance is still in progress and the notifications are
applied as usual. If the server has closed it, the maintenance is over: the stale
notifications are discarded and the connection is reconnected, through the
configured address, before the command is sent - so resuming traffic after an idle
period does not count as a failure towards a failover.
The failure detector and the circuit breakers are not involved in any of this: a
maintenance that the notifications cover produces neither the command failures nor
the failed health checks that would open a circuit.
Without the notifications
^^^^^^^^^^^^^^^^^^^^^^^^^
Without the notifications the client learns of a maintenance only through its effects:
connections closed by an endpoint rebind, and commands or health check probes stalled
by a shard switch for longer than their timeouts. Whether that turns into a geo
failover depends on the operation and on the configuration - the size of the shard,
`socket_timeout`, `health_check_timeout`, `min_num_failures` and the failure rate
threshold. A failover to a healthy database is the intended fallback when it does, and
the client falls back to the higher-weighted database once it is healthy again (see
`Failover and automatic fallback`_), but it costs a reconnect of every command in
flight and a switch of the database the application talks to, for an operation the
cluster could have resolved on its own. Enabling the notifications is the way to avoid
that.
Observing a maintenance
^^^^^^^^^^^^^^^^^^^^^^^
The phases the underlying clients go through are observable: the handlers dispatch a
`MaintenanceStartedEvent` when a notification starts relaxing timeouts and a
`MaintenanceCompletedEvent` when the relaxation is reverted, to the `event_dispatcher`
of the `MaintNotificationsConfig` passed in `client_kwargs`. Both events carry the
`state` (`MaintenanceState.MOVING` for a pool-level handoff,
`MaintenanceState.MAINTENANCE` for a connection-level migration or failover), the
affected `connection_pool` and `connection`, the `notification` and the `config` acted
with; `notification` is `None` when the relaxation ended because a connection still
under maintenance was closed. The events are dispatched synchronously from the
connection's read path, so listeners must not block.
.. code-block:: python
from redis.event import (
EventDispatcher,
EventListenerInterface,
MaintenanceCompletedEvent,
MaintenanceStartedEvent,
)
from redis.maint_notifications import MaintNotificationsConfig
class LogMaintenanceListener(EventListenerInterface):
def listen(self, event):
print(f"{type(event).__name__}: {event.state} {event.notification}")
dispatcher = EventDispatcher()
listener = LogMaintenanceListener()
dispatcher.register_listeners(
{MaintenanceStartedEvent: [listener], MaintenanceCompletedEvent: [listener]}
)
maint_config = MaintNotificationsConfig(enabled=True, event_dispatcher=dispatcher)
A geo failover, should one happen anyway, is reported through the
`ActiveDatabaseChanged` event described in the next section.
Custom failover callbacks
-------------------------
You may want to activate custom actions when failover happens. For example, you may want to collect some metrics,
logs or externally persist a connection state.
You can register your own event listener for the `ActiveDatabaseChanged` event (which is emitted when a failover happens) using
the `EventDispatcher`.
.. code-block:: python
class LogFailoverEventListener(EventListenerInterface):
def __init__(self, logger: Logger):
self.logger = logger
def listen(self, event: ActiveDatabaseChanged):
self.logger.warning(
f"Failover happened. Active database switched from {event.old_database} to {event.new_database}"
)
event_dispatcher = EventDispatcher()
listener = LogFailoverEventListener(logging.getLogger(__name__))
# Register custom listener
event_dispatcher.register_listeners(
{
ActiveDatabaseChanged: [listener],
}
)
config = MultiDbConfig(
client_class=client_class,
databases_config=db_configs,
command_retry=command_retry,
min_num_failures=min_num_failures,
health_check_probes=3,
health_check_interval=health_check_interval,
event_dispatcher=event_dispatcher,
health_check_probes_delay=health_check_delay,
)
client = MultiDBClient(config)
Managing databases at runtime
-----------------------------
You can manually add/remove databases, update weights, and promote a database if it’s healthy.
.. code-block:: python
from redis.multidb.client import MultiDBClient
from redis.multidb.config import MultiDbConfig, DatabaseConfig
from redis.multidb.database import Database
from redis.multidb.circuit import PBCircuitBreakerAdapter
import pybreaker
from redis import Redis
cfg = MultiDbConfig(
databases_config=[DatabaseConfig(from_url="redis://db-a:6379/0", weight=1.0)]
)
client = MultiDBClient(cfg)
# Add a database programmatically
other = Database(
client=Redis.from_url("redis://db-b:6379/0"),
circuit=PBCircuitBreakerAdapter(pybreaker.CircuitBreaker(reset_timeout=5.0)),
weight=0.5,
health_check_url=None,
)
client.add_database(other)
# Update weight; if it becomes the highest and healthy, it may become active
client.update_database_weight(other, 0.9)
# Promote a specific healthy database to active
client.set_active_database(other)
# Remove a database
client.remove_database(other)
Pub/Sub and re-subscription
--------------------------
The MultiDBClient offers Pub/Sub functionality with automatic re-subscription
to channels during failover events. For optimal failover handling,
both publishers and subscribers should use MultiDBClient instances.
1. **Subscriber failover**: Automatically reconnects to an alternative database
and re-subscribes to the same channels
2. **Publisher failover**: Seamlessly switches to an alternative database and
continues publishing to the same channels
**Note**: Message loss may occur if failover events happen in reverse order
(publisher fails before subscriber).
.. code-block:: python
pubsub = client.pubsub()
pubsub.subscribe("news", "alerts")
# If failover happens here, subscriptions are re-established on the new active DB.
msg = pubsub.get_message(timeout=1.0)
if msg:
print(msg)
Pipelines and transactions
--------------------------
Pipelines and transactions are executed against the active database at execution time. The client ensures
the active database is healthy and up-to-date before running the stack.
.. code-block:: python
with client.pipeline() as pipe:
pipe.set("x", 1)
pipe.incr("x")
results = pipe.execute()
def txn(pipe):
pipe.multi()
pipe.set("y", "42")
client.transaction(txn)
Best practices
--------------
- Assign the highest weight to your primary database and lower weights to replicas or disaster recovery sites.
- Keep `health_check_interval` short enough to promptly detect failures but avoid excessive load.
- Tune `command_retry` and failover attempts to your SLA and workload profile.
- Use `auto_fallback_interval` if you want the client to fail over back to your primary automatically.
- Handle `TemporaryUnavailableException` to be able to recover before giving up. In the meantime, you
can switch the data source (e.g. cache). `NoValidDatabaseException` indicates that there are no healthy
databases to operate.
Troubleshooting
---------------
- NoValidDatabaseException:
Indicates no healthy database is available. Check circuit breaker states and health checks.
- TemporaryUnavailableException
Indicates that currently there are no healthy databases, but you can still send requests until
`NoValidDatabaseException` is thrown. Probe interval is configured with `failure_attemtps`
- Health checks always failing:
Verify connectivity and, for clusters, that all nodes are reachable. For `LagAwareHealthCheck`,
ensure `health_check_url` points to your Redis Enterprise endpoint and authentication/TLS options
are configured properly.
- Pub/Sub not receiving messages after failover:
Ensure you are using the client’s Pub/Sub helper. The client re-subscribes automatically on switch.
---
### Index (`docs/index.rst`)
.. redis-py documentation master file, created by
sphinx-quickstart on Thu Jul 28 13:55:57 2011.
You can adapt this file completely to your liking, but it should at least
contain the root `toctree` directive.
redis-py - Python Client for Redis
====================================
Getting Started
****************
`redis-py `_ requires a running Redis server, and Python 3.7+. See the `Redis
quickstart `_ for Redis installation instructions.
redis-py can be installed using pip via ``pip install redis``.
Quickly connecting to redis
***************************
There are two quick ways to connect to Redis.
**Assuming you run Redis on localhost:6379 (the default)**
.. code-block:: python
import redis
r = redis.Redis()
r.ping()
**Running redis on foo.bar.com, port 12345**
.. code-block:: python
import redis
r = redis.Redis(host='foo.bar.com', port=12345)
r.ping()
**Another example with foo.bar.com, port 12345**
.. code-block:: python
import redis
r = redis.from_url('redis://foo.bar.com:12345')
r.ping()
After that, you probably want to `run redis commands `_.
.. toctree::
:hidden:
genindex
Redis Command Functions
***********************
.. toctree::
:maxdepth: 2
commands
redismodules
Module Documentation
********************
.. toctree::
:maxdepth: 1
connections
clustering
geographic_failover
exceptions
backoff
lock
retry
lua_scripting
opentelemetry
resp3_features
unified_responses
advanced_features
examples
Contributing
*************
- `How to contribute `_
- `Issue Tracker `_
- `Source Code `_
- `Release History `_
License
*******
This project is licensed under the `MIT license `_.
---
### Lock (`docs/lock.rst`)
Lock
#########
.. automodule:: redis.lock
:members:
---
### Lua Scripting (`docs/lua_scripting.rst`)
Lua Scripting
===
`Lua Scripting <#lua-scripting-in-default-connections>`__ \|
`Pipelines <#pipelines>`__ \| `Cluster mode <#cluster-mode>`__
--------------
Lua Scripting in default connections
------------------------------------
redis-py supports the EVAL, EVALSHA, and SCRIPT commands. However, there
are a number of edge cases that make these commands tedious to use in
real world scenarios. Therefore, redis-py exposes a Script object that
makes scripting much easier to use. (RedisClusters have limited support
for scripting.)
To create a Script instance, use the register_script function on a
client instance passing the Lua code as the first argument.
register_script returns a Script instance that you can use throughout
your code.
The following trivial Lua script accepts two parameters: the name of a
key and a multiplier value. The script fetches the value stored in the
key, multiplies it with the multiplier value and returns the result.
.. code:: python
>>> r = redis.Redis()
>>> lua = """
... local value = redis.call('GET', KEYS[1])
... value = tonumber(value)
... return value * ARGV[1]"""
>>> multiply = r.register_script(lua)
multiply is now a Script instance that is invoked by calling it like a
function. Script instances accept the following optional arguments:
- **keys**: A list of key names that the script will access. This
becomes the KEYS list in Lua.
- **args**: A list of argument values. This becomes the ARGV list in
Lua.
- **client**: A redis-py Client or Pipeline instance that will invoke
the script. If client isn't specified, the client that initially
created the Script instance (the one that register_script was invoked
from) will be used.
Continuing the example from above:
.. code:: python
>>> r.set('foo', 2)
>>> multiply(keys=['foo'], args=[5])
10
The value of key 'foo' is set to 2. When multiply is invoked, the 'foo'
key is passed to the script along with the multiplier value of 5. Lua
executes the script and returns the result, 10.
Script instances can be executed using a different client instance, even
one that points to a completely different Redis server.
.. code:: python
>>> r2 = redis.Redis('redis2.example.com')
>>> r2.set('foo', 3)
>>> multiply(keys=['foo'], args=[5], client=r2)
15
The Script object ensures that the Lua script is loaded into Redis's
script cache. In the event of a NOSCRIPT error, it will load the script
and retry executing it.
Pipelines
---------
Script objects can also be used in pipelines. The pipeline instance
should be passed as the client argument when calling the script. Care is
taken to ensure that the script is registered in Redis's script cache
just prior to pipeline execution.
.. code:: python
>>> pipe = r.pipeline()
>>> pipe.set('foo', 5)
>>> multiply(keys=['foo'], args=[5], client=pipe)
>>> pipe.execute()
[True, 25]
Cluster Mode
------------
Cluster mode has limited support for lua scripting.
The following commands are supported, with caveats:
- ``EVAL`` and ``EVALSHA``: The command is sent to the relevant node,
depending on the keys (i.e., in ``EVAL "