### Connect to a Redis Cluster

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/index.rst

Initialize a RedisCluster client by providing a list of startup nodes. This example shows the basic setup for connecting to a cluster.

```python
import anyio
import coredis

async def main() -> None:
    client = coredis.RedisCluster(
        startup_nodes=[

```

--------------------------------

### Install Coredis with Recipes

Source: https://github.com/alisaifee/coredis/blob/master/README.md

Install coredis with dependencies required for recipes.

```console
pip install "coredis[recipes]"
```

--------------------------------

### Install Coredis with Recipe Dependencies

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/recipes/index.rst

Use this command to install coredis along with all optional dependencies required by the recipes.

```bash
pip install coredis[recipes]
```

--------------------------------

### Install Coredis

Source: https://github.com/alisaifee/coredis/blob/master/README.md

Install the coredis library using pip.

```console
pip install coredis
```

--------------------------------

### RediSearch Create Index Example

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/modules.rst

Provides an example of how to create a RediSearch index using the `coredis.modules.search.Field` class. This is a foundational step for using RediSearch's search and aggregation capabilities.

```python
from coredis.modules.search import Field

# Example schema definition (actual index creation not shown)
# schema = [
#     Field("title", as_name="title"),
#     Field("body", as_name="body"),
#     Field("tags", as_name="tags",)
# ]
# await client.search.create("my_index", schema)
```

--------------------------------

### Initialize Redis Client with LRUCache

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/caching.rst

Demonstrates how to initialize a Redis client with an LRUCache instance for caching responses. This example also shows how to use it in a cluster mode.

```python
import asyncio
import coredis
from coredis.patterns.cache import LRUCache

cached_client = coredis.Redis(cache=LRUCache())
regular_client = coredis.Redis()

# or in cluster mode
# cached_client = coredis.RedisCluster("localhost", 7000, cache=LRUCache())
# regular_client = coredis.RedisCluster("localhost", 7000)
```

--------------------------------

### Configure and Execute Coredis with OpenTelemetry

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/observability.rst

Example demonstrating how to enable OpenTelemetry instrumentation, configure in-memory exporters for traces and metrics, and execute Redis commands to inspect emitted telemetry.

```python
import anyio
import coredis
from opentelemetry import metrics, trace
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import InMemoryMetricReader
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter

async def main() -> None:
    coredis.Config.otel_enabled = True
    coredis.Config.otel_capture_command_args = True

    span_exporter = InMemorySpanExporter()
    tracer_provider = TracerProvider()
    tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
    trace.set_tracer_provider(tracer_provider)

    metric_reader = InMemoryMetricReader()
    metrics.set_meter_provider(MeterProvider(metric_readers=[metric_reader]))

    client = coredis.Redis(host="127.0.0.1", port=6379)
    async with client:
        await client.set("example:key", "value")
        await client.get("example:key")

    spans = span_exporter.get_finished_spans()
    metric_data = metric_reader.get_metrics_data()

    print(f"spans={len(spans)} metric_resources={len(metric_data.resource_metrics)}")

anyio.run(main)
```

--------------------------------

### Install Coredis with OpenTelemetry

Source: https://github.com/alisaifee/coredis/blob/master/README.md

Install coredis with optional OpenTelemetry support.

```console
pip install "coredis[otel]"
```

--------------------------------

### Coredis Pub/Sub Subscription (5.x)

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/migration_guide.rst

Example of subscribing to a channel using Coredis Pub/Sub in version 5.x. PubSub classes must be used as async context managers.

```python
import asyncio
import coredis

async def main():
    client = coredis.Redis()
    pubsub = client.pubsub()
    await pubsub.subscribe("channel")
    async for message in pubsub:
        print(message)

asyncio.run(main())
```

--------------------------------

### Create a Simple Stream Consumer

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/streams.rst

Instantiate a simple consumer to read from one or more streams. It defaults to starting from the latest entry.

```python
consumer = client.xconsumer(streams=["one", "two", "three"])
# or directly
# import coredis.patterns.streams
# consumer = coredis.patterns.streams.Consumer(client, streams=["one", "two", "three"])
```

--------------------------------

### Consume from Backlog with Coredis Consumer Group

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/streams.rst

Use `start_from_backlog=True` to process old entries before new ones. This example simulates processing with a 50% failure rate and then restarts without the bug.

```python
import random

async with client:
    async with client.xconsumer(
        streams=["one"],
        group = "group-a",
        consumer = "consumer-1",
        start_from_backlog = True
    ) as consumer:
        [await client.xadd("one", {"id": i}) for i in range(10)]


        # fetch all ten entries and simulate a bug occurring 50% of the time
        # when processing the entry
        async for stream, entry in consumer:
            if random.random() > 0.5:
                print("success", await client.xack(stream, consumer.group, [entry.identifier]))
            else:
                print("oh nos!")

        pending = await client.xpending("one", "group-a")
        assert pending.consumers[b"consumer-1"] > 0

    print("round two")
    # Let's pretend the consumer crashed and started again
    # and now doesn't have a bug that fails 50% of the time
    async with client.xconsumer(
        streams=["one"],
        group = "group-a",
        consumer = "consumer-1",
        start_from_backlog = True
    ) as consumer:
        async for stream, entry in consumer:
             await client.xack(stream, consumer.group, [entry.identifier])

        pending = await client.xpending("one", "group-a")
        assert pending.consumers.get(b"consumer-1") is None
```

