glide_sync

GlideClient

Bases: BaseClient, StandaloneCommands

Client used for connection to standalone servers. For full documentation, see https://glide.valkey.io/how-to/client-initialization/#standalone

Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/glide_client.py

class GlideClient(BaseClient, StandaloneCommands):
    """
    Client used for connection to standalone servers.
    For full documentation, see
    https://glide.valkey.io/how-to/client-initialization/#standalone
    """

GlideClusterClient

Bases: BaseClient, ClusterCommands

Client used for connection to cluster servers. For full documentation, see https://glide.valkey.io/how-to/client-initialization/#cluster

Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/glide_client.py

class GlideClusterClient(BaseClient, ClusterCommands):
    """
    Client used for connection to cluster servers.
    For full documentation, see
    https://glide.valkey.io/how-to/client-initialization/#cluster
    """

    def _build_cluster_scan_args(self, match, count, type, allow_non_covered_slots):
        args = []
        if match is not None:
            # Inline _encode_arg logic
            if isinstance(match, str):
                encoded_match = match.encode(ENCODING)
            else:
                encoded_match = match
            args.extend([b"MATCH", encoded_match])

        if count is not None:
            args.extend([b"COUNT", str(count).encode(ENCODING)])
        if type is not None:
            args.extend([b"TYPE", type.value.encode(ENCODING)])
        if allow_non_covered_slots:
            args.extend([b"ALLOW_NON_COVERED_SLOTS"])

        return args

    def _cluster_scan(
        self,
        cursor: ClusterScanCursor,
        match: Optional[TEncodable] = None,
        count: Optional[int] = None,
        type: Optional[ObjectType] = None,
        allow_non_covered_slots: bool = False,
    ) -> List[Union[ClusterScanCursor, List[bytes]]]:
        if self._is_closed:
            raise ClosingError(
                "Unable to execute requests; the client is closed. Please create a new client."
            )

        client_adapter_ptr = self._core_client
        if client_adapter_ptr == self._ffi.NULL:
            raise ValueError("Invalid client pointer.")

        # Use helper method to build args
        args = self._build_cluster_scan_args(
            match, count, type, allow_non_covered_slots
        )
        # Convert cursor to C string
        cursor_string = cursor.get_cursor()
        cursor_bytes = cursor_string.encode(ENCODING) + b"\0"  # Null terminate for C

        # Keep references to prevent GC
        temp_buffers: List[Any] = [cursor_bytes]
        cursor_buffer = self._ffi.from_buffer(cursor_bytes)

        # Prepare FFI arguments
        if args:
            args_array, args_len_array, arg_buffers = self._to_c_strings(args)
            temp_buffers.extend(arg_buffers)  # Keep references alive
            arg_count = len(args)
        else:
            args_array = self._ffi.NULL
            args_len_array = self._ffi.NULL
            arg_count = 0

        result_ptr = self._lib.request_cluster_scan(
            client_adapter_ptr,
            0,
            cursor_buffer,
            arg_count,
            args_array,
            args_len_array,
        )

        response_data = self._handle_cmd_result(result_ptr)

        if not isinstance(response_data, list) or len(response_data) != 2:
            raise RequestError("Unexpected cluster scan response format")

        new_cursor = response_data[0]
        if isinstance(new_cursor, bytes):
            new_cursor = new_cursor.decode(ENCODING)

        keys_list = response_data[1] if response_data[1] is not None else []

        return [ClusterScanCursor(new_cursor), keys_list]

Logger

A singleton class that allows logging which is consistent with logs from the internal GLIDE core. The logger can be set up in 2 ways - 1. By calling Logger.init, which configures the logger only if it wasn't previously configured. 2. By calling Logger.set_logger_config, which replaces the existing configuration, and means that new logs will not be saved with the logs that were sent before the call. If none of these functions are called, the first log attempt will initialize a new logger with default configuration.

Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/logger.py

