glide.glide_client

BaseClient

Bases: CoreCommands

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

class BaseClient(CoreCommands):
    def __init__(self, config: BaseClientConfiguration):
        """To create a new client, use the `create` classmethod"""
        _glide_ffi = _ASYNC_FFI
        self._ffi = _glide_ffi.ffi
        self._lib = _glide_ffi.lib
        self.config: BaseClientConfiguration = config
        self._is_closed: bool = False
        self._core_client = None
        self._loop: Optional[asyncio.AbstractEventLoop] = None  # set in create()
        self._pending_futures: Dict[int, "TFuture"] = {}
        self._callback_id_gen = itertools.count(1)
        self._lock = threading.Lock()
        self._address_resolver_callback_ref = None
        self._pubsub_futures: List["TFuture"] = []
        self._pubsub_lock = threading.Lock()
        self._pending_push_notifications: List[PubSubMsg] = []
        self._pipe_client_id: int = 0
        self._is_asyncio: bool = True

    @classmethod
    async def create(cls, config: BaseClientConfiguration) -> Self:
        """Creates a Glide client.

        Args:
            config (ClientConfiguration): The configuration options for the client, including cluster addresses,
            authentication credentials, TLS settings, periodic checks, and Pub/Sub subscriptions.

        Returns:
            Self: A promise that resolves to a connected client instance.

        Examples:
            # Connecting to a Standalone Server
            >>> from glide import GlideClientConfiguration, NodeAddress, GlideClient, ServerCredentials, BackoffStrategy
            >>> config = GlideClientConfiguration(
            ...     [
            ...         NodeAddress('primary.example.com', 6379),
            ...         NodeAddress('replica1.example.com', 6379),
            ...     ],
            ...     use_tls = True,
            ...     database_id = 1,
            ...     credentials = ServerCredentials(username = 'user1', password = 'passwordA'),
            ...     reconnect_strategy = BackoffStrategy(num_of_retries = 5, factor = 1000, exponent_base = 2),
            ...     pubsub_subscriptions = GlideClientConfiguration.PubSubSubscriptions(
            ...         channels_and_patterns = {GlideClientConfiguration.PubSubChannelModes.Exact: {'updates'}},
            ...         callback = lambda message,context : print(message),
            ...     ),
            ... )
            >>> client = await GlideClient.create(config)

            # Connecting to a Cluster
            >>> from glide import GlideClusterClientConfiguration, NodeAddress, GlideClusterClient,
            ... PeriodicChecksManualInterval
            >>> config = GlideClusterClientConfiguration(
            ...     [
            ...         NodeAddress('address1.example.com', 6379),
            ...         NodeAddress('address2.example.com', 6379),
            ...     ],
            ...     use_tls = True,
            ...     periodic_checks = PeriodicChecksManualInterval(duration_in_sec = 30),
            ...     credentials = ServerCredentials(username = 'user1', password = 'passwordA'),
            ...     reconnect_strategy = BackoffStrategy(num_of_retries = 5, factor = 1000, exponent_base = 2),
            ...     pubsub_subscriptions = GlideClusterClientConfiguration.PubSubSubscriptions(
            ...         channels_and_patterns = {
            ...             GlideClusterClientConfiguration.PubSubChannelModes.Exact: {'updates'},
            ...             GlideClusterClientConfiguration.PubSubChannelModes.Sharded: {'sharded_channel'},
            ...         },
            ...         callback = lambda message,context : print(message),
            ...     ),
            ... )
            >>> client = await GlideClusterClient.create(config)

        Remarks:
            Use this static method to create and connect a client to a Valkey server.
            The client will automatically handle connection establishment, including cluster topology discovery and
            handling of authentication and TLS configurations.

                - **Cluster Topology Discovery**: The client will automatically discover the cluster topology based
                  on the seed addresses provided.
                - **Authentication**: If `ServerCredentials` are provided, the client will attempt to authenticate
                  using the specified username and password.
                - **TLS**: If `use_tls` is set to `true`, the client will establish secure connections using TLS.
                - **Periodic Checks**: The `periodic_checks` setting allows you to configure how often the client
                  checks for cluster topology changes.
                - **Reconnection Strategy**: The `BackoffStrategy` settings define how the client will attempt to
                  reconnect in case of disconnections.
                - **Pub/Sub Subscriptions**: Any channels or patterns specified in `PubSubSubscriptions` will be
                  subscribed to upon connection.
        """
        self = cls(config)

        try:
            self._is_asyncio = sniffio.current_async_library() == "asyncio"
        except sniffio.AsyncLibraryNotFoundError:
            self._is_asyncio = True

        self._loop = asyncio.get_running_loop() if self._is_asyncio else None

        # Build connection request
        conn_req = config._create_a_protobuf_conn_request(
            cluster_mode=isinstance(config, GlideClusterClientConfiguration)
        )
        conn_req.lib_name = "GlidePy"
        conn_req_bytes = conn_req.SerializeToString()

        # Create AsyncClient type
        client_type = self._ffi.new(
            "ClientType*",
            {
                "_type": self._ffi.cast("ClientTypeEnum", FFIClientTypeEnum.Async),
                "async_client": {
                    "success_callback": self._lib.noop_success_callback,
                    "failure_callback": self._lib.noop_failure_callback,
                    "allow_stack_response": False,
                },
            },
        )

        # Pubsub messages are delivered via the shared pipe — no callback needed.
        pubsub_callback = self._ffi.cast("PubSubCallback", 0)

        # Create address resolver callback if configured
        from glide_shared.ffi_helpers import create_address_resolver_callback

        address_resolver_callback = create_address_resolver_callback(
            self._ffi, self.config.address_resolver
        )
        if self.config.address_resolver is not None:
            self._address_resolver_callback_ref = address_resolver_callback

        # Set pipe_client_id before create_client so Rust routes responses
        # through the pipe from the very first command — no race window.
        self._pipe_client_id = next(_next_client_id)

        client_response_ptr = self._lib.create_client(
            conn_req_bytes,
            len(conn_req_bytes),
            client_type,
            pubsub_callback,
            address_resolver_callback,
            self._pipe_client_id,
        )

        ClientLogger.log(LogLevel.INFO, "connection info", "new connection established")

        if client_response_ptr == self._ffi.NULL:
            raise ClosingError("Failed to create client, response pointer is NULL.")

        client_response = self._ffi.cast("ConnectionResponse*", client_response_ptr)
        if client_response.conn_ptr != self._ffi.NULL:
            self._core_client = client_response.conn_ptr
        else:
            error_msg = (
                self._ffi.string(client_response.connection_error_message).decode(
                    ENCODING
                )
                if client_response.connection_error_message != self._ffi.NULL
                else "Unknown error"
            )
            self._lib.free_connection_response(client_response_ptr)
            raise ClosingError(error_msg)

        self._lib.free_connection_response(client_response_ptr)

        self._setup_pipe()

        return self

    def _setup_pipe(self) -> None:
        """Initialize and register the shared response pipe."""
        global _async_pipe_read_fd, _async_pipe_registered, _async_pipe_loop
        global _pipe_remainder, _trio_pipe_active
        with _async_pipe_lock:
            if _async_pipe_read_fd < 0:
                try:
                    _async_pipe_read_fd, pw = os.pipe()
                    os.set_blocking(_async_pipe_read_fd, False)
                    self._lib.init_async_pipe(pw)
                except OSError:
                    _async_pipe_read_fd = -1
                    self._pipe_client_id = 0
            # Detect stale registration: the loop that originally called
            # add_reader has been closed/destroyed (e.g. between anyio.run()
            # calls in benchmarks).  Reset so we re-register below.
            if _async_pipe_registered and _async_pipe_loop is not None:
                if _async_pipe_loop.is_closed():
                    _async_pipe_registered = False
                    _async_pipe_loop = None
                    _pipe_remainder = b""
                    _drain_stale_pipe_frames()
            if (
                _async_pipe_registered
                and not _trio_pipe_active
                and _async_pipe_loop is None
            ):
                # Trio task exited (back-to-back trio.run)
                _async_pipe_registered = False
                _pipe_remainder = b""
                _drain_stale_pipe_frames()
            if _async_pipe_read_fd >= 0 and self._pipe_client_id:
                _client_registry[self._pipe_client_id] = self
                if not _async_pipe_registered:
                    if self._is_asyncio:
                        assert self._loop is not None
                        self._loop.add_reader(
                            _async_pipe_read_fd, _on_async_pipe_readable
                        )
                        _async_pipe_loop = self._loop
                    else:
                        # For trio: spawn a background task that polls the pipe
                        import trio

                        _trio_pipe_active = True
                        trio.lowlevel.spawn_system_task(
                            _trio_pipe_reader, _async_pipe_read_fd
                        )
                    _async_pipe_registered = True

    # ==================== Callback Handling ====================

    def _get_callback_id(self) -> int:
        return next(self._callback_id_gen)

    def _complete_pubsub_futures_safe(self):
        """Complete pending pubsub futures with available messages. Must hold _pubsub_lock."""
        loop = self._loop
        while self._pending_push_notifications and self._pubsub_futures:
            fut = self._pubsub_futures[0]
            if fut.done():
                self._pubsub_futures.pop(0)
                continue
            msg = self._pending_push_notifications.pop(0)
            self._pubsub_futures.pop(0)
            if isinstance(fut, _CompatFuture):
                fut.set_result(msg)
            elif loop and not loop.is_closed():
                loop.call_soon_threadsafe(fut.set_result, msg)

    async def get_pubsub_message(self) -> PubSubMsg:
        if self._is_closed:
            raise ClosingError("Client is closed.")
        if self.config._get_pubsub_callback_and_context()[0] is not None:
            raise ConfigurationError(
                "The operation will never complete since messages will be passed to the configured callback."
            )
        fut: "TFuture" = _get_new_future_instance()
        with self._pubsub_lock:
            self._pubsub_futures.append(fut)
            self._complete_pubsub_futures_safe()
        return await fut

    def try_get_pubsub_message(self) -> Optional[PubSubMsg]:
        if self._is_closed:
            raise ClosingError("Client is closed.")
        if self.config._get_pubsub_callback_and_context()[0] is not None:
            raise ConfigurationError(
                "The operation will never complete since messages will be passed to the configured callback."
            )
        with self._pubsub_lock:
            if self._pending_push_notifications:
                return self._pending_push_notifications.pop(0)
            return None

    # ==================== Response Parsing ====================

    def _handle_response(self, message):
        """Parse a CommandResponse pointer into a Python object.

        For the async client, NULL means no response (returns None) and the arena
        is freed here since responses arrive via the pipe path without automatic
        cleanup. The sync client's _handle_response raises on NULL and relies on
        free_command_result to free the arena.
        """
        if message == self._ffi.NULL:
            return None
        addr = int(self._ffi.cast("uintptr_t", message))
        result, _arena_ptr = _c_parse_response(addr)
        # Arena is freed by the caller (free_command_result for sync path,
        # explicit free_response_arena for pipe path)
        return result

    # ==================== FFI Helpers ====================

    def _to_c_strings(self, args):
        return to_c_strings(self._ffi, args)

    def _to_c_route_ptr_and_len(self, route):
        return to_c_route_ptr_and_len(self._ffi, route)

    # ==================== Command Execution ====================

    async def _execute_command(
        self,
        request_type: int,
        args: List[TEncodable],
        route: Optional[Route] = None,
    ) -> TResult:
        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

        c_args, c_lengths, buffers = self._to_c_strings(args)

        # OTel span creation only when initialized (rare)
        span = 0
        if OpenTelemetry._instance is not None and OpenTelemetry.should_sample():
            span_name_cstr = self._ffi.new(
                "char[]", RequestType.Name(request_type).encode()
            )
            span = self._lib.create_named_otel_span(span_name_cstr)

        if route is None:
            self._lib.command(
                self._core_client,
                callback_id,
                request_type,
                len(args),
                c_args,
                c_lengths,
                self._ffi.NULL,
                0,
                span,
            )
        else:
            route_ptr, route_len, route_bytes = self._to_c_route_ptr_and_len(route)
            self._lib.command(
                self._core_client,
                callback_id,
                request_type,
                len(args),
                c_args,
                c_lengths,
                route_ptr,
                route_len,
                span,
            )

        try:
            return await fut
        finally:
            if span:
                self._lib.drop_otel_span(span)

    async def _execute_batch(
        self,
        commands: List[Tuple[int, List[TEncodable]]],
        is_atomic: bool,
        raise_on_error: bool = False,
        retry_server_error: bool = False,
        retry_connection_error: bool = False,
        route: Optional[Route] = None,
        timeout: Optional[int] = None,
    ) -> List[TResult]:
        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

        span = 0
        if OpenTelemetry.should_sample():
            span = self._lib.create_batch_otel_span()

        batch_info, batch_refs = convert_commands_to_c_batch_info(
            self._ffi, commands, is_atomic
        )
        batch_options, opts_refs = create_c_batch_options(
            self._ffi,
            route,
            retry_server_error=retry_server_error,
            retry_connection_error=retry_connection_error,
            timeout=timeout,
        )
        _refs = batch_refs + opts_refs  # noqa: F841  prevent GC

        self._lib.batch(
            self._core_client,
            callback_id,
            batch_info,
            raise_on_error,
            batch_options,
            span,
        )

        try:
            return await fut
        finally:
            if span != 0:
                self._lib.drop_otel_span(span)

    async def _execute_script(
        self,
        hash: str,
        keys: Optional[List[TEncodable]] = None,
        args: Optional[List[TEncodable]] = None,
        route: Optional[Route] = None,
    ) -> TResult:
        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

        if keys is None:
            keys = []
        if args is None:
            args = []

        keys_c_args, keys_c_lengths, keys_buffers = self._to_c_strings(keys)
        args_c_args, args_c_lengths, args_buffers = self._to_c_strings(args)

        hash_bytes = hash.encode(ENCODING) + b"\0"
        hash_buffer = self._ffi.from_buffer(hash_bytes)

        route_ptr, route_len, route_bytes = self._to_c_route_ptr_and_len(route)

        self._lib.invoke_script(
            self._core_client,
            callback_id,
            hash_buffer,
            len(keys),
            keys_c_args,
            keys_c_lengths,
            len(args),
            args_c_args,
            args_c_lengths,
            route_ptr,
            route_len,
            0,
        )

        return await fut

    # ==================== Cache Metrics ====================

    def _get_cache_metrics(self, metrics_type: int) -> "TResult":
        """
        Get cache metrics via FFI.

        Args:
            metrics_type: Type of metric to retrieve.

        Returns:
            The requested cache metric.

        Raises:
            RequestError: If client-side caching is not enabled or metrics tracking is disabled.
        """
        if self._is_closed:
            raise ClosingError("Client is closed.")
        if self._core_client == self._ffi.NULL:
            raise ValueError("Invalid client pointer.")

        from glide_shared.ffi_helpers import handle_command_result

        command_result = self._lib.get_cache_metrics(self._core_client, 0, metrics_type)
        return handle_command_result(
            self._ffi, self._lib, command_result, self._handle_response
        )

    # ==================== Connection Management ====================

    async def _update_connection_password(
        self, password: Optional[str], immediate_auth: bool
    ) -> TResult:
        if self._is_closed:
            raise ClosingError("Client is closed.")

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

        self._pending_futures[callback_id] = fut

        c_password = (
            self._ffi.new("char[]", password.encode(ENCODING))
            if password is not None
            else self._ffi.new("char[]", b"")
        )

        self._lib.update_connection_password(
            self._core_client,
            callback_id,
            c_password,
            immediate_auth,
        )

        result = await fut
        if result is OK:
            if self.config.credentials is None:
                self.config.credentials = ServerCredentials(password=password or "")
            self.config.credentials.password = password or ""
        return result

    async def _refresh_iam_token(self) -> TResult:
        if self._is_closed:
            raise ClosingError("Client is closed.")

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

        self._pending_futures[callback_id] = fut

        self._lib.refresh_iam_token(
            self._core_client,
            callback_id,
        )

        return await fut

    async def get_statistics(self) -> dict:
        """
        Get compression and connection statistics for this client.

        Returns:
            dict: A dictionary containing statistics with integer values:
                - total_connections: Total number of connections
                - total_clients: Total number of clients
                - total_values_compressed: Number of values successfully compressed
                - total_values_decompressed: Number of values successfully decompressed
                - total_original_bytes: Total bytes of original data before compression
                - total_bytes_compressed: Total bytes after compression
                - total_bytes_decompressed: Total bytes after decompression
                - compression_skipped_count: Number of times compression was skipped
                - subscription_out_of_sync_count: Number of times subscriptions were out of sync
                - subscription_last_sync_timestamp: Timestamp of last successful subscription sync
        """
        stats = self._lib.get_statistics()
        return {
            "total_connections": stats.total_connections,
            "total_clients": stats.total_clients,
            "total_values_compressed": stats.total_values_compressed,
            "total_values_decompressed": stats.total_values_decompressed,
            "total_original_bytes": stats.total_original_bytes,
            "total_bytes_compressed": stats.total_bytes_compressed,
            "total_bytes_decompressed": stats.total_bytes_decompressed,
            "compression_skipped_count": stats.compression_skipped_count,
            "subscription_out_of_sync_count": stats.subscription_out_of_sync_count,
            "subscription_last_sync_timestamp": stats.subscription_last_sync_timestamp,
        }

    def _parse_pubsub_state(self, result: TResult, is_cluster: bool):
        if not isinstance(result, list) or len(result) != 4:
            raise RequestError("Invalid response format from GetSubscriptions")

        desired_dict = result[1]
        actual_dict = result[3]

        if is_cluster:
            PubSubChannelModes = GlideClusterClientConfiguration.PubSubChannelModes
            StateClass = GlideClusterClientConfiguration.PubSubState
            mode_map = {
                "Exact": PubSubChannelModes.Exact,
                "Pattern": PubSubChannelModes.Pattern,
                "Sharded": PubSubChannelModes.Sharded,
            }
        else:
            PubSubChannelModes = GlideClientConfiguration.PubSubChannelModes  # type: ignore[assignment]
            StateClass = GlideClientConfiguration.PubSubState  # type: ignore[assignment]
            mode_map = {
                "Exact": PubSubChannelModes.Exact,
                "Pattern": PubSubChannelModes.Pattern,
            }

        desired_subscriptions = {}
        actual_subscriptions = {}

        for key_bytes, value_list in desired_dict.items():  # type: ignore[union-attr]
            key = key_bytes.decode() if isinstance(key_bytes, bytes) else key_bytes
            if key in mode_map:
                values = {v.decode() if isinstance(v, bytes) else v for v in value_list}
                desired_subscriptions[mode_map[key]] = values

        for key_bytes, value_list in actual_dict.items():  # type: ignore[union-attr]
            key = key_bytes.decode() if isinstance(key_bytes, bytes) else key_bytes
            if key in mode_map:
                values = {v.decode() if isinstance(v, bytes) else v for v in value_list}
                actual_subscriptions[mode_map[key]] = values

        return StateClass(
            desired_subscriptions=desired_subscriptions,
            actual_subscriptions=actual_subscriptions,
        )

    async def close(self, err_message: Optional[str] = None) -> None:
        """
        Terminate the client by closing all associated resources and any active futures.
        All open futures will be closed with an exception.

        Args:
            err_message (Optional[str]): If not None, this error message will be passed along with
            the exceptions when closing all open futures. Defaults to None.
        """
        if not self._is_closed:
            self._is_closed = True
            err_message = "" if err_message is None else err_message

            with self._lock:
                for fut in self._pending_futures.values():
                    if not fut.done():
                        fut.set_exception(ClosingError(err_message))
                self._pending_futures.clear()

            with self._pubsub_lock:
                for fut in self._pubsub_futures:
                    if not fut.done():
                        fut.set_exception(ClosingError(err_message))
                self._pubsub_futures.clear()

            _client_registry.pop(getattr(self, "_pipe_client_id", 0), None)

            if self._core_client is not None:
                self._lib.close_client(self._core_client)
                self._core_client = None

    async def aclose(self, err_message: Optional[str] = None) -> None:
        """
        Terminate the client by closing all associated resources and any active futures.
        All open futures will be closed with an exception.

        This is an alias for close() for compatibility with async context managers.

        Args:
            err_message (Optional[str]): If not None, this error message will be passed along with
            the exceptions when closing all open futures. Defaults to None.
        """
        await self.close(err_message)

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

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

