coredis

coredis

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

Features

• Fully typed, even when using pipelines, Lua scripts, and libraries
• Redis Cluster and Sentinel support
• Built with structured concurrency on anyio, supports both asyncio and trio
• Smart command routing: multiplexing when possible, pooling otherwise
• Server-assisted client-side caching implementation
• Redis Stack modules support
• Redis PubSub
• Pipelining
• Lua scripts and Redis functions with support for typing
• Convenient Stream Consumers implementation
• Comprehensive documentation
• Optional runtime type validation (via beartype)
• Observability with OpenTelemetry

Installation

$ pip install coredis

Optional extras

For OpenTelemetry support

$ pip install "coredis[otel]"

To install any dependencies required by the recipes

$ pip install "coredis[recipes]"

Getting started

Single node or cluster

import anyio
import coredis

async def main() -> None:
    client = coredis.Redis(host='127.0.0.1', port=6379, db=0, decode_responses=True)
    # or cluster
    # client = coredis.RedisCluster(startup_nodes=[coredis.connection.TCPLocation("127.0.0.1", 6379)], 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
import coredis

async def main() -> None:
    sentinel = coredis.Sentinel(sentinels=[("localhost", 26379)])
    async with sentinel:
        primary: coredis.Redis  = sentinel.primary_for("myservice")
        replica: coredis.Redis  = 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

To see a full list of supported Redis commands refer to the Command compatibility documentation. Details about supported Redis modules and their commands can be found here.

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.11

Redis API compatible databases

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

• Dragonfly
• Valkey

References

• Documentation (Stable)
• Documentation (Latest)
• Changelog
• Project History