Skip to content

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 
1003
1004
1005
1006
1007
1008
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 
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
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 
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
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 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
@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 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
@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 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
@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 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
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 
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
@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 
125
126
127
128
129
130
131
132
133
@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 
135
136
137
138
139
140
141
142
143
144
145
146
@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 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
@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 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
@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 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
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 
35
36
37
38
39
40
41
42
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 
44
45
46
47
48
49
50
51
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 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
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 
61
62
63
64
65
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 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
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()