hiredis

import hiredis

from hiredis import Reader, pack_command, HiredisError, ProtocolError, ReplyError, PushNotification

import hiredis

# Create a reader for parsing Redis replies
reader = hiredis.Reader()

# Feed data from Redis connection
reader.feed("$5\r\nhello\r\n")

# Extract parsed reply
reply = reader.gets()  # Returns b'hello'

# Pack commands into Redis protocol format  
packed = hiredis.pack_command(("SET", "key", "value"))
# Returns: b'*3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$5\r\nvalue\r\n'

class Reader:
    def __init__(
        self,
        protocolError: Callable[[str], Exception] = None,
        replyError: Callable[[str], Exception] = None,
        encoding: Optional[str] = None,
        errors: Optional[str] = None,
        notEnoughData: Any = False,
    ) -> None:
        """
        Create a new Reader for parsing Redis replies.

        Parameters:
        - protocolError: Custom exception class/callable for protocol errors
        - replyError: Custom exception class/callable for reply errors  
        - encoding: Text encoding for bulk data decoding (e.g., 'utf-8')
        - errors: Error handling mode ('strict', 'replace', 'ignore', 'backslashreplace')
        - notEnoughData: Custom sentinel returned when buffer lacks complete reply
        """

    def feed(self, buf: Union[str, bytes], off: int = 0, len: int = -1) -> None:
        """
        Feed data to the internal buffer for parsing.

        Parameters:
        - buf: Data to append to buffer (string or bytes)
        - off: Starting offset in buffer (default: 0)
        - len: Number of bytes to read (default: -1 for all)
        """

    def gets(self, shouldDecode: bool = True) -> Any:
        """
        Extract a reply from the buffer.

        Parameters:
        - shouldDecode: Whether to decode bulk data using configured encoding

        Returns:
        - Parsed reply data, or False if buffer lacks complete reply
        - Returns notEnoughData sentinel if configured
        """

    def setmaxbuf(self, maxbuf: Optional[int]) -> None:
        """
        Set maximum buffer size to control memory usage.

        Parameters:
        - maxbuf: Maximum buffer size in bytes, None for unlimited
        """

    def getmaxbuf(self) -> int:
        """
        Get current maximum buffer size.

        Returns:
        - Current maximum buffer size in bytes
        """

    def len(self) -> int:
        """
        Get length of data in internal buffer.

        Returns:
        - Number of bytes in buffer
        """

    def has_data(self) -> bool:
        """
        Check if buffer contains any data.

        Returns:
        - True if buffer has data, False otherwise
        """

    def set_encoding(
        self, encoding: Optional[str] = None, errors: Optional[str] = None
    ) -> None:
        """
        Update encoding configuration for bulk data decoding.

        Parameters:
        - encoding: Text encoding (e.g., 'utf-8'), None to disable decoding
        - errors: Error handling mode ('strict', 'replace', 'ignore', 'backslashreplace')
        """

def pack_command(cmd: Tuple[Union[str, int, float, bytes, memoryview], ...]) -> bytes:
    """
    Pack command arguments into Redis protocol format.

    Parameters:
    - cmd: Tuple containing command and arguments
           Supports: str, int, float, bytes, memoryview

    Returns:
    - Bytes object containing packed Redis protocol command

    Raises:
    - TypeError: If command contains unsupported argument types
    """

class HiredisError(Exception):
    """Base exception class for all hiredis errors."""

class ProtocolError(HiredisError):
    """
    Raised when protocol parsing fails due to corrupt data stream.
    Indicates unrecoverable protocol state requiring reconnection.
    """

class ReplyError(HiredisError):
    """
    Represents Redis error replies (-ERR responses).
    This exception is returned by gets(), not raised.
    """

class PushNotification(list):
    """
    Represents Redis push notifications in RESP3 protocol.
    Extends list to contain notification data.
    """

__version__: str
# Current package version string (e.g., "3.2.1")

import hiredis

# Configure reader for UTF-8 decoding
reader = hiredis.Reader(encoding="utf-8", errors="strict")
reader.feed(b"$3\r\n\xe2\x98\x83\r\n")  # UTF-8 snowman
reply = reader.gets()  # Returns '☃' (string, not bytes)

import hiredis

# Use custom exception classes
class MyProtocolError(Exception):
    pass

class MyReplyError(Exception):
    pass

reader = hiredis.Reader(
    protocolError=MyProtocolError,
    replyError=MyReplyError
)

# Protocol errors will raise MyProtocolError
# Reply errors will return MyReplyError instances

import hiredis

reader = hiredis.Reader()

# Set buffer size limit
reader.setmaxbuf(1024 * 1024)  # 1MB max

# Monitor buffer usage
print(f"Buffer size: {reader.len()} bytes")
print(f"Has data: {reader.has_data()}")
print(f"Max buffer: {reader.getmaxbuf()} bytes")

import hiredis

reader = hiredis.Reader()

# Feed data containing multiple replies
reader.feed("$5\r\nhello\r\n$5\r\nworld\r\n")

# Extract all replies
replies = []
while True:
    reply = reader.gets()
    if reply is False:  # No more complete replies
        break
    replies.append(reply)

print(replies)  # [b'hello', b'world']

import hiredis

# Basic command
cmd = hiredis.pack_command(("GET", "mykey"))
# Returns: b'*2\r\n$3\r\nGET\r\n$5\r\nmykey\r\n'

# Complex command with different data types
cmd = hiredis.pack_command((
    "HSET", "hash", "field1", "value1", 
    b"field2", b"binary_value", 
    "field3", 42,
    "field4", 3.14159
))

# Command with memoryview
data = memoryview(b"binary data")
cmd = hiredis.pack_command(("SET", "key", data))