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

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

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

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

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

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."
            )
        self._check_same_process()

        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

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. The configuration's lib_name and client_info_tag identify the dedicated monitor connection, using GlidePy as the default library name. 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

class MonitorClient:
    """
    An async client that streams all commands processed by the server via MONITOR.

    Must be used with a standalone (non-cluster) configuration.
    The configuration's ``lib_name`` and ``client_info_tag`` identify the dedicated
    monitor connection, using ``GlidePy`` as the default library name.
    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).
                    Its library-name override and client information tag are applied
                    to the dedicated monitor connection.
            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 = _create_async_connection_request(config)
        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). Its library-name override and client information tag are applied to the dedicated monitor connection.	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

@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).
                Its library-name override and client information tag are applied
                to the dedicated monitor connection.
        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 = _create_async_connection_request(config)
    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

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

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

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

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

class OpenTelemetry(_BaseOpenTelemetry):
    """Async client OpenTelemetry singleton (separate from sync)."""

    _instance = None
    _config = None