coredis¶

Attention

To learn about breaking changes and migration steps for version 6 please see Migrating from 5.x to 6.0.

Note

The documentation for 5.x can be found here

Fast, async, fully-typed Redis client with support for cluster and sentinel

The client API uses the specifications in the Redis command documentation to define the API by using the following conventions:

The coredis Clients use the specifications in the Redis command documentation to define the API by using the following conventions:

Arguments retain naming from Redis as much as possible
Only optional variadic arguments are mapped to position or keyword variadic arguments. When the variable length arguments are not optional the expected argument is an iterable of type Parameters or Mapping.
Pure tokens used as flags are mapped to boolean arguments
One of arguments accepting pure tokens are collapsed and accept a PureToken
Responses are mapped as closely from RESP3 to python types as possible (See Response Types).

For higher level concepts such as Pipelines, LUA Scripts, PubSub abstractions are provided to encapsulate recommended patterns.

Feature Summary¶

Clients for different topologies
Application patterns
Server side scripting
- LUA Scripts
- Library Functions
Redis Modules
Miscellaneous
- Strict Type Annotations for public API
- Runtime Type checking

Installation¶

Released versions of coredis published on pypi contain precompiled native extensions for various architectures that provide significant performance improvements especially when dealing with large bulk responses.

$ pip install coredis

Getting started¶

Single Node

import anyio
import coredis

async def main() -> None:
    client = coredis.Redis(host='127.0.0.1', port=6379, db=0, decode_responses=True)
    async with client:
        await client.flushdb()

        await client.set("foo", 1)
        assert await client.exists(["foo"]) == 1
        assert await client.incr("foo") == 2
        assert await client.expire("foo", 1)
        await anyio.sleep(0.1)
        assert await client.ttl("foo") == 1
        await anyio.sleep(1)
        assert not await client.exists(["foo"])

        async with client.pipeline() as pipeline:
            pipeline.incr("foo")
            value = pipeline.get("foo")
            pipeline.delete(["foo"])

        assert await value == "1"

anyio.run(main, backend="asyncio") # or trio

Redis Cluster

import anyio
import coredis

async def main() -> None:
    client = coredis.RedisCluster(
        startup_nodes=[
            coredis.connection.TCPLocation("127.0.0.1", 7000)
        ],
        db=0, decode_responses=True
    )
    async with client:
        await client.flushdb()

        await client.set("foo", 1)
        assert await client.exists(["foo"]) == 1
        assert await client.incr("foo") == 2
        assert await client.expire("foo", 1)
        await anyio.sleep(0.1)
        assert await client.ttl("foo") == 1
        await anyio.sleep(1)
        assert not await client.exists(["foo"])

        async with client.pipeline() as pipeline:
            pipeline.incr("foo")
            value = pipeline.get("foo")
            pipeline.delete(["foo"])

        assert await value == "1"

anyio.run(main, backend="asyncio") # or trio

Sentinel

import anyio
from coredis.sentinel import Sentinel

async def main() -> None:
    sentinel = Sentinel(sentinels=[("localhost", 26379)])
    async with sentinel:
        primary = sentinel.primary_for("myservice")
        replica = sentinel.replica_for("myservice")

        async with primary, replica:
            assert await primary.set("fubar", 1)
            assert int(await replica.get("fubar")) == 1

anyio.run(main, backend="asyncio") # or trio

Compatibility¶

coredis is tested against redis versions >= 7.0 The test matrix status can be reviewed here

coredis is additionally tested against:

uvloop >= 0.15.0
trio

Supported python versions¶

3.10
3.11
3.12
3.13
3.14
PyPy 3.10
PyPy 3.11

Support for Redis API compatible databases¶

coredis is known to work with the following databases that have redis protocol compatibility:

Compatibility tests for the above are included in the continuous integration test matrix here.