--------------------------------

### Coredis Scripting with Wraps (5.x)

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/migration_guide.rst

Example of registering and using a Lua script with the 'wraps' decorator in Coredis version 5.x. Key arguments must be annotated with KeyT.

```python
import asyncio
import coredis

async def main():
    client = coredis.Redis()

    @client.register_script("return {KEYS[1], ARGV[1]}").wraps(key_spec=["key"])
    async def echo_key_value(key: str, value: str) -> list[bytes]:
      ...

    k, v = await echo_key_value("co", "redis")
    print(f"{k!r}={v!r}")

asyncio.run(main())
```

--------------------------------

### Use a Python-Wrapped Library

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/scripting.rst

Instantiate a Python class that wraps a Lua library and use its methods. This example demonstrates using the library with a client and includes setting data and calling a complex function.

```python
client = coredis.Redis()
async with client:
    lib = await MyLib(client, replace=True)
    await lib.ping()
    # b"pong"
    await lib.echo("hello world")
    # b"hello world"
    await client.hset("k1", {"a": 10, "b": 20})
    await client.hset("k2", {"c": 30, "d": 40})

    await lib.hmmget("k1", "k2", a=1, b=2, c=3, d=4, e=5, f=6)
    # [b"10", b"20", b"30", b"40", b"5", b"6"]
```

--------------------------------

### Enable Coredis Runtime Type Checks with Beartype

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/typing.rst

Shows how to enable runtime type checking in Coredis by setting the COREDIS_RUNTIME_CHECKS environment variable to 1 and installing beartype. This example demonstrates a type violation when passing an integer to the 'set' command's key parameter.

```bash
$ COREDIS_RUNTIME_CHECKS=1 python -c "
import coredis
import asyncio
async def test():
    async with coredis.Redis() as client:
        await client.set(1,1)
asyncio.run(test())
"
    Traceback (most recent call last):
      File "<@beartype(coredis.commands.core.CoreCommands.set) at 0x10c403130>", line 33, in set
    beartype.roar.BeartypeCallHintParamViolation: @beartyped coroutine CoreCommands.set() parameter key=1 violates type hint typing.Union[str, bytes], as 1 not str or bytes.
```

--------------------------------

### RedisJSON Set and Get Operations

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/modules.rst

Illustrates basic RedisJSON operations for setting and retrieving JSON documents. Uses the standard Python json module for serialization by default, or orjson if installed.

```python
import coredis
client = coredis.Redis()

async with client:
    await client.json.set(
        "key1", ".", {"a": 1, "b": [1, 2, 3], "c": "str"}
    )
    assert 1 == await client.json.get("key1", ".a")
    assert [1,2,3] == await client.json.get("key1", ".b")
    assert "str" == await client.json.get("key1", ".c")

    await client.json.set("key2", ".", {"a": 2, "b": [4,5,6], "c": ["str"]})

    # multi get
    assert ["str", ["str"]] == await client.json.mget(["key1", "key2"], ".c")
```

--------------------------------

### Add Entries to Streams

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/streams.rst

Add multiple entries to specified streams using `xadd`. This example adds 10 entries to each of the 'one', 'two', and 'three' streams.

```python
async with client:
    [await client.xadd("one", {"id": i}) for i in range(10)]
    [await client.xadd("two", {"id": i}) for i in range(10)]
    [await client.xadd("three", {"id": i}) for i in range(10)]
```

--------------------------------

### Coredis Pub/Sub Subscription (6.0)

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/migration_guide.rst

Example of subscribing to a channel using Coredis Pub/Sub in version 6.0. PubSub classes must be used as async context managers and the client must be used as an async context manager.

```python
import anyio
import coredis

async def main():
    client = coredis.Redis()
    async with client:
        async with client.pubsub() as pubsub:
            await pubsub.subscribe("channel")
            async for message in pubsub:
                print(message)

anyio.run(main)
```

--------------------------------

### Coredis Basic Usage (Single Node/Cluster)

Source: https://github.com/alisaifee/coredis/blob/master/README.md

Connect to a single Redis node or cluster, perform basic operations like SET, GET, INCR, EXPIRE, and use pipelines. Requires `anyio` and `coredis` to be installed.

```python
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

```

--------------------------------

### Coredis Sentinel Usage (6.0)

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/migration_guide.rst

Example of using Coredis Sentinel in version 6.0. Sentinel instances must be used as async context managers.

