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 
1090
1091
1092
1093
1094
1095
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 
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
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]

MonitorClient

A client that streams all commands processed by the server via the MONITOR command.

Must be used with a standalone (non-cluster) configuration.

Warning: MONITOR is a debugging tool with performance implications. Do not use in production environments.

Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/monitor_client.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
 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
class MonitorClient:
    """
    A client that streams all commands processed by the server via the MONITOR command.

    Must be used with a standalone (non-cluster) configuration.

    Warning: MONITOR is a debugging tool with performance implications.
    Do not use in production environments.
    """

    def __init__(self) -> None:
        self._ffi = GlideFFI.ffi
        self._lib = GlideFFI.lib
        self._core_client = self._ffi.NULL
        self._callback_ref = None
        self._queue: List[MonitorMsg] = []
        self._lock = threading.Lock()
        self._condition = threading.Condition(self._lock)
        self._is_closed = False
        self._user_callback: Optional[Callable[[MonitorMsg], None]] = None

    @classmethod
    def create(
        cls,
        config: GlideClientConfiguration,
        callback: Optional[Callable[[MonitorMsg], None]] = None,
    ) -> "MonitorClient":
        """
        Create a new MonitorClient connected to the server.

        Args:
            config: Standalone client configuration (must be GlideClientConfiguration).
            callback: Optional callback invoked for each MonitorMsg. If None, messages
                      are queued and retrievable via get_monitor_message().

        Returns:
            A MonitorClient instance.
        """
        if not isinstance(config, GlideClientConfiguration):
            raise TypeError(
                "MonitorClient requires a GlideClientConfiguration (standalone only)"
            )
        instance = cls()
        instance._user_callback = callback
        conn_req = config._create_a_protobuf_conn_request(cluster_mode=False)
        conn_req_bytes = conn_req.SerializeToString()

        @instance._ffi.callback("MonitorCallback")
        def _monitor_callback(
            client_ptr,
            timestamp,
            db,
            client_addr_ptr,
            client_addr_len,
            command_ptr,
            command_len,
            args_json_ptr,
            args_json_len,
        ):
            try:
                client_addr = bytes(
                    instance._ffi.buffer(client_addr_ptr, client_addr_len)
                ).decode("utf-8", errors="replace")
                command = bytes(instance._ffi.buffer(command_ptr, command_len)).decode(
                    "utf-8", errors="replace"
                )
                args_json_str = bytes(
                    instance._ffi.buffer(args_json_ptr, args_json_len)
                ).decode("utf-8", errors="replace")
                try:
                    args: List[str] = (
                        json.loads(args_json_str) if args_json_len > 0 else []
                    )
                except (json.JSONDecodeError, ValueError):
                    args = []
                msg = MonitorMsg(
                    timestamp=timestamp,
                    db=db,
                    client_addr=client_addr,
                    command=command,
                    args=args,
                )
                if instance._user_callback is not None:
                    instance._user_callback(msg)
                else:
                    with instance._condition:
                        instance._queue.append(msg)
                        instance._condition.notify()
            except Exception:
                pass  # Suppress callback errors to avoid crashing the Rust FFI layer

        instance._callback_ref = _monitor_callback
        client_response = instance._lib.create_monitor_client(
            conn_req_bytes, len(conn_req_bytes), _monitor_callback
        )
        if client_response == instance._ffi.NULL:
            raise RuntimeError("Failed to create monitor client: null response")
        if client_response.connection_error_message != instance._ffi.NULL:
            error = instance._ffi.string(
                client_response.connection_error_message
            ).decode()
            instance._lib.free_connection_response(client_response)
            raise RuntimeError(f"Failed to create monitor client: {error}")
        instance._core_client = client_response.conn_ptr
        instance._lib.free_connection_response(client_response)
        return instance

    def get_monitor_message(
        self, timeout: Optional[float] = None
    ) -> Optional[MonitorMsg]:
        """
        Block until a MonitorMsg is available, then return it.

        Args:
            timeout: Optional timeout in seconds. Returns None on timeout or if closed.
        """
        deadline = time.monotonic() + timeout if timeout is not None else None
        with self._condition:
            while not self._queue and not self._is_closed:
                remaining = None
                if deadline is not None:
                    remaining = deadline - time.monotonic()
                    if remaining <= 0:
                        break
                self._condition.wait(timeout=remaining)
            return self._queue.pop(0) if self._queue else None

    def try_get_monitor_message(self) -> Optional[MonitorMsg]:
        """Non-blocking retrieval. Returns None if no message is available."""
        with self._condition:
            return self._queue.pop(0) if self._queue else None

    def close(self) -> None:
        """Stop monitoring and release resources."""
        with self._condition:
            if self._is_closed:
                return
            self._is_closed = True
            self._condition.notify_all()
        if self._core_client != self._ffi.NULL:
            client = self._core_client
            self._core_client = self._ffi.NULL
            self._lib.close_monitor_client(client)
        self._callback_ref = None

    def stop(self) -> None:
        """Alias for close(). Stop monitoring and release resources."""
        self.close()

    def __enter__(self) -> "MonitorClient":
        return self

    def __exit__(self, *args) -> None:
        self.close()

create(config, callback=None) classmethod

Create a new MonitorClient connected to the server.

Parameters:

Name Type Description Default
config GlideClientConfiguration

Standalone client configuration (must be GlideClientConfiguration).

 required
callback Optional[Callable[[MonitorMsg], None]]

Optional callback invoked for each MonitorMsg. If None, messages are queued and retrievable via get_monitor_message().

 None

Returns:

Type Description
MonitorClient

A MonitorClient instance.
Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/monitor_client.py 
 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