class Logger:
    """
    A singleton class that allows logging which is consistent with logs from the internal GLIDE core.
    The logger can be set up in 2 ways -
    1. By calling Logger.init, which configures the logger only if it wasn't previously configured.
    2. By calling Logger.set_logger_config, which replaces the existing configuration, and means that new logs will not be
    saved with the logs that were sent before the call.
    If none of these functions are called, the first log attempt will initialize a new logger with default configuration.
    """

    _instance: Logger | None = None
    _glide_ffi = _GlideFFI()
    _ffi = _glide_ffi.ffi
    _lib = _glide_ffi.lib
    logger_level: Level = Level.OFF

    def __init__(self, level: Optional[Level] = None, file_name: Optional[str] = None):
        c_level = (
            Logger._ffi.new("Level*", level.value)
            if level is not None
            else Logger._ffi.NULL
        )
        c_file_name = (
            Logger._ffi.new("char[]", file_name.encode(ENCODING))
            if file_name
            else Logger._ffi.NULL
        )

        result_ptr = Logger._lib.init(c_level, c_file_name)

        if result_ptr != Logger._ffi.NULL:
            try:
                if result_ptr.log_error != Logger._ffi.NULL:
                    error_str = cast(
                        bytes, Logger._ffi.string(result_ptr.log_error)
                    ).decode(ENCODING)
                    raise LoggerError(f"Logger initialization failed: {error_str}")
                else:
                    Logger.logger_level = Level(result_ptr.level)

            finally:
                Logger._lib.free_log_result(result_ptr)
        else:
            raise LoggerError("Logger init received a null pointer")

    @classmethod
    def init(cls, level: Optional[Level] = None, file_name: Optional[str] = None):
        """
        Initialize a logger if it wasn't initialized before - this method is meant to be used when there is no intention to
        replace an existing logger. Otherwise, use `set_logger_config` for overriding the existing logger configs.
        The logger will filter all logs with a level lower than the given level.
        If given a file_name argument, will write the logs to files postfixed with file_name. If file_name isn't provided,
        the logs will be written to the console.

        Args:
            level (Optional[Level]): Set the logger level to one of [ERROR, WARN, INFO, DEBUG, TRACE, OFF].
                If log level isn't provided, the logger will be configured with default configuration.
                To turn off logging completely, set the level to Level.OFF.
            file_name (Optional[str]): If provided the target of the logs will be the file mentioned.
                Otherwise, logs will be printed to the console.
        """
        if cls._instance is None:
            cls._instance = cls(level, file_name)

    @classmethod
    def log(
        cls,
        log_level: Level,
        log_identifier: str,
        message: str,
        err: Optional[Exception] = None,
    ):
        """
        Logs the provided message if the provided log level is lower then the logger level.

        Args:
            log_level (Level): The log level of the provided message.
            log_identifier (str): The log identifier should give the log a context.
            message (str): The message to log.
            err (Optional[Exception]): The exception or error to log.
        """
        if not cls._instance:
            cls._instance = cls(None)
        if log_level.value > Logger.logger_level.value:
            return
        if err:
            message = f"{message}: {traceback.format_exception(type(err), err, err.__traceback__)}"
        c_identifier = Logger._ffi.new("char[]", log_identifier.encode(ENCODING))
        c_message = Logger._ffi.new("char[]", message.encode(ENCODING))
        result_ptr = Logger._lib.glide_log(log_level.value, c_identifier, c_message)

        if result_ptr != Logger._ffi.NULL:
            try:
                if result_ptr.log_error != Logger._ffi.NULL:
                    error_str = cast(
                        bytes, Logger._ffi.string(result_ptr.log_error)
                    ).decode(ENCODING)

                    # If the log failed due to invalid provided identifier or message,
                    # Log the FFI log error using logger_core directly
                    error_identifier = Logger._ffi.new(
                        "char[]", "Logger".encode(ENCODING)
                    )
                    error_message = Logger._ffi.new(
                        "char[]", f"log error: {error_str}".encode(ENCODING)
                    )
                    error_result_ptr = Logger._lib.glide_log(
                        Level.ERROR.value, error_identifier, error_message
                    )
                    if error_result_ptr != Logger._ffi.NULL:
                        Logger._lib.free_log_result(error_result_ptr)

            finally:
                Logger._lib.free_log_result(result_ptr)

        else:
            # Log the null pointer error using logger_core directly
            error_identifier = Logger._ffi.new("char[]", "Logger".encode(ENCODING))
            error_message = Logger._ffi.new(
                "char[]", "Log function returned a null pointer".encode(ENCODING)
            )
            error_result_ptr = Logger._lib.glide_log(
                Level.ERROR.value, error_identifier, error_message
            )
            if error_result_ptr != Logger._ffi.NULL:
                Logger._lib.free_log_result(error_result_ptr)

    @classmethod
    def set_logger_config(
        cls, level: Optional[Level] = None, file_name: Optional[str] = None
    ):
        """
        Creates a new logger instance and configure it with the provided log level and file name.

        Args:
            level (Optional[Level]): Set the logger level to one of [ERROR, WARN, INFO, DEBUG, TRACE, OFF].
                If log level isn't provided, the logger will be configured with default configuration.
                To turn off logging completely, set the level to OFF.
            file_name (Optional[str]): If provided the target of the logs will be the file mentioned.
                Otherwise, logs will be printed to the console.
        """
        Logger._instance = cls(level, file_name)