create(config) async classmethod

Creates a Glide client.

Parameters:

Name	Type	Description	Default
config	ClientConfiguration	The configuration options for the client, including cluster addresses,	required

Returns:

Name	Type	Description
Self	Self	A promise that resolves to a connected client instance.

Examples:

Connecting to a Standalone Server

>>> from glide import GlideClientConfiguration, NodeAddress, GlideClient, ServerCredentials, BackoffStrategy
>>> config = GlideClientConfiguration(
...     [
...         NodeAddress('primary.example.com', 6379),
...         NodeAddress('replica1.example.com', 6379),
...     ],
...     use_tls = True,
...     database_id = 1,
...     credentials = ServerCredentials(username = 'user1', password = 'passwordA'),
...     reconnect_strategy = BackoffStrategy(num_of_retries = 5, factor = 1000, exponent_base = 2),
...     pubsub_subscriptions = GlideClientConfiguration.PubSubSubscriptions(
...         channels_and_patterns = {GlideClientConfiguration.PubSubChannelModes.Exact: {'updates'}},
...         callback = lambda message,context : print(message),
...     ),
... )
>>> client = await GlideClient.create(config)

Connecting to a Cluster

>>> from glide import GlideClusterClientConfiguration, NodeAddress, GlideClusterClient,
... PeriodicChecksManualInterval
>>> config = GlideClusterClientConfiguration(
...     [
...         NodeAddress('address1.example.com', 6379),
...         NodeAddress('address2.example.com', 6379),
...     ],
...     use_tls = True,
...     periodic_checks = PeriodicChecksManualInterval(duration_in_sec = 30),
...     credentials = ServerCredentials(username = 'user1', password = 'passwordA'),
...     reconnect_strategy = BackoffStrategy(num_of_retries = 5, factor = 1000, exponent_base = 2),
...     pubsub_subscriptions = GlideClusterClientConfiguration.PubSubSubscriptions(
...         channels_and_patterns = {
...             GlideClusterClientConfiguration.PubSubChannelModes.Exact: {'updates'},
...             GlideClusterClientConfiguration.PubSubChannelModes.Sharded: {'sharded_channel'},
...         },
...         callback = lambda message,context : print(message),
...     ),
... )
>>> client = await GlideClusterClient.create(config)

