libvalkey
libvalkey-py is a Python extension that wraps the protocol parsing code from the underlying libvalkey C library. It primarily speeds up the parsing of multi bulk replies from a Valkey server. The library is actively maintained, with its current version being 4.0.1, and releases occurring regularly to introduce new features, improvements, and bug fixes.
Warnings
- breaking The library was renamed from `hiredis-py` to `libvalkey-py` in version 3.0.0. All `hiredis` imports must be updated to `libvalkey`.
- gotcha Building `libvalkey` from source (e.g., if a pre-compiled wheel is not available for your environment) requires Python development headers (e.g., `python3-dev` on Debian/Ubuntu) and a C compiler like `gcc`.
- gotcha For error handling, `libvalkey.ProtocolError` is raised for invalid protocol states. However, `libvalkey.ReplyError` (for server-side error replies like `-ERR`) is *returned* by `gets()` and not raised as an exception by default.
- gotcha When `libvalkey.Reader` is initialized with an `encoding` (e.g., 'utf-8'), bulk string replies are decoded. If the decoding fails and the `errors` parameter is set to 'strict' (the default), a `UnicodeDecodeError` will be raised.
- breaking Version 4.0.0 included changes to the `Reader`'s behavior, with a subsequent patch (v4.0.0b1) to 'partly restore the previous behaviour'. Users upgrading to or through v4.0.0 should review the changelog carefully for potential subtle changes in parsing logic.
Install
-
pip install libvalkey
Imports
- Reader
from libvalkey import Reader
Quickstart
import libvalkey
# The Reader class is responsible for parsing replies from a stream of data.
# It does not handle I/O itself.
reader = libvalkey.Reader()
# Feed raw Valkey protocol bytes to the reader
reader.feed(b"$5\r\nhello\r\n")
# Retrieve the parsed reply
reply = reader.gets()
print(f"Parsed reply: {reply}")
# Example with a list reply
reader.feed(b"*2\r\n$5\r\nhello\r\n$5\r\nworld\r\n")
list_reply = reader.gets()
print(f"Parsed list reply: {list_reply}")
# Example with custom sentinel for 'not enough data'
reader_ellipsis = libvalkey.Reader(notEnoughData=Ellipsis)
reader_ellipsis.feed(b"*2\r\n$5\r\nhello\r\n")
partial_reply = reader_ellipsis.gets()
print(f"Partial reply (needs more data): {partial_reply}")
reader_ellipsis.feed(b"$5\r\nworld\r\n")
full_reply = reader_ellipsis.gets()
print(f"Full reply: {full_reply}")