Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/redis/asyncio/client.py: 22%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1import asyncio
2import copy
3import inspect
4import math
5import re
6import time
7import warnings
8from typing import (
9 TYPE_CHECKING,
10 Any,
11 AsyncIterator,
12 Awaitable,
13 Callable,
14 Dict,
15 Iterable,
16 List,
17 Literal,
18 Mapping,
19 MutableMapping,
20 Optional,
21 Protocol,
22 Set,
23 Tuple,
24 Type,
25 TypedDict,
26 TypeVar,
27 Union,
28 cast,
29)
31from redis._defaults import (
32 DEFAULT_RETRY_BASE,
33 DEFAULT_RETRY_CAP,
34 DEFAULT_RETRY_COUNT,
35 DEFAULT_SOCKET_CONNECT_TIMEOUT,
36 DEFAULT_SOCKET_READ_SIZE,
37 DEFAULT_SOCKET_TIMEOUT,
38)
39from redis._parsers.helpers import bool_ok, get_response_callbacks
40from redis.asyncio.connection import (
41 Connection,
42 ConnectionPool,
43 SSLConnection,
44 UnixDomainSocketConnection,
45)
46from redis.asyncio.lock import Lock
47from redis.asyncio.observability.recorder import (
48 record_error_count,
49 record_operation_duration,
50 record_pubsub_message,
51)
52from redis.asyncio.retry import Retry
53from redis.backoff import ExponentialWithJitterBackoff
54from redis.client import (
55 EMPTY_RESPONSE,
56 NEVER_DECODE,
57 AbstractRedis,
58 CaseInsensitiveDict,
59)
60from redis.commands import (
61 AsyncCoreCommands,
62 AsyncRedisModuleCommands,
63 AsyncSentinelCommands,
64 list_or_args,
65)
66from redis.commands.helpers import parse_pubsub_subscriptions, pubsub_subscription_args
67from redis.credentials import CredentialProvider
68from redis.driver_info import DriverInfo, resolve_driver_info
69from redis.event import (
70 AfterPooledConnectionsInstantiationEvent,
71 AfterPubSubConnectionInstantiationEvent,
72 AfterSingleConnectionInstantiationEvent,
73 ClientType,
74 EventDispatcher,
75)
76from redis.exceptions import (
77 ConnectionError,
78 ExecAbortError,
79 PubSubError,
80 RedisError,
81 ResponseError,
82 WatchError,
83)
84from redis.maint_notifications import MaintNotificationsConfig
85from redis.observability.attributes import PubSubDirection
86from redis.typing import ChannelT, EncodableT, KeyT, PubSubHandler, Subscription
87from redis.utils import (
88 SENTINEL,
89 SSL_AVAILABLE,
90 _set_info_logger,
91 check_protocol_version,
92 deprecated_args,
93 deprecated_function,
94 safe_str,
95 str_if_bytes,
96 truncate_text,
97)
99if TYPE_CHECKING and SSL_AVAILABLE:
100 from ssl import TLSVersion, VerifyFlags, VerifyMode
101else:
102 TLSVersion = None
103 VerifyMode = None
104 VerifyFlags = None
106_KeyT = TypeVar("_KeyT", bound=KeyT)
107_ArgT = TypeVar("_ArgT", KeyT, EncodableT)
108_RedisT = TypeVar("_RedisT", bound="Redis")
109_NormalizeKeysT = TypeVar("_NormalizeKeysT", bound=Mapping[ChannelT, object])
110if TYPE_CHECKING:
111 from redis.asyncio.keyspace_notifications import AsyncKeyspaceNotifications
112 from redis.commands.core import Script
115class ResponseCallbackProtocol(Protocol):
116 def __call__(self, response: Any, **kwargs): ...
119class AsyncResponseCallbackProtocol(Protocol):
120 async def __call__(self, response: Any, **kwargs): ...
123ResponseCallbackT = Union[ResponseCallbackProtocol, AsyncResponseCallbackProtocol]
126class Redis(
127 AbstractRedis, AsyncRedisModuleCommands, AsyncCoreCommands, AsyncSentinelCommands
128):
129 """
130 Implementation of the Redis protocol.
132 This abstract class provides a Python interface to all Redis commands
133 and an implementation of the Redis protocol.
135 Pipelines derive from this, implementing how
136 the commands are sent and received to the Redis server. Based on
137 configuration, an instance will either use a ConnectionPool, or
138 Connection object to talk to redis.
139 """
141 # Type discrimination marker for @overload self-type pattern
142 _is_async_client: Literal[True] = True
144 response_callbacks: MutableMapping[Union[str, bytes], ResponseCallbackT]
146 @classmethod
147 def from_url(
148 cls: Type["Redis"],
149 url: str,
150 single_connection_client: bool = False,
151 auto_close_connection_pool: Optional[bool] = None,
152 **kwargs,
153 ) -> "Redis":
154 """
155 Return a Redis client object configured from the given URL
157 For example::
159 redis://[[username]:[password]]@localhost:6379/0
160 rediss://[[username]:[password]]@localhost:6379/0
161 unix://[username@]/path/to/socket.sock?db=0[&password=password]
163 Three URL schemes are supported:
165 - `redis://` creates a TCP socket connection. See more at:
166 <https://www.iana.org/assignments/uri-schemes/prov/redis>
167 - `rediss://` creates a SSL wrapped TCP socket connection. See more at:
168 <https://www.iana.org/assignments/uri-schemes/prov/rediss>
169 - ``unix://``: creates a Unix Domain Socket connection.
171 The username, password, hostname, path and all querystring values
172 are passed through urllib.parse.unquote in order to replace any
173 percent-encoded values with their corresponding characters.
175 There are several ways to specify a database number. The first value
176 found will be used:
178 1. A ``db`` querystring option, e.g. redis://localhost?db=0
180 2. If using the redis:// or rediss:// schemes, the path argument
181 of the url, e.g. redis://localhost/0
183 3. A ``db`` keyword argument to this function.
185 If none of these options are specified, the default db=0 is used.
187 All querystring options are cast to their appropriate Python types.
188 Boolean arguments can be specified with string values "True"/"False"
189 or "Yes"/"No". Values that cannot be properly cast cause a
190 ``ValueError`` to be raised. Once parsed, the querystring arguments
191 and keyword arguments are passed to the ``ConnectionPool``'s
192 class initializer. In the case of conflicting arguments, querystring
193 arguments always win.
195 """
196 connection_pool = ConnectionPool.from_url(url, **kwargs)
197 client = cls(
198 connection_pool=connection_pool,
199 single_connection_client=single_connection_client,
200 )
201 if auto_close_connection_pool is not None:
202 warnings.warn(
203 DeprecationWarning(
204 '"auto_close_connection_pool" is deprecated '
205 "since version 5.0.1. "
206 "Please create a ConnectionPool explicitly and "
207 "provide to the Redis() constructor instead."
208 )
209 )
210 else:
211 auto_close_connection_pool = True
212 client.auto_close_connection_pool = auto_close_connection_pool
213 return client
215 @classmethod
216 def from_pool(
217 cls: Type["Redis"],
218 connection_pool: ConnectionPool,
219 ) -> "Redis":
220 """
221 Return a Redis client from the given connection pool.
222 The Redis client will take ownership of the connection pool and
223 close it when the Redis client is closed.
225 Because the client closes (disconnects all connections in) the pool
226 when it is closed or garbage-collected, the pool must not be shared
227 with other clients. Constructing multiple clients from the same pool
228 via ``from_pool`` -- for example one per request across tasks -- is
229 not safe: when one client is closed it will disconnect connections
230 still in use by the others.
232 To share a single pool across clients, construct the pool explicitly
233 and manage its lifecycle instead. Unlike ``from_pool``, the plain
234 ``Redis(connection_pool=pool)`` constructor does not take ownership of
235 the pool and will not close it, so a pool created this way can be
236 safely shared across clients. ``ConnectionPool`` supports the async
237 context manager protocol for this::
239 async with ConnectionPool.from_url(url) as pool:
240 r = Redis(connection_pool=pool)
241 """
242 client = cls(
243 connection_pool=connection_pool,
244 )
245 client.auto_close_connection_pool = True
246 return client
248 @deprecated_args(
249 args_to_warn=["retry_on_timeout"],
250 reason="TimeoutError is included by default.",
251 version="6.0.0",
252 )
253 @deprecated_args(
254 args_to_warn=["lib_name", "lib_version"],
255 reason="Use 'driver_info' parameter instead. "
256 "lib_name and lib_version will be removed in a future version.",
257 )
258 def __init__(
259 self,
260 *,
261 host: str = "localhost",
262 port: int = 6379,
263 db: str | int = 0,
264 password: str | None = None,
265 socket_timeout: float | None = DEFAULT_SOCKET_TIMEOUT,
266 socket_connect_timeout: float | None = DEFAULT_SOCKET_CONNECT_TIMEOUT,
267 socket_read_size: int = DEFAULT_SOCKET_READ_SIZE,
268 socket_keepalive: bool | None = True,
269 socket_keepalive_options: Mapping[int, int | bytes] | object | None = SENTINEL,
270 connection_pool: ConnectionPool | None = None,
271 unix_socket_path: str | None = None,
272 encoding: str = "utf-8",
273 encoding_errors: str = "strict",
274 decode_responses: bool = False,
275 retry_on_timeout: bool = False,
276 retry: Retry = Retry(
277 backoff=ExponentialWithJitterBackoff(
278 base=DEFAULT_RETRY_BASE, cap=DEFAULT_RETRY_CAP
279 ),
280 retries=DEFAULT_RETRY_COUNT,
281 ),
282 retry_on_error: list | None = None,
283 ssl: bool = False,
284 ssl_keyfile: str | None = None,
285 ssl_certfile: str | None = None,
286 ssl_cert_reqs: "str | VerifyMode" = "required",
287 ssl_include_verify_flags: List["VerifyFlags"] | None = None,
288 ssl_exclude_verify_flags: List["VerifyFlags"] | None = None,
289 ssl_ca_certs: str | None = None,
290 ssl_ca_data: str | None = None,
291 ssl_ca_path: str | None = None,
292 ssl_check_hostname: bool = True,
293 ssl_min_version: "TLSVersion | None" = None,
294 ssl_ciphers: str | None = None,
295 ssl_password: str | None = None,
296 max_connections: int | None = None,
297 single_connection_client: bool = False,
298 health_check_interval: int = 0,
299 client_name: str | None = None,
300 lib_name: str | object | None = SENTINEL,
301 lib_version: str | object | None = SENTINEL,
302 driver_info: DriverInfo | object | None = SENTINEL,
303 username: str | None = None,
304 auto_close_connection_pool: bool | None = None,
305 redis_connect_func=None,
306 credential_provider: CredentialProvider | None = None,
307 protocol: int | None = None,
308 legacy_responses: bool = True,
309 event_dispatcher: EventDispatcher | None = None,
310 maint_notifications_config: MaintNotificationsConfig | None = None,
311 ):
312 """
313 Initialize a new Redis client.
315 To specify a retry policy for specific errors, you have two options:
317 1. Set the `retry_on_error` to a list of the error/s to retry on, and
318 you can also set `retry` to a valid `Retry` object(in case the default
319 one is not appropriate) - with this approach the retries will be triggered
320 on the default errors specified in the Retry object enriched with the
321 errors specified in `retry_on_error`.
323 2. Define a `Retry` object with configured 'supported_errors' and set
324 it to the `retry` parameter - with this approach you completely redefine
325 the errors on which retries will happen.
327 `retry_on_timeout` is deprecated - please include the TimeoutError
328 either in the Retry object or in the `retry_on_error` list.
330 When 'connection_pool' is provided - the retry configuration of the
331 provided pool will be used.
333 Args:
335 socket_keepalive:
336 if `True`, TCP keepalive is enabled for TCP socket connections.
337 Argument is ignored when connection_pool is provided.
338 socket_keepalive_options:
339 mapping of TCP keepalive socket option constants to values, for
340 example `{socket.TCP_KEEPIDLE: 30}`. If left unspecified, redis-py
341 uses TCP keepalive defaults when `socket_keepalive` is enabled:
342 idle 30 seconds, interval 5 seconds, and 3 probes. Platform-specific
343 options that are not available are skipped. Pass `None` or `{}` to
344 avoid setting additional TCP keepalive options. Argument is ignored
345 when connection_pool is provided.
346 maint_notifications_config:
347 configures the pool to support maintenance notifications - see
348 `redis.maint_notifications.MaintNotificationsConfig` for details.
349 Only supported with RESP3
350 If not provided and protocol is RESP3, the maintenance notifications
351 will be enabled by default (logic is included in the connection pool
352 initialization).
353 Argument is ignored when connection_pool is provided.
354 """
355 kwargs: Dict[str, Any]
356 if event_dispatcher is None:
357 self._event_dispatcher = EventDispatcher()
358 else:
359 self._event_dispatcher = event_dispatcher
360 # auto_close_connection_pool only has an effect if connection_pool is
361 # None. It is assumed that if connection_pool is not None, the user
362 # wants to manage the connection pool themselves.
363 if auto_close_connection_pool is not None:
364 warnings.warn(
365 DeprecationWarning(
366 '"auto_close_connection_pool" is deprecated '
367 "since version 5.0.1. "
368 "Please create a ConnectionPool explicitly and "
369 "provide to the Redis() constructor instead."
370 )
371 )
372 else:
373 auto_close_connection_pool = True
375 if not connection_pool:
376 # Create internal connection pool, expected to be closed by Redis instance
377 if not retry_on_error:
378 retry_on_error = []
380 # Handle driver_info: if provided, use it; otherwise create from lib_name/lib_version.
381 computed_driver_info = resolve_driver_info(
382 driver_info, lib_name, lib_version
383 )
385 kwargs = {
386 "db": db,
387 "username": username,
388 "password": password,
389 "credential_provider": credential_provider,
390 "socket_timeout": socket_timeout,
391 "socket_read_size": socket_read_size,
392 "encoding": encoding,
393 "encoding_errors": encoding_errors,
394 "decode_responses": decode_responses,
395 "retry_on_error": retry_on_error,
396 "retry": copy.deepcopy(retry),
397 "max_connections": max_connections,
398 "health_check_interval": health_check_interval,
399 "client_name": client_name,
400 "driver_info": computed_driver_info,
401 "redis_connect_func": redis_connect_func,
402 "protocol": protocol,
403 "legacy_responses": legacy_responses,
404 }
405 # based on input, setup appropriate connection args
406 if unix_socket_path is not None:
407 if (
408 maint_notifications_config
409 and maint_notifications_config.enabled is True
410 ):
411 raise RedisError(
412 "Maintenance notifications are not supported with Unix "
413 "domain socket connections"
414 )
415 kwargs.update(
416 {
417 "path": unix_socket_path,
418 "connection_class": UnixDomainSocketConnection,
419 "maint_notifications_config": MaintNotificationsConfig(
420 enabled=False
421 ),
422 }
423 )
424 else:
425 # TCP specific options
426 kwargs.update(
427 {
428 "host": host,
429 "port": port,
430 "socket_connect_timeout": socket_connect_timeout,
431 "socket_keepalive": socket_keepalive,
432 "socket_keepalive_options": socket_keepalive_options,
433 }
434 )
436 if ssl:
437 kwargs.update(
438 {
439 "connection_class": SSLConnection,
440 "ssl_keyfile": ssl_keyfile,
441 "ssl_certfile": ssl_certfile,
442 "ssl_cert_reqs": ssl_cert_reqs,
443 "ssl_include_verify_flags": ssl_include_verify_flags,
444 "ssl_exclude_verify_flags": ssl_exclude_verify_flags,
445 "ssl_ca_certs": ssl_ca_certs,
446 "ssl_ca_data": ssl_ca_data,
447 "ssl_ca_path": ssl_ca_path,
448 "ssl_check_hostname": ssl_check_hostname,
449 "ssl_min_version": ssl_min_version,
450 "ssl_ciphers": ssl_ciphers,
451 "ssl_password": ssl_password,
452 }
453 )
454 maint_notifications_enabled = (
455 maint_notifications_config and maint_notifications_config.enabled
456 )
457 if maint_notifications_enabled and not check_protocol_version(protocol, 3):
458 raise RedisError(
459 "Maintenance notifications handlers on connection are only supported with RESP version 3"
460 )
461 if maint_notifications_config:
462 kwargs.update(
463 {
464 "maint_notifications_config": maint_notifications_config,
465 }
466 )
467 # This arg only used if no pool is passed in
468 self.auto_close_connection_pool = auto_close_connection_pool
469 connection_pool = ConnectionPool(**kwargs)
470 self._event_dispatcher.dispatch(
471 AfterPooledConnectionsInstantiationEvent(
472 [connection_pool], ClientType.ASYNC, credential_provider
473 )
474 )
475 else:
476 # If a pool is passed in, do not close it
477 self.auto_close_connection_pool = False
478 self._event_dispatcher.dispatch(
479 AfterPooledConnectionsInstantiationEvent(
480 [connection_pool], ClientType.ASYNC, credential_provider
481 )
482 )
484 self.connection_pool = connection_pool
485 self.single_connection_client = single_connection_client
486 self.connection: Optional[Connection] = None
488 connection_kwargs = self.connection_pool.connection_kwargs
489 self.response_callbacks = CaseInsensitiveDict(
490 get_response_callbacks(
491 user_protocol=connection_kwargs.get("protocol"),
492 legacy_responses=connection_kwargs.get("legacy_responses", True),
493 )
494 )
496 # If using a single connection client, we need to lock creation-of and use-of
497 # the client in order to avoid race conditions such as using asyncio.gather
498 # on a set of redis commands
499 self._single_conn_lock = asyncio.Lock()
501 # When used as an async context manager, we need to increment and decrement
502 # a usage counter so that we can close the connection pool when no one is
503 # using the client.
504 self._usage_counter = 0
505 self._usage_lock = asyncio.Lock()
507 def __repr__(self):
508 return (
509 f"<{self.__class__.__module__}.{self.__class__.__name__}"
510 f"({self.connection_pool!r})>"
511 )
513 def __await__(self):
514 return self.initialize().__await__()
516 async def initialize(self: _RedisT) -> _RedisT:
517 if self.single_connection_client:
518 async with self._single_conn_lock:
519 if self.connection is None:
520 self.connection = await self.connection_pool.get_connection()
522 self._event_dispatcher.dispatch(
523 AfterSingleConnectionInstantiationEvent(
524 self.connection, ClientType.ASYNC, self._single_conn_lock
525 )
526 )
527 return self
529 def set_response_callback(self, command: str, callback: ResponseCallbackT):
530 """Set a custom Response Callback"""
531 self.response_callbacks[command] = callback
533 def get_encoder(self):
534 """Get the connection pool's encoder"""
535 return self.connection_pool.get_encoder()
537 def get_connection_kwargs(self):
538 """Get the connection's key-word arguments"""
539 return self.connection_pool.connection_kwargs
541 def get_retry(self) -> Optional[Retry]:
542 return self.get_connection_kwargs().get("retry")
544 def set_retry(self, retry: Retry) -> None:
545 self.get_connection_kwargs().update({"retry": retry})
546 self.connection_pool.set_retry(retry)
548 def load_external_module(self, funcname, func):
549 """
550 This function can be used to add externally defined redis modules,
551 and their namespaces to the redis client.
553 funcname - A string containing the name of the function to create
554 func - The function, being added to this class.
556 ex: Assume that one has a custom redis module named foomod that
557 creates command named 'foo.dothing' and 'foo.anotherthing' in redis.
558 To load function functions into this namespace:
560 from redis import Redis
561 from foomodule import F
562 r = Redis()
563 r.load_external_module("foo", F)
564 r.foo().dothing('your', 'arguments')
566 For a concrete example see the reimport of the redisjson module in
567 tests/test_connection.py::test_loading_external_modules
568 """
569 setattr(self, funcname, func)
571 def pipeline(
572 self, transaction: bool = True, shard_hint: Optional[str] = None
573 ) -> "Pipeline":
574 """
575 Return a new pipeline object that can queue multiple commands for
576 later execution. ``transaction`` indicates whether all commands
577 should be executed atomically. Apart from making a group of operations
578 atomic, pipelines are useful for reducing the back-and-forth overhead
579 between the client and server.
580 """
581 return Pipeline(
582 self.connection_pool, self.response_callbacks, transaction, shard_hint
583 )
585 async def transaction(
586 self,
587 func: Callable[["Pipeline"], Union[Any, Awaitable[Any]]],
588 *watches: KeyT,
589 shard_hint: Optional[str] = None,
590 value_from_callable: bool = False,
591 watch_delay: Optional[float] = None,
592 ):
593 """
594 Convenience method for executing the callable `func` as a transaction
595 while watching all keys specified in `watches`. The 'func' callable
596 should expect a single argument which is a Pipeline object.
597 """
598 pipe: Pipeline
599 async with self.pipeline(True, shard_hint) as pipe:
600 while True:
601 try:
602 if watches:
603 await pipe.watch(*watches)
604 func_value = func(pipe)
605 if inspect.isawaitable(func_value):
606 func_value = await func_value
607 exec_value = await pipe.execute()
608 return func_value if value_from_callable else exec_value
609 except WatchError:
610 if watch_delay is not None and watch_delay > 0:
611 await asyncio.sleep(watch_delay)
612 continue
614 def lock(
615 self,
616 name: KeyT,
617 timeout: Optional[float] = None,
618 sleep: float = 0.1,
619 blocking: bool = True,
620 blocking_timeout: Optional[float] = None,
621 lock_class: Optional[Type[Lock]] = None,
622 thread_local: bool = True,
623 raise_on_release_error: bool = True,
624 ) -> Lock:
625 """
626 Return a new Lock object using key ``name`` that mimics
627 the behavior of threading.Lock.
629 If specified, ``timeout`` indicates a maximum life for the lock.
630 By default, it will remain locked until release() is called.
632 ``sleep`` indicates the amount of time to sleep per loop iteration
633 when the lock is in blocking mode and another client is currently
634 holding the lock.
636 ``blocking`` indicates whether calling ``acquire`` should block until
637 the lock has been acquired or to fail immediately, causing ``acquire``
638 to return False and the lock not being acquired. Defaults to True.
639 Note this value can be overridden by passing a ``blocking``
640 argument to ``acquire``.
642 ``blocking_timeout`` indicates the maximum amount of time in seconds to
643 spend trying to acquire the lock. A value of ``None`` indicates
644 continue trying forever. ``blocking_timeout`` can be specified as a
645 float or integer, both representing the number of seconds to wait.
647 ``lock_class`` forces the specified lock implementation. Note that as
648 of redis-py 3.0, the only lock class we implement is ``Lock`` (which is
649 a Lua-based lock). So, it's unlikely you'll need this parameter, unless
650 you have created your own custom lock class.
652 ``thread_local`` indicates whether the lock token is placed in
653 thread-local storage. By default, the token is placed in thread local
654 storage so that a thread only sees its token, not a token set by
655 another thread. Consider the following timeline:
657 time: 0, thread-1 acquires `my-lock`, with a timeout of 5 seconds.
658 thread-1 sets the token to "abc"
659 time: 1, thread-2 blocks trying to acquire `my-lock` using the
660 Lock instance.
661 time: 5, thread-1 has not yet completed. redis expires the lock
662 key.
663 time: 5, thread-2 acquired `my-lock` now that it's available.
664 thread-2 sets the token to "xyz"
665 time: 6, thread-1 finishes its work and calls release(). if the
666 token is *not* stored in thread local storage, then
667 thread-1 would see the token value as "xyz" and would be
668 able to successfully release the thread-2's lock.
670 ``raise_on_release_error`` indicates whether to raise an exception when
671 the lock is no longer owned when exiting the context manager. By default,
672 this is True, meaning an exception will be raised. If False, the warning
673 will be logged and the exception will be suppressed.
675 In some use cases it's necessary to disable thread local storage. For
676 example, if you have code where one thread acquires a lock and passes
677 that lock instance to a worker thread to release later. If thread
678 local storage isn't disabled in this case, the worker thread won't see
679 the token set by the thread that acquired the lock. Our assumption
680 is that these cases aren't common and as such default to using
681 thread local storage."""
682 if lock_class is None:
683 lock_class = Lock
684 return lock_class(
685 self,
686 name,
687 timeout=timeout,
688 sleep=sleep,
689 blocking=blocking,
690 blocking_timeout=blocking_timeout,
691 thread_local=thread_local,
692 raise_on_release_error=raise_on_release_error,
693 )
695 def pubsub(self, **kwargs) -> "PubSub":
696 """
697 Return a Publish/Subscribe object. With this object, you can
698 subscribe to channels and listen for messages that get published to
699 them.
700 """
701 return PubSub(
702 self.connection_pool, event_dispatcher=self._event_dispatcher, **kwargs
703 )
705 def keyspace_notifications(
706 self,
707 key_prefix: Union[str, bytes, None] = None,
708 ignore_subscribe_messages: bool = True,
709 ) -> "AsyncKeyspaceNotifications":
710 """
711 Return an :class:`~redis.asyncio.keyspace_notifications.AsyncKeyspaceNotifications`
712 object for subscribing to keyspace and keyevent notifications.
714 Note: Keyspace notifications must be enabled on the Redis server via
715 the ``notify-keyspace-events`` configuration option.
717 Args:
718 key_prefix: Optional prefix to filter and strip from keys in
719 notifications.
720 ignore_subscribe_messages: If True, subscribe/unsubscribe
721 confirmations are not returned by
722 get_message/listen.
723 """
724 from redis.asyncio.keyspace_notifications import AsyncKeyspaceNotifications
726 return AsyncKeyspaceNotifications(
727 self,
728 key_prefix=key_prefix,
729 ignore_subscribe_messages=ignore_subscribe_messages,
730 )
732 def monitor(self) -> "Monitor":
733 return Monitor(self.connection_pool)
735 def client(self) -> "Redis":
736 return self.__class__(
737 connection_pool=self.connection_pool, single_connection_client=True
738 )
740 async def __aenter__(self: _RedisT) -> _RedisT:
741 """
742 Async context manager entry. Increments a usage counter so that the
743 connection pool is only closed (via aclose()) when no context is using
744 the client.
745 """
746 await self._increment_usage()
747 try:
748 # Initialize the client (i.e. establish connection, etc.)
749 return await self.initialize()
750 except Exception:
751 # If initialization fails, decrement the counter to keep it in sync
752 await self._decrement_usage()
753 raise
755 async def _increment_usage(self) -> int:
756 """
757 Helper coroutine to increment the usage counter while holding the lock.
758 Returns the new value of the usage counter.
759 """
760 async with self._usage_lock:
761 self._usage_counter += 1
762 return self._usage_counter
764 async def _decrement_usage(self) -> int:
765 """
766 Helper coroutine to decrement the usage counter while holding the lock.
767 Returns the new value of the usage counter.
768 """
769 async with self._usage_lock:
770 self._usage_counter -= 1
771 return self._usage_counter
773 async def __aexit__(self, exc_type, exc_value, traceback):
774 """
775 Async context manager exit. Decrements a usage counter. If this is the
776 last exit (counter becomes zero), the client closes its connection pool.
777 """
778 current_usage = await asyncio.shield(self._decrement_usage())
779 if current_usage == 0:
780 # This was the last active context, so disconnect the pool.
781 await asyncio.shield(self.aclose())
783 _DEL_MESSAGE = "Unclosed Redis client"
785 # passing _warnings and _grl as argument default since they may be gone
786 # by the time __del__ is called at shutdown
787 def __del__(
788 self,
789 _warn: Any = warnings.warn,
790 _grl: Any = asyncio.get_running_loop,
791 ) -> None:
792 if hasattr(self, "connection") and (self.connection is not None):
793 _warn(f"Unclosed client session {self!r}", ResourceWarning, source=self)
794 try:
795 context = {"client": self, "message": self._DEL_MESSAGE}
796 _grl().call_exception_handler(context)
797 except RuntimeError:
798 pass
799 self.connection._close()
801 async def aclose(self, close_connection_pool: Optional[bool] = None) -> None:
802 """
803 Closes Redis client connection
805 Args:
806 close_connection_pool:
807 decides whether to close the connection pool used by this Redis client,
808 overriding Redis.auto_close_connection_pool.
809 By default, let Redis.auto_close_connection_pool decide
810 whether to close the connection pool.
811 """
812 conn = self.connection
813 if conn:
814 self.connection = None
815 await self.connection_pool.release(conn)
816 if close_connection_pool or (
817 close_connection_pool is None and self.auto_close_connection_pool
818 ):
819 await self.connection_pool.aclose()
821 @deprecated_function(version="5.0.1", reason="Use aclose() instead", name="close")
822 async def close(self, close_connection_pool: Optional[bool] = None) -> None:
823 """
824 Alias for aclose(), for backwards compatibility
825 """
826 await self.aclose(close_connection_pool)
828 async def _send_command_parse_response(self, conn, command_name, *args, **options):
829 """
830 Send a command and parse the response
831 """
832 await conn.send_command(*args)
833 return await self.parse_response(conn, command_name, **options)
835 async def _close_connection(
836 self,
837 conn: Connection,
838 error: Optional[BaseException] = None,
839 failure_count: Optional[int] = None,
840 start_time: Optional[float] = None,
841 command_name: Optional[str] = None,
842 ):
843 """
844 Close the connection before retrying.
846 The supported exceptions are already checked in the
847 retry object so we don't need to do it here.
849 After we disconnect the connection, it will try to reconnect and
850 do a health check as part of the send_command logic(on connection level).
851 """
852 if (
853 error
854 and failure_count is not None
855 and failure_count <= conn.retry.get_retries()
856 ):
857 await record_operation_duration(
858 command_name=command_name,
859 duration_seconds=time.monotonic() - start_time,
860 server_address=getattr(conn, "host", None),
861 server_port=getattr(conn, "port", None),
862 db_namespace=str(conn.db),
863 error=error,
864 retry_attempts=failure_count,
865 )
867 await conn.disconnect(error=error, failure_count=failure_count)
869 # COMMAND EXECUTION AND PROTOCOL PARSING
870 async def execute_command(self, *args, **options):
871 """Execute a command and return a parsed response"""
872 await self.initialize()
873 pool = self.connection_pool
874 command_name = args[0]
875 conn = self.connection or await pool.get_connection()
877 # Start timing for observability
878 start_time = time.monotonic()
879 # Track actual retry attempts for error reporting
880 actual_retry_attempts = 0
882 def failure_callback(error, failure_count):
883 nonlocal actual_retry_attempts
884 actual_retry_attempts = failure_count
885 return self._close_connection(
886 conn, error, failure_count, start_time, command_name
887 )
889 if self.single_connection_client:
890 await self._single_conn_lock.acquire()
891 try:
892 result = await conn.retry.call_with_retry(
893 lambda: self._send_command_parse_response(
894 conn, command_name, *args, **options
895 ),
896 failure_callback,
897 with_failure_count=True,
898 )
900 await record_operation_duration(
901 command_name=command_name,
902 duration_seconds=time.monotonic() - start_time,
903 server_address=getattr(conn, "host", None),
904 server_port=getattr(conn, "port", None),
905 db_namespace=str(conn.db),
906 )
907 return result
908 except Exception as e:
909 await record_error_count(
910 server_address=getattr(conn, "host", None),
911 server_port=getattr(conn, "port", None),
912 network_peer_address=getattr(conn, "host", None),
913 network_peer_port=getattr(conn, "port", None),
914 error_type=e,
915 retry_attempts=actual_retry_attempts,
916 is_internal=False,
917 )
918 raise
919 finally:
920 try:
921 if self.single_connection_client and conn and conn.should_reconnect():
922 await self._close_connection(conn)
923 await conn.connect()
924 finally:
925 if self.single_connection_client:
926 self._single_conn_lock.release()
927 if not self.connection:
928 await pool.release(conn)
930 async def parse_response(
931 self, connection: Connection, command_name: Union[str, bytes], **options
932 ):
933 """Parses a response from the Redis server"""
934 try:
935 if NEVER_DECODE in options:
936 response = await connection.read_response(disable_decoding=True)
937 options.pop(NEVER_DECODE)
938 else:
939 response = await connection.read_response()
940 except ResponseError:
941 if EMPTY_RESPONSE in options:
942 return options[EMPTY_RESPONSE]
943 raise
945 if EMPTY_RESPONSE in options:
946 options.pop(EMPTY_RESPONSE)
948 # Remove keys entry, it needs only for cache.
949 options.pop("keys", None)
951 if command_name in self.response_callbacks:
952 # Mypy bug: https://github.com/python/mypy/issues/10977
953 command_name = cast(str, command_name)
954 retval = self.response_callbacks[command_name](response, **options)
955 return await retval if inspect.isawaitable(retval) else retval
956 return response
959StrictRedis = Redis
962class MonitorCommandInfo(TypedDict):
963 time: float
964 db: int
965 client_address: str
966 client_port: str
967 client_type: str
968 command: str
971class Monitor:
972 """
973 Monitor is useful for handling the MONITOR command to the redis server.
974 next_command() method returns one command from monitor
975 listen() method yields commands from monitor.
976 """
978 monitor_re = re.compile(r"\[(\d+) (.*?)\] (.*)")
979 command_re = re.compile(r'"(.*?)(?<!\\)"')
981 def __init__(self, connection_pool: ConnectionPool):
982 self.connection_pool = connection_pool
983 self.connection: Optional[Connection] = None
985 async def connect(self):
986 if self.connection is None:
987 self.connection = await self.connection_pool.get_connection()
989 async def __aenter__(self):
990 await self.connect()
991 await self.connection.send_command("MONITOR")
992 # check that monitor returns 'OK', but don't return it to user
993 response = await self.connection.read_response()
994 if not bool_ok(response):
995 raise RedisError(f"MONITOR failed: {response}")
996 return self
998 async def __aexit__(self, *args):
999 await self.connection.disconnect()
1000 await self.connection_pool.release(self.connection)
1002 async def next_command(self) -> MonitorCommandInfo:
1003 """Parse the response from a monitor command"""
1004 await self.connect()
1005 response = await self.connection.read_response()
1006 if isinstance(response, bytes):
1007 response = self.connection.encoder.decode(response, force=True)
1008 command_time, command_data = response.split(" ", 1)
1009 m = self.monitor_re.match(command_data)
1010 db_id, client_info, command = m.groups()
1011 command = " ".join(self.command_re.findall(command))
1012 # Redis escapes double quotes because each piece of the command
1013 # string is surrounded by double quotes. We don't have that
1014 # requirement so remove the escaping and leave the quote.
1015 command = command.replace('\\"', '"')
1017 if client_info == "lua":
1018 client_address = "lua"
1019 client_port = ""
1020 client_type = "lua"
1021 elif client_info.startswith("unix"):
1022 client_address = "unix"
1023 client_port = client_info[5:]
1024 client_type = "unix"
1025 else:
1026 # use rsplit as ipv6 addresses contain colons
1027 client_address, client_port = client_info.rsplit(":", 1)
1028 client_type = "tcp"
1029 return {
1030 "time": float(command_time),
1031 "db": int(db_id),
1032 "client_address": client_address,
1033 "client_port": client_port,
1034 "client_type": client_type,
1035 "command": command,
1036 }
1038 async def listen(self) -> AsyncIterator[MonitorCommandInfo]:
1039 """Listen for commands coming to the server."""
1040 while True:
1041 yield await self.next_command()
1044class PubSub:
1045 """
1046 PubSub provides publish, subscribe and listen support to Redis channels.
1048 After subscribing to one or more channels, the listen() method will block
1049 until a message arrives on one of the subscribed channels. That message
1050 will be returned and it's safe to start listening again.
1051 """
1053 PUBLISH_MESSAGE_TYPES = ("message", "pmessage", "smessage")
1054 UNSUBSCRIBE_MESSAGE_TYPES = ("unsubscribe", "punsubscribe", "sunsubscribe")
1055 HEALTH_CHECK_MESSAGE = "redis-py-health-check"
1057 def __init__(
1058 self,
1059 connection_pool: ConnectionPool,
1060 shard_hint: Optional[str] = None,
1061 ignore_subscribe_messages: bool = False,
1062 encoder=None,
1063 push_handler_func: Optional[Callable] = None,
1064 event_dispatcher: Optional["EventDispatcher"] = None,
1065 ):
1066 if event_dispatcher is None:
1067 self._event_dispatcher = EventDispatcher()
1068 else:
1069 self._event_dispatcher = event_dispatcher
1070 self.connection_pool = connection_pool
1071 self.shard_hint = shard_hint
1072 self.ignore_subscribe_messages = ignore_subscribe_messages
1073 self.connection = None
1074 # we need to know the encoding options for this connection in order
1075 # to lookup channel and pattern names for callback handlers.
1076 self.encoder = encoder
1077 self.push_handler_func = push_handler_func
1078 if self.encoder is None:
1079 self.encoder = self.connection_pool.get_encoder()
1080 if self.encoder.decode_responses:
1081 self.health_check_response = [
1082 ["pong", self.HEALTH_CHECK_MESSAGE],
1083 self.HEALTH_CHECK_MESSAGE,
1084 ]
1085 else:
1086 self.health_check_response = [
1087 [b"pong", self.encoder.encode(self.HEALTH_CHECK_MESSAGE)],
1088 self.encoder.encode(self.HEALTH_CHECK_MESSAGE),
1089 ]
1090 if self.push_handler_func is None:
1091 _set_info_logger()
1092 self.channels = {}
1093 self.pending_unsubscribe_channels = set()
1094 self.patterns = {}
1095 self.pending_unsubscribe_patterns = set()
1096 self.shard_channels = {}
1097 self.pending_unsubscribe_shard_channels = set()
1098 self._lock = asyncio.Lock()
1100 async def __aenter__(self):
1101 return self
1103 async def __aexit__(self, exc_type, exc_value, traceback):
1104 await self.aclose()
1106 def __del__(self):
1107 if self.connection:
1108 self.connection.deregister_connect_callback(self.on_connect)
1110 async def aclose(self):
1111 # In case a connection property does not yet exist
1112 # (due to a crash earlier in the Redis() constructor), return
1113 # immediately as there is nothing to clean-up.
1114 if not hasattr(self, "connection"):
1115 return
1116 async with self._lock:
1117 if self.connection:
1118 # Use nowait=True to avoid awaiting StreamWriter.wait_closed(),
1119 # which can deadlock when a concurrent reader task (e.g. one
1120 # running pubsub.run() or get_message(block=True)) still holds
1121 # the transport. See https://github.com/redis/redis-py/issues/3941
1122 await self.connection.disconnect(nowait=True)
1123 self.connection.deregister_connect_callback(self.on_connect)
1124 await self.connection_pool.release(self.connection)
1125 self.connection = None
1126 self.channels = {}
1127 self.pending_unsubscribe_channels = set()
1128 self.patterns = {}
1129 self.pending_unsubscribe_patterns = set()
1130 self.shard_channels = {}
1131 self.pending_unsubscribe_shard_channels = set()
1133 @deprecated_function(version="5.0.1", reason="Use aclose() instead", name="close")
1134 async def close(self) -> None:
1135 """Alias for aclose(), for backwards compatibility"""
1136 await self.aclose()
1138 @deprecated_function(version="5.0.1", reason="Use aclose() instead", name="reset")
1139 async def reset(self) -> None:
1140 """Alias for aclose(), for backwards compatibility"""
1141 await self.aclose()
1143 async def _resubscribe(self, subscribed, subscribe_fn) -> None:
1144 # Replay handler-backed subscriptions as positional Subscription objects
1145 # so binary names never need to be decoded into keyword argument keys.
1146 subscriptions = pubsub_subscription_args(subscribed)
1147 if subscriptions:
1148 await subscribe_fn(*subscriptions)
1150 async def _resubscribe_shard_channels(self) -> None:
1151 await self._resubscribe(self.shard_channels, self.ssubscribe)
1153 async def on_connect(self, connection: Connection):
1154 """Re-subscribe to any channels and patterns previously subscribed to"""
1155 self.pending_unsubscribe_channels.clear()
1156 self.pending_unsubscribe_patterns.clear()
1157 self.pending_unsubscribe_shard_channels.clear()
1158 if self.channels:
1159 await self._resubscribe(self.channels, self.subscribe)
1160 if self.patterns:
1161 await self._resubscribe(self.patterns, self.psubscribe)
1162 if self.shard_channels:
1163 await self._resubscribe_shard_channels()
1165 @property
1166 def subscribed(self):
1167 """Indicates if there are subscriptions to any channels or patterns"""
1168 return bool(self.channels or self.patterns or self.shard_channels)
1170 async def execute_command(self, *args: EncodableT):
1171 """Execute a publish/subscribe command"""
1173 # NOTE: don't parse the response in this function -- it could pull a
1174 # legitimate message off the stack if the connection is already
1175 # subscribed to one or more channels
1177 await self.connect()
1178 connection = self.connection
1179 kwargs = {"check_health": not self.subscribed}
1180 await self._execute(connection, connection.send_command, *args, **kwargs)
1182 async def connect(self):
1183 """
1184 Ensure that the PubSub is connected
1185 """
1186 if self.connection is None:
1187 self.connection = await self.connection_pool.get_connection()
1188 # register a callback that re-subscribes to any channels we
1189 # were listening to when we were disconnected
1190 self.connection.register_connect_callback(self.on_connect)
1191 else:
1192 await self.connection.connect()
1193 if self.push_handler_func is not None:
1194 self.connection._parser.set_pubsub_push_handler(self.push_handler_func)
1196 self._event_dispatcher.dispatch(
1197 AfterPubSubConnectionInstantiationEvent(
1198 self.connection, self.connection_pool, ClientType.ASYNC, self._lock
1199 )
1200 )
1202 async def _reconnect(
1203 self,
1204 conn,
1205 error: Optional[BaseException] = None,
1206 failure_count: Optional[int] = None,
1207 start_time: Optional[float] = None,
1208 command_name: Optional[str] = None,
1209 ):
1210 """
1211 The supported exceptions are already checked in the
1212 retry object so we don't need to do it here.
1214 In this error handler we are trying to reconnect to the server.
1215 """
1216 if (
1217 error
1218 and failure_count is not None
1219 and failure_count <= conn.retry.get_retries()
1220 ):
1221 if command_name:
1222 await record_operation_duration(
1223 command_name=command_name,
1224 duration_seconds=time.monotonic() - start_time,
1225 server_address=getattr(conn, "host", None),
1226 server_port=getattr(conn, "port", None),
1227 db_namespace=str(conn.db),
1228 error=error,
1229 retry_attempts=failure_count,
1230 )
1231 await conn.disconnect(error=error, failure_count=failure_count)
1232 await conn.connect()
1234 async def _execute(self, conn, command, *args, **kwargs):
1235 """
1236 Connect manually upon disconnection. If the Redis server is down,
1237 this will fail and raise a ConnectionError as desired.
1238 After reconnection, the ``on_connect`` callback should have been
1239 called by the # connection to resubscribe us to any channels and
1240 patterns we were previously listening to
1241 """
1242 if not len(args) == 0:
1243 command_name = args[0]
1244 else:
1245 command_name = None
1247 # Start timing for observability
1248 start_time = time.monotonic()
1249 # Track actual retry attempts for error reporting
1250 actual_retry_attempts = 0
1252 def failure_callback(error, failure_count):
1253 nonlocal actual_retry_attempts
1254 actual_retry_attempts = failure_count
1255 return self._reconnect(conn, error, failure_count, start_time, command_name)
1257 try:
1258 response = await conn.retry.call_with_retry(
1259 lambda: command(*args, **kwargs),
1260 failure_callback,
1261 with_failure_count=True,
1262 )
1264 if command_name:
1265 await record_operation_duration(
1266 command_name=command_name,
1267 duration_seconds=time.monotonic() - start_time,
1268 server_address=getattr(conn, "host", None),
1269 server_port=getattr(conn, "port", None),
1270 db_namespace=str(conn.db),
1271 )
1273 return response
1274 except Exception as e:
1275 await record_error_count(
1276 server_address=getattr(conn, "host", None),
1277 server_port=getattr(conn, "port", None),
1278 network_peer_address=getattr(conn, "host", None),
1279 network_peer_port=getattr(conn, "port", None),
1280 error_type=e,
1281 retry_attempts=actual_retry_attempts,
1282 is_internal=False,
1283 )
1284 raise
1286 async def parse_response(self, block: bool = True, timeout: float = 0):
1287 """
1288 Parse the response from a publish/subscribe command.
1290 Args:
1291 block: If True, block indefinitely until a message is available.
1292 If False, return immediately if no message is available.
1293 Default: True
1294 timeout: The timeout in seconds for reading a response when block=False.
1295 This parameter is ignored when block=True.
1296 Default: 0 (return immediately if no data available)
1298 Returns:
1299 The parsed response from the server, or None if no message is available
1300 within the timeout period (when block=False).
1302 Important:
1303 The block and timeout parameters work together:
1304 - When block=True: timeout is IGNORED, method blocks indefinitely
1305 - When block=False: timeout is USED, method returns after timeout expires
1307 Typically, you should use get_message(timeout=X) instead of calling
1308 parse_response() directly. The get_message() method automatically sets
1309 block=False when a timeout is provided, and block=True when timeout=None.
1311 Example:
1312 # Block indefinitely (timeout is ignored)
1313 response = await pubsub.parse_response(block=True, timeout=0.1)
1315 # Non-blocking with 0.1 second timeout
1316 response = await pubsub.parse_response(block=False, timeout=0.1)
1318 # Non-blocking, return immediately
1319 response = await pubsub.parse_response(block=False, timeout=0)
1321 # Recommended: use get_message() instead
1322 msg = await pubsub.get_message(timeout=0.1) # automatically sets block=False
1323 msg = await pubsub.get_message(timeout=None) # automatically sets block=True
1324 """
1325 conn = self.connection
1326 if conn is None:
1327 raise RuntimeError(
1328 "pubsub connection not set: "
1329 "did you forget to call subscribe() or psubscribe()?"
1330 )
1332 await self.check_health()
1334 if not conn.is_connected:
1335 await conn.connect()
1337 # Block=True: signal "no timeout" to conn.read_response via
1338 # math.inf. The connection treats math.inf as the per-read
1339 # opt-in for blocking indefinitely without falling back to
1340 # self.socket_timeout. Reconnect/AUTH/HELLO/resubscribe
1341 # operations performed by the retry layer continue to honor
1342 # self.socket_timeout because they do not pass math.inf.
1343 #
1344 # TODO(next-major): when the async Connection.read_response
1345 # default for ``timeout`` is changed to SENTINEL, passing
1346 # ``timeout=None`` from this method will become the natural
1347 # "no timeout" signal and the math.inf hand-off can be
1348 # removed. That swap is a breaking change to the
1349 # Connection.read_response signature so it must wait for a
1350 # major release.
1351 read_timeout = math.inf if block else timeout
1352 response = await self._execute(
1353 conn,
1354 conn.read_response,
1355 timeout=read_timeout,
1356 disconnect_on_error=False,
1357 push_request=True,
1358 )
1360 if conn.health_check_interval and response in self.health_check_response:
1361 # ignore the health check message as user might not expect it
1362 return None
1363 return response
1365 async def check_health(self):
1366 conn = self.connection
1367 if conn is None:
1368 raise RuntimeError(
1369 "pubsub connection not set: "
1370 "did you forget to call subscribe() or psubscribe()?"
1371 )
1373 if (
1374 conn.health_check_interval
1375 and asyncio.get_running_loop().time() > conn.next_health_check
1376 ):
1377 await conn.send_command(
1378 "PING", self.HEALTH_CHECK_MESSAGE, check_health=False
1379 )
1381 def _normalize_keys(self, data: _NormalizeKeysT) -> _NormalizeKeysT:
1382 """
1383 normalize channel/pattern names to be either bytes or strings
1384 based on whether responses are automatically decoded. this saves us
1385 from coercing the value for each message coming in.
1386 """
1387 encode = self.encoder.encode
1388 decode = self.encoder.decode
1389 return {decode(encode(k)): v for k, v in data.items()} # type: ignore[return-value] # noqa: E501
1391 async def psubscribe(
1392 self, *args: ChannelT | Subscription, **kwargs: PubSubHandler
1393 ) -> None:
1394 """
1395 Subscribe to channel patterns.
1396 Patterns supplied as keyword arguments expect a pattern name as the
1397 key and a callable as the value.
1398 ``Subscription`` objects can also be supplied positionally with an
1399 optional handler.
1400 A pattern's callable will be invoked automatically
1401 when a message is received on that pattern rather than producing a
1402 message via ``listen()``.
1403 """
1404 new_patterns = parse_pubsub_subscriptions(args, kwargs)
1405 ret_val = await self.execute_command("PSUBSCRIBE", *new_patterns.keys())
1406 # update the patterns dict AFTER we send the command. we don't want to
1407 # subscribe twice to these patterns, once for the command and again
1408 # for the reconnection.
1409 new_patterns = self._normalize_keys(new_patterns)
1410 self.patterns.update(new_patterns)
1411 self.pending_unsubscribe_patterns.difference_update(new_patterns)
1412 return ret_val
1414 def punsubscribe(self, *args: ChannelT) -> Awaitable:
1415 """
1416 Unsubscribe from the supplied patterns. If empty, unsubscribe from
1417 all patterns.
1418 """
1419 patterns: Iterable[ChannelT]
1420 if args:
1421 parsed_args = list_or_args((args[0],), args[1:])
1422 patterns = self._normalize_keys(dict.fromkeys(parsed_args)).keys()
1423 else:
1424 parsed_args = []
1425 patterns = self.patterns
1426 self.pending_unsubscribe_patterns.update(patterns)
1427 return self.execute_command("PUNSUBSCRIBE", *parsed_args)
1429 async def subscribe(
1430 self, *args: ChannelT | Subscription, **kwargs: PubSubHandler
1431 ) -> None:
1432 """
1433 Subscribe to channels.
1434 Channels supplied as keyword arguments expect
1435 a channel name as the key and a callable as the value.
1436 ``Subscription`` objects can also be supplied positionally with an
1437 optional handler.
1438 A channel's callable will be invoked automatically
1439 when a message is received on that channel rather than producing a
1440 message via ``listen()`` or ``get_message()``.
1441 """
1442 new_channels = parse_pubsub_subscriptions(args, kwargs)
1443 ret_val = await self.execute_command("SUBSCRIBE", *new_channels.keys())
1444 # update the channels dict AFTER we send the command. we don't want to
1445 # subscribe twice to these channels, once for the command and again
1446 # for the reconnection.
1447 new_channels = self._normalize_keys(new_channels)
1448 self.channels.update(new_channels)
1449 self.pending_unsubscribe_channels.difference_update(new_channels)
1450 return ret_val
1452 def unsubscribe(self, *args) -> Awaitable:
1453 """
1454 Unsubscribe from the supplied channels. If empty, unsubscribe from
1455 all channels
1456 """
1457 if args:
1458 parsed_args = list_or_args(args[0], args[1:])
1459 channels = self._normalize_keys(dict.fromkeys(parsed_args))
1460 else:
1461 parsed_args = []
1462 channels = self.channels
1463 self.pending_unsubscribe_channels.update(channels)
1464 return self.execute_command("UNSUBSCRIBE", *parsed_args)
1466 async def ssubscribe(
1467 self,
1468 *args: ChannelT | Subscription,
1469 target_node: Any = None,
1470 **kwargs: PubSubHandler,
1471 ) -> None:
1472 """
1473 Subscribes the client to the specified shard channels.
1474 Channels supplied as keyword arguments expect a channel name as the key
1475 and a callable as the value.
1476 ``Subscription`` objects can also be supplied positionally
1477 with an optional handler.
1478 A channel's callable will be invoked automatically when a message
1479 is received on that channel rather than producing a message
1480 via ``listen()`` or ``get_sharded_message()``.
1481 """
1482 new_s_channels = parse_pubsub_subscriptions(args, kwargs)
1483 ret_val = await self.execute_command("SSUBSCRIBE", *new_s_channels.keys())
1484 # update the s_channels dict AFTER we send the command. we don't want to
1485 # subscribe twice to these channels, once for the command and again
1486 # for the reconnection.
1487 new_s_channels = self._normalize_keys(new_s_channels)
1488 self.shard_channels.update(new_s_channels)
1489 self.pending_unsubscribe_shard_channels.difference_update(new_s_channels)
1490 return ret_val
1492 def sunsubscribe(self, *args, target_node=None) -> Awaitable:
1493 """
1494 Unsubscribe from the supplied shard_channels. If empty, unsubscribe from
1495 all shard_channels
1496 """
1497 if args:
1498 args = list_or_args(args[0], args[1:])
1499 s_channels = self._normalize_keys(dict.fromkeys(args))
1500 else:
1501 s_channels = self.shard_channels
1502 self.pending_unsubscribe_shard_channels.update(s_channels)
1503 return self.execute_command("SUNSUBSCRIBE", *args)
1505 async def listen(self) -> AsyncIterator:
1506 """Listen for messages on channels this client has been subscribed to"""
1507 while self.subscribed:
1508 response = await self.handle_message(await self.parse_response(block=True))
1509 if response is not None:
1510 yield response
1512 async def get_message(
1513 self, ignore_subscribe_messages: bool = False, timeout: Optional[float] = 0.0
1514 ):
1515 """
1516 Get the next message if one is available, otherwise None.
1518 If timeout is specified, the system will wait for `timeout` seconds
1519 before returning. Timeout should be specified as a floating point
1520 number or None to wait indefinitely.
1521 """
1522 response = await self.parse_response(block=(timeout is None), timeout=timeout)
1523 if response:
1524 return await self.handle_message(response, ignore_subscribe_messages)
1525 return None
1527 def ping(self, message=None) -> Awaitable[bool]:
1528 """
1529 Ping the Redis server to test connectivity.
1531 Sends a PING command to the Redis server and returns True if the server
1532 responds with "PONG".
1533 """
1534 args = ["PING", message] if message is not None else ["PING"]
1535 return self.execute_command(*args)
1537 async def handle_message(self, response, ignore_subscribe_messages=False):
1538 """
1539 Parses a pub/sub message. If the channel or pattern was subscribed to
1540 with a message handler, the handler is invoked instead of a parsed
1541 message being returned.
1542 """
1543 if response is None:
1544 return None
1545 if isinstance(response, bytes):
1546 response = [b"pong", response] if response != b"PONG" else [b"pong", b""]
1547 message_type = str_if_bytes(response[0])
1548 if message_type == "pmessage":
1549 message = {
1550 "type": message_type,
1551 "pattern": response[1],
1552 "channel": response[2],
1553 "data": response[3],
1554 }
1555 elif message_type == "pong":
1556 message = {
1557 "type": message_type,
1558 "pattern": None,
1559 "channel": None,
1560 "data": response[1],
1561 }
1562 else:
1563 message = {
1564 "type": message_type,
1565 "pattern": None,
1566 "channel": response[1],
1567 "data": response[2],
1568 }
1570 if message_type in ["message", "pmessage"]:
1571 channel = str_if_bytes(message["channel"])
1572 await record_pubsub_message(
1573 direction=PubSubDirection.RECEIVE,
1574 channel=channel,
1575 )
1576 elif message_type == "smessage":
1577 channel = str_if_bytes(message["channel"])
1578 await record_pubsub_message(
1579 direction=PubSubDirection.RECEIVE,
1580 channel=channel,
1581 sharded=True,
1582 )
1584 # if this is an unsubscribe message, remove it from memory
1585 if message_type in self.UNSUBSCRIBE_MESSAGE_TYPES:
1586 if message_type == "punsubscribe":
1587 pattern = response[1]
1588 if pattern in self.pending_unsubscribe_patterns:
1589 self.pending_unsubscribe_patterns.remove(pattern)
1590 self.patterns.pop(pattern, None)
1591 elif message_type == "sunsubscribe":
1592 s_channel = response[1]
1593 if s_channel in self.pending_unsubscribe_shard_channels:
1594 self.pending_unsubscribe_shard_channels.remove(s_channel)
1595 self.shard_channels.pop(s_channel, None)
1596 else:
1597 channel = response[1]
1598 if channel in self.pending_unsubscribe_channels:
1599 self.pending_unsubscribe_channels.remove(channel)
1600 self.channels.pop(channel, None)
1602 if message_type in self.PUBLISH_MESSAGE_TYPES:
1603 # if there's a message handler, invoke it
1604 if message_type == "pmessage":
1605 handler = self.patterns.get(message["pattern"], None)
1606 elif message_type == "smessage":
1607 handler = self.shard_channels.get(message["channel"], None)
1608 else:
1609 handler = self.channels.get(message["channel"], None)
1610 if handler:
1611 if inspect.iscoroutinefunction(handler):
1612 await handler(message)
1613 else:
1614 handler(message)
1615 return None
1616 elif message_type != "pong":
1617 # this is a subscribe/unsubscribe message. ignore if we don't
1618 # want them
1619 if ignore_subscribe_messages or self.ignore_subscribe_messages:
1620 return None
1622 return message
1624 async def run(
1625 self,
1626 *,
1627 exception_handler: Optional["PSWorkerThreadExcHandlerT"] = None,
1628 poll_timeout: float = 1.0,
1629 pubsub=None,
1630 ) -> None:
1631 """Process pub/sub messages using registered callbacks.
1633 This is the equivalent of :py:meth:`redis.PubSub.run_in_thread` in
1634 redis-py, but it is a coroutine. To launch it as a separate task, use
1635 ``asyncio.create_task``:
1637 >>> task = asyncio.create_task(pubsub.run())
1639 To shut it down, use asyncio cancellation:
1641 >>> task.cancel()
1642 >>> await task
1643 """
1644 for channel, handler in self.channels.items():
1645 if handler is None:
1646 raise PubSubError(f"Channel: '{channel}' has no handler registered")
1647 for pattern, handler in self.patterns.items():
1648 if handler is None:
1649 raise PubSubError(f"Pattern: '{pattern}' has no handler registered")
1651 await self.connect()
1652 while True:
1653 try:
1654 if pubsub is None:
1655 await self.get_message(
1656 ignore_subscribe_messages=True, timeout=poll_timeout
1657 )
1658 else:
1659 await pubsub.get_message(
1660 ignore_subscribe_messages=True, timeout=poll_timeout
1661 )
1662 except asyncio.CancelledError:
1663 raise
1664 except BaseException as e:
1665 if exception_handler is None:
1666 raise
1667 res = exception_handler(e, self)
1668 if inspect.isawaitable(res):
1669 await res
1670 # Ensure that other tasks on the event loop get a chance to run
1671 # if we didn't have to block for I/O anywhere.
1672 await asyncio.sleep(0)
1675class PubsubWorkerExceptionHandler(Protocol):
1676 def __call__(self, e: BaseException, pubsub: PubSub): ...
1679class AsyncPubsubWorkerExceptionHandler(Protocol):
1680 async def __call__(self, e: BaseException, pubsub: PubSub): ...
1683PSWorkerThreadExcHandlerT = Union[
1684 PubsubWorkerExceptionHandler, AsyncPubsubWorkerExceptionHandler
1685]
1688CommandT = Tuple[Tuple[Union[str, bytes], ...], Mapping[str, Any]]
1689CommandStackT = List[CommandT]
1692class Pipeline(Redis): # lgtm [py/init-calls-subclass]
1693 """
1694 Pipelines provide a way to transmit multiple commands to the Redis server
1695 in one transmission. This is convenient for batch processing, such as
1696 saving all the values in a list to Redis.
1698 All commands executed within a pipeline(when running in transactional mode,
1699 which is the default behavior) are wrapped with MULTI and EXEC
1700 calls. This guarantees all commands executed in the pipeline will be
1701 executed atomically.
1703 Any command raising an exception does *not* halt the execution of
1704 subsequent commands in the pipeline. Instead, the exception is caught
1705 and its instance is placed into the response list returned by execute().
1706 Code iterating over the response list should be able to deal with an
1707 instance of an exception as a potential value. In general, these will be
1708 ResponseError exceptions, such as those raised when issuing a command
1709 on a key of a different datatype.
1710 """
1712 UNWATCH_COMMANDS = {"DISCARD", "EXEC", "UNWATCH"}
1714 def __init__(
1715 self,
1716 connection_pool: ConnectionPool,
1717 response_callbacks: MutableMapping[Union[str, bytes], ResponseCallbackT],
1718 transaction: bool,
1719 shard_hint: Optional[str],
1720 ):
1721 self.connection_pool = connection_pool
1722 self.connection = None
1723 self.response_callbacks = response_callbacks
1724 self.is_transaction = transaction
1725 self.shard_hint = shard_hint
1726 self.watching = False
1727 self.command_stack: CommandStackT = []
1728 self.scripts: Set[Script] = set()
1729 self.explicit_transaction = False
1731 async def __aenter__(self: _RedisT) -> _RedisT:
1732 return self
1734 async def __aexit__(self, exc_type, exc_value, traceback):
1735 await self.reset()
1737 def __await__(self):
1738 return self._async_self().__await__()
1740 _DEL_MESSAGE = "Unclosed Pipeline client"
1742 def __len__(self):
1743 return len(self.command_stack)
1745 def __bool__(self):
1746 """Pipeline instances should always evaluate to True"""
1747 return True
1749 async def _async_self(self):
1750 return self
1752 async def reset(self):
1753 self.command_stack = []
1754 self.scripts = set()
1755 try:
1756 # make sure to reset the connection state in the event that we were
1757 # watching something
1758 if self.watching and self.connection:
1759 try:
1760 # call this manually since our unwatch or
1761 # immediate_execute_command methods can call reset()
1762 await self.connection.send_command("UNWATCH")
1763 await self.connection.read_response()
1764 except ConnectionError:
1765 # disconnect will also remove any previous WATCHes
1766 if self.connection:
1767 await self.connection.disconnect()
1768 except asyncio.CancelledError:
1769 # Disconnect so any unread UNWATCH reply does not get
1770 # served to the next caller that takes the connection.
1771 if self.connection:
1772 await self.connection.disconnect()
1773 raise
1774 finally:
1775 self.watching = False
1776 self.explicit_transaction = False
1777 # We can safely return the connection to the pool here since we're
1778 # sure we're no longer WATCHing anything. Detach self.connection
1779 # before awaiting release: if a second cancel aborts the await,
1780 # the pipeline must not be left holding a reference to a
1781 # connection that is being returned to the pool. Shield the
1782 # release itself so a second cancel cannot split the pool's
1783 # internal in-use/available bookkeeping mid-update.
1784 if self.connection:
1785 connection, self.connection = self.connection, None
1786 await asyncio.shield(self.connection_pool.release(connection))
1788 async def aclose(self) -> None:
1789 """Alias for reset(), a standard method name for cleanup"""
1790 await self.reset()
1792 def multi(self):
1793 """
1794 Start a transactional block of the pipeline after WATCH commands
1795 are issued. End the transactional block with `execute`.
1796 """
1797 if self.explicit_transaction:
1798 raise RedisError("Cannot issue nested calls to MULTI")
1799 if self.command_stack:
1800 raise RedisError(
1801 "Commands without an initial WATCH have already been issued"
1802 )
1803 self.explicit_transaction = True
1805 def execute_command(
1806 self, *args, **kwargs
1807 ) -> Union["Pipeline", Awaitable["Pipeline"]]:
1808 if (self.watching or args[0] == "WATCH") and not self.explicit_transaction:
1809 return self.immediate_execute_command(*args, **kwargs)
1810 return self.pipeline_execute_command(*args, **kwargs)
1812 async def _disconnect_reset_raise_on_watching(
1813 self,
1814 conn: Connection,
1815 error: Exception,
1816 failure_count: Optional[int] = None,
1817 start_time: Optional[float] = None,
1818 command_name: Optional[str] = None,
1819 ) -> None:
1820 """
1821 Close the connection reset watching state and
1822 raise an exception if we were watching.
1824 The supported exceptions are already checked in the
1825 retry object so we don't need to do it here.
1827 After we disconnect the connection, it will try to reconnect and
1828 do a health check as part of the send_command logic(on connection level).
1829 """
1830 if (
1831 error
1832 and failure_count is not None
1833 and failure_count <= conn.retry.get_retries()
1834 ):
1835 await record_operation_duration(
1836 command_name=command_name,
1837 duration_seconds=time.monotonic() - start_time,
1838 server_address=getattr(conn, "host", None),
1839 server_port=getattr(conn, "port", None),
1840 db_namespace=str(conn.db),
1841 error=error,
1842 retry_attempts=failure_count,
1843 )
1844 await conn.disconnect(error=error, failure_count=failure_count)
1845 # if we were already watching a variable, the watch is no longer
1846 # valid since this connection has died. raise a WatchError, which
1847 # indicates the user should retry this transaction.
1848 if self.watching:
1849 await self.reset()
1850 raise WatchError(
1851 f"A {type(error).__name__} occurred while watching one or more keys"
1852 )
1854 async def immediate_execute_command(self, *args, **options):
1855 """
1856 Execute a command immediately, but don't auto-retry on the supported
1857 errors for retry if we're already WATCHing a variable.
1858 Used when issuing WATCH or subsequent commands retrieving their values but before
1859 MULTI is called.
1860 """
1861 command_name = args[0]
1862 conn = self.connection
1863 # if this is the first call, we need a connection
1864 if not conn:
1865 conn = await self.connection_pool.get_connection()
1866 self.connection = conn
1868 # Start timing for observability
1869 start_time = time.monotonic()
1870 # Track actual retry attempts for error reporting
1871 actual_retry_attempts = 0
1873 def failure_callback(error, failure_count):
1874 nonlocal actual_retry_attempts
1875 actual_retry_attempts = failure_count
1876 return self._disconnect_reset_raise_on_watching(
1877 conn, error, failure_count, start_time, command_name
1878 )
1880 try:
1881 response = await conn.retry.call_with_retry(
1882 lambda: self._send_command_parse_response(
1883 conn, command_name, *args, **options
1884 ),
1885 failure_callback,
1886 with_failure_count=True,
1887 )
1889 await record_operation_duration(
1890 command_name=command_name,
1891 duration_seconds=time.monotonic() - start_time,
1892 server_address=getattr(conn, "host", None),
1893 server_port=getattr(conn, "port", None),
1894 db_namespace=str(conn.db),
1895 )
1897 return response
1898 except Exception as e:
1899 await record_error_count(
1900 server_address=getattr(conn, "host", None),
1901 server_port=getattr(conn, "port", None),
1902 network_peer_address=getattr(conn, "host", None),
1903 network_peer_port=getattr(conn, "port", None),
1904 error_type=e,
1905 retry_attempts=actual_retry_attempts,
1906 is_internal=False,
1907 )
1908 raise
1910 def pipeline_execute_command(self, *args, **options):
1911 """
1912 Stage a command to be executed when execute() is next called
1914 Returns the current Pipeline object back so commands can be
1915 chained together, such as:
1917 pipe = pipe.set('foo', 'bar').incr('baz').decr('bang')
1919 At some other point, you can then run: pipe.execute(),
1920 which will execute all commands queued in the pipe.
1921 """
1922 self.command_stack.append((args, options))
1923 return self
1925 async def _execute_transaction( # noqa: C901
1926 self, connection: Connection, commands: CommandStackT, raise_on_error
1927 ):
1928 pre: CommandT = (("MULTI",), {})
1929 post: CommandT = (("EXEC",), {})
1930 cmds = (pre, *commands, post)
1931 all_cmds = connection.pack_commands(
1932 args for args, options in cmds if EMPTY_RESPONSE not in options
1933 )
1934 await connection.send_packed_command(all_cmds)
1935 errors = []
1937 # parse off the response for MULTI
1938 # NOTE: we need to handle ResponseErrors here and continue
1939 # so that we read all the additional command messages from
1940 # the socket
1941 try:
1942 await self.parse_response(connection, "_")
1943 except ResponseError as err:
1944 errors.append((0, err))
1946 # and all the other commands
1947 for i, command in enumerate(commands):
1948 if EMPTY_RESPONSE in command[1]:
1949 errors.append((i, command[1][EMPTY_RESPONSE]))
1950 else:
1951 try:
1952 await self.parse_response(connection, "_")
1953 except ResponseError as err:
1954 self.annotate_exception(err, i + 1, command[0])
1955 errors.append((i, err))
1957 # parse the EXEC.
1958 try:
1959 response = await self.parse_response(connection, "_")
1960 except ExecAbortError as err:
1961 if errors:
1962 raise errors[0][1] from err
1963 raise
1965 # EXEC clears any watched keys
1966 self.watching = False
1968 if response is None:
1969 raise WatchError("Watched variable changed.") from None
1971 # put any parse errors into the response
1972 for i, e in errors:
1973 response.insert(i, e)
1975 if len(response) != len(commands):
1976 if self.connection:
1977 await self.connection.disconnect()
1978 raise ResponseError(
1979 "Wrong number of response items from pipeline execution"
1980 ) from None
1982 # find any errors in the response and raise if necessary
1983 if raise_on_error:
1984 self.raise_first_error(commands, response)
1986 # We have to run response callbacks manually
1987 data = []
1988 for r, cmd in zip(response, commands):
1989 if not isinstance(r, Exception):
1990 args, options = cmd
1991 command_name = args[0]
1993 # Remove keys entry, it needs only for cache.
1994 options.pop("keys", None)
1996 if command_name in self.response_callbacks:
1997 r = self.response_callbacks[command_name](r, **options)
1998 if inspect.isawaitable(r):
1999 r = await r
2000 data.append(r)
2001 return data
2003 async def _execute_pipeline(
2004 self, connection: Connection, commands: CommandStackT, raise_on_error: bool
2005 ):
2006 # build up all commands into a single request to increase network perf
2007 all_cmds = connection.pack_commands([args for args, _ in commands])
2008 await connection.send_packed_command(all_cmds)
2010 response = []
2011 for args, options in commands:
2012 try:
2013 response.append(
2014 await self.parse_response(connection, args[0], **options)
2015 )
2016 except ResponseError as e:
2017 response.append(e)
2019 if raise_on_error:
2020 self.raise_first_error(commands, response)
2021 return response
2023 def raise_first_error(self, commands: CommandStackT, response: Iterable[Any]):
2024 for i, r in enumerate(response):
2025 if isinstance(r, ResponseError):
2026 self.annotate_exception(r, i + 1, commands[i][0])
2027 raise r
2029 def annotate_exception(
2030 self, exception: Exception, number: int, command: Iterable[object]
2031 ) -> None:
2032 cmd = " ".join(map(safe_str, command))
2033 msg = (
2034 f"Command # {number} ({truncate_text(cmd)}) "
2035 f"of pipeline caused error: {exception.args}"
2036 )
2037 exception.args = (msg,) + exception.args[1:]
2039 async def parse_response(
2040 self, connection: Connection, command_name: Union[str, bytes], **options
2041 ):
2042 result = await super().parse_response(connection, command_name, **options)
2043 if command_name in self.UNWATCH_COMMANDS:
2044 self.watching = False
2045 elif command_name == "WATCH":
2046 self.watching = True
2047 return result
2049 async def load_scripts(self):
2050 # make sure all scripts that are about to be run on this pipeline exist
2051 scripts = list(self.scripts)
2052 immediate = self.immediate_execute_command
2053 shas = [s.sha for s in scripts]
2054 # we can't use the normal script_* methods because they would just
2055 # get buffered in the pipeline.
2056 exists = await immediate("SCRIPT EXISTS", *shas)
2057 if not all(exists):
2058 for s, exist in zip(scripts, exists):
2059 if not exist:
2060 s.sha = await immediate("SCRIPT LOAD", s.script)
2062 async def _disconnect_raise_on_watching(
2063 self,
2064 conn: Connection,
2065 error: Exception,
2066 failure_count: Optional[int] = None,
2067 start_time: Optional[float] = None,
2068 command_name: Optional[str] = None,
2069 ):
2070 """
2071 Close the connection, raise an exception if we were watching.
2073 The supported exceptions are already checked in the
2074 retry object so we don't need to do it here.
2076 After we disconnect the connection, it will try to reconnect and
2077 do a health check as part of the send_command logic(on connection level).
2078 """
2079 if (
2080 error
2081 and failure_count is not None
2082 and failure_count <= conn.retry.get_retries()
2083 ):
2084 await record_operation_duration(
2085 command_name=command_name,
2086 duration_seconds=time.monotonic() - start_time,
2087 server_address=getattr(conn, "host", None),
2088 server_port=getattr(conn, "port", None),
2089 db_namespace=str(conn.db),
2090 error=error,
2091 retry_attempts=failure_count,
2092 )
2093 await conn.disconnect(error=error, failure_count=failure_count)
2094 # if we were watching a variable, the watch is no longer valid
2095 # since this connection has died. raise a WatchError, which
2096 # indicates the user should retry this transaction.
2097 if self.watching:
2098 raise WatchError(
2099 f"A {type(error).__name__} occurred while watching one or more keys"
2100 )
2102 async def execute(self, raise_on_error: bool = True) -> List[Any]:
2103 """Execute all the commands in the current pipeline"""
2104 stack = self.command_stack
2105 if not stack and not self.watching:
2106 return []
2107 if self.scripts:
2108 await self.load_scripts()
2109 if self.is_transaction or self.explicit_transaction:
2110 execute = self._execute_transaction
2111 operation_name = "MULTI"
2112 else:
2113 execute = self._execute_pipeline
2114 operation_name = "PIPELINE"
2116 conn = self.connection
2117 if not conn:
2118 conn = await self.connection_pool.get_connection()
2119 # assign to self.connection so reset() releases the connection
2120 # back to the pool after we're done
2121 self.connection = conn
2122 conn = cast(Connection, conn)
2124 # Start timing for observability
2125 start_time = time.monotonic()
2126 # Track actual retry attempts for error reporting
2127 actual_retry_attempts = 0
2129 def failure_callback(error, failure_count):
2130 nonlocal actual_retry_attempts
2131 actual_retry_attempts = failure_count
2132 return self._disconnect_raise_on_watching(
2133 conn, error, failure_count, start_time, operation_name
2134 )
2136 try:
2137 response = await conn.retry.call_with_retry(
2138 lambda: execute(conn, stack, raise_on_error),
2139 failure_callback,
2140 with_failure_count=True,
2141 )
2143 await record_operation_duration(
2144 command_name=operation_name,
2145 duration_seconds=time.monotonic() - start_time,
2146 server_address=getattr(conn, "host", None),
2147 server_port=getattr(conn, "port", None),
2148 db_namespace=str(conn.db),
2149 )
2150 return response
2151 except Exception as e:
2152 await record_error_count(
2153 server_address=getattr(conn, "host", None),
2154 server_port=getattr(conn, "port", None),
2155 network_peer_address=getattr(conn, "host", None),
2156 network_peer_port=getattr(conn, "port", None),
2157 error_type=e,
2158 retry_attempts=actual_retry_attempts,
2159 is_internal=False,
2160 )
2161 raise
2162 finally:
2163 await self.reset()
2165 async def discard(self):
2166 """Flushes all previously queued commands
2167 See: https://redis.io/commands/DISCARD
2168 """
2169 await self.execute_command("DISCARD")
2171 async def watch(self, *names: KeyT):
2172 """Watches the values at keys ``names``"""
2173 if self.explicit_transaction:
2174 raise RedisError("Cannot issue a WATCH after a MULTI")
2175 return await self.execute_command("WATCH", *names)
2177 async def unwatch(self):
2178 """Unwatches all previously specified keys"""
2179 return self.watching and await self.execute_command("UNWATCH") or True