Remarks

Use this static method to create and connect a client to a Valkey server. The client will automatically handle connection establishment, including cluster topology discovery and handling of authentication and TLS configurations.

- **Cluster Topology Discovery**: The client will automatically discover the cluster topology based
  on the seed addresses provided.
- **Authentication**: If `ServerCredentials` are provided, the client will attempt to authenticate
  using the specified username and password.
- **TLS**: If `use_tls` is set to `true`, the client will establish secure connections using TLS.
- **Periodic Checks**: The `periodic_checks` setting allows you to configure how often the client
  checks for cluster topology changes.
- **Reconnection Strategy**: The `BackoffStrategy` settings define how the client will attempt to
  reconnect in case of disconnections.
- **Pub/Sub Subscriptions**: Any channels or patterns specified in `PubSubSubscriptions` will be
  subscribed to upon connection.

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

@classmethod
async def create(cls, config: BaseClientConfiguration) -> Self:
    """Creates a Glide client.

    Args:
        config (ClientConfiguration): The configuration options for the client, including cluster addresses,
        authentication credentials, TLS settings, periodic checks, and Pub/Sub subscriptions.

    Returns:
        Self: A promise that resolves to a connected client instance.

    Examples:
        # Connecting to a Standalone Server
        >>> from glide import GlideClientConfiguration, NodeAddress, GlideClient, ServerCredentials, BackoffStrategy
        >>> config = GlideClientConfiguration(
        ...     [
        ...         NodeAddress('primary.example.com', 6379),
        ...         NodeAddress('replica1.example.com', 6379),
        ...     ],
        ...     use_tls = True,
        ...     database_id = 1,
        ...     credentials = ServerCredentials(username = 'user1', password = 'passwordA'),
        ...     reconnect_strategy = BackoffStrategy(num_of_retries = 5, factor = 1000, exponent_base = 2),
        ...     pubsub_subscriptions = GlideClientConfiguration.PubSubSubscriptions(
        ...         channels_and_patterns = {GlideClientConfiguration.PubSubChannelModes.Exact: {'updates'}},
        ...         callback = lambda message,context : print(message),
        ...     ),
        ... )
        >>> client = await GlideClient.create(config)

        # Connecting to a Cluster
        >>> from glide import GlideClusterClientConfiguration, NodeAddress, GlideClusterClient,
        ... PeriodicChecksManualInterval
        >>> config = GlideClusterClientConfiguration(
        ...     [
        ...         NodeAddress('address1.example.com', 6379),
        ...         NodeAddress('address2.example.com', 6379),
        ...     ],
        ...     use_tls = True,
        ...     periodic_checks = PeriodicChecksManualInterval(duration_in_sec = 30),
        ...     credentials = ServerCredentials(username = 'user1', password = 'passwordA'),
        ...     reconnect_strategy = BackoffStrategy(num_of_retries = 5, factor = 1000, exponent_base = 2),
        ...     pubsub_subscriptions = GlideClusterClientConfiguration.PubSubSubscriptions(
        ...         channels_and_patterns = {
        ...             GlideClusterClientConfiguration.PubSubChannelModes.Exact: {'updates'},
        ...             GlideClusterClientConfiguration.PubSubChannelModes.Sharded: {'sharded_channel'},
        ...         },
        ...         callback = lambda message,context : print(message),
        ...     ),
        ... )
        >>> client = await GlideClusterClient.create(config)

    Remarks:
        Use this static method to create and connect a client to a Valkey server.
        The client will automatically handle connection establishment, including cluster topology discovery and
        handling of authentication and TLS configurations.

            - **Cluster Topology Discovery**: The client will automatically discover the cluster topology based
              on the seed addresses provided.
            - **Authentication**: If `ServerCredentials` are provided, the client will attempt to authenticate
              using the specified username and password.
            - **TLS**: If `use_tls` is set to `true`, the client will establish secure connections using TLS.
            - **Periodic Checks**: The `periodic_checks` setting allows you to configure how often the client
              checks for cluster topology changes.
            - **Reconnection Strategy**: The `BackoffStrategy` settings define how the client will attempt to
              reconnect in case of disconnections.
            - **Pub/Sub Subscriptions**: Any channels or patterns specified in `PubSubSubscriptions` will be
              subscribed to upon connection.
    """
    self = cls(config)

    try:
        self._is_asyncio = sniffio.current_async_library() == "asyncio"
    except sniffio.AsyncLibraryNotFoundError:
        self._is_asyncio = True

    self._loop = asyncio.get_running_loop() if self._is_asyncio else None

    # Build connection request
    conn_req = config._create_a_protobuf_conn_request(
        cluster_mode=isinstance(config, GlideClusterClientConfiguration)
    )
    conn_req.lib_name = "GlidePy"
    conn_req_bytes = conn_req.SerializeToString()

    # Create AsyncClient type
    client_type = self._ffi.new(
        "ClientType*",
        {
            "_type": self._ffi.cast("ClientTypeEnum", FFIClientTypeEnum.Async),
            "async_client": {
                "success_callback": self._lib.noop_success_callback,
                "failure_callback": self._lib.noop_failure_callback,
                "allow_stack_response": False,
            },
        },
    )

    # Pubsub messages are delivered via the shared pipe — no callback needed.
    pubsub_callback = self._ffi.cast("PubSubCallback", 0)

    # Create address resolver callback if configured
    from glide_shared.ffi_helpers import create_address_resolver_callback

    address_resolver_callback = create_address_resolver_callback(
        self._ffi, self.config.address_resolver
    )
    if self.config.address_resolver is not None:
        self._address_resolver_callback_ref = address_resolver_callback

    # Set pipe_client_id before create_client so Rust routes responses
    # through the pipe from the very first command — no race window.
    self._pipe_client_id = next(_next_client_id)

    client_response_ptr = self._lib.create_client(
        conn_req_bytes,
        len(conn_req_bytes),
        client_type,
        pubsub_callback,
        address_resolver_callback,
        self._pipe_client_id,
    )

    ClientLogger.log(LogLevel.INFO, "connection info", "new connection established")

    if client_response_ptr == self._ffi.NULL:
        raise ClosingError("Failed to create client, response pointer is NULL.")

    client_response = self._ffi.cast("ConnectionResponse*", client_response_ptr)
    if client_response.conn_ptr != self._ffi.NULL:
        self._core_client = client_response.conn_ptr
    else:
        error_msg = (
            self._ffi.string(client_response.connection_error_message).decode(
                ENCODING
            )
            if client_response.connection_error_message != self._ffi.NULL
            else "Unknown error"
        )
        self._lib.free_connection_response(client_response_ptr)
        raise ClosingError(error_msg)

    self._lib.free_connection_response(client_response_ptr)

    self._setup_pipe()

    return self

