Skip to content

glide

ClusterScanCursor

Bases: ClusterScanCursor

ClusterScanCursor using the async package's FFI instance.

Source code in doc-gen/valkey-glide/python/glide-async/python/glide/_ffi_wrappers.py
19
20
21
22
23
class ClusterScanCursor(_ClusterScanCursorBase):
    """ClusterScanCursor using the async package's FFI instance."""

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

Script

Bases: Script

Script using the async package's FFI instance.

Source code in doc-gen/valkey-glide/python/glide-async/python/glide/_ffi_wrappers.py
12
13
14
15
16
class Script(_ScriptBase):
    """Script using the async package's FFI instance."""

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

GlideClient

Bases: BaseClient, StandaloneCommands

Client used for connection to standalone servers. Use :func:~BaseClient.create to request a client. For full documentation, see Valkey GLIDE Documentation

Source code in doc-gen/valkey-glide/python/glide-async/python/glide/glide_client.py
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
class GlideClient(BaseClient, StandaloneCommands):
    """
    Client used for connection to standalone servers.
    Use :func:`~BaseClient.create` to request a client.
    For full documentation, see
    [Valkey GLIDE Documentation](https://glide.valkey.io/how-to/client-initialization/#standalone)
    """

    async def get_subscriptions(
        self,
    ) -> GlideClientConfiguration.PubSubState:
        """
        Retrieves both the desired and current subscription states as tracked by the client.

        This allows verification of synchronization between what the client intends to be
        subscribed to (desired) and what it is actually subscribed to on the server (actual).

        Returns:
            GlideClientConfiguration.PubSubState: An object containing two attributes:
                - desired_subscriptions: Dict[PubSubChannelModes, Set[str]]
                - actual_subscriptions: Dict[PubSubChannelModes, Set[str]]

        Examples:
            >>> from glide import GlideClientConfiguration
            >>> PubSubChannelModes = GlideClientConfiguration.PubSubChannelModes
            >>>
            >>> # Get both subscription states
            >>> state = await client.get_subscriptions()
            >>> desired = state.desired_subscriptions
            >>> actual = state.actual_subscriptions
            >>>
            >>> # Check if subscribed to specific channel
            >>> if "channel1" in actual.get(PubSubChannelModes.Exact, set()):
            >>>     print("Subscribed to channel1")
            >>>
            >>> # Check if synchronized
            >>> if desired == actual:
            >>>     print("Subscriptions are synchronized")
            >>>
            >>> # Find missing subscriptions
            >>> missing = desired.get(PubSubChannelModes.Exact, set()) - actual.get(PubSubChannelModes.Exact, set())
            >>> if missing:
            >>>     print(f"Not yet subscribed to: {missing}")
        """
        result = await self._execute_command(RequestType.GetSubscriptions, [])
        return cast(
            GlideClientConfiguration.PubSubState,
            self._parse_pubsub_state(result, is_cluster=False),
        )

get_subscriptions() async

Retrieves both the desired and current subscription states as tracked by the client.

This allows verification of synchronization between what the client intends to be subscribed to (desired) and what it is actually subscribed to on the server (actual).

Returns:

Type Description
PubSubState

GlideClientConfiguration.PubSubState: An object containing two attributes: - desired_subscriptions: Dict[PubSubChannelModes, Set[str]] - actual_subscriptions: Dict[PubSubChannelModes, Set[str]]

Examples:

>>> from glide import GlideClientConfiguration
>>> PubSubChannelModes = GlideClientConfiguration.PubSubChannelModes
>>>
>>> # Get both subscription states
>>> state = await client.get_subscriptions()
>>> desired = state.desired_subscriptions
>>> actual = state.actual_subscriptions
>>>
>>> # Check if subscribed to specific channel
>>> if "channel1" in actual.get(PubSubChannelModes.Exact, set()):
>>>     print("Subscribed to channel1")
>>>
>>> # Check if synchronized
>>> if desired == actual:
>>>     print("Subscriptions are synchronized")
>>>
>>> # Find missing subscriptions
>>> missing = desired.get(PubSubChannelModes.Exact, set()) - actual.get(PubSubChannelModes.Exact, set())
>>> if missing:
>>>     print(f"Not yet subscribed to: {missing}")
Source code in doc-gen/valkey-glide/python/glide-async/python/glide/glide_client.py
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
async def get_subscriptions(
    self,
) -> GlideClientConfiguration.PubSubState:
    """
    Retrieves both the desired and current subscription states as tracked by the client.

    This allows verification of synchronization between what the client intends to be
    subscribed to (desired) and what it is actually subscribed to on the server (actual).

    Returns:
        GlideClientConfiguration.PubSubState: An object containing two attributes:
            - desired_subscriptions: Dict[PubSubChannelModes, Set[str]]
            - actual_subscriptions: Dict[PubSubChannelModes, Set[str]]

    Examples:
        >>> from glide import GlideClientConfiguration
        >>> PubSubChannelModes = GlideClientConfiguration.PubSubChannelModes
        >>>
        >>> # Get both subscription states
        >>> state = await client.get_subscriptions()
        >>> desired = state.desired_subscriptions
        >>> actual = state.actual_subscriptions
        >>>
        >>> # Check if subscribed to specific channel
        >>> if "channel1" in actual.get(PubSubChannelModes.Exact, set()):
        >>>     print("Subscribed to channel1")
        >>>
        >>> # Check if synchronized
        >>> if desired == actual:
        >>>     print("Subscriptions are synchronized")
        >>>
        >>> # Find missing subscriptions
        >>> missing = desired.get(PubSubChannelModes.Exact, set()) - actual.get(PubSubChannelModes.Exact, set())
        >>> if missing:
        >>>     print(f"Not yet subscribed to: {missing}")
    """
    result = await self._execute_command(RequestType.GetSubscriptions, [])
    return cast(
        GlideClientConfiguration.PubSubState,
        self._parse_pubsub_state(result, is_cluster=False),
    )

GlideClusterClient

Bases: BaseClient, ClusterCommands

Client used for connection to cluster servers. Use :func:~BaseClient.create to request a client. For full documentation, see Valkey GLIDE Documentation

Source code in doc-gen/valkey-glide/python/glide-async/python/glide/glide_client.py
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
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
class GlideClusterClient(BaseClient, ClusterCommands):
    """
    Client used for connection to cluster servers.
    Use :func:`~BaseClient.create` to request a client.
    For full documentation, see
    [Valkey GLIDE Documentation](https://glide.valkey.io/how-to/client-initialization/#cluster)
    """

    async 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."
            )

        callback_id = self._get_callback_id()
        fut = _get_new_future_instance()

        self._pending_futures[callback_id] = fut

        # Build scan args
        args = []
        if match is not None:
            encoded_match = match.encode(ENCODING) if isinstance(match, str) else 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"])

        cursor_string = cursor.get_cursor()
        cursor_bytes = cursor_string.encode(ENCODING) + b"\0"
        cursor_buffer = self._ffi.from_buffer(cursor_bytes)

        if args:
            args_array, args_len_array, arg_buffers = self._to_c_strings(args)
            arg_count = len(args)
        else:
            args_array = self._ffi.NULL
            args_len_array = self._ffi.NULL
            arg_count = 0

        self._lib.request_cluster_scan(
            self._core_client,
            callback_id,
            cursor_buffer,
            arg_count,
            args_array,
            args_len_array,
        )

        response_data = await fut

        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]

    async def get_subscriptions(
        self,
    ) -> GlideClusterClientConfiguration.PubSubState:
        """
        Retrieves both the desired and current subscription states as tracked by the client.

        This allows verification of synchronization between what the client intends to be
        subscribed to (desired) and what it is actually subscribed to on the server (actual).

        Returns:
            GlideClusterClientConfiguration.PubSubState: An object containing two attributes:
                - desired_subscriptions: Dict[PubSubChannelModes, Set[str]]
                - actual_subscriptions: Dict[PubSubChannelModes, Set[str]]

        Examples:
            >>> from glide import GlideClusterClientConfiguration
            >>> PubSubChannelModes = GlideClusterClientConfiguration.PubSubChannelModes
            >>>
            >>> # Get both subscription states
            >>> state = await client.get_subscriptions()
            >>> desired = state.desired_subscriptions
            >>> actual = state.actual_subscriptions
            >>>
            >>> # Check if subscribed to specific channel
            >>> if "channel1" in actual.get(PubSubChannelModes.Exact, set()):
            >>>     print("Subscribed to channel1")
            >>>
            >>> # Check if synchronized
            >>> if desired == actual:
            >>>     print("Subscriptions are synchronized")
            >>>
            >>> # Find missing subscriptions
            >>> missing = desired.get(PubSubChannelModes.Exact, set()) - actual.get(PubSubChannelModes.Exact, set())
            >>> if missing:
            >>>     print(f"Not yet subscribed to: {missing}")
        """
        result = await self._execute_command(RequestType.GetSubscriptions, [])
        return cast(
            GlideClusterClientConfiguration.PubSubState,
            self._parse_pubsub_state(result, is_cluster=True),
        )