@classmethod
def create(
    cls,
    config: GlideClientConfiguration,
    callback: Optional[Callable[[MonitorMsg], None]] = None,
) -> "MonitorClient":
    """
    Create a new MonitorClient connected to the server.

    Args:
        config: Standalone client configuration (must be GlideClientConfiguration).
        callback: Optional callback invoked for each MonitorMsg. If None, messages
                  are queued and retrievable via get_monitor_message().

    Returns:
        A MonitorClient instance.
    """
    if not isinstance(config, GlideClientConfiguration):
        raise TypeError(
            "MonitorClient requires a GlideClientConfiguration (standalone only)"
        )
    instance = cls()
    instance._user_callback = callback
    conn_req = config._create_a_protobuf_conn_request(cluster_mode=False)
    conn_req_bytes = conn_req.SerializeToString()

    @instance._ffi.callback("MonitorCallback")
    def _monitor_callback(
        client_ptr,
        timestamp,
        db,
        client_addr_ptr,
        client_addr_len,
        command_ptr,
        command_len,
        args_json_ptr,
        args_json_len,
    ):
        try:
            client_addr = bytes(
                instance._ffi.buffer(client_addr_ptr, client_addr_len)
            ).decode("utf-8", errors="replace")
            command = bytes(instance._ffi.buffer(command_ptr, command_len)).decode(
                "utf-8", errors="replace"
            )
            args_json_str = bytes(
                instance._ffi.buffer(args_json_ptr, args_json_len)
            ).decode("utf-8", errors="replace")
            try:
                args: List[str] = (
                    json.loads(args_json_str) if args_json_len > 0 else []
                )
            except (json.JSONDecodeError, ValueError):
                args = []
            msg = MonitorMsg(
                timestamp=timestamp,
                db=db,
                client_addr=client_addr,
                command=command,
                args=args,
            )
            if instance._user_callback is not None:
                instance._user_callback(msg)
            else:
                with instance._condition:
                    instance._queue.append(msg)
                    instance._condition.notify()
        except Exception:
            pass  # Suppress callback errors to avoid crashing the Rust FFI layer

    instance._callback_ref = _monitor_callback
    client_response = instance._lib.create_monitor_client(
        conn_req_bytes, len(conn_req_bytes), _monitor_callback
    )
    if client_response == instance._ffi.NULL:
        raise RuntimeError("Failed to create monitor client: null response")
    if client_response.connection_error_message != instance._ffi.NULL:
        error = instance._ffi.string(
            client_response.connection_error_message
        ).decode()
        instance._lib.free_connection_response(client_response)
        raise RuntimeError(f"Failed to create monitor client: {error}")
    instance._core_client = client_response.conn_ptr
    instance._lib.free_connection_response(client_response)
    return instance

get_monitor_message(timeout=None)

Block until a MonitorMsg is available, then return it.

Parameters:

Name Type Description Default
timeout Optional[float]

Optional timeout in seconds. Returns None on timeout or if closed.

 None
Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/monitor_client.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def get_monitor_message(
    self, timeout: Optional[float] = None
) -> Optional[MonitorMsg]:
    """
    Block until a MonitorMsg is available, then return it.

    Args:
        timeout: Optional timeout in seconds. Returns None on timeout or if closed.
    """
    deadline = time.monotonic() + timeout if timeout is not None else None
    with self._condition:
        while not self._queue and not self._is_closed:
            remaining = None
            if deadline is not None:
                remaining = deadline - time.monotonic()
                if remaining <= 0:
                    break
            self._condition.wait(timeout=remaining)
        return self._queue.pop(0) if self._queue else None

try_get_monitor_message()

Non-blocking retrieval. Returns None if no message is available.

Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/monitor_client.py 
140
141
142
143
def try_get_monitor_message(self) -> Optional[MonitorMsg]:
    """Non-blocking retrieval. Returns None if no message is available."""
    with self._condition:
        return self._queue.pop(0) if self._queue else None

close()

Stop monitoring and release resources.

Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/monitor_client.py 
145
146
147
148
149
150
151
152
153
154
155
156
def close(self) -> None:
    """Stop monitoring and release resources."""
    with self._condition:
        if self._is_closed:
            return
        self._is_closed = True
        self._condition.notify_all()
    if self._core_client != self._ffi.NULL:
        client = self._core_client
        self._core_client = self._ffi.NULL
        self._lib.close_monitor_client(client)
    self._callback_ref = None

stop()

Alias for close(). Stop monitoring and release resources.

Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/monitor_client.py 
158
159
160
def stop(self) -> None:
    """Alias for close(). Stop monitoring and release resources."""
    self.close()

OpenTelemetry

Bases: OpenTelemetry

Sync client OpenTelemetry singleton (separate from async).

Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/opentelemetry.py 
11
12
13
14
15
class OpenTelemetry(_BaseOpenTelemetry):
    """Sync client OpenTelemetry singleton (separate from async)."""

    _instance = None
    _config = None

ClusterScanCursor

Bases: ClusterScanCursor

ClusterScanCursor using the sync package's FFI instance.

Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/sync_commands/cluster_scan_cursor.py 
 7
 8
 9
10
11
class ClusterScanCursor(_ClusterScanCursorBase):
    """ClusterScanCursor using the sync package's FFI instance."""

    def __init__(self, new_cursor=None):
        super().__init__(new_cursor, _ffi=_SYNC_FFI.ffi, _lib=_SYNC_FFI.lib)

Script

Bases: Script

Script using the sync package's FFI instance.

Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/sync_commands/script.py 
 7
 8
 9
10
11
class Script(_ScriptBase):
    """Script using the sync package's FFI instance."""

    def __init__(self, code):
        super().__init__(code, _ffi=_SYNC_FFI.ffi, _lib=_SYNC_FFI.lib)

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 = _SYNC_FFI
    return _glide_ffi.lib.get_min_compressed_size()