```python
import anyio
import coredis

async def main():
    sentinel = coredis.Sentinel(sentinels=[("localhost", 26379)])
    async with sentinel:
        primary = sentinel.primary_for("svc")
        replica = sentinel.replica_for("svc")
        async with primary, replica:
            await primary.set("fubar", 1)
            await replica.get("fubar")

anyio.run(main)
```

--------------------------------

### failover

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/compatibility.rst

Starts a coordinated failover from a server to one of its replicas.

```APIDOC
## failover

### Description
Starts a coordinated failover from a server to one of its replicas.
```

--------------------------------

### Coredis Stream Consumer (5.x)

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/migration_guide.rst

Example of using a stream consumer in Coredis version 5.x. Stream consumer instances must be async context managers.

```python
import asyncio
import coredis
from coredis.stream import Consumer

async def main():
    client = coredis.Redis()
    consumer = await Consumer(client, streams=["one", "two"])
    stream, entry = await consumer.get_entry()

asyncio.run(main())
```

--------------------------------

### Pub/Sub Channel and Subscription Info

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/pubsub.rst

Get information about subscribed channels and the number of subscribers for specific channels or all patterns.

```python
await client.pubsub_channels()
# ['foo', 'bar']
```

```python
await client.pubsub_numsub('foo', 'bar')
# [('foo', 9001), ('bar', 42)]
```

```python
await client.pubsub_numsub('baz')
# [('baz', 0)]
```

```python
await client.pubsub_numpat()
# 1204
```

--------------------------------

### Client-Side Caching with TrackingCache (5.x)

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/migration_guide.rst

Demonstrates using `TrackingCache` for client-side caching with a specified maximum size in bytes. It shows setting, getting, and retrieving cached values, and printing cache statistics.

```python
import asyncio
import coredis
from coredis.cache import TrackingCache

async def main():
    cache=TrackingCache(max_size_bytes=128 * 1024 * 1024)
    client = coredis.Redis(cache=cache)
    await client.set("fubar", 1)
    await client.get("fubar")
    await client.get("fubar")
    print(cache.stats)

asyncio.run(main())
```

--------------------------------

### Coredis Sentinel Usage

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/index.rst

Shows how to connect to Redis primary and replica instances using Coredis Sentinel. Demonstrates setting and getting a value through primary and replica clients.

```python
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")
```

--------------------------------

### Coredis Scripting with Wraps (6.0)

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/migration_guide.rst

Example of registering and using a Lua script with the 'wraps' decorator in Coredis version 6.0. Key arguments must be annotated with KeyT and the client must be used as an async context manager.

```python
import asyncio
import coredis
from coredis.typing import KeyT
from coredis.commands import CommandRequest

async def main():
    client = coredis.Redis()

    @client.register_script("return {KEYS[1], ARGV[1]}").wraps()
    def echo_key_value(key: KeyT, value: str) -> CommandRequest[list[bytes]]:
      ...

    async with client:
        k, v = await echo_key_value("co", "redis")
        print(f"{k!r}={v!r}")

asyncio.run(main())
```

--------------------------------

### Client-Side Caching with LRUCache (6.0)

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/migration_guide.rst

Demonstrates using `LRUCache` for client-side caching with a specified maximum number of keys. It shows setting, getting, and retrieving cached values within an async context manager, and printing cache statistics.

```python
import anyio
import coredis
from coredis.patterns.cache import LRUCache

async def main():
    cache=LRUCache(max_keys=10000)
    client = coredis.Redis(cache=cache)
    async with client:
      await client.set("fubar", 1)
      await client.get("fubar")
      await client.get("fubar")
      print(cache.stats)

anyio.run(main)
```

--------------------------------

### Coredis Stream Consumer (6.0)

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/migration_guide.rst

Example of using stream consumers in Coredis version 6.0. Stream consumers have moved to coredis.patterns.streams and must be used as async context managers.

```python
import anyio
import coredis

async def main():
    async with coredis.Redis() as client:
        async with client.xconsumer(streams=["one", "two"]) as consumer:
            stream, entry = await consumer.get_entry()
            async for stream, entry in consumer:
                print(stream, entry)

        async with client.xconsumer(
            streams=["one", "two"],
            group="group-a",
            consumer="consumer-1",
        ) as group_consumer:
            async for stream, entry in group_consumer:
                print(stream, entry)

anyio.run(main)
```

--------------------------------

### Connect to a Single Node Redis Instance

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/index.rst

Connect to a single Redis node, perform basic operations like set, get, incr, and expire. Demonstrates pipeline usage for atomic operations. Ensure Redis is running on localhost:6379.

```python
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
```

--------------------------------

### Create and Use Bloom Filter

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/modules.rst

Demonstrates creating a Bloom filter, adding items individually and in batches, and checking for their existence. Also shows how to use the BloomFilter class directly.