get_subscriptions() async

Retrieves both the desired and current subscription states as tracked by the client.

This allows verification of synchronization between what the client intends to be subscribed to (desired) and what it is actually subscribed to on the server (actual).

Returns:

Type Description
PubSubState

GlideClusterClientConfiguration.PubSubState: An object containing two attributes: - desired_subscriptions: Dict[PubSubChannelModes, Set[str]] - actual_subscriptions: Dict[PubSubChannelModes, Set[str]]

Examples:

>>> from glide import GlideClusterClientConfiguration
>>> PubSubChannelModes = GlideClusterClientConfiguration.PubSubChannelModes
>>>
>>> # Get both subscription states
>>> state = await client.get_subscriptions()
>>> desired = state.desired_subscriptions
>>> actual = state.actual_subscriptions
>>>
>>> # Check if subscribed to specific channel
>>> if "channel1" in actual.get(PubSubChannelModes.Exact, set()):
>>>     print("Subscribed to channel1")
>>>
>>> # Check if synchronized
>>> if desired == actual:
>>>     print("Subscriptions are synchronized")
>>>
>>> # Find missing subscriptions
>>> missing = desired.get(PubSubChannelModes.Exact, set()) - actual.get(PubSubChannelModes.Exact, set())
>>> if missing:
>>>     print(f"Not yet subscribed to: {missing}")
Source code in doc-gen/valkey-glide/python/glide-async/python/glide/glide_client.py
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
async def get_subscriptions(
    self,
) -> GlideClusterClientConfiguration.PubSubState:
    """
    Retrieves both the desired and current subscription states as tracked by the client.

    This allows verification of synchronization between what the client intends to be
    subscribed to (desired) and what it is actually subscribed to on the server (actual).

    Returns:
        GlideClusterClientConfiguration.PubSubState: An object containing two attributes:
            - desired_subscriptions: Dict[PubSubChannelModes, Set[str]]
            - actual_subscriptions: Dict[PubSubChannelModes, Set[str]]

    Examples:
        >>> from glide import GlideClusterClientConfiguration
        >>> PubSubChannelModes = GlideClusterClientConfiguration.PubSubChannelModes
        >>>
        >>> # Get both subscription states
        >>> state = await client.get_subscriptions()
        >>> desired = state.desired_subscriptions
        >>> actual = state.actual_subscriptions
        >>>
        >>> # Check if subscribed to specific channel
        >>> if "channel1" in actual.get(PubSubChannelModes.Exact, set()):
        >>>     print("Subscribed to channel1")
        >>>
        >>> # Check if synchronized
        >>> if desired == actual:
        >>>     print("Subscriptions are synchronized")
        >>>
        >>> # Find missing subscriptions
        >>> missing = desired.get(PubSubChannelModes.Exact, set()) - actual.get(PubSubChannelModes.Exact, set())
        >>> if missing:
        >>>     print(f"Not yet subscribed to: {missing}")
    """
    result = await self._execute_command(RequestType.GetSubscriptions, [])
    return cast(
        GlideClusterClientConfiguration.PubSubState,
        self._parse_pubsub_state(result, is_cluster=True),
    )