get_statistics() async

Get compression and connection statistics for this client.

Returns:

Name	Type	Description
dict	dict	A dictionary containing statistics with integer values: - total_connections: Total number of connections - total_clients: Total number of clients - total_values_compressed: Number of values successfully compressed - total_values_decompressed: Number of values successfully decompressed - total_original_bytes: Total bytes of original data before compression - total_bytes_compressed: Total bytes after compression - total_bytes_decompressed: Total bytes after decompression - compression_skipped_count: Number of times compression was skipped - subscription_out_of_sync_count: Number of times subscriptions were out of sync - subscription_last_sync_timestamp: Timestamp of last successful subscription sync

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

async def get_statistics(self) -> dict:
    """
    Get compression and connection statistics for this client.

    Returns:
        dict: A dictionary containing statistics with integer values:
            - total_connections: Total number of connections
            - total_clients: Total number of clients
            - total_values_compressed: Number of values successfully compressed
            - total_values_decompressed: Number of values successfully decompressed
            - total_original_bytes: Total bytes of original data before compression
            - total_bytes_compressed: Total bytes after compression
            - total_bytes_decompressed: Total bytes after decompression
            - compression_skipped_count: Number of times compression was skipped
            - subscription_out_of_sync_count: Number of times subscriptions were out of sync
            - subscription_last_sync_timestamp: Timestamp of last successful subscription sync
    """
    stats = self._lib.get_statistics()
    return {
        "total_connections": stats.total_connections,
        "total_clients": stats.total_clients,
        "total_values_compressed": stats.total_values_compressed,
        "total_values_decompressed": stats.total_values_decompressed,
        "total_original_bytes": stats.total_original_bytes,
        "total_bytes_compressed": stats.total_bytes_compressed,
        "total_bytes_decompressed": stats.total_bytes_decompressed,
        "compression_skipped_count": stats.compression_skipped_count,
        "subscription_out_of_sync_count": stats.subscription_out_of_sync_count,
        "subscription_last_sync_timestamp": stats.subscription_last_sync_timestamp,
    }