init(level=None, file_name=None) classmethod

Initialize a logger if it wasn't initialized before - this method is meant to be used when there is no intention to replace an existing logger. Otherwise, use set_logger_config for overriding the existing logger configs. The logger will filter all logs with a level lower than the given level. If given a file_name argument, will write the logs to files postfixed with file_name. If file_name isn't provided, the logs will be written to the console.

Parameters:

Name	Type	Description	Default
level	Optional[Level]	Set the logger level to one of [ERROR, WARN, INFO, DEBUG, TRACE, OFF]. If log level isn't provided, the logger will be configured with default configuration. To turn off logging completely, set the level to Level.OFF.	None
file_name	Optional[str]	If provided the target of the logs will be the file mentioned. Otherwise, logs will be printed to the console.	None

Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/logger.py

@classmethod
def init(cls, level: Optional[Level] = None, file_name: Optional[str] = None):
    """
    Initialize a logger if it wasn't initialized before - this method is meant to be used when there is no intention to
    replace an existing logger. Otherwise, use `set_logger_config` for overriding the existing logger configs.
    The logger will filter all logs with a level lower than the given level.
    If given a file_name argument, will write the logs to files postfixed with file_name. If file_name isn't provided,
    the logs will be written to the console.

    Args:
        level (Optional[Level]): Set the logger level to one of [ERROR, WARN, INFO, DEBUG, TRACE, OFF].
            If log level isn't provided, the logger will be configured with default configuration.
            To turn off logging completely, set the level to Level.OFF.
        file_name (Optional[str]): If provided the target of the logs will be the file mentioned.
            Otherwise, logs will be printed to the console.
    """
    if cls._instance is None:
        cls._instance = cls(level, file_name)

log(log_level, log_identifier, message, err=None) classmethod

Logs the provided message if the provided log level is lower then the logger level.

Parameters:

Name	Type	Description	Default
log_level	Level	The log level of the provided message.	required
log_identifier	str	The log identifier should give the log a context.	required
message	str	The message to log.	required
err	Optional[Exception]	The exception or error to log.	None

Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/logger.py