MonitorClient

An async client that streams all commands processed by the server via MONITOR.

Must be used with a standalone (non-cluster) configuration. Supports both asyncio and trio backends via anyio.

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

Source code in doc-gen/valkey-glide/python/glide-async/python/glide/monitor_client.py
 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
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
class MonitorClient:
    """
    An async client that streams all commands processed by the server via MONITOR.

    Must be used with a standalone (non-cluster) configuration.
    Supports both asyncio and trio backends via anyio.

    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._is_closed = False
        self._stop_lock = threading.Lock()
        self._user_callback: Optional[Callable[[MonitorMsg], None]] = None
        self._is_asyncio: bool = True
        # asyncio path
        self._asyncio_queue: Optional[asyncio.Queue[MonitorMsg]] = None
        self._asyncio_loop: Optional[asyncio.AbstractEventLoop] = None
        # trio path
        self._trio_send: Any = None
        self._trio_receive: Any = None
        self._trio_token: Any = None

    def _setup_queue(self) -> None:
        """Initialize the backend-specific message queue."""
        if not self._is_asyncio:
            import math

            import trio

            self._trio_token = trio.lowlevel.current_trio_token()
            # Unbounded — matches asyncio.Queue() default behavior for this debug-only client
            self._trio_send, self._trio_receive = trio.open_memory_channel(math.inf)
        else:
            self._asyncio_loop = asyncio.get_running_loop()
            self._asyncio_queue = asyncio.Queue()

    def _enqueue_message(self, msg: MonitorMsg) -> None:
        """Thread-safe enqueue from the FFI callback thread."""
        if self._is_closed:
            return
        if not self._is_asyncio:
            import trio

            def _safe_send(msg=msg):
                try:
                    self._trio_send.send_nowait(msg)
                except (trio.ClosedResourceError, trio.BrokenResourceError):
                    pass  # channel torn down, discard

            try:
                self._trio_token.run_sync_soon(_safe_send)
            except trio.RunFinishedError:
                pass  # trio loop exited, discard
        else:
            loop = self._asyncio_loop
            if loop is not None and not loop.is_closed():
                loop.call_soon_threadsafe(
                    self._asyncio_queue.put_nowait, msg  # type: ignore[union-attr]
                )

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

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

        Returns:
            A MonitorClient instance.
        """
        if not isinstance(config, GlideClientConfiguration):
            raise TypeError(
                "MonitorClient requires a GlideClientConfiguration (standalone only)"
            )
        instance = cls()
        try:
            instance._is_asyncio = sniffio.current_async_library() == "asyncio"
        except sniffio.AsyncLibraryNotFoundError:
            instance._is_asyncio = True
        instance._user_callback = callback
        instance._setup_queue()

        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:
                    instance._enqueue_message(msg)
            except Exception:
                pass  # Suppress to avoid crashing the 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

    async def get_monitor_message(self) -> MonitorMsg:
        """Wait for and return the next MonitorMsg."""
        if not self._is_asyncio:
            return await self._trio_receive.receive()
        # asyncio path also covers uvloop: uvloop is a drop-in asyncio replacement
        # and sniffio reports it as "asyncio"
        return await self._asyncio_queue.get()  # type: ignore[union-attr]

    def try_get_monitor_message(self) -> Optional[MonitorMsg]:
        """Non-blocking retrieval. Returns None if queue is empty."""
        if not self._is_asyncio:
            import trio

            try:
                return self._trio_receive.receive_nowait()
            except trio.WouldBlock:
                return None
        try:
            return self._asyncio_queue.get_nowait()  # type: ignore[union-attr]
        except asyncio.QueueEmpty:
            return None

    async def stop(self) -> None:
        """Stop monitoring and release resources."""
        with self._stop_lock:
            if self._is_closed:
                return
            self._is_closed = True
            core_client, self._core_client = self._core_client, self._ffi.NULL
        if core_client != self._ffi.NULL:
            self._lib.close_monitor_client(core_client)
        if not self._is_asyncio and self._trio_send is not None:
            await self._trio_send.aclose()
            if self._trio_receive is not None:
                await self._trio_receive.aclose()
        self._callback_ref = None  # clear after Rust is done and channels are closed

    async def aclose(self) -> None:
        """Alias for stop()."""
        await self.stop()

    async def __aenter__(self) -> "MonitorClient":
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
        await self.stop()

