Stream

Stream class for managing data in Redis.

Source code in blue/stream.py

class Stream:
    """
    Stream class for managing data in Redis.
    """

    def __init__(self, cid, properties={}):
        """
        Initialize the Stream with a unique identifier and optional properties.
        """
        self.cid = cid
        self._initialize(properties=properties)

        self._start()

    def _initialize(self, properties=None):
        """
        Initialize the stream with default and provided properties.

        Parameters:
            properties: Optional dictionary of properties to override defaults.
        """
        self._initialize_properties()
        self._update_properties(properties=properties)

    def _initialize_properties(self):
        """
        Initialize the default properties for the stream.
        """
        self.properties = {}

        # db connectivity
        self.properties['db.host'] = 'localhost'
        self.properties['db.port'] = 6379

    def _update_properties(self, properties=None):
        """
        Update the properties of the stream.

        Parameters:
            properties: Optional dictionary of properties to update.
        """
        if properties is None:
            return

        # override
        for p in properties:
            self.properties[p] = properties[p]

    ##  data
    def _get_data_namespace(self):
        """
        Get the data namespace for the stream.

        Returns:
            (str): The data namespace string.
        """
        return self.cid + ":DATA"

    def _init_data_namespace(self):
        """
        Initialize the data namespace for the stream.
        """
        # create namespaces for stream-specific data
        return self.connection.json().set(
            self._get_data_namespace(),
            "$",
            {},
            nx=True,
        )

    def set_data(self, key, value):
        """
        Set the data for a specific key in the stream.

        Parameters:
            key: The key to set.
            value: The value to set.
        """
        self.connection.json().set(
            self._get_data_namespace(),
            "$." + key,
            value,
        )

    def get_data(self, key):
        """
        Get the data for a specific key in the stream.

        Parameters:
            key: The key to get.

        Returns:
            The value associated with the key, or None if the key does not exist.
        """
        value = self.connection.json().get(
            self._get_data_namespace(),
            Path("$." + key),
        )
        return self.__get_json_value(value)

    def get_all_data(self):
        """
        Get all data for the stream.

        Returns:
            (dict): A dictionary containing all data in the stream.
        """
        value = self.connection.json().get(
            self._get_data_namespace(),
            Path("$"),
        )
        return self.__get_json_value(value)

    def append_data(self, key, value):
        """
        Append data to a specific key in the stream.

        Parameters:
            key: The key to append to.
            value: The value to append.
        """
        self.connection.json().arrappend(
            self._get_data_namespace(),
            "$." + key,
            value,
        )

    def get_data_len(self, key):
        """
        Get the length of the data array for a specific key in the stream.

        Parameters:
            key: The key to get the length for.
        Returns:
            (int): The length of the data array, or 0 if the key does not exist.
        """
        return self.connection.json().arrlen(
            self._get_data_namespace(),
            Path("$." + key),
        )

    ##  metadata
    def _get_metadata_namespace(self):
        """Get the metadata namespace for the stream.

        Returns:
            (str): The metadata namespace string.
        """
        return self.cid + ":METADATA"

    def _init_metadata_namespace(self):
        """Initialize the metadata namespace for the stream."""
        # create metadata namespace
        return self.connection.json().set(self._get_metadata_namespace(), "$", {"created_by": "", "id": "", "tags": {}, "consumers": {}, "producers": {}}, nx=True)

    def set_metadata(self, key, value, nx=False):
        """
        Set the metadata for a specific key in the stream.

        Parameters:
            key: The key to set.
            value: The value to set.
            nx (bool): If True, set the value only if it does not already exist.
        """
        self.connection.json().set(self._get_metadata_namespace(), "$." + key, value, nx=nx)

    def get_metadata(self, key=""):
        """
        Get the metadata for a specific key in the stream.

        Parameters:
            key: The key to get.
        Returns:
            The value associated with the key, or None if the key does not exist.
        """
        value = self.connection.json().get(
            self._get_metadata_namespace(),
            Path("$" + ("" if pydash.is_empty(key) else ".") + key),
        )
        return self.__get_json_value(value)

    def __get_json_value(self, value):
        if value is None:
            return None
        if type(value) is list:
            if len(value) == 0:
                return None
            else:
                return value[0]
        else:
            return value

    def _start(self):
        """Start the stream by establishing a database connection and initializing metadata and data namespaces."""
        self._start_connection()

        # initialize session metadata
        self._init_metadata_namespace()

        # initialize session data
        self._init_data_namespace()

    def _start_connection(self):
        """Establish a database connection using the provided properties."""
        self.connection_factory = PooledConnectionFactory(properties=self.properties)
        self.connection = self.connection_factory.get_connection()

