DOCSEARCHbyAlgolia
Submit Docs

valkey-py

repository·main·Indexed 42 days ago

https://github.com/valkey-io/valkey-py

The official Python interface for the Valkey key-value store, forked from redis-py. It provides high-performance connectivity with support for RESP3, connection pooling, pipelines, Pub/Sub, and Valkey Cluster. Includes features such as OpenTelemetry instrumentation, atomic transactions via WATCH, and support for CLIENT CAPA redirect in Valkey 8.0.

Tokens
64.8K
Snippets
258
Records
334
Agent score
54%

What's inside valkey-py

  1. Perform atomic and non-atomic multi-key operations

    main

    Valkey Cluster supports multi-key commands (like mget and mset) under two conditions:

    1. Atomic Operations: Use standard methods like mget or mset when all keys are guaranteed to be in the same hash slot (e.g., by using hash tags like {foo}1 and {foo}2). If keys belong to different slots, a ValkeyClusterException is raised.
    2. Non-atomic Operations: Use mget_nonatomic or mset_nonatomic to handle keys across different slots. The client batches keys by their hash value and sends separate commands to the respective slot owners. This is not an atomic operation.

    To force keys into the same slot, use hash tags in the key names.

    # Atomic operations (requires hash tags for same slot)
rc.mset({'{foo}1': 'bar1', '{foo}2': 'bar2'})
rc.mget('{foo}1', '{foo}2')

# Non-atomic operations (keys can be in different slots)
rc.mset_nonatomic({'foo': 'value1', 'bar': 'value2', 'zzz': 'value3'})
rc.mget_nonatomic('foo', 'bar', 'zzz')

  2. Three strategies for reading PubSub messages

    main

    There are three ways to consume messages from a PubSub instance:

    1. get_message() (Polling): Uses the system's select module to poll the socket. It returns the message dictionary if data is available, or immediately returns None if no data is waiting. This is ideal for integration into an existing event loop.
    2. listen() (Blocking Generator): A generator that blocks the current thread until a message is available. Use this if your application's primary purpose is to react to messages.
    3. run_in_thread() (Background Thread): Spawns a new thread that runs a non-blocking event loop. This is useful for background processing. You can control the loop frequency with sleep_time and handle errors using the exception_handler argument.
    # Strategy 1: Polling
while True:
    message = p.get_message()
    if message:
        process(message)
    time.sleep(0.001)

# Strategy 2: Blocking Generator
for message in p.listen():
    process(message)

# Strategy 3: Background Thread
thread = p.run_in_thread(sleep_time=0.001, exception_handler=my_err_handler)
# ... later ...
thread.stop()

  3. Monitor Valkey Server performance with OpenTelemetry Collector

    main

    To monitor the performance of the Valkey Server itself (rather than just the client), use an OpenTelemetry Collector Agent. The Collector acts as a proxy that receives telemetry data, processes it, and exports it to APM tools like Uptrace or Jaeger.

    You can use the OpenTelemetry Redis receiver within the Collector to gather Valkey metrics.

  4. Use Valkey Pub/Sub

    main

    Pub/Sub is implemented via a separate PubSub class because subscribing puts the underlying connection into a state where it cannot execute standard commands. You can obtain a PubSub instance by calling .pubsub() on your Valkey client. Note that while you can subscribe and listen via the PubSub instance, you should use the main Valkey client to call PUBLISH.

    import valkey

r = valkey.Valkey(host='localhost', port=6379, db=0)
p = r.pubsub()

# Subscribe to channels
p.subscribe('my-first-channel', 'my-second-channel')

# Retrieve a message
message = p.get_message()
# Example output: {'pattern': None, 'type': 'subscribe', 'channel': b'my-second-channel', 'data': 1}

  5. Thread safety and the SELECT command

    main

    Valkey client instances are thread-safe and can be shared between threads because connections are retrieved from the pool only during command execution and returned immediately after.

    Important Caveat: The SELECT command is not implemented on client instances. Because SELECT changes the state of a connection (the database index) and that state persists until changed or closed, it could lead to connections being returned to the pool in an unexpected database state.

    Best Practice: If your application uses multiple Valkey databases, create a separate client instance (and potentially a separate connection pool) for each database.

  6. Use Valkey Pipelines to optimize round-trips

    main

    Pipelines allow you to batch multiple commands together, reducing network round-trips. The Pipeline class implements MULTI/EXEC logic by default; when execute() is called, the commands are wrapped in MULTI and EXEC statements. You can disable this transactional behavior by setting transaction=False.

    # Assuming 'r' is an existing Valkey instance
