coredis

coredis

coredis is an async redis client with support for redis server, cluster & sentinel.

• The client API uses 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 variadic positional or keyword arguments. When the variable length arguments are not optional (which is almost always the case) 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 between RESP and python types as closely as possible.

• For higher level concepts such as Pipelines, LUA Scripts, PubSub & Streams abstractions are provided to encapsulate recommended patterns. See the Handbook and the API Documentation for more details.

• Installation
• Feature Summary
  • Deployment topologies
  • Application patterns
  • Server side scripting
  • Redis Modules
  • Miscellaneous
• Quick start
  • Single Node or Cluster client
  • Sentinel
• Compatibility
  • Supported python versions
  • Redis API compatible databases backends
• References

Installation

To install coredis:

$ pip install coredis

Feature Summary

Deployment topologies

• Redis Cluster
• Sentinel

Application patterns

• Connection Pooling
• PubSub
• Sharded PubSub [>= Redis 7.0]
• Stream Consumers
• Pipelining
• Client side caching

Server side scripting

• LUA Scripting
• Redis Libraries and functions [>= Redis 7.0]

Redis Modules

• RedisJSON
• RediSearch
• RedisBloom
• RedisTimeSeries

Miscellaneous

• Public API annotated with type annotations
• Optional Runtime Type Validation (via beartype)

Quick start

Single Node or Cluster client

import asyncio
from coredis import Redis, RedisCluster

async def example():
    client = Redis(host='127.0.0.1', port=6379, db=0)
    # or with redis cluster
    # client = RedisCluster(startup_nodes=[{"host": "127.0.01", "port": 7001}])
    await client.flushdb()
    await client.set('foo', 1)
    assert await client.exists(['foo']) == 1
    assert await client.incr('foo') == 2
    assert await client.incrby('foo', increment=100) == 102
    assert int(await client.get('foo')) == 102

    assert await client.expire('foo', 1)
    await asyncio.sleep(0.1)
    assert await client.ttl('foo') == 1
    assert await client.pttl('foo') < 1000
    await asyncio.sleep(1)
    assert not await client.exists(['foo'])

asyncio.run(example())

Sentinel

import asyncio
from coredis.sentinel import Sentinel

async def example():
    sentinel = Sentinel(sentinels=[("localhost", 26379)])
    primary = sentinel.primary_for("myservice")
    replica = sentinel.replica_for("myservice")

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

asyncio.run(example())

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

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

Supported python versions

• 3.10
• 3.11
• 3.12
• 3.13
• PyPy 3.10

Redis API compatible databases

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

• Dragonfly
• Redict
• Valkey

References

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