@classmethod
def log(
    cls,
    log_level: Level,
    log_identifier: str,
    message: str,
    err: Optional[Exception] = None,
):
    """
    Logs the provided message if the provided log level is lower then the logger level.

    Args:
        log_level (Level): The log level of the provided message.
        log_identifier (str): The log identifier should give the log a context.
        message (str): The message to log.
        err (Optional[Exception]): The exception or error to log.
    """
    if not cls._instance:
        cls._instance = cls(None)
    if log_level.value > Logger.logger_level.value:
        return
    if err:
        message = f"{message}: {traceback.format_exception(type(err), err, err.__traceback__)}"
    c_identifier = Logger._ffi.new("char[]", log_identifier.encode(ENCODING))
    c_message = Logger._ffi.new("char[]", message.encode(ENCODING))
    result_ptr = Logger._lib.glide_log(log_level.value, c_identifier, c_message)

    if result_ptr != Logger._ffi.NULL:
        try:
            if result_ptr.log_error != Logger._ffi.NULL:
                error_str = cast(
                    bytes, Logger._ffi.string(result_ptr.log_error)
                ).decode(ENCODING)

                # If the log failed due to invalid provided identifier or message,
                # Log the FFI log error using logger_core directly
                error_identifier = Logger._ffi.new(
                    "char[]", "Logger".encode(ENCODING)
                )
                error_message = Logger._ffi.new(
                    "char[]", f"log error: {error_str}".encode(ENCODING)
                )
                error_result_ptr = Logger._lib.glide_log(
                    Level.ERROR.value, error_identifier, error_message
                )
                if error_result_ptr != Logger._ffi.NULL:
                    Logger._lib.free_log_result(error_result_ptr)

        finally:
            Logger._lib.free_log_result(result_ptr)

    else:
        # Log the null pointer error using logger_core directly
        error_identifier = Logger._ffi.new("char[]", "Logger".encode(ENCODING))
        error_message = Logger._ffi.new(
            "char[]", "Log function returned a null pointer".encode(ENCODING)
        )
        error_result_ptr = Logger._lib.glide_log(
            Level.ERROR.value, error_identifier, error_message
        )
        if error_result_ptr != Logger._ffi.NULL:
            Logger._lib.free_log_result(error_result_ptr)

set_logger_config(level=None, file_name=None) classmethod

Creates a new logger instance and configure it with the provided log level and file name.

Parameters:

Name	Type	Description	Default
level	Optional[Level]	Set the logger level to one of [ERROR, WARN, INFO, DEBUG, TRACE, OFF]. If log level isn't provided, the logger will be configured with default configuration. To turn off logging completely, set the level to OFF.	None
file_name	Optional[str]	If provided the target of the logs will be the file mentioned. Otherwise, logs will be printed to the console.	None

Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/logger.py

@classmethod
def set_logger_config(
    cls, level: Optional[Level] = None, file_name: Optional[str] = None
):
    """
    Creates a new logger instance and configure it with the provided log level and file name.

    Args:
        level (Optional[Level]): Set the logger level to one of [ERROR, WARN, INFO, DEBUG, TRACE, OFF].
            If log level isn't provided, the logger will be configured with default configuration.
            To turn off logging completely, set the level to OFF.
        file_name (Optional[str]): If provided the target of the logs will be the file mentioned.
            Otherwise, logs will be printed to the console.
    """
    Logger._instance = cls(level, file_name)

OpenTelemetry

Singleton class for managing OpenTelemetry configuration and operations.

This class provides a centralized way to initialize OpenTelemetry and control sampling behavior at runtime.

Example usage

from glide_sync import OpenTelemetry, OpenTelemetryConfig, OpenTelemetryTracesConfig, OpenTelemetryMetricsConfig

OpenTelemetry.init(OpenTelemetryConfig(
    traces=OpenTelemetryTracesConfig(
        endpoint="http://localhost:4318/v1/traces",
        sample_percentage=10  # Optional, defaults to 1. Can also be changed at runtime via set_sample_percentage().
    ),
    metrics=OpenTelemetryMetricsConfig(
        endpoint="http://localhost:4318/v1/metrics"
    ),
    flush_interval_ms=1000  # Optional, defaults to 5000
))

Note

OpenTelemetry can only be initialized once per process. Subsequent calls to init() will be ignored. This is by design, as OpenTelemetry is a global resource that should be configured once at application startup.

Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/opentelemetry.py