create(config, callback=None) async classmethod

Create a new async MonitorClient.

Parameters:

Name Type Description Default
config GlideClientConfiguration

Standalone client configuration (must be GlideClientConfiguration).

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

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

None

Returns:

Type Description
MonitorClient

A MonitorClient instance.

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

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

    Returns:
        A MonitorClient instance.
    """
    if not isinstance(config, GlideClientConfiguration):
        raise TypeError(
            "MonitorClient requires a GlideClientConfiguration (standalone only)"
        )
    instance = cls()
    try:
        instance._is_asyncio = sniffio.current_async_library() == "asyncio"
    except sniffio.AsyncLibraryNotFoundError:
        instance._is_asyncio = True
    instance._user_callback = callback
    instance._setup_queue()

    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:
                instance._enqueue_message(msg)
        except Exception:
            pass  # Suppress to avoid crashing the 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() async

Wait for and return the next MonitorMsg.

Source code in doc-gen/valkey-glide/python/glide-async/python/glide/monitor_client.py
170
171
172
173
174
175
176
async def get_monitor_message(self) -> MonitorMsg:
    """Wait for and return the next MonitorMsg."""
    if not self._is_asyncio:
        return await self._trio_receive.receive()
    # asyncio path also covers uvloop: uvloop is a drop-in asyncio replacement
    # and sniffio reports it as "asyncio"
    return await self._asyncio_queue.get()  # type: ignore[union-attr]

try_get_monitor_message()

Non-blocking retrieval. Returns None if queue is empty.

Source code in doc-gen/valkey-glide/python/glide-async/python/glide/monitor_client.py
178
179
180
181
182
183
184
185
186
187
188
189
190
def try_get_monitor_message(self) -> Optional[MonitorMsg]:
    """Non-blocking retrieval. Returns None if queue is empty."""
    if not self._is_asyncio:
        import trio

        try:
            return self._trio_receive.receive_nowait()
        except trio.WouldBlock:
            return None
    try:
        return self._asyncio_queue.get_nowait()  # type: ignore[union-attr]
    except asyncio.QueueEmpty:
        return None

stop() async

Stop monitoring and release resources.

Source code in doc-gen/valkey-glide/python/glide-async/python/glide/monitor_client.py
192
193
194
195
196
197
198
199
200
201
202
203
204
205
async def stop(self) -> None:
    """Stop monitoring and release resources."""
    with self._stop_lock:
        if self._is_closed:
            return
        self._is_closed = True
        core_client, self._core_client = self._core_client, self._ffi.NULL
    if core_client != self._ffi.NULL:
        self._lib.close_monitor_client(core_client)
    if not self._is_asyncio and self._trio_send is not None:
        await self._trio_send.aclose()
        if self._trio_receive is not None:
            await self._trio_receive.aclose()
    self._callback_ref = None  # clear after Rust is done and channels are closed

aclose() async

Alias for stop().

Source code in doc-gen/valkey-glide/python/glide-async/python/glide/monitor_client.py
207
208
209
async def aclose(self) -> None:
    """Alias for stop()."""
    await self.stop()

OpenTelemetry

Bases: OpenTelemetry

Async client OpenTelemetry singleton (separate from sync).

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

    _instance = None
    _config = None