__init__(cid, properties={})

Initialize the Stream with a unique identifier and optional properties.

Source code in blue/stream.py

def __init__(self, cid, properties={}):
    """
    Initialize the Stream with a unique identifier and optional properties.
    """
    self.cid = cid
    self._initialize(properties=properties)

    self._start()

append_data(key, value)

Append data to a specific key in the stream.

Parameters:

Name	Type	Description	Default
key		The key to append to.	required
value		The value to append.	required

Source code in blue/stream.py

def append_data(self, key, value):
    """
    Append data to a specific key in the stream.

    Parameters:
        key: The key to append to.
        value: The value to append.
    """
    self.connection.json().arrappend(
        self._get_data_namespace(),
        "$." + key,
        value,
    )

get_all_data()

Get all data for the stream.

Returns:

Type	Description
dict	A dictionary containing all data in the stream.

Source code in blue/stream.py

def get_all_data(self):
    """
    Get all data for the stream.

    Returns:
        (dict): A dictionary containing all data in the stream.
    """
    value = self.connection.json().get(
        self._get_data_namespace(),
        Path("$"),
    )
    return self.__get_json_value(value)

get_data(key)

Get the data for a specific key in the stream.

Parameters:

Name	Type	Description	Default
key		The key to get.	required

Returns:

Type	Description
	The value associated with the key, or None if the key does not exist.

Source code in blue/stream.py

def get_data(self, key):
    """
    Get the data for a specific key in the stream.

    Parameters:
        key: The key to get.

    Returns:
        The value associated with the key, or None if the key does not exist.
    """
    value = self.connection.json().get(
        self._get_data_namespace(),
        Path("$." + key),
    )
    return self.__get_json_value(value)

get_data_len(key)

Get the length of the data array for a specific key in the stream.

Parameters:

Name	Type	Description	Default
key		The key to get the length for.	required

Returns: (int): The length of the data array, or 0 if the key does not exist.

Source code in blue/stream.py

def get_data_len(self, key):
    """
    Get the length of the data array for a specific key in the stream.

    Parameters:
        key: The key to get the length for.
    Returns:
        (int): The length of the data array, or 0 if the key does not exist.
    """
    return self.connection.json().arrlen(
        self._get_data_namespace(),
        Path("$." + key),
    )

get_metadata(key='')

Get the metadata for a specific key in the stream.

Parameters:

Name	Type	Description	Default
key		The key to get.	''

Returns: The value associated with the key, or None if the key does not exist.

Source code in blue/stream.py

def get_metadata(self, key=""):
    """
    Get the metadata for a specific key in the stream.

    Parameters:
        key: The key to get.
    Returns:
        The value associated with the key, or None if the key does not exist.
    """
    value = self.connection.json().get(
        self._get_metadata_namespace(),
        Path("$" + ("" if pydash.is_empty(key) else ".") + key),
    )
    return self.__get_json_value(value)

set_data(key, value)

Set the data for a specific key in the stream.

Parameters:

Name	Type	Description	Default
key		The key to set.	required
value		The value to set.	required

Source code in blue/stream.py

def set_data(self, key, value):
    """
    Set the data for a specific key in the stream.

    Parameters:
        key: The key to set.
        value: The value to set.
    """
    self.connection.json().set(
        self._get_data_namespace(),
        "$." + key,
        value,
    )

set_metadata(key, value, nx=False)

Set the metadata for a specific key in the stream.

Parameters:

Name	Type	Description	Default
key		The key to set.	required
value		The value to set.	required
nx	bool	If True, set the value only if it does not already exist.	False

Source code in blue/stream.py

def set_metadata(self, key, value, nx=False):
    """
    Set the metadata for a specific key in the stream.

    Parameters:
        key: The key to set.
        value: The value to set.
        nx (bool): If True, set the value only if it does not already exist.
    """
    self.connection.json().set(self._get_metadata_namespace(), "$." + key, value, nx=nx)

Last update: 2025-10-08