```python
import coredis
client = coredis.Redis()
async with client:
    # create filter
    await client.bf.reserve("filter", 0.1, 1000)

    # add items
    await client.bf.add("filter", 1)
    await client.bf.madd("filter", [2,3,4])

    # test for inclusion
    assert await client.bf.exists("filter", 1)
    assert (True, False) == await client.bf.mexists("filter", [2,5])

    # or
    assert await coredis.modules.BloomFilter(client).exists("filter", 1)
```

--------------------------------

### Coredis Sentinel Usage

Source: https://github.com/alisaifee/coredis/blob/master/README.md

Connect to Redis via Sentinel to get primary and replica clients for a given service. Requires `anyio` and `coredis` to be installed.

```python
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

```

--------------------------------

### Instantiating Module Command Groups Directly

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/api/modules.rst

Shows how to instantiate module command group classes directly and bind them to a client.

```APIDOC
## Instantiating Module Command Groups Directly

### Description
Module command groups can also be instantiated directly by passing a client instance to their constructor. This provides an alternative way to access module commands.

### Example
```python
import coredis
import coredis.modules

client = coredis.Redis()

# Instantiating Json command group directly
json_module = coredis.modules.Json(client)
await json_module.get("mykey", "$")

# Instantiating BloomFilter command group directly
bloom_filter = coredis.modules.BloomFilter(client)
await bloom_filter.add("mybloom", "item")
```
```

--------------------------------

### Configure Consumer with Internal Buffer

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/streams.rst

Set up a consumer with a buffer size to pre-fetch entries and a timeout for blocking operations.

```python
consumer = client.xconsumer(
    client,
    streams=["one", "two", "three"],
    # Will fetch upto 10 extra entries per stream
    # on every request to redis
    buffer_size=10,
    timeout=30*1000  # 30 seconds
)
```

--------------------------------

### Coredis Span Examples

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/observability.rst

Examples of span names emitted by Coredis for different operations. These are used for tracing command executions, pipelines, and transactions.

```text
SET
GET
```

```text
SET key1 ?
GET key1
```

--------------------------------

### Subscribe to Channels and Patterns on Instantiation

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/pubsub.rst

Instantiate a Pub/Sub consumer and subscribe to specific channels and patterns simultaneously. The async context manager handles subscription and cleanup.

```python
async with client.pubsub(
    channels=["my-first-channel", "my-second-channel"], patterns=["my-*"]
) as consumer:
    ...
```

--------------------------------

### Initialize and Use Count-Min Sketch

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/modules.rst

Illustrates initializing a Count-Min Sketch with dimensions, incrementing counts for multiple entries, and querying counts for entries.

```python
# create a sketch
await client.cms.initbydim("sketch", 2, 50)

# increment the counts for multiple entries
assert (1, 2) == await client.cms.incrby("sketch", {"a": 1, "b": 2})

# query the count for multiple entries
assert (1, 2, 0) == await client.cms.query("sketch", ["a", "b", "c"])
```

--------------------------------

### Initialize Coredis Client

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/streams.rst

Import and initialize a Coredis client for regular or cluster connections.

```python
import coredis
client = coredis.Redis()
# or cluster
# client = coredis.RedisCluster("localhost", 7000)
```

--------------------------------

### Get Message with Handler

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/pubsub.rst

Retrieve a message and process it with a handler. The message variable will be None if a handler successfully processed it.

```python
await message = consumer.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
```

--------------------------------

### Manage Connections with Blocking Pool (5.x)

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/migration_guide.rst

Demonstrates creating a blocking connection pool with a maximum of 2 connections and executing commands that might block.

```python
import coredis
import asyncio

from coredis import ConnectionPool, BlockingConnectionPool

async def main():
    # Blocking pool
    client = coredis.Redis(connection_pool=BlockingConnectionPool(max_connections=2))
    await asyncio.gather(*(client.blpop(["fubar"], timeout=1) for _ in range(3)))
```

--------------------------------

### Query Latest Temperature in Each Room with Coredis TimeSeries

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/modules.rst

Retrieves the most recent temperature reading for each specified room using the 'get' method.

```python
for room in rooms:
    print(await client.timeseries.get(f"temp:{room}"))
```

--------------------------------

### Perform Geo Filtered Search

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/modules.rst

Conduct a search with geographical filtering. This example finds documents within a specified radius of a given coordinate.

```python
results = await client.search.search(
    "json_index",
    "*",
    geo_filters={"location": ((67.0011, 24.8607), 1, coredis.PureToken.KM)},
    returns={"name": None},
)
assert results.total == 1
assert results.documents[0].properties["name"] == "karachi"
```

--------------------------------

### Implementing a Custom Cache

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/api/caching.rst

Guidelines for creating your own cache implementations.