pipe = r.pipeline()
pipe.set('foo', 5)
pipe.set('bar', 18.5)
pipe.set('blee', "hello world!")

# Executes all commands in a batch
results = pipe.execute()
# Returns: [True, True, True]

  7. PubSub in Valkey Cluster

    main

    When creating a ClusterPubSub instance without a specific node, the client transparently selects a node for the connection on the first command execution by hashing the channel name to find its keyslot.

    Limitations: pattern subscribe and publish are not currently recommended because pattern matching (e.g., fo*) involves an unknowable number of potential channel names across different keyslots.

    # Connection is set to the node holding the 'foo' keyslot
p1 = rc.pubsub()
p1.subscribe('foo')

# Connection is explicitly set to a specific node
p2 = rc.pubsub(rc.get_node('localhost', 6379))

  8. Manage Connection Pools

    main

    To manage multiple connections efficiently, use connection pooling. valkey-py provides both synchronous and asynchronous pools:

    • valkey.connection.ConnectionPool: For standard synchronous usage.
    • valkey.asyncio.connection.ConnectionPool: For asynchronous usage.

  9. Use ClusterPipelines in Valkey Cluster

    main

    The ClusterPipeline class (a subclass of ValkeyCluster) supports pipelining in cluster mode. When execute() is called, commands are grouped by the node they belong to and executed in parallel across the cluster nodes.

    Constraints and Behavior:

    • Key-based commands only: ClusterPipeline currently only supports commands that operate on specific keys.
    • No Transactions: The transaction option (which uses MULTI/EXEC) is NOT supported in cluster mode because commands are partitioned across different nodes and cannot be wrapped in a single global transaction block.
    • Read from Replicas: The pipeline inherits the read_from_replicas setting from the cluster instance. If enabled on the cluster, the pipeline will direct read commands to replicas.
    # Assuming rc is a ValkeyCluster instance
with rc.pipeline() as pipe:
    pipe.set('foo', 'value1')
    pipe.set('bar', 'value2')
    pipe.get('foo')
    pipe.get('bar')
    print(pipe.execute())
# Output: [True, True, b'value1', b'value2']

  10. Create a ValkeyCluster instance

    main

    To connect to a Valkey Cluster, use the ValkeyCluster class. You must provide at least one node for cluster discovery. You can initialize the client using host/port arguments, a URL, or a list of ClusterNode objects.

    If no startup nodes are reachable, a ValkeyClusterException is thrown. Upon successful connection, the client initializes caches for slots, nodes, and supported commands.

    from valkey.cluster import ValkeyCluster as Valkey
from valkey.cluster import ClusterNode

# Option 1: Using host and port
rc = Valkey(host='localhost', port=6379)

# Option 2: Using a URL
rc = Valkey.from_url("valkey://localhost:6379/0")

# Option 3: Using ClusterNode objects
nodes = [ClusterNode('localhost', 6379), ClusterNode('localhost', 6378)]
rc = Valkey(startup_nodes=nodes)

  11. Handle RESP 3 push notifications with a custom function

    main

    RESP 3 includes a push type that allows the client to intercept out-of-band messages. While valkey-py logs these messages by default, you can provide a custom processor by passing a function to the push_handler_func argument when calling .pubsub() on a RESP 3 connection.

    from valkey import Valkey

def our_func(message):
    if message.find("This special thing happened"):
        raise IOError("This was the message: \n" + message)

# Ensure protocol=3 is enabled to support push notifications
r = Valkey(protocol=3)
p = r.pubsub(push_handler_func=our_func)

  12. Use Pipelines to buffer commands

    main

    Pipelines allow you to buffer multiple commands and send them to the server in a single request, significantly improving performance by reducing TCP round trips.

    Key Features:

    • Chaining: All commands buffered into a pipeline return the pipeline object itself, allowing for method chaining.
    • Atomicity: By default, pipelines execute commands atomically as a group (using MULTI/EXEC). To disable this and just use the pipeline for buffering, use r.pipeline(transaction=False).
    • Execution: Use .execute() to send the buffered commands. It returns a list of responses corresponding to each command in the order they were sent.
    r = valkey.Valkey(host='localhost', port=6379)

# Basic usage
pipe = r.pipeline()
pipe.set('foo', 'bar')
pipe.get('bing')
responses = pipe.execute()

# Chained usage
responses = r.pipeline().set('foo', 'bar').sadd('faz', 'baz').incr('auto_number').execute()