class OpenTelemetry:
    """
    Singleton class for managing OpenTelemetry configuration and operations.

    This class provides a centralized way to initialize OpenTelemetry and control
    sampling behavior at runtime.

    Example usage:
        ```python
        from glide_sync import OpenTelemetry, OpenTelemetryConfig, OpenTelemetryTracesConfig, OpenTelemetryMetricsConfig

        OpenTelemetry.init(OpenTelemetryConfig(
            traces=OpenTelemetryTracesConfig(
                endpoint="http://localhost:4318/v1/traces",
                sample_percentage=10  # Optional, defaults to 1. Can also be changed at runtime via set_sample_percentage().
            ),
            metrics=OpenTelemetryMetricsConfig(
                endpoint="http://localhost:4318/v1/metrics"
            ),
            flush_interval_ms=1000  # Optional, defaults to 5000
        ))
        ```

    Note:
        OpenTelemetry can only be initialized once per process. Subsequent calls to
        init() will be ignored. This is by design, as OpenTelemetry is a global
        resource that should be configured once at application startup.
    """

    _instance: Optional["OpenTelemetry"] = None
    _config: Optional[OpenTelemetryConfig] = None

    @classmethod
    def init(cls, config: OpenTelemetryConfig) -> None:
        """
        Initialize the OpenTelemetry instance.

        Args:
            config: The OpenTelemetry configuration

        Note:
            OpenTelemetry can only be initialized once per process.
            Subsequent calls will be ignored and a warning will be logged.
        """
        if not cls._instance:
            cls._config = config
            ffi = GlideFFI.ffi
            lib = GlideFFI.lib

            # Build FFI config struct - keep string references alive
            traces_ptr = ffi.NULL
            traces_endpoint_cstr = None
            if config.traces:
                traces_endpoint_cstr = ffi.new(
                    "char[]", config.traces.endpoint.encode()
                )
                traces_config = ffi.new("OpenTelemetryTracesConfig*")
                traces_config.endpoint = traces_endpoint_cstr
                traces_config.has_sample_percentage = True
                traces_config.sample_percentage = config.traces.sample_percentage
                traces_ptr = traces_config

            metrics_ptr = ffi.NULL
            metrics_endpoint_cstr = None
            if config.metrics:
                metrics_endpoint_cstr = ffi.new(
                    "char[]", config.metrics.endpoint.encode()
                )
                metrics_config = ffi.new("OpenTelemetryMetricsConfig*")
                metrics_config.endpoint = metrics_endpoint_cstr
                metrics_ptr = metrics_config

            otel_config = ffi.new("OpenTelemetryConfig*")
            otel_config.traces = traces_ptr
            otel_config.metrics = metrics_ptr
            otel_config.has_flush_interval_ms = config.flush_interval_ms is not None
            otel_config.flush_interval_ms = (
                config.flush_interval_ms if config.flush_interval_ms else 0
            )

            error = lib.init_open_telemetry(otel_config)
            if error != ffi.NULL:
                error_msg = ffi.string(error).decode()
                lib.free_c_string(error)
                raise ConfigurationError(
                    f"Failed to initialize OpenTelemetry: {error_msg}"
                )

            cls._instance = OpenTelemetry()
            Logger.log(
                Level.INFO,
                "GlideOpenTelemetry",
                "OpenTelemetry initialized successfully",
            )
            return

        Logger.log(
            Level.WARN,
            "GlideOpenTelemetry",
            "OpenTelemetry already initialized - ignoring new configuration",
        )

    @classmethod
    def is_initialized(cls) -> bool:
        """
        Check if the OpenTelemetry instance is initialized.

        Returns:
            bool: True if the OpenTelemetry instance is initialized, False otherwise
        """
        return cls._instance is not None

    @classmethod
    def get_sample_percentage(cls) -> Optional[int]:
        """
        Get the sample percentage for traces.

        Returns:
            Optional[int]: The sample percentage for traces only if OpenTelemetry is initialized
                and the traces config is set, otherwise None.
        """
        if cls._config and cls._config.traces:
            return cls._config.traces.sample_percentage
        return None

    @classmethod
    def should_sample(cls) -> bool:
        """
        Determines if the current request should be sampled for OpenTelemetry tracing.
        Uses the configured sample percentage to randomly decide whether to create a span for this request.

        Returns:
            bool: True if the request should be sampled, False otherwise
        """
        percentage = cls.get_sample_percentage()
        return (
            cls.is_initialized()
            and percentage is not None
            and random.random() * 100 < percentage
        )

    @classmethod
    def set_sample_percentage(cls, percentage: int) -> None:
        """
        Set the percentage of requests to be sampled and traced. Must be a value between 0 and 100.
        This setting only affects traces, not metrics.

        Args:
            percentage: The sample percentage 0-100

        Raises:
            ConfigurationError: If OpenTelemetry is not initialized or traces config is not set

        Remarks:
            This method can be called at runtime to change the sampling percentage
            without reinitializing OpenTelemetry.
        """
        if not cls._config or not cls._config.traces:
            raise ConfigurationError("OpenTelemetry traces not initialized")

        if percentage < 0 or percentage > 100:
            raise ConfigurationError("Sample percentage must be between 0 and 100")

        # Update the shared config directly
        cls._config.traces.sample_percentage = percentage