```APIDOC
## Implementing a Custom Cache

### Description
To create a custom cache compatible with `coredis.Redis` or `coredis.RedisCluster`, you must implement the `AbstractCache` interface.

### Abstract Cache Interface
`coredis.patterns.cache.AbstractCache`

### Cache Statistics
`coredis.patterns.cache.CacheStats`
```

--------------------------------

### Manage Connections with Blocking Pool and Timeout (6.0)

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/migration_guide.rst

Demonstrates creating a connection pool with a maximum of 2 connections and a timeout, using async context managers, and handling `TimeoutError` when the timeout is exceeded.

```python
import coredis
import asyncio

async def main():
    async with coredis.Redis(
        connection_pool=coredis.pool.ConnectionPool(max_connections=2, timeout=1)
    ) as client:
        try:
            print(await asyncio.gather(*(client.blpop(["fubar"], timeout=5) for _ in range(3))))
        except TimeoutError:  # Note that a `TimeoutError` is raised instead of `ConnectionError` 
            print("Failed with timeout")

asyncio.run(main())
```

--------------------------------

### coredis.Redis.hello

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/compatibility.rst

Handshakes with the Redis server.

```APIDOC
## coredis.Redis.hello

### Description
Handshakes with the Redis server. New in redis 6.0.0.
```

--------------------------------

### Create and Use TopK

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/modules.rst

Demonstrates reserving space for a TopK structure, adding a large number of entries, and retrieving the top 3 most frequent items.

```python
import string
import itertools
import random

# create a top-3
await client.topk.reserve("top3", 3)

# add entries
letters = list(itertools.chain(*[k[0]*k[1] for k in list(enumerate(string.ascii_lowercase))]))
random.shuffle(letters)
await client.topk.add("top3", letters)

# get top 3 letters
assert (b'z', b'y', b'x') == await client.topk.list("top3")
```

--------------------------------

### Create and Use Cuckoo Filter

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/modules.rst

Shows how to create a Cuckoo filter, add items, check for existence, count items, and delete items. Demonstrates the addnx command for adding only if not present.

```python
# create filter
await client.cf.reserve("filter", 1000)

# add items
assert await client.cf.add("filter", 1)
assert not await client.cf.addnx("filter", 1)

# test for inclusion
assert await client.cf.exists("filter", 1)
assert 1 == await client.cf.count("filter", 1)

# delete an item
assert await client.cf.delete("filter", 1)

# test for inclusion
assert not await client.cf.exists("filter", 1)
assert 0 == await client.cf.count("filter", 1)
```

--------------------------------

### Initialize and Cleanup Redis Client (6.0)

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/migration_guide.rst

Version 6.0 mandates the use of async context managers for Redis clients, ensuring proper initialization and automatic cleanup of resources.

```python
import anyio
import coredis

async def main():
    client = coredis.Redis(host="127.0.0.1", port=6379)
    async with client:
        await client.set("key", "1")
    # Client and connection pool are automatically cleaned up

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

--------------------------------

### Lua Library Definition

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/scripting.rst

Defines a Lua library named 'mylib' with several functions: 'echo', 'ping', 'get', and 'hmmget'. This library can be registered with coredis for server-side execution.

```lua
#!lua name=mylib

redis.register_function('echo', function(k, a)
    return a[1]
end)
redis.register_function('ping', function()
    return "PONG"
end)
redis.register_function('get', function(k, a)
    return redis.call("GET", k[1])
end)
redis.register_function('hmmget', function(k, a)
    local values = {}
    local fields = {}
    local response = {}
    local i = 1
    local j = 1

    while a[i] do
        fields[j] = a[i]
        i = i + 2
        j = j + 1
    end

    for idx, key in ipairs(k) do
        values = redis.call("HMGET", key, unpack(fields))
        for idx, value in ipairs(values) do
            if not response[idx] and value then
                response[idx] = value
            end
        end
    end
    for idx, value in ipairs(fields) do
        if not response[idx] then
            response[idx] = a[idx*2]
        end
    end
    return response
end)
```

--------------------------------

### Manage Connections with Blocking Pool and Timeout (5.x)

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/migration_guide.rst

Demonstrates creating a blocking connection pool with a maximum of 2 connections and a timeout, and handling connection errors when the timeout is exceeded.

```python
import coredis
import asyncio

from coredis import ConnectionPool, BlockingConnectionPool

async def main():
    # Blocking pool with timeout
    client = coredis.Redis(connection_pool=BlockingConnectionPool(max_connections=2, timeout=1))
    try:
        await asyncio.gather(*(client.blpop(["fubar"], timeout=5) for _ in range(3)))
    except coredis.exceptions.ConnectionError as err:
        print("Failed with timeout")

asyncio.run(main())
```

--------------------------------

### Create Redis Client with Max Connections

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/connections.rst

Demonstrates creating a Redis client with a specified maximum number of connections and handling concurrent blocking requests that exceed this limit. The command will block until a connection is available.

```python
import asyncio
import coredis

