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