init(config) classmethod

Initialize the OpenTelemetry instance.

Parameters:

Name	Type	Description	Default
config	OpenTelemetryConfig	The OpenTelemetry configuration	required

Note

OpenTelemetry can only be initialized once per process. Subsequent calls will be ignored and a warning will be logged.

Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/opentelemetry.py

@classmethod
def init(cls, config: OpenTelemetryConfig) -> None:
    """
    Initialize the OpenTelemetry instance.

    Args:
        config: The OpenTelemetry configuration

    Note:
        OpenTelemetry can only be initialized once per process.
        Subsequent calls will be ignored and a warning will be logged.
    """
    if not cls._instance:
        cls._config = config
        ffi = GlideFFI.ffi
        lib = GlideFFI.lib

        # Build FFI config struct - keep string references alive
        traces_ptr = ffi.NULL
        traces_endpoint_cstr = None
        if config.traces:
            traces_endpoint_cstr = ffi.new(
                "char[]", config.traces.endpoint.encode()
            )
            traces_config = ffi.new("OpenTelemetryTracesConfig*")
            traces_config.endpoint = traces_endpoint_cstr
            traces_config.has_sample_percentage = True
            traces_config.sample_percentage = config.traces.sample_percentage
            traces_ptr = traces_config

        metrics_ptr = ffi.NULL
        metrics_endpoint_cstr = None
        if config.metrics:
            metrics_endpoint_cstr = ffi.new(
                "char[]", config.metrics.endpoint.encode()
            )
            metrics_config = ffi.new("OpenTelemetryMetricsConfig*")
            metrics_config.endpoint = metrics_endpoint_cstr
            metrics_ptr = metrics_config

        otel_config = ffi.new("OpenTelemetryConfig*")
        otel_config.traces = traces_ptr
        otel_config.metrics = metrics_ptr
        otel_config.has_flush_interval_ms = config.flush_interval_ms is not None
        otel_config.flush_interval_ms = (
            config.flush_interval_ms if config.flush_interval_ms else 0
        )

        error = lib.init_open_telemetry(otel_config)
        if error != ffi.NULL:
            error_msg = ffi.string(error).decode()
            lib.free_c_string(error)
            raise ConfigurationError(
                f"Failed to initialize OpenTelemetry: {error_msg}"
            )

        cls._instance = OpenTelemetry()
        Logger.log(
            Level.INFO,
            "GlideOpenTelemetry",
            "OpenTelemetry initialized successfully",
        )
        return

    Logger.log(
        Level.WARN,
        "GlideOpenTelemetry",
        "OpenTelemetry already initialized - ignoring new configuration",
    )

is_initialized() classmethod

Check if the OpenTelemetry instance is initialized.

Returns:

Name	Type	Description
bool	bool	True if the OpenTelemetry instance is initialized, False otherwise

Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/opentelemetry.py

@classmethod
def is_initialized(cls) -> bool:
    """
    Check if the OpenTelemetry instance is initialized.

    Returns:
        bool: True if the OpenTelemetry instance is initialized, False otherwise
    """
    return cls._instance is not None

get_sample_percentage() classmethod

Get the sample percentage for traces.

Returns:

Type	Description
Optional[int]	Optional[int]: The sample percentage for traces only if OpenTelemetry is initialized and the traces config is set, otherwise None.

Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/opentelemetry.py