async def test():
    client = coredis.Redis(max_connections=8)
    # or with cluster
    # client = coredis.RedisCluster(
    #   "localhost", 7000,
    #   max_connections=8, max_connections_per_node=True
    # )

    async with client:
        results = await asyncio.gather(
            *[client.blpop(["fubar"], 3) for _ in range(10)],
        )

asyncio.run(test())
```

--------------------------------

### Create TimeSeries Compaction Rules

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/modules.rst

Shows how to create compaction rules for hourly and daily averages for time series, specifying the aggregation type and time interval.

```python
for room in rooms:
    assert await client.timeseries.create(
        f"temp:{room}:hourly:avg", labels={"room": room, "compaction": "hourly"}
    )
    assert await client.timeseries.create(
        f"temp:{room}:daily:avg", labels={"room": room, "compaction": "daily"}
    )

    assert await client.timeseries.createrule(
        f"temp:{room}", f"temp:{room}:hourly:avg",
        coredis.PureToken.AVG, timedelta(hours=1)
    )
    assert await client.timeseries.createrule(
        f"temp:{room}", f"temp:{room}:daily:avg",
        coredis.PureToken.AVG, timedelta(days=1)
    )
```

--------------------------------

### Run Specific Pytest Tests

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/development.rst

Execute specific tests by filtering based on markers. This example runs 'basic' tests while excluding 'raw', 'resp2', and 'cached' client types for a specific test file.

```bash
$ pytest -m 'basic and not (raw or resp2 or cached)' tests/commands/test_string.py
```

--------------------------------

### Create TimeSeries with Labels

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/modules.rst

Demonstrates creating multiple time series, each with a unique label indicating the room, using the coredis client.

```python
import coredis
from datetime import datetime, timedelta

rooms = {"bedroom", "lounge", "bathroom"}
client = coredis.Redis(port=9379)

async with client:
    for room in rooms:
        assert await client.timeseries.create(f"temp:{room}", labels={"room": room})
```

--------------------------------

### Test Redis Caching with LRUCache

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/caching.rst

Tests the caching mechanism by performing get, set, and delete operations. It verifies cache invalidation when the server data changes and immediate invalidation upon local delete.

```python
async def test():
    async with cached_client, regular_client:
        assert not await cached_client.get("fubar") # None response cached
        await regular_client.set("fubar", "bar") # <- triggers a push message to cached_client
        await asyncio.sleep(0.01)
        assert b"bar" == await cached_client.get("fubar") # Cache should be invalidated
        assert b"bar" == await cached_client.get("fubar") # Fetched from local cache
        await cached_client.delete(["fubar"]) # Invalidates local cache immediately
        assert not await cached_client.get("fubar")

asyncio.run(test())
```

--------------------------------

### Basic Redis Operations with Coredis

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/index.rst

Demonstrates setting, checking existence, incrementing, expiring, and deleting keys in Redis using Coredis. Includes pipeline usage for atomic operations.

```python
import anyio
from coredis.connection import TCPLocation
from coredis import Redis

