Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/redis/client.py: 20%
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 copy
2import logging
3import re
4import threading
5import time
6from itertools import chain
7from typing import (
8 TYPE_CHECKING,
9 Any,
10 Callable,
11 Dict,
12 Iterable,
13 List,
14 Literal,
15 Mapping,
16 Optional,
17 Set,
18 Type,
19 Union,
20)
22from redis import _himport_exec
23from redis._defaults import (
24 DEFAULT_RETRY_BASE,
25 DEFAULT_RETRY_CAP,
26 DEFAULT_RETRY_COUNT,
27 DEFAULT_SOCKET_CONNECT_TIMEOUT,
28 DEFAULT_SOCKET_READ_SIZE,
29 DEFAULT_SOCKET_TIMEOUT,
30)
31from redis._parsers.encoders import Encoder
32from redis._parsers.helpers import bool_ok, get_response_callbacks
33from redis.backoff import ExponentialWithJitterBackoff
34from redis.cache import CacheConfig, CacheInterface
35from redis.commands import (
36 CoreCommands,
37 RedisModuleCommands,
38 SentinelCommands,
39 list_or_args,
40)
41from redis.commands.core import Script
42from redis.commands.helpers import parse_pubsub_subscriptions, pubsub_subscription_args
43from redis.connection import (
44 AbstractConnection,
45 Connection,
46 ConnectionPool,
47 SSLConnection,
48 UnixDomainSocketConnection,
49)
50from redis.credentials import CredentialProvider
51from redis.driver_info import DriverInfo, resolve_driver_info
52from redis.event import (
53 AfterPooledConnectionsInstantiationEvent,
54 AfterPubSubConnectionInstantiationEvent,
55 AfterSingleConnectionInstantiationEvent,
56 ClientType,
57 EventDispatcher,
58)
59from redis.exceptions import (
60 ConnectionError,
61 ExecAbortError,
62 PubSubError,
63 RedisError,
64 ResponseError,
65 WatchError,
66)
67from redis.himport import HImportRegistry, parse_himport_set_args
68from redis.lock import Lock
69from redis.maint_notifications import (
70 MaintNotificationsConfig,
71 OSSMaintNotificationsHandler,
72)
73from redis.observability.attributes import PubSubDirection
74from redis.observability.recorder import (
75 record_error_count,
76 record_operation_duration,
77 record_pubsub_message,
78)
79from redis.retry import Retry
80from redis.typing import (
81 ChannelT,
82 FieldT,
83 PubSubHandler,
84 Subscription,
85)
86from redis.utils import (
87 SENTINEL,
88 _set_info_logger,
89 check_protocol_version,
90 deprecated_args,
91 experimental_method,
92 safe_str,
93 str_if_bytes,
94 truncate_text,
95)
97if TYPE_CHECKING:
98 import ssl
100 import OpenSSL
102 from redis.keyspace_notifications import KeyspaceNotifications
104SYM_EMPTY = b""
105EMPTY_RESPONSE = "EMPTY_RESPONSE"
107# some responses (ie. dump) are binary, and just meant to never be decoded
108NEVER_DECODE = "NEVER_DECODE"
111logger = logging.getLogger(__name__)
114def is_debug_log_enabled():
115 return logger.isEnabledFor(logging.DEBUG)
118def add_debug_log_for_operation_failure(connection: "AbstractConnection"):
119 logger.debug(
120 f"Operation failed, "
121 f"with connection: {connection}, details: {connection.extract_connection_details() if connection else 'no connection'}",
122 )
125class CaseInsensitiveDict(dict):
126 "Case insensitive dict implementation. Assumes string keys only."
128 def __init__(self, data: Dict[str, str]) -> None:
129 for k, v in data.items():
130 self[k.upper()] = v
132 def __contains__(self, k):
133 return super().__contains__(k.upper())
135 def __delitem__(self, k):
136 super().__delitem__(k.upper())
138 def __getitem__(self, k):
139 return super().__getitem__(k.upper())
141 def get(self, k, default=None):
142 return super().get(k.upper(), default)
144 def __setitem__(self, k, v):
145 super().__setitem__(k.upper(), v)
147 def update(self, data):
148 data = CaseInsensitiveDict(data)
149 super().update(data)
152class AbstractRedis:
153 pass
156class Redis(RedisModuleCommands, CoreCommands, SentinelCommands):
157 """
158 Implementation of the Redis protocol.
160 This abstract class provides a Python interface to all Redis commands
161 and an implementation of the Redis protocol.
163 Pipelines derive from this, implementing how
164 the commands are sent and received to the Redis server. Based on
165 configuration, an instance will either use a ConnectionPool, or
166 Connection object to talk to redis.
168 It is not safe to pass PubSub or Pipeline objects between threads.
169 """
171 # Type discrimination marker for @overload self-type pattern
172 _is_async_client: Literal[False] = False
174 @classmethod
175 def from_url(cls, url: str, **kwargs) -> "Redis":
176 """
177 Return a Redis client object configured from the given URL
179 For example::
181 redis://[[username]:[password]]@localhost:6379/0
182 rediss://[[username]:[password]]@localhost:6379/0
183 unix://[username@]/path/to/socket.sock?db=0[&password=password]
185 Three URL schemes are supported:
187 - `redis://` creates a TCP socket connection. See more at:
188 <https://www.iana.org/assignments/uri-schemes/prov/redis>
189 - `rediss://` creates a SSL wrapped TCP socket connection. See more at:
190 <https://www.iana.org/assignments/uri-schemes/prov/rediss>
191 - ``unix://``: creates a Unix Domain Socket connection.
193 The username, password, hostname and path are passed through
194 urllib.parse.unquote in order to replace any percent-encoded values
195 with their corresponding characters. Querystring values are decoded
196 by urllib.parse.parse_qs and are not unquoted again.
198 There are several ways to specify a database number. The first value
199 found will be used:
201 1. A ``db`` querystring option, e.g. redis://localhost?db=0
202 2. If using the redis:// or rediss:// schemes, the path argument
203 of the url, e.g. redis://localhost/0
204 3. A ``db`` keyword argument to this function.
206 If none of these options are specified, the default db=0 is used.
208 All querystring options are cast to their appropriate Python types.
209 Boolean arguments can be specified with string values "True"/"False"
210 or "Yes"/"No". Values that cannot be properly cast cause a
211 ``ValueError`` to be raised. Once parsed, the querystring arguments
212 and keyword arguments are passed to the ``ConnectionPool``'s
213 class initializer. In the case of conflicting arguments, querystring
214 arguments always win.
216 """
217 single_connection_client = kwargs.pop("single_connection_client", False)
218 connection_pool = ConnectionPool.from_url(url, **kwargs)
219 client = cls(
220 connection_pool=connection_pool,
221 single_connection_client=single_connection_client,
222 )
223 client.auto_close_connection_pool = True
224 return client
226 @classmethod
227 def from_pool(
228 cls: Type["Redis"],
229 connection_pool: ConnectionPool,
230 ) -> "Redis":
231 """
232 Return a Redis client from the given connection pool.
233 The Redis client will take ownership of the connection pool and
234 close it when the Redis client is closed.
236 Because the client closes (disconnects all connections in) the pool
237 when it is closed or garbage-collected, the pool must not be shared
238 with other clients. Constructing multiple clients from the same pool
239 via ``from_pool`` -- for example one per request across threads -- is
240 not thread safe: when one client is closed it will disconnect
241 connections still in use by the others.
243 To share a single pool across clients, construct the pool explicitly
244 and manage its lifecycle instead. Unlike ``from_pool``, the plain
245 ``Redis(connection_pool=pool)`` constructor does not take ownership of
246 the pool and will not close it, so a pool created this way can be
247 safely shared across clients. ``ConnectionPool`` supports the context
248 manager protocol for this::
250 with ConnectionPool.from_url(url) as pool:
251 r = Redis(connection_pool=pool)
252 """
253 client = cls(
254 connection_pool=connection_pool,
255 )
256 client.auto_close_connection_pool = True
257 return client
259 @deprecated_args(
260 args_to_warn=["retry_on_timeout"],
261 reason="TimeoutError is included by default.",
262 version="6.0.0",
263 )
264 @deprecated_args(
265 args_to_warn=["lib_name", "lib_version"],
266 reason="Use 'driver_info' parameter instead. "
267 "lib_name and lib_version will be removed in a future version.",
268 )
269 def __init__(
270 self,
271 host: str = "localhost",
272 port: int = 6379,
273 db: int = 0,
274 password: str | None = None,
275 socket_timeout: float | None = DEFAULT_SOCKET_TIMEOUT,
276 socket_connect_timeout: float | None = DEFAULT_SOCKET_CONNECT_TIMEOUT,
277 socket_read_size: int = DEFAULT_SOCKET_READ_SIZE,
278 socket_keepalive: bool | None = True,
279 socket_keepalive_options: Mapping[int, int | bytes] | object | None = SENTINEL,
280 connection_pool: ConnectionPool | None = None,
281 unix_socket_path: str | None = None,
282 encoding: str = "utf-8",
283 encoding_errors: str = "strict",
284 decode_responses: bool = False,
285 retry_on_timeout: bool = False,
286 retry: Retry = Retry(
287 backoff=ExponentialWithJitterBackoff(
288 base=DEFAULT_RETRY_BASE, cap=DEFAULT_RETRY_CAP
289 ),
290 retries=DEFAULT_RETRY_COUNT,
291 ),
292 retry_on_error: List[Type[Exception]] | None = None,
293 ssl: bool = False,
294 ssl_keyfile: str | None = None,
295 ssl_certfile: str | None = None,
296 ssl_cert_reqs: "str | ssl.VerifyMode" = "required",
297 ssl_include_verify_flags: List["ssl.VerifyFlags"] | None = None,
298 ssl_exclude_verify_flags: List["ssl.VerifyFlags"] | None = None,
299 ssl_ca_certs: str | None = None,
300 ssl_ca_path: str | None = None,
301 ssl_ca_data: str | None = None,
302 ssl_check_hostname: bool = True,
303 ssl_password: str | None = None,
304 ssl_validate_ocsp: bool = False,
305 ssl_validate_ocsp_stapled: bool = False,
306 ssl_ocsp_context: "OpenSSL.SSL.Context | None" = None,
307 ssl_ocsp_expected_cert: str | None = None,
308 ssl_min_version: "ssl.TLSVersion | None" = None,
309 ssl_ciphers: str | None = None,
310 max_connections: int | None = None,
311 single_connection_client: bool = False,
312 health_check_interval: int = 0,
313 client_name: str | None = None,
314 lib_name: str | object | None = SENTINEL,
315 lib_version: str | object | None = SENTINEL,
316 driver_info: DriverInfo | object | None = SENTINEL,
317 username: str | None = None,
318 redis_connect_func: Callable[[], None] | None = None,
319 credential_provider: CredentialProvider | None = None,
320 protocol: int | None = None,
321 legacy_responses: bool = True,
322 cache: CacheInterface | None = None,
323 cache_config: CacheConfig | None = None,
324 event_dispatcher: EventDispatcher | None = None,
325 maint_notifications_config: MaintNotificationsConfig | None = None,
326 oss_cluster_maint_notifications_handler: OSSMaintNotificationsHandler
327 | None = None,
328 ) -> None:
329 """
330 Initialize a new Redis client.
332 To specify a retry policy for specific errors, you have two options:
334 1. Set the `retry_on_error` to a list of the error/s to retry on, and
335 you can also set `retry` to a valid `Retry` object(in case the default
336 one is not appropriate) - with this approach the retries will be triggered
337 on the default errors specified in the Retry object enriched with the
338 errors specified in `retry_on_error`.
340 2. Define a `Retry` object with configured 'supported_errors' and set
341 it to the `retry` parameter - with this approach you completely redefine
342 the errors on which retries will happen.
344 `retry_on_timeout` is deprecated - please include the TimeoutError
345 either in the Retry object or in the `retry_on_error` list.
347 When 'connection_pool' is provided - the retry configuration of the
348 provided pool will be used.
350 Args:
352 socket_keepalive:
353 if `True`, TCP keepalive is enabled for TCP socket connections.
354 Argument is ignored when connection_pool is provided.
355 socket_keepalive_options:
356 mapping of TCP keepalive socket option constants to values, for
357 example `{socket.TCP_KEEPIDLE: 30}`. If left unspecified, redis-py
358 uses TCP keepalive defaults when `socket_keepalive` is enabled:
359 idle 30 seconds, interval 5 seconds, and 3 probes. Platform-specific
360 options that are not available are skipped. Pass `None` or `{}` to
361 avoid setting additional TCP keepalive options. Argument is ignored
362 when connection_pool is provided.
363 single_connection_client:
364 if `True`, connection pool is not used. In that case `Redis`
365 instance use is not thread safe.
366 decode_responses:
367 if `True`, the response will be decoded to utf-8.
368 Argument is ignored when connection_pool is provided.
369 driver_info:
370 Optional DriverInfo object to identify upstream libraries.
371 If provided, lib_name and lib_version are ignored.
372 If not provided, a DriverInfo will be created from lib_name and lib_version.
373 Explicit None disables CLIENT SETINFO.
374 Argument is ignored when connection_pool is provided.
375 lib_name:
376 **Deprecated.** Use driver_info instead. Library name for CLIENT SETINFO.
377 lib_version:
378 **Deprecated.** Use driver_info instead. Library version for CLIENT SETINFO.
379 maint_notifications_config:
380 configures the pool to support maintenance notifications - see
381 `redis.maint_notifications.MaintNotificationsConfig` for details.
382 Only supported with RESP3
383 If not provided and protocol is RESP3, the maintenance notifications
384 will be enabled by default (logic is included in the connection pool
385 initialization).
386 Argument is ignored when connection_pool is provided.
387 oss_cluster_maint_notifications_handler:
388 handler for OSS cluster notifications - see
389 `redis.maint_notifications.OSSMaintNotificationsHandler` for details.
390 Only supported with RESP3
391 Argument is ignored when connection_pool is provided.
392 """
393 if event_dispatcher is None:
394 self._event_dispatcher = EventDispatcher()
395 else:
396 self._event_dispatcher = event_dispatcher
397 if not connection_pool:
398 if not retry_on_error:
399 retry_on_error = []
401 # Handle driver_info: if provided, use it; otherwise create from lib_name/lib_version.
402 computed_driver_info = resolve_driver_info(
403 driver_info, lib_name, lib_version
404 )
406 kwargs = {
407 "db": db,
408 "username": username,
409 "password": password,
410 "socket_timeout": socket_timeout,
411 "socket_read_size": socket_read_size,
412 "encoding": encoding,
413 "encoding_errors": encoding_errors,
414 "decode_responses": decode_responses,
415 "retry_on_error": retry_on_error,
416 "retry": copy.deepcopy(retry),
417 "max_connections": max_connections,
418 "health_check_interval": health_check_interval,
419 "client_name": client_name,
420 "driver_info": computed_driver_info,
421 "redis_connect_func": redis_connect_func,
422 "credential_provider": credential_provider,
423 "protocol": protocol,
424 "legacy_responses": legacy_responses,
425 }
426 # based on input, setup appropriate connection args
427 if unix_socket_path is not None:
428 if (
429 maint_notifications_config
430 and maint_notifications_config.enabled is True
431 ):
432 raise RedisError(
433 "Maintenance notifications are not supported with Unix "
434 "domain socket connections"
435 )
436 kwargs.update(
437 {
438 "path": unix_socket_path,
439 "connection_class": UnixDomainSocketConnection,
440 "maint_notifications_config": MaintNotificationsConfig(
441 enabled=False
442 ),
443 }
444 )
445 else:
446 # TCP specific options
447 kwargs.update(
448 {
449 "host": host,
450 "port": port,
451 "socket_connect_timeout": socket_connect_timeout,
452 "socket_keepalive": socket_keepalive,
453 "socket_keepalive_options": socket_keepalive_options,
454 }
455 )
457 if ssl:
458 kwargs.update(
459 {
460 "connection_class": SSLConnection,
461 "ssl_keyfile": ssl_keyfile,
462 "ssl_certfile": ssl_certfile,
463 "ssl_cert_reqs": ssl_cert_reqs,
464 "ssl_include_verify_flags": ssl_include_verify_flags,
465 "ssl_exclude_verify_flags": ssl_exclude_verify_flags,
466 "ssl_ca_certs": ssl_ca_certs,
467 "ssl_ca_data": ssl_ca_data,
468 "ssl_check_hostname": ssl_check_hostname,
469 "ssl_password": ssl_password,
470 "ssl_ca_path": ssl_ca_path,
471 "ssl_validate_ocsp_stapled": ssl_validate_ocsp_stapled,
472 "ssl_validate_ocsp": ssl_validate_ocsp,
473 "ssl_ocsp_context": ssl_ocsp_context,
474 "ssl_ocsp_expected_cert": ssl_ocsp_expected_cert,
475 "ssl_min_version": ssl_min_version,
476 "ssl_ciphers": ssl_ciphers,
477 }
478 )
479 if (cache_config or cache) and check_protocol_version(protocol, 3):
480 kwargs.update(
481 {
482 "cache": cache,
483 "cache_config": cache_config,
484 }
485 )
486 maint_notifications_enabled = (
487 maint_notifications_config and maint_notifications_config.enabled
488 )
489 if maint_notifications_enabled and not check_protocol_version(
490 protocol, 3
491 ):
492 raise RedisError(
493 "Maintenance notifications handlers on connection are only supported with RESP version 3"
494 )
495 if maint_notifications_config:
496 kwargs.update(
497 {
498 "maint_notifications_config": maint_notifications_config,
499 }
500 )
501 if oss_cluster_maint_notifications_handler:
502 kwargs.update(
503 {
504 "oss_cluster_maint_notifications_handler": oss_cluster_maint_notifications_handler,
505 }
506 )
507 connection_pool = ConnectionPool(**kwargs)
508 self._event_dispatcher.dispatch(
509 AfterPooledConnectionsInstantiationEvent(
510 [connection_pool], ClientType.SYNC, credential_provider
511 )
512 )
513 self.auto_close_connection_pool = True
514 else:
515 self.auto_close_connection_pool = False
516 self._event_dispatcher.dispatch(
517 AfterPooledConnectionsInstantiationEvent(
518 [connection_pool], ClientType.SYNC, credential_provider
519 )
520 )
522 self.connection_pool = connection_pool
524 if (cache_config or cache) and not check_protocol_version(
525 self.connection_pool.get_protocol(), 3
526 ):
527 raise RedisError("Client caching is only supported with RESP version 3")
529 self.single_connection_lock = threading.RLock()
530 self.connection = None
531 self._single_connection_client = single_connection_client
532 if self._single_connection_client:
533 self.connection = self.connection_pool.get_connection()
534 self._event_dispatcher.dispatch(
535 AfterSingleConnectionInstantiationEvent(
536 self.connection, ClientType.SYNC, self.single_connection_lock
537 )
538 )
540 connection_kwargs = self.connection_pool.connection_kwargs
541 self.response_callbacks = CaseInsensitiveDict(
542 get_response_callbacks(
543 user_protocol=connection_kwargs.get("protocol"),
544 legacy_responses=connection_kwargs.get("legacy_responses", True),
545 )
546 )
548 def __repr__(self) -> str:
549 return (
550 f"<{type(self).__module__}.{type(self).__name__}"
551 f"({repr(self.connection_pool)})>"
552 )
554 def get_encoder(self) -> "Encoder":
555 """Get the connection pool's encoder"""
556 return self.connection_pool.get_encoder()
558 def get_connection_kwargs(self) -> Dict:
559 """Get the connection's key-word arguments"""
560 return self.connection_pool.connection_kwargs
562 @property
563 def himport_registry(self) -> HImportRegistry:
564 """The client's HIMPORT fieldset registry (contains empty
565 schema registry if none was declared).
567 Read-only: the registry is mutated only through the HIMPORT command methods.
568 """
569 return self.connection_pool.himport_registry
571 def get_retry(self) -> Optional[Retry]:
572 return self.get_connection_kwargs().get("retry")
574 def set_retry(self, retry: Retry) -> None:
575 self.get_connection_kwargs().update({"retry": retry})
576 self.connection_pool.set_retry(retry)
578 def set_response_callback(self, command: str, callback: Callable) -> None:
579 """Set a custom Response Callback"""
580 self.response_callbacks[command] = callback
582 def load_external_module(self, funcname, func) -> None:
583 """
584 This function can be used to add externally defined redis modules,
585 and their namespaces to the redis client.
587 funcname - A string containing the name of the function to create
588 func - The function, being added to this class.
590 ex: Assume that one has a custom redis module named foomod that
591 creates command named 'foo.dothing' and 'foo.anotherthing' in redis.
592 To load function functions into this namespace:
594 from redis import Redis
595 from foomodule import F
596 r = Redis()
597 r.load_external_module("foo", F)
598 r.foo().dothing('your', 'arguments')
600 For a concrete example see the reimport of the redisjson module in
601 tests/test_connection.py::test_loading_external_modules
602 """
603 setattr(self, funcname, func)
605 def pipeline(self, transaction=True, shard_hint=None) -> "Pipeline":
606 """
607 Return a new pipeline object that can queue multiple commands for
608 later execution. ``transaction`` indicates whether all commands
609 should be executed atomically. Apart from making a group of operations
610 atomic, pipelines are useful for reducing the back-and-forth overhead
611 between the client and server.
612 """
613 return Pipeline(
614 self.connection_pool, self.response_callbacks, transaction, shard_hint
615 )
617 def transaction(
618 self, func: Callable[["Pipeline"], None], *watches, **kwargs
619 ) -> Union[List[Any], Any, None]:
620 """
621 Convenience method for executing the callable `func` as a transaction
622 while watching all keys specified in `watches`. The 'func' callable
623 should expect a single argument which is a Pipeline object.
624 """
625 shard_hint = kwargs.pop("shard_hint", None)
626 value_from_callable = kwargs.pop("value_from_callable", False)
627 watch_delay = kwargs.pop("watch_delay", None)
628 with self.pipeline(True, shard_hint) as pipe:
629 while True:
630 try:
631 if watches:
632 pipe.watch(*watches)
633 func_value = func(pipe)
634 exec_value = pipe.execute()
635 return func_value if value_from_callable else exec_value
636 except WatchError:
637 if watch_delay is not None and watch_delay > 0:
638 time.sleep(watch_delay)
639 continue
641 def lock(
642 self,
643 name: str,
644 timeout: Optional[float] = None,
645 sleep: float = 0.1,
646 blocking: bool = True,
647 blocking_timeout: Optional[float] = None,
648 lock_class: Union[None, Any] = None,
649 thread_local: bool = True,
650 raise_on_release_error: bool = True,
651 ):
652 """
653 Return a new Lock object using key ``name`` that mimics
654 the behavior of threading.Lock.
656 If specified, ``timeout`` indicates a maximum life for the lock.
657 By default, it will remain locked until release() is called.
659 ``sleep`` indicates the amount of time to sleep per loop iteration
660 when the lock is in blocking mode and another client is currently
661 holding the lock.
663 ``blocking`` indicates whether calling ``acquire`` should block until
664 the lock has been acquired or to fail immediately, causing ``acquire``
665 to return False and the lock not being acquired. Defaults to True.
666 Note this value can be overridden by passing a ``blocking``
667 argument to ``acquire``.
669 ``blocking_timeout`` indicates the maximum amount of time in seconds to
670 spend trying to acquire the lock. A value of ``None`` indicates
671 continue trying forever. ``blocking_timeout`` can be specified as a
672 float or integer, both representing the number of seconds to wait.
674 ``lock_class`` forces the specified lock implementation. Note that as
675 of redis-py 3.0, the only lock class we implement is ``Lock`` (which is
676 a Lua-based lock). So, it's unlikely you'll need this parameter, unless
677 you have created your own custom lock class.
679 ``thread_local`` indicates whether the lock token is placed in
680 thread-local storage. By default, the token is placed in thread local
681 storage so that a thread only sees its token, not a token set by
682 another thread. Consider the following timeline:
684 time: 0, thread-1 acquires `my-lock`, with a timeout of 5 seconds.
685 thread-1 sets the token to "abc"
686 time: 1, thread-2 blocks trying to acquire `my-lock` using the
687 Lock instance.
688 time: 5, thread-1 has not yet completed. redis expires the lock
689 key.
690 time: 5, thread-2 acquired `my-lock` now that it's available.
691 thread-2 sets the token to "xyz"
692 time: 6, thread-1 finishes its work and calls release(). if the
693 token is *not* stored in thread local storage, then
694 thread-1 would see the token value as "xyz" and would be
695 able to successfully release the thread-2's lock.
697 ``raise_on_release_error`` indicates whether to raise an exception when
698 the lock is no longer owned when exiting the context manager. By default,
699 this is True, meaning an exception will be raised. If False, the warning
700 will be logged and the exception will be suppressed.
702 In some use cases it's necessary to disable thread local storage. For
703 example, if you have code where one thread acquires a lock and passes
704 that lock instance to a worker thread to release later. If thread
705 local storage isn't disabled in this case, the worker thread won't see
706 the token set by the thread that acquired the lock. Our assumption
707 is that these cases aren't common and as such default to using
708 thread local storage."""
709 if lock_class is None:
710 lock_class = Lock
711 return lock_class(
712 self,
713 name,
714 timeout=timeout,
715 sleep=sleep,
716 blocking=blocking,
717 blocking_timeout=blocking_timeout,
718 thread_local=thread_local,
719 raise_on_release_error=raise_on_release_error,
720 )
722 def pubsub(self, **kwargs):
723 """
724 Return a Publish/Subscribe object. With this object, you can
725 subscribe to channels and listen for messages that get published to
726 them.
727 """
728 return PubSub(
729 self.connection_pool, event_dispatcher=self._event_dispatcher, **kwargs
730 )
732 def keyspace_notifications(
733 self,
734 key_prefix: Union[str, bytes, None] = None,
735 ignore_subscribe_messages: bool = True,
736 ) -> "KeyspaceNotifications":
737 """
738 Return a :class:`~redis.keyspace_notifications.KeyspaceNotifications`
739 object for subscribing to keyspace and keyevent notifications.
741 Note: Keyspace notifications must be enabled on the Redis server via
742 the ``notify-keyspace-events`` configuration option.
744 Args:
745 key_prefix: Optional prefix to filter and strip from keys in
746 notifications.
747 ignore_subscribe_messages: If True, subscribe/unsubscribe
748 confirmations are not returned by
749 get_message/listen.
750 """
751 from redis.keyspace_notifications import KeyspaceNotifications
753 return KeyspaceNotifications(
754 self,
755 key_prefix=key_prefix,
756 ignore_subscribe_messages=ignore_subscribe_messages,
757 )
759 def monitor(self):
760 return Monitor(self.connection_pool)
762 def client(self):
763 return self.__class__(
764 connection_pool=self.connection_pool,
765 single_connection_client=True,
766 )
768 def __enter__(self):
769 return self
771 def __exit__(self, exc_type, exc_value, traceback):
772 self.close()
774 def __del__(self):
775 try:
776 self.close()
777 except Exception:
778 pass
780 def close(self) -> None:
781 # In case a connection property does not yet exist
782 # (due to a crash earlier in the Redis() constructor), return
783 # immediately as there is nothing to clean-up.
784 if not hasattr(self, "connection"):
785 return
787 conn = self.connection
788 if conn:
789 self.connection = None
790 self.connection_pool.release(conn)
792 if self.auto_close_connection_pool:
793 self.connection_pool.close()
795 def _send_command_parse_response(self, conn, command_name, *args, **options):
796 """
797 Send a command and parse the response
798 """
799 # HIMPORT SET is the one command whose wire form depends on per-connection
800 # state: the fieldset must be PREPAREd on this connection first, and any
801 # fieldset discarded since this connection last reconciled must be dropped.
802 # Handling it here (rather than in himport_set) lets himport_set reuse the
803 # full execute_command machinery — retry, disconnect-on-error, pooling — so
804 # a failed HIMPORT SET disconnects the connection like any other command.
805 # This per-command branch in the hot dispatch path is deliberate and has no
806 # cleaner alternative: this is the only seam where the concrete borrowed
807 # connection is known, and connection-scoped session setup can only happen
808 # once that connection is chosen. The overhead is one string compare per
809 # command.
810 himport_set = parse_himport_set_args(args)
811 if himport_set is not None:
812 # ``args`` is an HIMPORT SET in either the joined ("HIMPORT SET", key,
813 # ...) or split ("HIMPORT", "SET", key, ...) raw form; the operands come
814 # back at the right offsets for the form. A command with too few operands
815 # returns None and falls through to the normal send path so the server
816 # returns its arity error instead of a client-side IndexError here.
817 key, fieldset_name, values = himport_set
818 return self._himport_execute_set(conn, key, fieldset_name, values)
819 conn.send_command(*args, **options)
820 return self.parse_response(conn, command_name, **options)
822 def _himport_reconcile_discards(self, conn):
823 """Delegate to the shared sync HIMPORT executor."""
824 return _himport_exec.reconcile_discards(self, conn)
826 def _himport_prepare_and_set(self, conn, key, fieldset_name, values, fieldset):
827 """Delegate to the shared sync HIMPORT executor."""
828 return _himport_exec.prepare_and_set(
829 self, conn, key, fieldset_name, values, fieldset
830 )
832 def _himport_execute_set(self, conn, key, fieldset_name, values):
833 """Delegate to the shared sync HIMPORT executor."""
834 return _himport_exec.execute_set(self, conn, key, fieldset_name, values)
836 def _close_connection(
837 self,
838 conn,
839 error: Optional[Exception] = None,
840 failure_count: Optional[int] = None,
841 start_time: Optional[float] = None,
842 command_name: Optional[str] = None,
843 ) -> None:
844 """
845 Close the connection before retrying.
847 The supported exceptions are already checked in the
848 retry object so we don't need to do it here.
850 After we disconnect the connection, it will try to reconnect and
851 do a health check as part of the send_command logic(on connection level).
852 """
853 if error and failure_count <= conn.retry.get_retries():
854 record_operation_duration(
855 command_name=command_name,
856 duration_seconds=time.monotonic() - start_time,
857 server_address=getattr(conn, "host", None),
858 server_port=getattr(conn, "port", None),
859 db_namespace=str(conn.db),
860 error=error,
861 retry_attempts=failure_count,
862 )
864 conn.disconnect()
866 # COMMAND EXECUTION AND PROTOCOL PARSING
867 def execute_command(self, *args, **options):
868 return self._execute_command(*args, **options)
870 def _execute_command(self, *args, **options):
871 """Execute a command and return a parsed response"""
872 pool = self.connection_pool
873 command_name = args[0]
874 conn = self.connection or pool.get_connection()
876 # Start timing for observability
877 start_time = time.monotonic()
878 # Track actual retry attempts for error reporting
879 actual_retry_attempts = [0]
881 def failure_callback(error, failure_count):
882 if is_debug_log_enabled():
883 add_debug_log_for_operation_failure(conn)
884 actual_retry_attempts[0] = failure_count
885 self._close_connection(conn, error, failure_count, start_time, command_name)
887 if self._single_connection_client:
888 self.single_connection_lock.acquire()
889 try:
890 result = conn.retry.call_with_retry(
891 lambda: self._send_command_parse_response(
892 conn, command_name, *args, **options
893 ),
894 failure_callback,
895 with_failure_count=True,
896 )
898 record_operation_duration(
899 command_name=command_name,
900 duration_seconds=time.monotonic() - start_time,
901 server_address=getattr(conn, "host", None),
902 server_port=getattr(conn, "port", None),
903 db_namespace=str(conn.db),
904 )
905 return result
906 except Exception as e:
907 record_error_count(
908 server_address=getattr(conn, "host", None),
909 server_port=getattr(conn, "port", None),
910 network_peer_address=getattr(conn, "host", None),
911 network_peer_port=getattr(conn, "port", None),
912 error_type=e,
913 retry_attempts=actual_retry_attempts[0],
914 is_internal=False,
915 )
916 raise
918 finally:
919 try:
920 if conn and conn.should_reconnect():
921 self._close_connection(conn)
922 conn.connect()
923 finally:
924 if self._single_connection_client:
925 self.single_connection_lock.release()
926 if not self.connection:
927 pool.release(conn)
929 def parse_response(self, connection, command_name, **options):
930 """Parses a response from the Redis server"""
931 try:
932 if NEVER_DECODE in options:
933 response = connection.read_response(disable_decoding=True)
934 options.pop(NEVER_DECODE)
935 else:
936 response = connection.read_response()
937 except ResponseError:
938 if EMPTY_RESPONSE in options:
939 return options[EMPTY_RESPONSE]
940 raise
942 if EMPTY_RESPONSE in options:
943 options.pop(EMPTY_RESPONSE)
945 # Remove keys entry, it needs only for cache.
946 options.pop("keys", None)
948 if command_name in self.response_callbacks:
949 return self.response_callbacks[command_name](response, **options)
950 return response
952 def get_cache(self) -> Optional[CacheInterface]:
953 return self.connection_pool.cache
955 # HIMPORT orchestration. The registry lives on the shared HImportRegistry; the
956 # server-side effect is applied lazily per connection (PREPARE bundled into the
957 # first himport_set; DISCARD reconciled when a connection is next borrowed for a
958 # himport_set). The connection carries the per-connection HIMPORT state; a
959 # CacheProxyConnection transparently delegates it to the wrapped connection, so
960 # this code never needs to know which connection type it holds.
962 @experimental_method()
963 def himport_prepare(self, fieldset_name: str, fields: Iterable[FieldT]) -> bool:
964 """Declare an HIMPORT fieldset for use by :meth:`himport_set`.
966 Registers ``fieldset_name`` (ordered ``fields``, verbatim) in the client's
967 shared registry. On a pooled client the server-side ``PREPARE`` is deferred
968 and bundled into the next ``himport_set`` per connection. On a single
969 connection client it is run immediately when the pinned connection is live;
970 while that connection is not connected there is no session state to prepare,
971 so the next ``himport_set`` prepares it lazily instead.
972 """
973 fieldset = self.himport_registry.prepare(fieldset_name, fields)
974 conn = self.connection
975 if self._single_connection_client and conn is not None and conn.is_connected:
976 self.himport_prepare_internal(fieldset_name, fieldset.fields)
977 conn._himport_prepared[fieldset_name] = fieldset.version
978 return True
980 @experimental_method()
981 def himport_discard(self, fieldset_name: str) -> int:
982 """Remove a fieldset from the registry.
984 Returns ``1`` if it was registered, ``0`` otherwise. On a pooled client the
985 server-side ``DISCARD`` is reconciled lazily when each connection is next
986 used for ``himport_set``. On a single connection client it runs immediately
987 on the pinned connection when it is live; while that connection is not
988 connected there is nothing prepared on the server to discard (its tracking is
989 reset on connect), so no server call is made.
990 """
991 removed = self.himport_registry.discard(fieldset_name)
992 conn = self.connection
993 if self._single_connection_client and conn is not None and conn.is_connected:
994 if removed:
995 self.himport_discard_internal(fieldset_name)
996 conn._himport_prepared.pop(fieldset_name, None)
997 conn._himport_reconciled_revision = self.himport_registry.revision
998 return 1 if removed else 0
1000 @experimental_method()
1001 def himport_discard_all(self) -> int:
1002 """Remove all fieldsets from the registry.
1004 Returns the number removed from the registry. Server-side removal follows the
1005 same live/lazy rule as :meth:`himport_discard`.
1006 """
1007 count = self.himport_registry.discard_all()
1008 conn = self.connection
1009 if self._single_connection_client and conn is not None and conn.is_connected:
1010 if count:
1011 self.himport_discard_all_internal()
1012 conn._himport_prepared.clear()
1013 conn._himport_reconciled_revision = self.himport_registry.revision
1014 return count
1017StrictRedis = Redis
1020class Monitor:
1021 """
1022 Monitor is useful for handling the MONITOR command to the redis server.
1023 next_command() method returns one command from monitor
1024 listen() method yields commands from monitor.
1025 """
1027 monitor_re = re.compile(r"\[(\d+) (.*?)\] (.*)")
1028 command_re = re.compile(r'"(.*?)(?<!\\)"')
1030 def __init__(self, connection_pool):
1031 self.connection_pool = connection_pool
1032 self.connection = self.connection_pool.get_connection()
1034 def __enter__(self):
1035 self._start_monitor()
1036 return self
1038 def __exit__(self, *args):
1039 self.connection.disconnect()
1040 self.connection_pool.release(self.connection)
1042 def next_command(self):
1043 """Parse the response from a monitor command"""
1044 response = self.connection.read_response()
1046 if response is None:
1047 return None
1049 if isinstance(response, bytes):
1050 response = self.connection.encoder.decode(response, force=True)
1052 command_time, command_data = response.split(" ", 1)
1053 m = self.monitor_re.match(command_data)
1054 db_id, client_info, command = m.groups()
1055 command = " ".join(self.command_re.findall(command))
1056 # Redis escapes double quotes because each piece of the command
1057 # string is surrounded by double quotes. We don't have that
1058 # requirement so remove the escaping and leave the quote.
1059 command = command.replace('\\"', '"')
1061 if client_info == "lua":
1062 client_address = "lua"
1063 client_port = ""
1064 client_type = "lua"
1065 elif client_info.startswith("unix"):
1066 client_address = "unix"
1067 client_port = client_info[5:]
1068 client_type = "unix"
1069 else:
1070 if client_info == "":
1071 client_address = ""
1072 client_port = ""
1073 client_type = "unknown"
1074 else:
1075 # use rsplit as ipv6 addresses contain colons
1076 client_address, client_port = client_info.rsplit(":", 1)
1077 client_type = "tcp"
1078 return {
1079 "time": float(command_time),
1080 "db": int(db_id),
1081 "client_address": client_address,
1082 "client_port": client_port,
1083 "client_type": client_type,
1084 "command": command,
1085 }
1087 def listen(self):
1088 """Listen for commands coming to the server."""
1089 while True:
1090 yield self.next_command()
1092 def _start_monitor(self):
1093 self.connection.send_command("MONITOR")
1094 # check that monitor returns 'OK', but don't return it to user
1095 response = self.connection.read_response()
1097 if not bool_ok(response):
1098 raise RedisError(f"MONITOR failed: {response}")
1101class PubSub:
1102 """
1103 PubSub provides publish, subscribe and listen support to Redis channels.
1105 After subscribing to one or more channels, the listen() method will block
1106 until a message arrives on one of the subscribed channels. That message
1107 will be returned and it's safe to start listening again.
1108 """
1110 PUBLISH_MESSAGE_TYPES = ("message", "pmessage", "smessage")
1111 UNSUBSCRIBE_MESSAGE_TYPES = ("unsubscribe", "punsubscribe", "sunsubscribe")
1112 HEALTH_CHECK_MESSAGE = "redis-py-health-check"
1114 def __init__(
1115 self,
1116 connection_pool,
1117 shard_hint=None,
1118 ignore_subscribe_messages: bool = False,
1119 encoder: Optional["Encoder"] = None,
1120 push_handler_func: Union[None, Callable[[str], None]] = None,
1121 event_dispatcher: Optional["EventDispatcher"] = None,
1122 ):
1123 self.connection_pool = connection_pool
1124 self.shard_hint = shard_hint
1125 self.ignore_subscribe_messages = ignore_subscribe_messages
1126 self.connection = None
1127 self.subscribed_event = threading.Event()
1128 # we need to know the encoding options for this connection in order
1129 # to lookup channel and pattern names for callback handlers.
1130 self.encoder = encoder
1131 self.push_handler_func = push_handler_func
1132 if event_dispatcher is None:
1133 self._event_dispatcher = EventDispatcher()
1134 else:
1135 self._event_dispatcher = event_dispatcher
1137 self._lock = threading.RLock()
1138 if self.encoder is None:
1139 self.encoder = self.connection_pool.get_encoder()
1140 self.health_check_response_b = self.encoder.encode(self.HEALTH_CHECK_MESSAGE)
1141 if self.encoder.decode_responses:
1142 self.health_check_response = ["pong", self.HEALTH_CHECK_MESSAGE]
1143 else:
1144 self.health_check_response = [b"pong", self.health_check_response_b]
1145 if self.push_handler_func is None:
1146 _set_info_logger()
1147 self.reset()
1149 def __enter__(self) -> "PubSub":
1150 return self
1152 def __exit__(self, exc_type, exc_value, traceback) -> None:
1153 self.reset()
1155 def __del__(self) -> None:
1156 try:
1157 # if this object went out of scope prior to shutting down
1158 # subscriptions, close the connection manually before
1159 # returning it to the connection pool
1160 self.reset()
1161 except Exception:
1162 pass
1164 def reset(self) -> None:
1165 if self.connection:
1166 self.connection.disconnect()
1167 self.connection.deregister_connect_callback(self.on_connect)
1168 self.connection_pool.release(self.connection)
1169 self.connection = None
1170 self.health_check_response_counter = 0
1171 self.channels = {}
1172 self.pending_unsubscribe_channels = set()
1173 self.shard_channels = {}
1174 self.pending_unsubscribe_shard_channels = set()
1175 self.patterns = {}
1176 self.pending_unsubscribe_patterns = set()
1177 self.subscribed_event.clear()
1179 def close(self) -> None:
1180 self.reset()
1182 def _resubscribe(self, subscribed, subscribe_fn) -> None:
1183 # Replay handler-backed subscriptions as positional Subscription objects
1184 # so binary names never need to be decoded into keyword argument keys.
1185 subscriptions = pubsub_subscription_args(subscribed)
1186 if subscriptions:
1187 subscribe_fn(*subscriptions)
1189 def _resubscribe_shard_channels(self) -> None:
1190 self._resubscribe(self.shard_channels, self.ssubscribe)
1192 def on_connect(self, connection) -> None:
1193 "Re-subscribe to any channels and patterns previously subscribed to"
1194 self.pending_unsubscribe_channels.clear()
1195 self.pending_unsubscribe_patterns.clear()
1196 self.pending_unsubscribe_shard_channels.clear()
1197 if self.channels:
1198 self._resubscribe(self.channels, self.subscribe)
1199 if self.patterns:
1200 self._resubscribe(self.patterns, self.psubscribe)
1201 if self.shard_channels:
1202 self._resubscribe_shard_channels()
1204 @property
1205 def subscribed(self) -> bool:
1206 """Indicates if there are subscriptions to any channels or patterns"""
1207 return self.subscribed_event.is_set()
1209 def execute_command(self, *args):
1210 """Execute a publish/subscribe command"""
1212 # NOTE: don't parse the response in this function -- it could pull a
1213 # legitimate message off the stack if the connection is already
1214 # subscribed to one or more channels
1216 if self.connection is None:
1217 self.connection = self.connection_pool.get_connection()
1218 # register a callback that re-subscribes to any channels we
1219 # were listening to when we were disconnected
1220 self.connection.register_connect_callback(self.on_connect)
1221 if self.push_handler_func is not None:
1222 self.connection._parser.set_pubsub_push_handler(self.push_handler_func)
1223 self._event_dispatcher.dispatch(
1224 AfterPubSubConnectionInstantiationEvent(
1225 self.connection, self.connection_pool, ClientType.SYNC, self._lock
1226 )
1227 )
1228 connection = self.connection
1229 kwargs = {"check_health": not self.subscribed}
1230 if not self.subscribed:
1231 self.clean_health_check_responses()
1232 with self._lock:
1233 self._execute(connection, connection.send_command, *args, **kwargs)
1235 def clean_health_check_responses(self) -> None:
1236 """
1237 If any health check responses are present, clean them
1238 """
1239 ttl = 10
1240 conn = self.connection
1241 while conn and self.health_check_response_counter > 0 and ttl > 0:
1242 if self._execute(conn, conn.can_read, timeout=conn.socket_timeout):
1243 response = self._execute(conn, conn.read_response)
1244 if self.is_health_check_response(response):
1245 self.health_check_response_counter -= 1
1246 else:
1247 raise PubSubError(
1248 "A non health check response was cleaned by "
1249 "execute_command: {}".format(response)
1250 )
1251 ttl -= 1
1253 def _reconnect(
1254 self,
1255 conn,
1256 error: Optional[Exception] = None,
1257 failure_count: Optional[int] = None,
1258 start_time: Optional[float] = None,
1259 command_name: Optional[str] = None,
1260 ) -> None:
1261 """
1262 The supported exceptions are already checked in the
1263 retry object so we don't need to do it here.
1265 In this error handler we are trying to reconnect to the server.
1266 """
1267 if error and failure_count <= conn.retry.get_retries():
1268 if command_name:
1269 record_operation_duration(
1270 command_name=command_name,
1271 duration_seconds=time.monotonic() - start_time,
1272 server_address=getattr(conn, "host", None),
1273 server_port=getattr(conn, "port", None),
1274 db_namespace=str(conn.db),
1275 error=error,
1276 retry_attempts=failure_count,
1277 )
1278 conn.disconnect()
1279 conn.connect()
1281 def _execute(self, conn, command, *args, **kwargs):
1282 """
1283 Connect manually upon disconnection. If the Redis server is down,
1284 this will fail and raise a ConnectionError as desired.
1285 After reconnection, the ``on_connect`` callback should have been
1286 called by the # connection to resubscribe us to any channels and
1287 patterns we were previously listening to
1288 """
1290 if conn.should_reconnect():
1291 self._reconnect(conn)
1293 if not len(args) == 0:
1294 command_name = args[0]
1295 else:
1296 command_name = None
1298 # Start timing for observability
1299 start_time = time.monotonic()
1300 # Track actual retry attempts for error reporting
1301 actual_retry_attempts = [0]
1303 def failure_callback(error, failure_count):
1304 actual_retry_attempts[0] = failure_count
1305 self._reconnect(conn, error, failure_count, start_time, command_name)
1307 try:
1308 response = conn.retry.call_with_retry(
1309 lambda: command(*args, **kwargs),
1310 failure_callback,
1311 with_failure_count=True,
1312 )
1314 if command_name:
1315 record_operation_duration(
1316 command_name=command_name,
1317 duration_seconds=time.monotonic() - start_time,
1318 server_address=getattr(conn, "host", None),
1319 server_port=getattr(conn, "port", None),
1320 db_namespace=str(conn.db),
1321 )
1323 return response
1324 except Exception as e:
1325 record_error_count(
1326 server_address=getattr(conn, "host", None),
1327 server_port=getattr(conn, "port", None),
1328 network_peer_address=getattr(conn, "host", None),
1329 network_peer_port=getattr(conn, "port", None),
1330 error_type=e,
1331 retry_attempts=actual_retry_attempts[0],
1332 is_internal=False,
1333 )
1334 raise
1336 def parse_response(self, block=True, timeout=0):
1337 """
1338 Parse the response from a publish/subscribe command.
1340 Args:
1341 block: If True, block indefinitely until a message is available.
1342 If False, return immediately if no message is available.
1343 Default: True
1344 timeout: The timeout in seconds for reading a response when block=False.
1345 This parameter is ignored when block=True.
1346 Default: 0 (return immediately if no data available)
1348 Returns:
1349 The parsed response from the server, or None if no message is available
1350 within the timeout period (when block=False).
1352 Important:
1353 The block and timeout parameters work together:
1354 - When block=True: timeout is IGNORED, method blocks indefinitely
1355 - When block=False: timeout is USED, method returns after timeout expires
1357 Typically, you should use get_message(timeout=X) instead of calling
1358 parse_response() directly. The get_message() method automatically sets
1359 block=False when a timeout is provided, and block=True when timeout=None.
1361 Example:
1362 # Block indefinitely (timeout is ignored)
1363 response = pubsub.parse_response(block=True, timeout=0.1)
1365 # Non-blocking with 0.1 second timeout
1366 response = pubsub.parse_response(block=False, timeout=0.1)
1368 # Non-blocking, return immediately
1369 response = pubsub.parse_response(block=False, timeout=0)
1371 # Recommended: use get_message() instead
1372 msg = pubsub.get_message(timeout=0.1) # automatically sets block=False
1373 msg = pubsub.get_message(timeout=None) # automatically sets block=True
1374 """
1375 conn = self.connection
1376 if conn is None:
1377 raise RuntimeError(
1378 "pubsub connection not set: "
1379 "did you forget to call subscribe() or psubscribe()?"
1380 )
1382 self.check_health()
1384 def try_read():
1385 if not block:
1386 if not conn.can_read(timeout=timeout):
1387 return None
1388 read_timeout = timeout
1389 else:
1390 conn.connect()
1391 # Block indefinitely waiting for a pubsub message. timeout=None
1392 # makes the socket layer call sock.settimeout(None) for this read
1393 # (and restore the original socket_timeout afterwards), so the
1394 # configured socket_timeout does not abort the read.
1395 read_timeout = None
1396 return conn.read_response(
1397 disconnect_on_error=False, push_request=True, timeout=read_timeout
1398 )
1400 response = self._execute(conn, try_read)
1402 if self.is_health_check_response(response):
1403 # ignore the health check message as user might not expect it
1404 self.health_check_response_counter -= 1
1405 return None
1406 return response
1408 def is_health_check_response(self, response) -> bool:
1409 """
1410 Check if the response is a health check response.
1411 If there are no subscriptions redis responds to PING command with a
1412 bulk response, instead of a multi-bulk with "pong" and the response.
1413 """
1414 if self.encoder.decode_responses:
1415 return (
1416 response
1417 in [
1418 self.health_check_response, # If there is a subscription
1419 self.HEALTH_CHECK_MESSAGE, # If there are no subscriptions and decode_responses=True
1420 ]
1421 )
1422 else:
1423 return (
1424 response
1425 in [
1426 self.health_check_response, # If there is a subscription
1427 self.health_check_response_b, # If there isn't a subscription and decode_responses=False
1428 ]
1429 )
1431 def check_health(self) -> None:
1432 conn = self.connection
1433 if conn is None:
1434 raise RuntimeError(
1435 "pubsub connection not set: "
1436 "did you forget to call subscribe() or psubscribe()?"
1437 )
1439 if conn.health_check_interval and time.monotonic() > conn.next_health_check:
1440 conn.send_command("PING", self.HEALTH_CHECK_MESSAGE, check_health=False)
1441 self.health_check_response_counter += 1
1443 def _normalize_keys(self, data) -> Dict:
1444 """
1445 normalize channel/pattern names to be either bytes or strings
1446 based on whether responses are automatically decoded. this saves us
1447 from coercing the value for each message coming in.
1448 """
1449 encode = self.encoder.encode
1450 decode = self.encoder.decode
1451 return {decode(encode(k)): v for k, v in data.items()}
1453 def psubscribe(
1454 self, *args: ChannelT | Subscription, **kwargs: PubSubHandler
1455 ) -> None:
1456 """
1457 Subscribe to channel patterns.
1458 Patterns supplied as keyword arguments expect a pattern name as the
1459 key and a callable as the value.
1460 ``Subscription`` objects can also be supplied positionally with an
1461 optional handler.
1462 A pattern's callable will be invoked automatically
1463 when a message is received on that pattern rather than producing a
1464 message via ``listen()``.
1465 """
1466 new_patterns = parse_pubsub_subscriptions(args, kwargs)
1467 ret_val = self.execute_command("PSUBSCRIBE", *new_patterns.keys())
1468 # update the patterns dict AFTER we send the command. we don't want to
1469 # subscribe twice to these patterns, once for the command and again
1470 # for the reconnection.
1471 new_patterns = self._normalize_keys(new_patterns)
1472 self.patterns.update(new_patterns)
1473 if not self.subscribed:
1474 # Set the subscribed_event flag to True
1475 self.subscribed_event.set()
1476 # Clear the health check counter
1477 self.health_check_response_counter = 0
1478 self.pending_unsubscribe_patterns.difference_update(new_patterns)
1479 return ret_val
1481 def punsubscribe(self, *args):
1482 """
1483 Unsubscribe from the supplied patterns. If empty, unsubscribe from
1484 all patterns.
1485 """
1486 if args:
1487 args = list_or_args(args[0], args[1:])
1488 patterns = self._normalize_keys(dict.fromkeys(args))
1489 else:
1490 patterns = self.patterns
1491 self.pending_unsubscribe_patterns.update(patterns)
1492 return self.execute_command("PUNSUBSCRIBE", *args)
1494 def subscribe(
1495 self, *args: ChannelT | Subscription, **kwargs: PubSubHandler
1496 ) -> None:
1497 """
1498 Subscribe to channels.
1499 Channels supplied as keyword arguments expect
1500 a channel name as the key and a callable as the value.
1501 ``Subscription`` objects can also be supplied positionally with an
1502 optional handler.
1503 A channel's callable will be invoked automatically
1504 when a message is received on that channel rather than producing a
1505 message via ``listen()`` or ``get_message()``.
1506 """
1507 new_channels = parse_pubsub_subscriptions(args, kwargs)
1508 ret_val = self.execute_command("SUBSCRIBE", *new_channels.keys())
1509 # update the channels dict AFTER we send the command. we don't want to
1510 # subscribe twice to these channels, once for the command and again
1511 # for the reconnection.
1512 new_channels = self._normalize_keys(new_channels)
1513 self.channels.update(new_channels)
1514 if not self.subscribed:
1515 # Set the subscribed_event flag to True
1516 self.subscribed_event.set()
1517 # Clear the health check counter
1518 self.health_check_response_counter = 0
1519 self.pending_unsubscribe_channels.difference_update(new_channels)
1520 return ret_val
1522 def unsubscribe(self, *args):
1523 """
1524 Unsubscribe from the supplied channels. If empty, unsubscribe from
1525 all channels
1526 """
1527 if args:
1528 args = list_or_args(args[0], args[1:])
1529 channels = self._normalize_keys(dict.fromkeys(args))
1530 else:
1531 channels = self.channels
1532 self.pending_unsubscribe_channels.update(channels)
1533 return self.execute_command("UNSUBSCRIBE", *args)
1535 def ssubscribe(
1536 self,
1537 *args: ChannelT | Subscription,
1538 target_node: Any = None,
1539 **kwargs: PubSubHandler,
1540 ) -> None:
1541 """
1542 Subscribes the client to the specified shard channels.
1543 Channels supplied as keyword arguments expect a channel name as the key
1544 and a callable as the value.
1545 ``Subscription`` objects can also be supplied positionally
1546 with an optional handler.
1547 A channel's callable will be invoked automatically when a message
1548 is received on that channel rather than producing a message
1549 via ``listen()`` or ``get_sharded_message()``.
1550 """
1551 new_s_channels = parse_pubsub_subscriptions(args, kwargs)
1552 ret_val = self.execute_command("SSUBSCRIBE", *new_s_channels.keys())
1553 # update the s_channels dict AFTER we send the command. we don't want to
1554 # subscribe twice to these channels, once for the command and again
1555 # for the reconnection.
1556 new_s_channels = self._normalize_keys(new_s_channels)
1557 self.shard_channels.update(new_s_channels)
1558 if not self.subscribed:
1559 # Set the subscribed_event flag to True
1560 self.subscribed_event.set()
1561 # Clear the health check counter
1562 self.health_check_response_counter = 0
1563 self.pending_unsubscribe_shard_channels.difference_update(new_s_channels)
1564 return ret_val
1566 def sunsubscribe(self, *args, target_node=None):
1567 """
1568 Unsubscribe from the supplied shard_channels. If empty, unsubscribe from
1569 all shard_channels
1570 """
1571 if args:
1572 args = list_or_args(args[0], args[1:])
1573 s_channels = self._normalize_keys(dict.fromkeys(args))
1574 else:
1575 s_channels = self.shard_channels
1576 self.pending_unsubscribe_shard_channels.update(s_channels)
1577 return self.execute_command("SUNSUBSCRIBE", *args)
1579 def listen(self):
1580 "Listen for messages on channels this client has been subscribed to"
1581 while self.subscribed:
1582 response = self.handle_message(self.parse_response(block=True))
1583 if response is not None:
1584 yield response
1586 def get_message(
1587 self, ignore_subscribe_messages: bool = False, timeout: float = 0.0
1588 ):
1589 """
1590 Get the next message if one is available, otherwise None.
1592 If timeout is specified, the system will wait for `timeout` seconds
1593 before returning. Timeout should be specified as a floating point
1594 number, or None, to wait indefinitely.
1595 """
1596 if not self.subscribed:
1597 # Wait for subscription
1598 start_time = time.monotonic()
1599 if self.subscribed_event.wait(timeout) is True:
1600 # The connection was subscribed during the timeout time frame.
1601 # The timeout should be adjusted based on the time spent
1602 # waiting for the subscription
1603 time_spent = time.monotonic() - start_time
1604 timeout = max(0.0, timeout - time_spent)
1605 else:
1606 # The connection isn't subscribed to any channels or patterns,
1607 # so no messages are available
1608 return None
1610 response = self.parse_response(block=(timeout is None), timeout=timeout)
1612 if response:
1613 return self.handle_message(response, ignore_subscribe_messages)
1614 return None
1616 get_sharded_message = get_message
1618 def ping(self, message: Union[str, None] = None) -> bool:
1619 """
1620 Ping the Redis server to test connectivity.
1622 Sends a PING command to the Redis server and returns True if the server
1623 responds with "PONG".
1624 """
1625 args = ["PING", message] if message is not None else ["PING"]
1626 return self.execute_command(*args)
1628 def handle_message(self, response, ignore_subscribe_messages=False):
1629 """
1630 Parses a pub/sub message. If the channel or pattern was subscribed to
1631 with a message handler, the handler is invoked instead of a parsed
1632 message being returned.
1633 """
1634 if response is None:
1635 return None
1636 if isinstance(response, bytes):
1637 response = [b"pong", response] if response != b"PONG" else [b"pong", b""]
1639 message_type = str_if_bytes(response[0])
1640 if message_type == "pmessage":
1641 message = {
1642 "type": message_type,
1643 "pattern": response[1],
1644 "channel": response[2],
1645 "data": response[3],
1646 }
1647 elif message_type == "pong":
1648 message = {
1649 "type": message_type,
1650 "pattern": None,
1651 "channel": None,
1652 "data": response[1],
1653 }
1654 else:
1655 message = {
1656 "type": message_type,
1657 "pattern": None,
1658 "channel": response[1],
1659 "data": response[2],
1660 }
1662 if message_type in ["message", "pmessage"]:
1663 channel = str_if_bytes(message["channel"])
1664 record_pubsub_message(
1665 direction=PubSubDirection.RECEIVE,
1666 channel=channel,
1667 )
1668 elif message_type == "smessage":
1669 channel = str_if_bytes(message["channel"])
1670 record_pubsub_message(
1671 direction=PubSubDirection.RECEIVE,
1672 channel=channel,
1673 sharded=True,
1674 )
1676 # if this is an unsubscribe message, remove it from memory
1677 if message_type in self.UNSUBSCRIBE_MESSAGE_TYPES:
1678 if message_type == "punsubscribe":
1679 pattern = response[1]
1680 if pattern in self.pending_unsubscribe_patterns:
1681 self.pending_unsubscribe_patterns.remove(pattern)
1682 self.patterns.pop(pattern, None)
1683 elif message_type == "sunsubscribe":
1684 s_channel = response[1]
1685 if s_channel in self.pending_unsubscribe_shard_channels:
1686 self.pending_unsubscribe_shard_channels.remove(s_channel)
1687 self.shard_channels.pop(s_channel, None)
1688 else:
1689 channel = response[1]
1690 if channel in self.pending_unsubscribe_channels:
1691 self.pending_unsubscribe_channels.remove(channel)
1692 self.channels.pop(channel, None)
1693 if not self.channels and not self.patterns and not self.shard_channels:
1694 # There are no subscriptions anymore, set subscribed_event flag
1695 # to false
1696 self.subscribed_event.clear()
1698 if message_type in self.PUBLISH_MESSAGE_TYPES:
1699 # if there's a message handler, invoke it
1700 if message_type == "pmessage":
1701 handler = self.patterns.get(message["pattern"], None)
1702 elif message_type == "smessage":
1703 handler = self.shard_channels.get(message["channel"], None)
1704 else:
1705 handler = self.channels.get(message["channel"], None)
1706 if handler:
1707 handler(message)
1708 return None
1709 elif message_type != "pong":
1710 # this is a subscribe/unsubscribe message. ignore if we don't
1711 # want them
1712 if ignore_subscribe_messages or self.ignore_subscribe_messages:
1713 return None
1715 return message
1717 def run_in_thread(
1718 self,
1719 sleep_time: float = 0.0,
1720 daemon: bool = False,
1721 exception_handler: Optional[Callable] = None,
1722 pubsub=None,
1723 sharded_pubsub: bool = False,
1724 ) -> "PubSubWorkerThread":
1725 for channel, handler in self.channels.items():
1726 if handler is None:
1727 raise PubSubError(f"Channel: '{channel}' has no handler registered")
1728 for pattern, handler in self.patterns.items():
1729 if handler is None:
1730 raise PubSubError(f"Pattern: '{pattern}' has no handler registered")
1731 for s_channel, handler in self.shard_channels.items():
1732 if handler is None:
1733 raise PubSubError(
1734 f"Shard Channel: '{s_channel}' has no handler registered"
1735 )
1737 pubsub = self if pubsub is None else pubsub
1738 thread = PubSubWorkerThread(
1739 pubsub,
1740 sleep_time,
1741 daemon=daemon,
1742 exception_handler=exception_handler,
1743 sharded_pubsub=sharded_pubsub,
1744 )
1745 thread.start()
1746 return thread
1749class PubSubWorkerThread(threading.Thread):
1750 def __init__(
1751 self,
1752 pubsub,
1753 sleep_time: float,
1754 daemon: bool = False,
1755 exception_handler: Union[
1756 Callable[[Exception, "PubSub", "PubSubWorkerThread"], None], None
1757 ] = None,
1758 sharded_pubsub: bool = False,
1759 ):
1760 super().__init__()
1761 self.daemon = daemon
1762 self.pubsub = pubsub
1763 self.sleep_time = sleep_time
1764 self.exception_handler = exception_handler
1765 self.sharded_pubsub = sharded_pubsub
1766 self._running = threading.Event()
1768 def run(self) -> None:
1769 if self._running.is_set():
1770 return
1771 self._running.set()
1772 pubsub = self.pubsub
1773 sleep_time = self.sleep_time
1774 while self._running.is_set():
1775 try:
1776 if not self.sharded_pubsub:
1777 pubsub.get_message(
1778 ignore_subscribe_messages=True, timeout=sleep_time
1779 )
1780 else:
1781 pubsub.get_sharded_message(
1782 ignore_subscribe_messages=True, timeout=sleep_time
1783 )
1784 except BaseException as e:
1785 if self.exception_handler is None:
1786 raise
1787 self.exception_handler(e, pubsub, self)
1788 pubsub.close()
1790 def stop(self) -> None:
1791 # trip the flag so the run loop exits. the run loop will
1792 # close the pubsub connection, which disconnects the socket
1793 # and returns the connection to the pool.
1794 self._running.clear()
1797class Pipeline(Redis):
1798 """
1799 Pipelines provide a way to transmit multiple commands to the Redis server
1800 in one transmission. This is convenient for batch processing, such as
1801 saving all the values in a list to Redis.
1803 All commands executed within a pipeline(when running in transactional mode,
1804 which is the default behavior) are wrapped with MULTI and EXEC
1805 calls. This guarantees all commands executed in the pipeline will be
1806 executed atomically.
1808 Any command raising an exception does *not* halt the execution of
1809 subsequent commands in the pipeline. Instead, the exception is caught
1810 and its instance is placed into the response list returned by execute().
1811 Code iterating over the response list should be able to deal with an
1812 instance of an exception as a potential value. In general, these will be
1813 ResponseError exceptions, such as those raised when issuing a command
1814 on a key of a different datatype.
1815 """
1817 UNWATCH_COMMANDS = {"DISCARD", "EXEC", "UNWATCH"}
1819 def __init__(
1820 self,
1821 connection_pool: ConnectionPool,
1822 response_callbacks,
1823 transaction,
1824 shard_hint,
1825 ):
1826 self.connection_pool = connection_pool
1827 self.connection: Optional[Connection] = None
1828 self.response_callbacks = response_callbacks
1829 self.transaction = transaction
1830 self.shard_hint = shard_hint
1831 self.watching = False
1832 self.command_stack = []
1833 self.scripts: Set[Script] = set()
1834 self.explicit_transaction = False
1836 def __enter__(self) -> "Pipeline":
1837 return self
1839 def __exit__(self, exc_type, exc_value, traceback):
1840 self.reset()
1842 def __del__(self):
1843 try:
1844 self.reset()
1845 except Exception:
1846 pass
1848 def __len__(self) -> int:
1849 return len(self.command_stack)
1851 def __bool__(self) -> bool:
1852 """Pipeline instances should always evaluate to True"""
1853 return True
1855 def reset(self) -> None:
1856 self.command_stack = []
1857 self.scripts = set()
1858 # make sure to reset the connection state in the event that we were
1859 # watching something
1860 if self.watching and self.connection:
1861 try:
1862 # call this manually since our unwatch or
1863 # immediate_execute_command methods can call reset()
1864 self.connection.send_command("UNWATCH")
1865 self.connection.read_response()
1866 except ConnectionError:
1867 # disconnect will also remove any previous WATCHes
1868 self.connection.disconnect()
1869 # clean up the other instance attributes
1870 self.watching = False
1871 self.explicit_transaction = False
1873 # we can safely return the connection to the pool here since we're
1874 # sure we're no longer WATCHing anything
1875 if self.connection:
1876 self.connection_pool.release(self.connection)
1877 self.connection = None
1879 def close(self) -> None:
1880 """Close the pipeline"""
1881 self.reset()
1883 def multi(self) -> None:
1884 """
1885 Start a transactional block of the pipeline after WATCH commands
1886 are issued. End the transactional block with `execute`.
1887 """
1888 if self.explicit_transaction:
1889 raise RedisError("Cannot issue nested calls to MULTI")
1890 if self.command_stack:
1891 raise RedisError(
1892 "Commands without an initial WATCH have already been issued"
1893 )
1894 self.explicit_transaction = True
1896 def execute_command(self, *args, **kwargs):
1897 if (self.watching or args[0] == "WATCH") and not self.explicit_transaction:
1898 return self.immediate_execute_command(*args, **kwargs)
1899 return self.pipeline_execute_command(*args, **kwargs)
1901 def _disconnect_reset_raise_on_watching(
1902 self,
1903 conn: AbstractConnection,
1904 error: Exception,
1905 failure_count: Optional[int] = None,
1906 start_time: Optional[float] = None,
1907 command_name: Optional[str] = None,
1908 ) -> None:
1909 """
1910 Close the connection reset watching state and
1911 raise an exception if we were watching.
1913 The supported exceptions are already checked in the
1914 retry object so we don't need to do it here.
1916 After we disconnect the connection, it will try to reconnect and
1917 do a health check as part of the send_command logic(on connection level).
1918 """
1919 if error and failure_count <= conn.retry.get_retries():
1920 record_operation_duration(
1921 command_name=command_name,
1922 duration_seconds=time.monotonic() - start_time,
1923 server_address=getattr(conn, "host", None),
1924 server_port=getattr(conn, "port", None),
1925 db_namespace=str(conn.db),
1926 error=error,
1927 retry_attempts=failure_count,
1928 )
1929 conn.disconnect()
1931 # if we were already watching a variable, the watch is no longer
1932 # valid since this connection has died. raise a WatchError, which
1933 # indicates the user should retry this transaction.
1934 if self.watching:
1935 self.reset()
1936 raise WatchError(
1937 f"A {type(error).__name__} occurred while watching one or more keys"
1938 )
1940 def immediate_execute_command(self, *args, **options):
1941 """
1942 Execute a command immediately, but don't auto-retry on the supported
1943 errors for retry if we're already WATCHing a variable.
1944 Used when issuing WATCH or subsequent commands retrieving their values but before
1945 MULTI is called.
1946 """
1947 command_name = args[0]
1948 conn = self.connection
1949 # if this is the first call, we need a connection
1950 if not conn:
1951 conn = self.connection_pool.get_connection()
1952 self.connection = conn
1954 # Start timing for observability
1955 start_time = time.monotonic()
1956 # Track actual retry attempts for error reporting
1957 actual_retry_attempts = [0]
1959 def failure_callback(error, failure_count):
1960 if is_debug_log_enabled():
1961 add_debug_log_for_operation_failure(conn)
1962 actual_retry_attempts[0] = failure_count
1963 self._disconnect_reset_raise_on_watching(
1964 conn, error, failure_count, start_time, command_name
1965 )
1967 try:
1968 response = conn.retry.call_with_retry(
1969 lambda: self._send_command_parse_response(
1970 conn, command_name, *args, **options
1971 ),
1972 failure_callback,
1973 with_failure_count=True,
1974 )
1976 record_operation_duration(
1977 command_name=command_name,
1978 duration_seconds=time.monotonic() - start_time,
1979 server_address=getattr(conn, "host", None),
1980 server_port=getattr(conn, "port", None),
1981 db_namespace=str(conn.db),
1982 )
1984 return response
1985 except Exception as e:
1986 record_error_count(
1987 server_address=getattr(conn, "host", None),
1988 server_port=getattr(conn, "port", None),
1989 network_peer_address=getattr(conn, "host", None),
1990 network_peer_port=getattr(conn, "port", None),
1991 error_type=e,
1992 retry_attempts=actual_retry_attempts[0],
1993 is_internal=False,
1994 )
1995 raise
1997 def pipeline_execute_command(self, *args, **options) -> "Pipeline":
1998 """
1999 Stage a command to be executed when execute() is next called
2001 Returns the current Pipeline object back so commands can be
2002 chained together, such as:
2004 pipe = pipe.set('foo', 'bar').incr('baz').decr('bang')
2006 At some other point, you can then run: pipe.execute(),
2007 which will execute all commands queued in the pipe.
2008 """
2009 self.command_stack.append((args, options))
2010 return self
2012 def _himport_prepare_pipeline(self, conn, commands):
2013 """Delegate to the shared sync HIMPORT executor."""
2014 _himport_exec.prepare_pipeline(self, conn, [args for args, _ in commands])
2016 def _execute_transaction(
2017 self, connection: Connection, commands, raise_on_error
2018 ) -> List:
2019 # Ensure fieldsets referenced by buffered HIMPORT SETs are prepared on this
2020 # connection before the MULTI/EXEC block (session state, not transactional).
2021 self._himport_prepare_pipeline(connection, commands)
2022 cmds = chain([(("MULTI",), {})], commands, [(("EXEC",), {})])
2023 all_cmds = connection.pack_commands(
2024 [args for args, options in cmds if EMPTY_RESPONSE not in options]
2025 )
2026 connection.send_packed_command(all_cmds)
2027 errors = []
2029 # parse off the response for MULTI
2030 # NOTE: we need to handle ResponseErrors here and continue
2031 # so that we read all the additional command messages from
2032 # the socket
2033 try:
2034 self.parse_response(connection, "_")
2035 except ResponseError as e:
2036 errors.append((0, e))
2038 # and all the other commands
2039 for i, command in enumerate(commands):
2040 if EMPTY_RESPONSE in command[1]:
2041 errors.append((i, command[1][EMPTY_RESPONSE]))
2042 else:
2043 try:
2044 self.parse_response(connection, "_")
2045 except ResponseError as e:
2046 self.annotate_exception(e, i + 1, command[0])
2047 errors.append((i, e))
2049 # parse the EXEC.
2050 try:
2051 response = self.parse_response(connection, "_")
2052 except ExecAbortError:
2053 if errors:
2054 raise errors[0][1]
2055 raise
2057 # EXEC clears any watched keys
2058 self.watching = False
2060 if response is None:
2061 raise WatchError("Watched variable changed.")
2063 # put any parse errors into the response
2064 for i, e in errors:
2065 response.insert(i, e)
2067 if len(response) != len(commands):
2068 self.connection.disconnect()
2069 raise ResponseError(
2070 "Wrong number of response items from pipeline execution"
2071 )
2073 # find any errors in the response and raise if necessary
2074 if raise_on_error:
2075 self.raise_first_error(commands, response)
2077 # We have to run response callbacks manually
2078 data = []
2079 for r, cmd in zip(response, commands):
2080 if not isinstance(r, Exception):
2081 args, options = cmd
2082 # Remove keys entry, it needs only for cache.
2083 options.pop("keys", None)
2084 command_name = args[0]
2085 if command_name in self.response_callbacks:
2086 r = self.response_callbacks[command_name](r, **options)
2087 data.append(r)
2089 return data
2091 def _execute_pipeline(self, connection, commands, raise_on_error):
2092 # Fold any first-use HIMPORT PREPAREs for referenced fieldsets into the same
2093 # packed write as the queued commands, so a pipeline that lands on a fresh or
2094 # reconnected connection stays a single round trip (the batched write bypasses
2095 # the per-command lazy PREPARE path). Deferred-discard reconciliation happens
2096 # inside pipeline_prepares and only touches the socket when discards are
2097 # actually pending.
2098 fieldsets = _himport_exec.pipeline_prepares(
2099 self, connection, [args for args, _ in commands]
2100 )
2101 preflight = _himport_exec.prepare_wire_commands(fieldsets)
2102 # build up all commands into a single request to increase network perf
2103 all_cmds = connection.pack_commands(preflight + [args for args, _ in commands])
2104 connection.send_packed_command(all_cmds)
2106 # Drain the leading PREPARE replies (bookkeeping + capture the first error)
2107 # before the queued replies. Everything on the wire is read before raising so
2108 # the pooled socket never desyncs.
2109 prep_error = _himport_exec.drain_pipeline_prepares(self, connection, fieldsets)
2111 responses = []
2112 for args, options in commands:
2113 try:
2114 responses.append(self.parse_response(connection, args[0], **options))
2115 except ResponseError as e:
2116 responses.append(e)
2118 # A PREPARE failure (rare: an invalid fieldset definition) is a hard error,
2119 # raised regardless of raise_on_error as it was before folding -- only now
2120 # every reply has already been drained.
2121 if prep_error is not None:
2122 raise prep_error
2123 if raise_on_error:
2124 self.raise_first_error(commands, responses)
2126 return responses
2128 def raise_first_error(self, commands, response):
2129 for i, r in enumerate(response):
2130 if isinstance(r, ResponseError):
2131 self.annotate_exception(r, i + 1, commands[i][0])
2132 raise r
2134 def annotate_exception(self, exception, number, command):
2135 cmd = " ".join(map(safe_str, command))
2136 msg = (
2137 f"Command # {number} ({truncate_text(cmd)}) of pipeline "
2138 f"caused error: {exception.args[0]}"
2139 )
2140 exception.args = (msg,) + exception.args[1:]
2142 def parse_response(self, connection, command_name, **options):
2143 result = Redis.parse_response(self, connection, command_name, **options)
2144 if command_name in self.UNWATCH_COMMANDS:
2145 self.watching = False
2146 elif command_name == "WATCH":
2147 self.watching = True
2148 return result
2150 def load_scripts(self):
2151 # make sure all scripts that are about to be run on this pipeline exist
2152 scripts = list(self.scripts)
2153 immediate = self.immediate_execute_command
2154 shas = [s.sha for s in scripts]
2155 # we can't use the normal script_* methods because they would just
2156 # get buffered in the pipeline.
2157 exists = immediate("SCRIPT EXISTS", *shas)
2158 if not all(exists):
2159 for s, exist in zip(scripts, exists):
2160 if not exist:
2161 s.sha = immediate("SCRIPT LOAD", s.script)
2163 def _disconnect_raise_on_watching(
2164 self,
2165 conn: AbstractConnection,
2166 error: Exception,
2167 failure_count: Optional[int] = None,
2168 start_time: Optional[float] = None,
2169 command_name: Optional[str] = None,
2170 ) -> None:
2171 """
2172 Close the connection, raise an exception if we were watching.
2174 The supported exceptions are already checked in the
2175 retry object so we don't need to do it here.
2177 After we disconnect the connection, it will try to reconnect and
2178 do a health check as part of the send_command logic(on connection level).
2179 """
2180 if error and failure_count <= conn.retry.get_retries():
2181 record_operation_duration(
2182 command_name=command_name,
2183 duration_seconds=time.monotonic() - start_time,
2184 server_address=getattr(conn, "host", None),
2185 server_port=getattr(conn, "port", None),
2186 db_namespace=str(conn.db),
2187 error=error,
2188 retry_attempts=failure_count,
2189 )
2190 conn.disconnect()
2191 # if we were watching a variable, the watch is no longer valid
2192 # since this connection has died. raise a WatchError, which
2193 # indicates the user should retry this transaction.
2194 if self.watching:
2195 raise WatchError(
2196 f"A {type(error).__name__} occurred while watching one or more keys"
2197 )
2199 def execute(self, raise_on_error: bool = True) -> List[Any]:
2200 """Execute all the commands in the current pipeline"""
2201 stack = self.command_stack
2202 if not stack and not self.watching:
2203 return []
2204 if self.scripts:
2205 self.load_scripts()
2206 if self.transaction or self.explicit_transaction:
2207 execute = self._execute_transaction
2208 operation_name = "MULTI"
2209 else:
2210 execute = self._execute_pipeline
2211 operation_name = "PIPELINE"
2213 conn = self.connection
2214 if not conn:
2215 conn = self.connection_pool.get_connection()
2216 # assign to self.connection so reset() releases the connection
2217 # back to the pool after we're done
2218 self.connection = conn
2220 # Start timing for observability
2221 start_time = time.monotonic()
2222 # Track actual retry attempts for error reporting
2223 actual_retry_attempts = [0]
2225 def failure_callback(error, failure_count):
2226 if is_debug_log_enabled():
2227 add_debug_log_for_operation_failure(conn)
2228 actual_retry_attempts[0] = failure_count
2229 self._disconnect_raise_on_watching(
2230 conn, error, failure_count, start_time, operation_name
2231 )
2233 try:
2234 response = conn.retry.call_with_retry(
2235 lambda: execute(conn, stack, raise_on_error),
2236 failure_callback,
2237 with_failure_count=True,
2238 )
2240 record_operation_duration(
2241 command_name=operation_name,
2242 duration_seconds=time.monotonic() - start_time,
2243 server_address=getattr(conn, "host", None),
2244 server_port=getattr(conn, "port", None),
2245 db_namespace=str(conn.db),
2246 )
2247 return response
2248 except Exception as e:
2249 record_error_count(
2250 server_address=getattr(conn, "host", None),
2251 server_port=getattr(conn, "port", None),
2252 network_peer_address=getattr(conn, "host", None),
2253 network_peer_port=getattr(conn, "port", None),
2254 error_type=e,
2255 retry_attempts=actual_retry_attempts[0],
2256 is_internal=False,
2257 )
2258 raise
2260 finally:
2261 # in reset() the connection is disconnected before returned to the pool if
2262 # it is marked for reconnect.
2263 self.reset()
2265 def discard(self):
2266 """
2267 Flushes all previously queued commands
2268 See: https://redis.io/commands/DISCARD
2269 """
2270 self.execute_command("DISCARD")
2272 def watch(self, *names):
2273 """Watches the values at keys ``names``"""
2274 if self.explicit_transaction:
2275 raise RedisError("Cannot issue a WATCH after a MULTI")
2276 return self.execute_command("WATCH", *names)
2278 def unwatch(self) -> bool:
2279 """Unwatches all previously specified keys"""
2280 return self.watching and self.execute_command("UNWATCH") or True