@classmethod
def get_sample_percentage(cls) -> Optional[int]:
    """
    Get the sample percentage for traces.

    Returns:
        Optional[int]: The sample percentage for traces only if OpenTelemetry is initialized
            and the traces config is set, otherwise None.
    """
    if cls._config and cls._config.traces:
        return cls._config.traces.sample_percentage
    return None

should_sample() classmethod

Determines if the current request should be sampled for OpenTelemetry tracing. Uses the configured sample percentage to randomly decide whether to create a span for this request.

Returns:

Name	Type	Description
bool	bool	True if the request should be sampled, False otherwise

Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/opentelemetry.py

@classmethod
def should_sample(cls) -> bool:
    """
    Determines if the current request should be sampled for OpenTelemetry tracing.
    Uses the configured sample percentage to randomly decide whether to create a span for this request.

    Returns:
        bool: True if the request should be sampled, False otherwise
    """
    percentage = cls.get_sample_percentage()
    return (
        cls.is_initialized()
        and percentage is not None
        and random.random() * 100 < percentage
    )

set_sample_percentage(percentage) classmethod

Set the percentage of requests to be sampled and traced. Must be a value between 0 and 100. This setting only affects traces, not metrics.

Parameters:

Name	Type	Description	Default
percentage	int	The sample percentage 0-100	required

Raises:

Type	Description
ConfigurationError	If OpenTelemetry is not initialized or traces config is not set

Remarks

This method can be called at runtime to change the sampling percentage without reinitializing OpenTelemetry.

Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/opentelemetry.py

@classmethod
def set_sample_percentage(cls, percentage: int) -> None:
    """
    Set the percentage of requests to be sampled and traced. Must be a value between 0 and 100.
    This setting only affects traces, not metrics.

    Args:
        percentage: The sample percentage 0-100

    Raises:
        ConfigurationError: If OpenTelemetry is not initialized or traces config is not set

    Remarks:
        This method can be called at runtime to change the sampling percentage
        without reinitializing OpenTelemetry.
    """
    if not cls._config or not cls._config.traces:
        raise ConfigurationError("OpenTelemetry traces not initialized")

    if percentage < 0 or percentage > 100:
        raise ConfigurationError("Sample percentage must be between 0 and 100")

    # Update the shared config directly
    cls._config.traces.sample_percentage = percentage

ClusterScanCursor

Represents a cursor for cluster scan operations.

This class manages the state of a cluster scan cursor and automatically cleans up resources when the cursor is no longer needed.

Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/sync_commands/cluster_scan_cursor.py

class ClusterScanCursor:
    """
    Represents a cursor for cluster scan operations.

    This class manages the state of a cluster scan cursor and automatically
    cleans up resources when the cursor is no longer needed.
    """

    def __init__(self, new_cursor: Optional[str] = None):
        """
        Create a new ClusterScanCursor instance.

        Args:
            new_cursor: Optional cursor string to initialize with. If None,
                       creates a cursor in the initial state.
        """
        _glide_ffi = _GlideFFI()
        self._ffi = _glide_ffi.ffi
        self._lib = _glide_ffi.lib

        self._cursor = new_cursor or "0"

    def get_cursor(self) -> str:
        """
        Get the current cursor string.

        Returns:
            str: The current cursor string
        """
        return self._cursor

    def is_finished(self) -> bool:
        """
        Check if the cluster scan is finished.

        Returns:
            bool: True if the scan is finished, False otherwise
        """
        return self._cursor == FINISHED_SCAN_CURSOR

    def __del__(self):
        """
        Clean up cluster scan cursor resources when the object is garbage collected.
        """
        try:
            if hasattr(self, "_cursor") and self._cursor:
                cursor_bytes = self._cursor.encode(ENCODING) + b"\0"
                cursor_buffer = self._ffi.from_buffer(cursor_bytes)

                self._lib.remove_cluster_scan_cursor(cursor_buffer)

        except Exception as e:
            Logger.log(
                Level.ERROR,
                "cluster_scan_cursor",
                f"Error during cluster scan cursor cleanup: {e}",
            )

get_cursor()

Get the current cursor string.

Returns:

Name	Type	Description
str	str	The current cursor string

Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/sync_commands/cluster_scan_cursor.py

def get_cursor(self) -> str:
    """
    Get the current cursor string.

    Returns:
        str: The current cursor string
    """
    return self._cursor

is_finished()

Check if the cluster scan is finished.

Returns:

Name	Type	Description
bool	bool	True if the scan is finished, False otherwise

Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/sync_commands/cluster_scan_cursor.py

def is_finished(self) -> bool:
    """
    Check if the cluster scan is finished.

    Returns:
        bool: True if the scan is finished, False otherwise
    """
    return self._cursor == FINISHED_SCAN_CURSOR

Script

A wrapper for a Script object for the invoke script function. As long as this object is alive, the script's code is saved in memory, and can be resent to the server.

Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/sync_commands/script.py

class Script:
    """
    A wrapper for a Script object for the `invoke script` function. As long as
    this object is alive, the script's code is saved in memory, and can be resent to the server.
    """

    def __init__(self, code: TEncodable):
        """
        Create a new Script instance and store it in the script cache.

        Args:
            code: The Lua script code as a string or bytes

        Raises:
            TypeError: If code is not a string or bytes
            RequestError: If script storage fails
        """
        _glide_ffi = _GlideFFI()
        self._ffi = _glide_ffi.ffi
        self._lib = _glide_ffi.lib
        self._hash = None

        # Convert code to bytes if it's a string
        if isinstance(code, str):
            script_bytes = code.encode(ENCODING)
        elif isinstance(code, bytes):
            script_bytes = code
        else:
            raise TypeError("code must be either a string or bytes")

        # Store script in cache
        script_ptr = self._ffi.from_buffer(script_bytes)
        hash_buffer_ptr = self._lib.store_script(script_ptr, len(script_bytes))

        if hash_buffer_ptr == self._ffi.NULL:
            raise RequestError(
                "Failed to store script - got null pointer from the `store script` function"
            )

        try:
            hash_buffer = self._ffi.cast("ScriptHashBuffer*", hash_buffer_ptr)

            # Extract the hash string from the buffer
            hash_bytes = self._ffi.buffer(hash_buffer.ptr, hash_buffer.len)[:]
            self.hash = hash_bytes.decode(ENCODING)
        finally:
            self._lib.free_script_hash_buffer(hash_buffer_ptr)

    def get_hash(self) -> str:
        """
        Get the SHA1 hash of the script.
        """
        return self.hash

    def __del__(self):
        """
        Clean up the script from the cache when the object is garbage collected.
        """
        try:
            hash_bytes = self.hash.encode(ENCODING)
            hash_buffer = self._ffi.from_buffer(hash_bytes)
            error_ptr = self._lib.drop_script(hash_buffer, len(hash_bytes))

            if error_ptr != self._ffi.NULL:
                try:
                    error_msg = self._ffi.string(error_ptr).decode(ENCODING)
                    Logger.log(
                        Level.WARN, "script, " f"Error dropping script: {error_msg}"
                    )
                finally:
                    self._lib.free_drop_script_error(error_ptr)

        except Exception as e:
            Logger.log(Level.WARN, "script", f"Error during script cleanup: {e}")

get_hash()

Get the SHA1 hash of the script.

Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/sync_commands/script.py

def get_hash(self) -> str:
    """
    Get the SHA1 hash of the script.
    """
    return self.hash

get_min_compressed_size()

Get the minimum compression size in bytes.

Returns the minimum size threshold for compression. Values smaller than this will not be compressed.

Returns:

Name	Type	Description
int	int	The minimum compression size in bytes (currently 6 bytes: 5-byte header + 1 byte data)

Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/sync_commands/utils.py

def get_min_compressed_size() -> int:
    """
    Get the minimum compression size in bytes.

    Returns the minimum size threshold for compression. Values smaller than this
    will not be compressed.

    Returns:
        int: The minimum compression size in bytes (currently 6 bytes: 5-byte header + 1 byte data)
    """
    _glide_ffi = _GlideFFI()
    return _glide_ffi.lib.get_min_compressed_size()