async def main() -> None:
    client = Redis(
        locations=[
            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")
```

--------------------------------

### Manage Connections with Non-blocking Pool (5.x)

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/migration_guide.rst

Demonstrates creating a non-blocking connection pool with a maximum of 2 connections and handling potential connection errors when attempting to exceed this limit.

```python
import coredis
import asyncio

from coredis import ConnectionPool, BlockingConnectionPool

async def main():
    # Non-blocking pool
    client = coredis.Redis(connection_pool=ConnectionPool(max_connections=2))
    try:
        await asyncio.gather(*(client.blpop(["fubar"], timeout=1) for _ in range(3)))
    except coredis.exceptions.ConnectionError as err:
        print("Failed with too many connections attempted")
```

--------------------------------

### Registering Decimal Serializer and Deserializer with TypeAdapter

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/response.rst

Register custom serializers and deserializers for types like Decimal to enable type-safe interactions with Redis. This setup allows Decimal values to be sent to and retrieved from Redis.

```python
from decimal import Decimal
from coredis import Redis
from coredis.typing import Serializable, TypeAdapter

adapter = TypeAdapter()

# Register a serializer so Decimal types can be passed into redis
# commands in a type safe manner.
@adapter.serializer
def decimal_to_str(value: Decimal) -> str:
    return str(value)

# registration can be done without using a decorator as well
# adapter.register_serializer(serializable_type=Decimal, serializer=decimal_to_str)


# Register a deserializer to retrieve string or byte values as
# Decimal types
@adapter.deserializer
def str_to_decimal(value: str | bytes) -> Decimal:
    return Decimal(value.decode("utf-8") if isinstance(value, bytes) else value)

# registration can be done without using a decorator as well
# adapter.register_deserializer(
#     deserialized_type=Decimal, deserializer=decimal_to_str, deserializable_type=str|bytes
# )

client = Redis(type_adapter=adapter, decode_responses=True)
async with client:
    await client.set("price", Serializable(Decimal("19.99")))
    value = await client.get("price").transform(Decimal)
    assert isinstance(value, Decimal)
```

--------------------------------

### Pipeline and Transactions

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/api/pipeline.rst

Coredis provides support for pipelining and transactions through the Pipeline and ClusterPipeline classes. These classes are instantiated by calling the `pipeline()` method on `coredis.Redis` or `coredis.RedisCluster` client objects, respectively. Refer to the handbook for detailed examples.

```APIDOC
## Pipeline and Transaction Support

### Description
Coredis exposes Pipelining and Transactions via the `Pipeline` and `ClusterPipeline` classes. These classes are returned by the `coredis.Redis.pipeline()` and `coredis.RedisCluster.pipeline()` methods.

Refer to the :ref:`handbook/pipelines:pipelines` for usage examples.

### Classes

#### `coredis.patterns.pipeline.Pipeline`

This class handles pipelining and transactions for a single Redis instance.

#### `coredis.patterns.pipeline.ClusterPipeline`

This class handles pipelining and transactions for a Redis Cluster setup.
```

--------------------------------

### coredis.Redis.command_docs

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/compatibility.rst

Returns documentation for commands.

```APIDOC
## coredis.Redis.command_docs

### Description
Returns documentary information about one, multiple or all commands.
```

--------------------------------

### Create and Use T-Digest

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/modules.rst

Shows how to create a T-Digest, add values, and retrieve data by rank, reverse rank, and quantile.

```python
# create a digest
await client.tdigest.create("digest")

# add some values
await client.tdigest.add("digest", 1, [1, 2, 3, 4])

# add some more values
await client.tdigest.add("digest", 1, [1, 2, 3, 4])

# get the rank & reverse ranks
assert (1.0, 1.0, 2.0) == await client.tdigest.byrank("digest", [0, 1, 2])
assert (6.0, 5.0, 4.0) == await client.tdigest.byrevrank("digest", [0, 1, 2])

# get the quantiles
assert (1.0, 3.0, 6.0) == await client.tdigest.quantile("digest", [0, 0.5, 1])
```

--------------------------------

### Atomic Increment with Distributed Lock

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/locks.rst

Implement an atomic increment operation using a distributed lock. This example demonstrates acquiring a lock, reading a value, incrementing it, and setting it back, ensuring atomicity across concurrent operations.

```python
import asyncio
import coredis

async def increment(client: coredis.Redis, key: str) -> int:
    async with client.lock(f"increment:{key}") as lock:
        value = int(await client.get(key) or 0)
        await client.set(key, int(value)+1)

async def test():
    async with coredis.Redis() as client:
        await client.delete(["fubar"])
        await asyncio.gather(
            *(increment(client, "fubar") for _ in range(64))
        )
        assert int(await client.get("fubar")) ==  64

asyncio.run(test())
```

--------------------------------

### Manage Connections with Non-blocking Pool (6.0)

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/migration_guide.rst

Demonstrates creating a non-blocking connection pool with a maximum of 2 connections using async context managers.

```python
import coredis
import asyncio

async def main():
    async with coredis.Redis(
        connection_pool=coredis.pool.ConnectionPool(max_connections=2)
    ) as client:
        print(await asyncio.gather(*(client.blpop(["fubar"], timeout=1) for _ in range(3))))

asyncio.run(main())
```

--------------------------------

### coredis.Redis.function_load

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/compatibility.rst

Creates a library.

```APIDOC
## coredis.Redis.function_load

### Description
Creates a library.
```

--------------------------------

### Explicitly Subscribe to Channels and Patterns

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/pubsub.rst

Create a Pub/Sub consumer and then explicitly subscribe to channels and patterns using the subscribe and psubscribe methods. The async context manager ensures proper cleanup.

```python
async with client.pubsub() as consumer:
    await consumer.subscribe("my-first-channel", "my-second-channel", ...)
    await consumer.psubscribe("my-*")
```

--------------------------------

### Register and Load a Lua Library

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/scripting.rst

Register a Lua script with a given name and then load it. If a library with the same name exists, this will raise an exception unless 'replace' is set to True.

```python
client = coredis.Redis()
library = await client.register_library("mylib", open("/var/tmp/library.lua").read())
```

```python
library = await client.load_library("mylib")
```

--------------------------------

### Initialize Pipeline with Transaction Enabled

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/pipelines.rst

Create a pipeline instance with `transaction=True` to enable atomic execution of buffered commands.

```python
pipe = r.pipeline(transaction=True)
```

--------------------------------

### Query Daily Averages by Individual Room (Compacted) with Coredis TimeSeries

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/modules.rst

Queries for daily average temperatures from a pre-compacted time-series using individual 'range' calls. Requires the compacted series to exist.

```python
# using individual range queries on the compacted timeseries
for room in rooms:
    print(await client.timeseries.range(
        f"temp:{room}:daily:avg", 0, datetime(1971, 1, 1),
    ))
```

--------------------------------

### Create Primary and Replica Client Connections

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/sentinel.rst

Obtain Redis client instances for primary (write) and replica (read-only) nodes managed by Sentinel. These clients use a SentinelConnectionPool.

```python
primary = sentinel.primary_for('myredis', stream_timeout=0.1)
replica = sentinel.replica_for('myredis', stream_timeout=0.1)
async with primary, replica:
    await primary.set('foo', 'bar')
    await replica.get('foo')
    # 'bar'
```

--------------------------------

### Execute Pipeline Commands (5.x)

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/migration_guide.rst

In version 5.x, pipelines were created, commands added, and then explicitly executed using `pipe.execute()`.

```python
import asyncio
import coredis

async def main():
    client = coredis.Redis()
    pipe = await client.pipeline()
    pipe.set("foo", 1)
    pipe.incr("foo")
    results = await pipe.execute()
    assert results == (True, 2)

asyncio.run(main())
```

--------------------------------

### Discover Primary and Replicas with Sentinel

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/sentinel.rst

Connect to Sentinel instances to discover the primary and replica nodes for a given Redis service name. Requires at least one Sentinel daemon running.

```python
from coredis import Sentinel
sentinel = Sentinel([('localhost', 26379)], stream_timeout=0.1)
async with sentinel:
    await sentinel.discover_primary('myredis')
    # ('127.0.0.1', 6379)
    await sentinel.discover_replicas('myredis')
    # [('127.0.0.1', 6380)]
```

--------------------------------

### Manage Shared Connection Pool Lifecycle

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/connections.rst

Demonstrates the correct way to manage the lifecycle of a shared connection pool by entering its async context manager in the parent task. This ensures the pool is properly initialized and closed.

```python
from typing import Any
import asyncio
import coredis
from coredis.connection import TCPLocation

async def worker(pool: coredis.pool.ConnectionPool[Any]) -> None:
   async with coredis.Redis(connection_pool=pool) as client:
       while True:
           await client.ping()

async def start_workers():
    pool = coredis.ConnectionPool(location=TCPLocation("localhost", 6379))
    # Entering the pool here means it's lifetime is now managed here
    async with pool:
        await asyncio.gather(*(worker(pool) for _ in range(1024)))

asyncio.run(start_workers())
```

--------------------------------

### Aggregate Data: Filter, Group, Apply, and Group Again

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/modules.rst

Demonstrates a complex aggregation pipeline: filters cities by population, groups by country to calculate count and average population, applies a logarithmic transformation, and then groups by the transformed average population to list countries.

```python
aggregations = await client.search.aggregate(
        "hash_index",
        "*",
        load="*",
        transforms=[
            # include only cities with population greater than 20 million
            coredis.modules.search.Filter(
                '@population > 20000000'
            ),
            # group by country=>{count, average_city_population}
            coredis.modules.search.Group(
                "@country", [
                    coredis.modules.search.Reduce("count", [0], "city_count"),
                    coredis.modules.search.Reduce("avg", [1, '@population'], "average_city_population")
                 ]
            ),
            # apply a transformation of average_city_population -> log10(average_city_population)
            coredis.modules.search.Apply(
                "floor(log(@average_city_population))",
                "average_population_bucket"
            ),
            # group by average_population_bucket=>countries
            coredis.modules.search.Group(
                "@average_population_bucket", [
                    coredis.modules.search.Reduce("tolist", [1, "@country"], "countries"),
                 ]
            ),
        ],
    )
    assert aggregations.results[0] == {
        'average_population_bucket': '16',
        'countries': ['Brazil', 'South Korea', 'Egypt', 'Mexico']
    }
    assert aggregations.results[1] == {
        'average_population_bucket': '17',
        'countries': ['Japan', 'Indonesia', 'China', 'Philippines', 'India']
    }
```

--------------------------------

### Enable Coredis OpenTelemetry Instrumentation

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/observability.rst

Enable OpenTelemetry instrumentation by setting the COREDIS_OTEL_ENABLED environment variable or the coredis.Config.otel_enabled configuration option.

```bash
export COREDIS_OTEL_ENABLED=true
```

--------------------------------

### Configure Consumer with Blocking Timeout

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/streams.rst

Initialize a consumer with a blocking timeout to control how long `get_entry` or iteration will wait for new entries.

```python
consumer = client.xconsumer(
    client,
    streams=["one", "two", "three"],
    timeout=30*1000  # 30 seconds
)
```

--------------------------------

### Pipeline with Error Handling and Type Hinting

Source: https://github.com/alisaifee/coredis/blob/master/docs/source/handbook/pipelines.rst

Demonstrates pipeline usage with `raise_on_error=False` and type hinting for awaited command results.

```python
async with client.pipeline(raise_on_error=False) as pipe:
    pipe.set("foo", 1)
    value = pipe.incr("foo")

v: int = await value
```