close(err_message=None) async

Terminate the client by closing all associated resources and any active futures. All open futures will be closed with an exception.

Parameters:

Name	Type	Description	Default
err_message	Optional[str]	If not None, this error message will be passed along with	None

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

async def close(self, err_message: Optional[str] = None) -> None:
    """
    Terminate the client by closing all associated resources and any active futures.
    All open futures will be closed with an exception.

    Args:
        err_message (Optional[str]): If not None, this error message will be passed along with
        the exceptions when closing all open futures. Defaults to None.
    """
    if not self._is_closed:
        self._is_closed = True
        err_message = "" if err_message is None else err_message

        with self._lock:
            for fut in self._pending_futures.values():
                if not fut.done():
                    fut.set_exception(ClosingError(err_message))
            self._pending_futures.clear()

        with self._pubsub_lock:
            for fut in self._pubsub_futures:
                if not fut.done():
                    fut.set_exception(ClosingError(err_message))
            self._pubsub_futures.clear()

        _client_registry.pop(getattr(self, "_pipe_client_id", 0), None)

        if self._core_client is not None:
            self._lib.close_client(self._core_client)
            self._core_client = None

aclose(err_message=None) async

Terminate the client by closing all associated resources and any active futures. All open futures will be closed with an exception.

This is an alias for close() for compatibility with async context managers.

Parameters:

Name	Type	Description	Default
err_message	Optional[str]	If not None, this error message will be passed along with	None

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

async def aclose(self, err_message: Optional[str] = None) -> None:
    """
    Terminate the client by closing all associated resources and any active futures.
    All open futures will be closed with an exception.

    This is an alias for close() for compatibility with async context managers.

    Args:
        err_message (Optional[str]): If not None, this error message will be passed along with
        the exceptions when closing all open futures. Defaults to None.
    """
    await self.close(err_message)

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

        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),
    )

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),
    )