Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/redis/client.py: 19%
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 List,
13 Literal,
14 Mapping,
15 Optional,
16 Set,
17 Type,
18 Union,
19)
21from redis._defaults import (
22 DEFAULT_RETRY_BASE,
23 DEFAULT_RETRY_CAP,
24 DEFAULT_RETRY_COUNT,
25 DEFAULT_SOCKET_CONNECT_TIMEOUT,
26 DEFAULT_SOCKET_READ_SIZE,
27 DEFAULT_SOCKET_TIMEOUT,
28)
29from redis._parsers.encoders import Encoder
30from redis._parsers.helpers import bool_ok, get_response_callbacks
31from redis.backoff import ExponentialWithJitterBackoff
32from redis.cache import CacheConfig, CacheInterface
33from redis.commands import (
34 CoreCommands,
35 RedisModuleCommands,
36 SentinelCommands,
37 list_or_args,
38)
39from redis.commands.core import Script
40from redis.commands.helpers import parse_pubsub_subscriptions, pubsub_subscription_args
41from redis.connection import (
42 AbstractConnection,
43 Connection,
44 ConnectionPool,
45 SSLConnection,
46 UnixDomainSocketConnection,
47)
48from redis.credentials import CredentialProvider
49from redis.driver_info import DriverInfo, resolve_driver_info
50from redis.event import (
51 AfterPooledConnectionsInstantiationEvent,
52 AfterPubSubConnectionInstantiationEvent,
53 AfterSingleConnectionInstantiationEvent,
54 ClientType,
55 EventDispatcher,
56)
57from redis.exceptions import (
58 ConnectionError,
59 ExecAbortError,
60 PubSubError,
61 RedisError,
62 ResponseError,
63 WatchError,
64)
65from redis.lock import Lock
66from redis.maint_notifications import (
67 MaintNotificationsConfig,
68 OSSMaintNotificationsHandler,
69)
70from redis.observability.attributes import PubSubDirection
71from redis.observability.recorder import (
72 record_error_count,
73 record_operation_duration,
74 record_pubsub_message,
75)
76from redis.retry import Retry
77from redis.typing import ChannelT, PubSubHandler, Subscription
78from redis.utils import (
79 SENTINEL,
80 _set_info_logger,
81 check_protocol_version,
82 deprecated_args,
83 safe_str,
84 str_if_bytes,
85 truncate_text,
86)
88if TYPE_CHECKING:
89 import ssl
91 import OpenSSL
93 from redis.keyspace_notifications import KeyspaceNotifications
95SYM_EMPTY = b""
96EMPTY_RESPONSE = "EMPTY_RESPONSE"
98# some responses (ie. dump) are binary, and just meant to never be decoded
99NEVER_DECODE = "NEVER_DECODE"
102logger = logging.getLogger(__name__)
105def is_debug_log_enabled():
106 return logger.isEnabledFor(logging.DEBUG)
109def add_debug_log_for_operation_failure(connection: "AbstractConnection"):
110 logger.debug(
111 f"Operation failed, "
112 f"with connection: {connection}, details: {connection.extract_connection_details() if connection else 'no connection'}",
113 )
116class CaseInsensitiveDict(dict):
117 "Case insensitive dict implementation. Assumes string keys only."
119 def __init__(self, data: Dict[str, str]) -> None:
120 for k, v in data.items():
121 self[k.upper()] = v
123 def __contains__(self, k):
124 return super().__contains__(k.upper())
126 def __delitem__(self, k):
127 super().__delitem__(k.upper())
129 def __getitem__(self, k):
130 return super().__getitem__(k.upper())
132 def get(self, k, default=None):
133 return super().get(k.upper(), default)
135 def __setitem__(self, k, v):
136 super().__setitem__(k.upper(), v)
138 def update(self, data):
139 data = CaseInsensitiveDict(data)
140 super().update(data)
143class AbstractRedis:
144 pass
147class Redis(RedisModuleCommands, CoreCommands, SentinelCommands):
148 """
149 Implementation of the Redis protocol.
151 This abstract class provides a Python interface to all Redis commands
152 and an implementation of the Redis protocol.
154 Pipelines derive from this, implementing how
155 the commands are sent and received to the Redis server. Based on
156 configuration, an instance will either use a ConnectionPool, or
157 Connection object to talk to redis.
159 It is not safe to pass PubSub or Pipeline objects between threads.
160 """
162 # Type discrimination marker for @overload self-type pattern
163 _is_async_client: Literal[False] = False
165 @classmethod
166 def from_url(cls, url: str, **kwargs) -> "Redis":
167 """
168 Return a Redis client object configured from the given URL
170 For example::
172 redis://[[username]:[password]]@localhost:6379/0
173 rediss://[[username]:[password]]@localhost:6379/0
174 unix://[username@]/path/to/socket.sock?db=0[&password=password]
176 Three URL schemes are supported:
178 - `redis://` creates a TCP socket connection. See more at:
179 <https://www.iana.org/assignments/uri-schemes/prov/redis>
180 - `rediss://` creates a SSL wrapped TCP socket connection. See more at:
181 <https://www.iana.org/assignments/uri-schemes/prov/rediss>
182 - ``unix://``: creates a Unix Domain Socket connection.
184 The username, password, hostname, path and all querystring values
185 are passed through urllib.parse.unquote in order to replace any
186 percent-encoded values with their corresponding characters.
188 There are several ways to specify a database number. The first value
189 found will be used:
191 1. A ``db`` querystring option, e.g. redis://localhost?db=0
192 2. If using the redis:// or rediss:// schemes, the path argument
193 of the url, e.g. redis://localhost/0
194 3. A ``db`` keyword argument to this function.
196 If none of these options are specified, the default db=0 is used.
198 All querystring options are cast to their appropriate Python types.
199 Boolean arguments can be specified with string values "True"/"False"
200 or "Yes"/"No". Values that cannot be properly cast cause a
201 ``ValueError`` to be raised. Once parsed, the querystring arguments
202 and keyword arguments are passed to the ``ConnectionPool``'s
203 class initializer. In the case of conflicting arguments, querystring
204 arguments always win.
206 """
207 single_connection_client = kwargs.pop("single_connection_client", False)
208 connection_pool = ConnectionPool.from_url(url, **kwargs)
209 client = cls(
210 connection_pool=connection_pool,
211 single_connection_client=single_connection_client,
212 )
213 client.auto_close_connection_pool = True
214 return client
216 @classmethod
217 def from_pool(
218 cls: Type["Redis"],
219 connection_pool: ConnectionPool,
220 ) -> "Redis":
221 """
222 Return a Redis client from the given connection pool.
223 The Redis client will take ownership of the connection pool and
224 close it when the Redis client is closed.
226 Because the client closes (disconnects all connections in) the pool
227 when it is closed or garbage-collected, the pool must not be shared
228 with other clients. Constructing multiple clients from the same pool
229 via ``from_pool`` -- for example one per request across threads -- is
230 not thread safe: when one client is closed it will disconnect
231 connections still in use by the others.
233 To share a single pool across clients, construct the pool explicitly
234 and manage its lifecycle instead. Unlike ``from_pool``, the plain
235 ``Redis(connection_pool=pool)`` constructor does not take ownership of
236 the pool and will not close it, so a pool created this way can be
237 safely shared across clients. ``ConnectionPool`` supports the context
238 manager protocol for this::
240 with ConnectionPool.from_url(url) as pool:
241 r = Redis(connection_pool=pool)
242 """
243 client = cls(
244 connection_pool=connection_pool,
245 )
246 client.auto_close_connection_pool = True
247 return client
249 @deprecated_args(
250 args_to_warn=["retry_on_timeout"],
251 reason="TimeoutError is included by default.",
252 version="6.0.0",
253 )
254 @deprecated_args(
255 args_to_warn=["lib_name", "lib_version"],
256 reason="Use 'driver_info' parameter instead. "
257 "lib_name and lib_version will be removed in a future version.",
258 )
259 def __init__(
260 self,
261 host: str = "localhost",
262 port: int = 6379,
263 db: int = 0,
264 password: str | None = None,
265 socket_timeout: float | None = DEFAULT_SOCKET_TIMEOUT,
266 socket_connect_timeout: float | None = DEFAULT_SOCKET_CONNECT_TIMEOUT,
267 socket_read_size: int = DEFAULT_SOCKET_READ_SIZE,
268 socket_keepalive: bool | None = True,
269 socket_keepalive_options: Mapping[int, int | bytes] | object | None = SENTINEL,
270 connection_pool: ConnectionPool | None = None,
271 unix_socket_path: str | None = None,
272 encoding: str = "utf-8",
273 encoding_errors: str = "strict",
274 decode_responses: bool = False,
275 retry_on_timeout: bool = False,
276 retry: Retry = Retry(
277 backoff=ExponentialWithJitterBackoff(
278 base=DEFAULT_RETRY_BASE, cap=DEFAULT_RETRY_CAP
279 ),
280 retries=DEFAULT_RETRY_COUNT,
281 ),
282 retry_on_error: List[Type[Exception]] | None = None,
283 ssl: bool = False,
284 ssl_keyfile: str | None = None,
285 ssl_certfile: str | None = None,
286 ssl_cert_reqs: "str | ssl.VerifyMode" = "required",
287 ssl_include_verify_flags: List["ssl.VerifyFlags"] | None = None,
288 ssl_exclude_verify_flags: List["ssl.VerifyFlags"] | None = None,
289 ssl_ca_certs: str | None = None,
290 ssl_ca_path: str | None = None,
291 ssl_ca_data: str | None = None,
292 ssl_check_hostname: bool = True,
293 ssl_password: str | None = None,
294 ssl_validate_ocsp: bool = False,
295 ssl_validate_ocsp_stapled: bool = False,
296 ssl_ocsp_context: "OpenSSL.SSL.Context | None" = None,
297 ssl_ocsp_expected_cert: str | None = None,
298 ssl_min_version: "ssl.TLSVersion | None" = None,
299 ssl_ciphers: str | None = None,
300 max_connections: int | None = None,
301 single_connection_client: bool = False,
302 health_check_interval: int = 0,
303 client_name: str | None = None,
304 lib_name: str | object | None = SENTINEL,
305 lib_version: str | object | None = SENTINEL,
306 driver_info: DriverInfo | object | None = SENTINEL,
307 username: str | None = None,
308 redis_connect_func: Callable[[], None] | None = None,
309 credential_provider: CredentialProvider | None = None,
310 protocol: int | None = None,
311 legacy_responses: bool = True,
312 cache: CacheInterface | None = None,
313 cache_config: CacheConfig | None = None,
314 event_dispatcher: EventDispatcher | None = None,
315 maint_notifications_config: MaintNotificationsConfig | None = None,
316 oss_cluster_maint_notifications_handler: OSSMaintNotificationsHandler
317 | None = None,
318 ) -> None:
319 """
320 Initialize a new Redis client.
322 To specify a retry policy for specific errors, you have two options:
324 1. Set the `retry_on_error` to a list of the error/s to retry on, and
325 you can also set `retry` to a valid `Retry` object(in case the default
326 one is not appropriate) - with this approach the retries will be triggered
327 on the default errors specified in the Retry object enriched with the
328 errors specified in `retry_on_error`.
330 2. Define a `Retry` object with configured 'supported_errors' and set
331 it to the `retry` parameter - with this approach you completely redefine
332 the errors on which retries will happen.
334 `retry_on_timeout` is deprecated - please include the TimeoutError
335 either in the Retry object or in the `retry_on_error` list.
337 When 'connection_pool' is provided - the retry configuration of the
338 provided pool will be used.
340 Args:
342 socket_keepalive:
343 if `True`, TCP keepalive is enabled for TCP socket connections.
344 Argument is ignored when connection_pool is provided.
345 socket_keepalive_options:
346 mapping of TCP keepalive socket option constants to values, for
347 example `{socket.TCP_KEEPIDLE: 30}`. If left unspecified, redis-py
348 uses TCP keepalive defaults when `socket_keepalive` is enabled:
349 idle 30 seconds, interval 5 seconds, and 3 probes. Platform-specific
350 options that are not available are skipped. Pass `None` or `{}` to
351 avoid setting additional TCP keepalive options. Argument is ignored
352 when connection_pool is provided.
353 single_connection_client:
354 if `True`, connection pool is not used. In that case `Redis`
355 instance use is not thread safe.
356 decode_responses:
357 if `True`, the response will be decoded to utf-8.
358 Argument is ignored when connection_pool is provided.
359 driver_info:
360 Optional DriverInfo object to identify upstream libraries.
361 If provided, lib_name and lib_version are ignored.
362 If not provided, a DriverInfo will be created from lib_name and lib_version.
363 Explicit None disables CLIENT SETINFO.
364 Argument is ignored when connection_pool is provided.
365 lib_name:
366 **Deprecated.** Use driver_info instead. Library name for CLIENT SETINFO.
367 lib_version:
368 **Deprecated.** Use driver_info instead. Library version for CLIENT SETINFO.
369 maint_notifications_config:
370 configures the pool to support maintenance notifications - see
371 `redis.maint_notifications.MaintNotificationsConfig` for details.
372 Only supported with RESP3
373 If not provided and protocol is RESP3, the maintenance notifications
374 will be enabled by default (logic is included in the connection pool
375 initialization).
376 Argument is ignored when connection_pool is provided.
377 oss_cluster_maint_notifications_handler:
378 handler for OSS cluster notifications - see
379 `redis.maint_notifications.OSSMaintNotificationsHandler` for details.
380 Only supported with RESP3
381 Argument is ignored when connection_pool is provided.
382 """
383 if event_dispatcher is None:
384 self._event_dispatcher = EventDispatcher()
385 else:
386 self._event_dispatcher = event_dispatcher
387 if not connection_pool:
388 if not retry_on_error:
389 retry_on_error = []
391 # Handle driver_info: if provided, use it; otherwise create from lib_name/lib_version.
392 computed_driver_info = resolve_driver_info(
393 driver_info, lib_name, lib_version
394 )
396 kwargs = {
397 "db": db,
398 "username": username,
399 "password": password,
400 "socket_timeout": socket_timeout,
401 "socket_read_size": socket_read_size,
402 "encoding": encoding,
403 "encoding_errors": encoding_errors,
404 "decode_responses": decode_responses,
405 "retry_on_error": retry_on_error,
406 "retry": copy.deepcopy(retry),
407 "max_connections": max_connections,
408 "health_check_interval": health_check_interval,
409 "client_name": client_name,
410 "driver_info": computed_driver_info,
411 "redis_connect_func": redis_connect_func,
412 "credential_provider": credential_provider,
413 "protocol": protocol,
414 "legacy_responses": legacy_responses,
415 }
416 # based on input, setup appropriate connection args
417 if unix_socket_path is not None:
418 if (
419 maint_notifications_config
420 and maint_notifications_config.enabled is True
421 ):
422 raise RedisError(
423 "Maintenance notifications are not supported with Unix "
424 "domain socket connections"
425 )
426 kwargs.update(
427 {
428 "path": unix_socket_path,
429 "connection_class": UnixDomainSocketConnection,
430 "maint_notifications_config": MaintNotificationsConfig(
431 enabled=False
432 ),
433 }
434 )
435 else:
436 # TCP specific options
437 kwargs.update(
438 {
439 "host": host,
440 "port": port,
441 "socket_connect_timeout": socket_connect_timeout,
442 "socket_keepalive": socket_keepalive,
443 "socket_keepalive_options": socket_keepalive_options,
444 }
445 )
447 if ssl:
448 kwargs.update(
449 {
450 "connection_class": SSLConnection,
451 "ssl_keyfile": ssl_keyfile,
452 "ssl_certfile": ssl_certfile,
453 "ssl_cert_reqs": ssl_cert_reqs,
454 "ssl_include_verify_flags": ssl_include_verify_flags,
455 "ssl_exclude_verify_flags": ssl_exclude_verify_flags,
456 "ssl_ca_certs": ssl_ca_certs,
457 "ssl_ca_data": ssl_ca_data,
458 "ssl_check_hostname": ssl_check_hostname,
459 "ssl_password": ssl_password,
460 "ssl_ca_path": ssl_ca_path,
461 "ssl_validate_ocsp_stapled": ssl_validate_ocsp_stapled,
462 "ssl_validate_ocsp": ssl_validate_ocsp,
463 "ssl_ocsp_context": ssl_ocsp_context,
464 "ssl_ocsp_expected_cert": ssl_ocsp_expected_cert,
465 "ssl_min_version": ssl_min_version,
466 "ssl_ciphers": ssl_ciphers,
467 }
468 )
469 if (cache_config or cache) and check_protocol_version(protocol, 3):
470 kwargs.update(
471 {
472 "cache": cache,
473 "cache_config": cache_config,
474 }
475 )
476 maint_notifications_enabled = (
477 maint_notifications_config and maint_notifications_config.enabled
478 )
479 if maint_notifications_enabled and not check_protocol_version(
480 protocol, 3
481 ):
482 raise RedisError(
483 "Maintenance notifications handlers on connection are only supported with RESP version 3"
484 )
485 if maint_notifications_config:
486 kwargs.update(
487 {
488 "maint_notifications_config": maint_notifications_config,
489 }
490 )
491 if oss_cluster_maint_notifications_handler:
492 kwargs.update(
493 {
494 "oss_cluster_maint_notifications_handler": oss_cluster_maint_notifications_handler,
495 }
496 )
497 connection_pool = ConnectionPool(**kwargs)
498 self._event_dispatcher.dispatch(
499 AfterPooledConnectionsInstantiationEvent(
500 [connection_pool], ClientType.SYNC, credential_provider
501 )
502 )
503 self.auto_close_connection_pool = True
504 else:
505 self.auto_close_connection_pool = False
506 self._event_dispatcher.dispatch(
507 AfterPooledConnectionsInstantiationEvent(
508 [connection_pool], ClientType.SYNC, credential_provider
509 )
510 )
512 self.connection_pool = connection_pool
514 if (cache_config or cache) and not check_protocol_version(
515 self.connection_pool.get_protocol(), 3
516 ):
517 raise RedisError("Client caching is only supported with RESP version 3")
519 self.single_connection_lock = threading.RLock()
520 self.connection = None
521 self._single_connection_client = single_connection_client
522 if self._single_connection_client:
523 self.connection = self.connection_pool.get_connection()
524 self._event_dispatcher.dispatch(
525 AfterSingleConnectionInstantiationEvent(
526 self.connection, ClientType.SYNC, self.single_connection_lock
527 )
528 )
530 connection_kwargs = self.connection_pool.connection_kwargs
531 self.response_callbacks = CaseInsensitiveDict(
532 get_response_callbacks(
533 user_protocol=connection_kwargs.get("protocol"),
534 legacy_responses=connection_kwargs.get("legacy_responses", True),
535 )
536 )
538 def __repr__(self) -> str:
539 return (
540 f"<{type(self).__module__}.{type(self).__name__}"
541 f"({repr(self.connection_pool)})>"
542 )
544 def get_encoder(self) -> "Encoder":
545 """Get the connection pool's encoder"""
546 return self.connection_pool.get_encoder()
548 def get_connection_kwargs(self) -> Dict:
549 """Get the connection's key-word arguments"""
550 return self.connection_pool.connection_kwargs
552 def get_retry(self) -> Optional[Retry]:
553 return self.get_connection_kwargs().get("retry")
555 def set_retry(self, retry: Retry) -> None:
556 self.get_connection_kwargs().update({"retry": retry})
557 self.connection_pool.set_retry(retry)
559 def set_response_callback(self, command: str, callback: Callable) -> None:
560 """Set a custom Response Callback"""
561 self.response_callbacks[command] = callback
563 def load_external_module(self, funcname, func) -> None:
564 """
565 This function can be used to add externally defined redis modules,
566 and their namespaces to the redis client.
568 funcname - A string containing the name of the function to create
569 func - The function, being added to this class.
571 ex: Assume that one has a custom redis module named foomod that
572 creates command named 'foo.dothing' and 'foo.anotherthing' in redis.
573 To load function functions into this namespace:
575 from redis import Redis
576 from foomodule import F
577 r = Redis()
578 r.load_external_module("foo", F)
579 r.foo().dothing('your', 'arguments')
581 For a concrete example see the reimport of the redisjson module in
582 tests/test_connection.py::test_loading_external_modules
583 """
584 setattr(self, funcname, func)
586 def pipeline(self, transaction=True, shard_hint=None) -> "Pipeline":
587 """
588 Return a new pipeline object that can queue multiple commands for
589 later execution. ``transaction`` indicates whether all commands
590 should be executed atomically. Apart from making a group of operations
591 atomic, pipelines are useful for reducing the back-and-forth overhead
592 between the client and server.
593 """
594 return Pipeline(
595 self.connection_pool, self.response_callbacks, transaction, shard_hint
596 )
598 def transaction(
599 self, func: Callable[["Pipeline"], None], *watches, **kwargs
600 ) -> Union[List[Any], Any, None]:
601 """
602 Convenience method for executing the callable `func` as a transaction
603 while watching all keys specified in `watches`. The 'func' callable
604 should expect a single argument which is a Pipeline object.
605 """
606 shard_hint = kwargs.pop("shard_hint", None)
607 value_from_callable = kwargs.pop("value_from_callable", False)
608 watch_delay = kwargs.pop("watch_delay", None)
609 with self.pipeline(True, shard_hint) as pipe:
610 while True:
611 try:
612 if watches:
613 pipe.watch(*watches)
614 func_value = func(pipe)
615 exec_value = pipe.execute()
616 return func_value if value_from_callable else exec_value
617 except WatchError:
618 if watch_delay is not None and watch_delay > 0:
619 time.sleep(watch_delay)
620 continue
622 def lock(
623 self,
624 name: str,
625 timeout: Optional[float] = None,
626 sleep: float = 0.1,
627 blocking: bool = True,
628 blocking_timeout: Optional[float] = None,
629 lock_class: Union[None, Any] = None,
630 thread_local: bool = True,
631 raise_on_release_error: bool = True,
632 ):
633 """
634 Return a new Lock object using key ``name`` that mimics
635 the behavior of threading.Lock.
637 If specified, ``timeout`` indicates a maximum life for the lock.
638 By default, it will remain locked until release() is called.
640 ``sleep`` indicates the amount of time to sleep per loop iteration
641 when the lock is in blocking mode and another client is currently
642 holding the lock.
644 ``blocking`` indicates whether calling ``acquire`` should block until
645 the lock has been acquired or to fail immediately, causing ``acquire``
646 to return False and the lock not being acquired. Defaults to True.
647 Note this value can be overridden by passing a ``blocking``
648 argument to ``acquire``.
650 ``blocking_timeout`` indicates the maximum amount of time in seconds to
651 spend trying to acquire the lock. A value of ``None`` indicates
652 continue trying forever. ``blocking_timeout`` can be specified as a
653 float or integer, both representing the number of seconds to wait.
655 ``lock_class`` forces the specified lock implementation. Note that as
656 of redis-py 3.0, the only lock class we implement is ``Lock`` (which is
657 a Lua-based lock). So, it's unlikely you'll need this parameter, unless
658 you have created your own custom lock class.
660 ``thread_local`` indicates whether the lock token is placed in
661 thread-local storage. By default, the token is placed in thread local
662 storage so that a thread only sees its token, not a token set by
663 another thread. Consider the following timeline:
665 time: 0, thread-1 acquires `my-lock`, with a timeout of 5 seconds.
666 thread-1 sets the token to "abc"
667 time: 1, thread-2 blocks trying to acquire `my-lock` using the
668 Lock instance.
669 time: 5, thread-1 has not yet completed. redis expires the lock
670 key.
671 time: 5, thread-2 acquired `my-lock` now that it's available.
672 thread-2 sets the token to "xyz"
673 time: 6, thread-1 finishes its work and calls release(). if the
674 token is *not* stored in thread local storage, then
675 thread-1 would see the token value as "xyz" and would be
676 able to successfully release the thread-2's lock.
678 ``raise_on_release_error`` indicates whether to raise an exception when
679 the lock is no longer owned when exiting the context manager. By default,
680 this is True, meaning an exception will be raised. If False, the warning
681 will be logged and the exception will be suppressed.
683 In some use cases it's necessary to disable thread local storage. For
684 example, if you have code where one thread acquires a lock and passes
685 that lock instance to a worker thread to release later. If thread
686 local storage isn't disabled in this case, the worker thread won't see
687 the token set by the thread that acquired the lock. Our assumption
688 is that these cases aren't common and as such default to using
689 thread local storage."""
690 if lock_class is None:
691 lock_class = Lock
692 return lock_class(
693 self,
694 name,
695 timeout=timeout,
696 sleep=sleep,
697 blocking=blocking,
698 blocking_timeout=blocking_timeout,
699 thread_local=thread_local,
700 raise_on_release_error=raise_on_release_error,
701 )
703 def pubsub(self, **kwargs):
704 """
705 Return a Publish/Subscribe object. With this object, you can
706 subscribe to channels and listen for messages that get published to
707 them.
708 """
709 return PubSub(
710 self.connection_pool, event_dispatcher=self._event_dispatcher, **kwargs
711 )
713 def keyspace_notifications(
714 self,
715 key_prefix: Union[str, bytes, None] = None,
716 ignore_subscribe_messages: bool = True,
717 ) -> "KeyspaceNotifications":
718 """
719 Return a :class:`~redis.keyspace_notifications.KeyspaceNotifications`
720 object for subscribing to keyspace and keyevent notifications.
722 Note: Keyspace notifications must be enabled on the Redis server via
723 the ``notify-keyspace-events`` configuration option.
725 Args:
726 key_prefix: Optional prefix to filter and strip from keys in
727 notifications.
728 ignore_subscribe_messages: If True, subscribe/unsubscribe
729 confirmations are not returned by
730 get_message/listen.
731 """
732 from redis.keyspace_notifications import KeyspaceNotifications
734 return KeyspaceNotifications(
735 self,
736 key_prefix=key_prefix,
737 ignore_subscribe_messages=ignore_subscribe_messages,
738 )
740 def monitor(self):
741 return Monitor(self.connection_pool)
743 def client(self):
744 return self.__class__(
745 connection_pool=self.connection_pool,
746 single_connection_client=True,
747 )
749 def __enter__(self):
750 return self
752 def __exit__(self, exc_type, exc_value, traceback):
753 self.close()
755 def __del__(self):
756 try:
757 self.close()
758 except Exception:
759 pass
761 def close(self) -> None:
762 # In case a connection property does not yet exist
763 # (due to a crash earlier in the Redis() constructor), return
764 # immediately as there is nothing to clean-up.
765 if not hasattr(self, "connection"):
766 return
768 conn = self.connection
769 if conn:
770 self.connection = None
771 self.connection_pool.release(conn)
773 if self.auto_close_connection_pool:
774 self.connection_pool.close()
776 def _send_command_parse_response(self, conn, command_name, *args, **options):
777 """
778 Send a command and parse the response
779 """
780 conn.send_command(*args, **options)
781 return self.parse_response(conn, command_name, **options)
783 def _close_connection(
784 self,
785 conn,
786 error: Optional[Exception] = None,
787 failure_count: Optional[int] = None,
788 start_time: Optional[float] = None,
789 command_name: Optional[str] = None,
790 ) -> None:
791 """
792 Close the connection before retrying.
794 The supported exceptions are already checked in the
795 retry object so we don't need to do it here.
797 After we disconnect the connection, it will try to reconnect and
798 do a health check as part of the send_command logic(on connection level).
799 """
800 if error and failure_count <= conn.retry.get_retries():
801 record_operation_duration(
802 command_name=command_name,
803 duration_seconds=time.monotonic() - start_time,
804 server_address=getattr(conn, "host", None),
805 server_port=getattr(conn, "port", None),
806 db_namespace=str(conn.db),
807 error=error,
808 retry_attempts=failure_count,
809 )
811 conn.disconnect()
813 # COMMAND EXECUTION AND PROTOCOL PARSING
814 def execute_command(self, *args, **options):
815 return self._execute_command(*args, **options)
817 def _execute_command(self, *args, **options):
818 """Execute a command and return a parsed response"""
819 pool = self.connection_pool
820 command_name = args[0]
821 conn = self.connection or pool.get_connection()
823 # Start timing for observability
824 start_time = time.monotonic()
825 # Track actual retry attempts for error reporting
826 actual_retry_attempts = [0]
828 def failure_callback(error, failure_count):
829 if is_debug_log_enabled():
830 add_debug_log_for_operation_failure(conn)
831 actual_retry_attempts[0] = failure_count
832 self._close_connection(conn, error, failure_count, start_time, command_name)
834 if self._single_connection_client:
835 self.single_connection_lock.acquire()
836 try:
837 result = conn.retry.call_with_retry(
838 lambda: self._send_command_parse_response(
839 conn, command_name, *args, **options
840 ),
841 failure_callback,
842 with_failure_count=True,
843 )
845 record_operation_duration(
846 command_name=command_name,
847 duration_seconds=time.monotonic() - start_time,
848 server_address=getattr(conn, "host", None),
849 server_port=getattr(conn, "port", None),
850 db_namespace=str(conn.db),
851 )
852 return result
853 except Exception as e:
854 record_error_count(
855 server_address=getattr(conn, "host", None),
856 server_port=getattr(conn, "port", None),
857 network_peer_address=getattr(conn, "host", None),
858 network_peer_port=getattr(conn, "port", None),
859 error_type=e,
860 retry_attempts=actual_retry_attempts[0],
861 is_internal=False,
862 )
863 raise
865 finally:
866 try:
867 if conn and conn.should_reconnect():
868 self._close_connection(conn)
869 conn.connect()
870 finally:
871 if self._single_connection_client:
872 self.single_connection_lock.release()
873 if not self.connection:
874 pool.release(conn)
876 def parse_response(self, connection, command_name, **options):
877 """Parses a response from the Redis server"""
878 try:
879 if NEVER_DECODE in options:
880 response = connection.read_response(disable_decoding=True)
881 options.pop(NEVER_DECODE)
882 else:
883 response = connection.read_response()
884 except ResponseError:
885 if EMPTY_RESPONSE in options:
886 return options[EMPTY_RESPONSE]
887 raise
889 if EMPTY_RESPONSE in options:
890 options.pop(EMPTY_RESPONSE)
892 # Remove keys entry, it needs only for cache.
893 options.pop("keys", None)
895 if command_name in self.response_callbacks:
896 return self.response_callbacks[command_name](response, **options)
897 return response
899 def get_cache(self) -> Optional[CacheInterface]:
900 return self.connection_pool.cache
903StrictRedis = Redis
906class Monitor:
907 """
908 Monitor is useful for handling the MONITOR command to the redis server.
909 next_command() method returns one command from monitor
910 listen() method yields commands from monitor.
911 """
913 monitor_re = re.compile(r"\[(\d+) (.*?)\] (.*)")
914 command_re = re.compile(r'"(.*?)(?<!\\)"')
916 def __init__(self, connection_pool):
917 self.connection_pool = connection_pool
918 self.connection = self.connection_pool.get_connection()
920 def __enter__(self):
921 self._start_monitor()
922 return self
924 def __exit__(self, *args):
925 self.connection.disconnect()
926 self.connection_pool.release(self.connection)
928 def next_command(self):
929 """Parse the response from a monitor command"""
930 response = self.connection.read_response()
932 if response is None:
933 return None
935 if isinstance(response, bytes):
936 response = self.connection.encoder.decode(response, force=True)
938 command_time, command_data = response.split(" ", 1)
939 m = self.monitor_re.match(command_data)
940 db_id, client_info, command = m.groups()
941 command = " ".join(self.command_re.findall(command))
942 # Redis escapes double quotes because each piece of the command
943 # string is surrounded by double quotes. We don't have that
944 # requirement so remove the escaping and leave the quote.
945 command = command.replace('\\"', '"')
947 if client_info == "lua":
948 client_address = "lua"
949 client_port = ""
950 client_type = "lua"
951 elif client_info.startswith("unix"):
952 client_address = "unix"
953 client_port = client_info[5:]
954 client_type = "unix"
955 else:
956 if client_info == "":
957 client_address = ""
958 client_port = ""
959 client_type = "unknown"
960 else:
961 # use rsplit as ipv6 addresses contain colons
962 client_address, client_port = client_info.rsplit(":", 1)
963 client_type = "tcp"
964 return {
965 "time": float(command_time),
966 "db": int(db_id),
967 "client_address": client_address,
968 "client_port": client_port,
969 "client_type": client_type,
970 "command": command,
971 }
973 def listen(self):
974 """Listen for commands coming to the server."""
975 while True:
976 yield self.next_command()
978 def _start_monitor(self):
979 self.connection.send_command("MONITOR")
980 # check that monitor returns 'OK', but don't return it to user
981 response = self.connection.read_response()
983 if not bool_ok(response):
984 raise RedisError(f"MONITOR failed: {response}")
987class PubSub:
988 """
989 PubSub provides publish, subscribe and listen support to Redis channels.
991 After subscribing to one or more channels, the listen() method will block
992 until a message arrives on one of the subscribed channels. That message
993 will be returned and it's safe to start listening again.
994 """
996 PUBLISH_MESSAGE_TYPES = ("message", "pmessage", "smessage")
997 UNSUBSCRIBE_MESSAGE_TYPES = ("unsubscribe", "punsubscribe", "sunsubscribe")
998 HEALTH_CHECK_MESSAGE = "redis-py-health-check"
1000 def __init__(
1001 self,
1002 connection_pool,
1003 shard_hint=None,
1004 ignore_subscribe_messages: bool = False,
1005 encoder: Optional["Encoder"] = None,
1006 push_handler_func: Union[None, Callable[[str], None]] = None,
1007 event_dispatcher: Optional["EventDispatcher"] = None,
1008 ):
1009 self.connection_pool = connection_pool
1010 self.shard_hint = shard_hint
1011 self.ignore_subscribe_messages = ignore_subscribe_messages
1012 self.connection = None
1013 self.subscribed_event = threading.Event()
1014 # we need to know the encoding options for this connection in order
1015 # to lookup channel and pattern names for callback handlers.
1016 self.encoder = encoder
1017 self.push_handler_func = push_handler_func
1018 if event_dispatcher is None:
1019 self._event_dispatcher = EventDispatcher()
1020 else:
1021 self._event_dispatcher = event_dispatcher
1023 self._lock = threading.RLock()
1024 if self.encoder is None:
1025 self.encoder = self.connection_pool.get_encoder()
1026 self.health_check_response_b = self.encoder.encode(self.HEALTH_CHECK_MESSAGE)
1027 if self.encoder.decode_responses:
1028 self.health_check_response = ["pong", self.HEALTH_CHECK_MESSAGE]
1029 else:
1030 self.health_check_response = [b"pong", self.health_check_response_b]
1031 if self.push_handler_func is None:
1032 _set_info_logger()
1033 self.reset()
1035 def __enter__(self) -> "PubSub":
1036 return self
1038 def __exit__(self, exc_type, exc_value, traceback) -> None:
1039 self.reset()
1041 def __del__(self) -> None:
1042 try:
1043 # if this object went out of scope prior to shutting down
1044 # subscriptions, close the connection manually before
1045 # returning it to the connection pool
1046 self.reset()
1047 except Exception:
1048 pass
1050 def reset(self) -> None:
1051 if self.connection:
1052 self.connection.disconnect()
1053 self.connection.deregister_connect_callback(self.on_connect)
1054 self.connection_pool.release(self.connection)
1055 self.connection = None
1056 self.health_check_response_counter = 0
1057 self.channels = {}
1058 self.pending_unsubscribe_channels = set()
1059 self.shard_channels = {}
1060 self.pending_unsubscribe_shard_channels = set()
1061 self.patterns = {}
1062 self.pending_unsubscribe_patterns = set()
1063 self.subscribed_event.clear()
1065 def close(self) -> None:
1066 self.reset()
1068 def _resubscribe(self, subscribed, subscribe_fn) -> None:
1069 # Replay handler-backed subscriptions as positional Subscription objects
1070 # so binary names never need to be decoded into keyword argument keys.
1071 subscriptions = pubsub_subscription_args(subscribed)
1072 if subscriptions:
1073 subscribe_fn(*subscriptions)
1075 def _resubscribe_shard_channels(self) -> None:
1076 self._resubscribe(self.shard_channels, self.ssubscribe)
1078 def on_connect(self, connection) -> None:
1079 "Re-subscribe to any channels and patterns previously subscribed to"
1080 self.pending_unsubscribe_channels.clear()
1081 self.pending_unsubscribe_patterns.clear()
1082 self.pending_unsubscribe_shard_channels.clear()
1083 if self.channels:
1084 self._resubscribe(self.channels, self.subscribe)
1085 if self.patterns:
1086 self._resubscribe(self.patterns, self.psubscribe)
1087 if self.shard_channels:
1088 self._resubscribe_shard_channels()
1090 @property
1091 def subscribed(self) -> bool:
1092 """Indicates if there are subscriptions to any channels or patterns"""
1093 return self.subscribed_event.is_set()
1095 def execute_command(self, *args):
1096 """Execute a publish/subscribe command"""
1098 # NOTE: don't parse the response in this function -- it could pull a
1099 # legitimate message off the stack if the connection is already
1100 # subscribed to one or more channels
1102 if self.connection is None:
1103 self.connection = self.connection_pool.get_connection()
1104 # register a callback that re-subscribes to any channels we
1105 # were listening to when we were disconnected
1106 self.connection.register_connect_callback(self.on_connect)
1107 if self.push_handler_func is not None:
1108 self.connection._parser.set_pubsub_push_handler(self.push_handler_func)
1109 self._event_dispatcher.dispatch(
1110 AfterPubSubConnectionInstantiationEvent(
1111 self.connection, self.connection_pool, ClientType.SYNC, self._lock
1112 )
1113 )
1114 connection = self.connection
1115 kwargs = {"check_health": not self.subscribed}
1116 if not self.subscribed:
1117 self.clean_health_check_responses()
1118 with self._lock:
1119 self._execute(connection, connection.send_command, *args, **kwargs)
1121 def clean_health_check_responses(self) -> None:
1122 """
1123 If any health check responses are present, clean them
1124 """
1125 ttl = 10
1126 conn = self.connection
1127 while conn and self.health_check_response_counter > 0 and ttl > 0:
1128 if self._execute(conn, conn.can_read, timeout=conn.socket_timeout):
1129 response = self._execute(conn, conn.read_response)
1130 if self.is_health_check_response(response):
1131 self.health_check_response_counter -= 1
1132 else:
1133 raise PubSubError(
1134 "A non health check response was cleaned by "
1135 "execute_command: {}".format(response)
1136 )
1137 ttl -= 1
1139 def _reconnect(
1140 self,
1141 conn,
1142 error: Optional[Exception] = None,
1143 failure_count: Optional[int] = None,
1144 start_time: Optional[float] = None,
1145 command_name: Optional[str] = None,
1146 ) -> None:
1147 """
1148 The supported exceptions are already checked in the
1149 retry object so we don't need to do it here.
1151 In this error handler we are trying to reconnect to the server.
1152 """
1153 if error and failure_count <= conn.retry.get_retries():
1154 if command_name:
1155 record_operation_duration(
1156 command_name=command_name,
1157 duration_seconds=time.monotonic() - start_time,
1158 server_address=getattr(conn, "host", None),
1159 server_port=getattr(conn, "port", None),
1160 db_namespace=str(conn.db),
1161 error=error,
1162 retry_attempts=failure_count,
1163 )
1164 conn.disconnect()
1165 conn.connect()
1167 def _execute(self, conn, command, *args, **kwargs):
1168 """
1169 Connect manually upon disconnection. If the Redis server is down,
1170 this will fail and raise a ConnectionError as desired.
1171 After reconnection, the ``on_connect`` callback should have been
1172 called by the # connection to resubscribe us to any channels and
1173 patterns we were previously listening to
1174 """
1176 if conn.should_reconnect():
1177 self._reconnect(conn)
1179 if not len(args) == 0:
1180 command_name = args[0]
1181 else:
1182 command_name = None
1184 # Start timing for observability
1185 start_time = time.monotonic()
1186 # Track actual retry attempts for error reporting
1187 actual_retry_attempts = [0]
1189 def failure_callback(error, failure_count):
1190 actual_retry_attempts[0] = failure_count
1191 self._reconnect(conn, error, failure_count, start_time, command_name)
1193 try:
1194 response = conn.retry.call_with_retry(
1195 lambda: command(*args, **kwargs),
1196 failure_callback,
1197 with_failure_count=True,
1198 )
1200 if command_name:
1201 record_operation_duration(
1202 command_name=command_name,
1203 duration_seconds=time.monotonic() - start_time,
1204 server_address=getattr(conn, "host", None),
1205 server_port=getattr(conn, "port", None),
1206 db_namespace=str(conn.db),
1207 )
1209 return response
1210 except Exception as e:
1211 record_error_count(
1212 server_address=getattr(conn, "host", None),
1213 server_port=getattr(conn, "port", None),
1214 network_peer_address=getattr(conn, "host", None),
1215 network_peer_port=getattr(conn, "port", None),
1216 error_type=e,
1217 retry_attempts=actual_retry_attempts[0],
1218 is_internal=False,
1219 )
1220 raise
1222 def parse_response(self, block=True, timeout=0):
1223 """
1224 Parse the response from a publish/subscribe command.
1226 Args:
1227 block: If True, block indefinitely until a message is available.
1228 If False, return immediately if no message is available.
1229 Default: True
1230 timeout: The timeout in seconds for reading a response when block=False.
1231 This parameter is ignored when block=True.
1232 Default: 0 (return immediately if no data available)
1234 Returns:
1235 The parsed response from the server, or None if no message is available
1236 within the timeout period (when block=False).
1238 Important:
1239 The block and timeout parameters work together:
1240 - When block=True: timeout is IGNORED, method blocks indefinitely
1241 - When block=False: timeout is USED, method returns after timeout expires
1243 Typically, you should use get_message(timeout=X) instead of calling
1244 parse_response() directly. The get_message() method automatically sets
1245 block=False when a timeout is provided, and block=True when timeout=None.
1247 Example:
1248 # Block indefinitely (timeout is ignored)
1249 response = pubsub.parse_response(block=True, timeout=0.1)
1251 # Non-blocking with 0.1 second timeout
1252 response = pubsub.parse_response(block=False, timeout=0.1)
1254 # Non-blocking, return immediately
1255 response = pubsub.parse_response(block=False, timeout=0)
1257 # Recommended: use get_message() instead
1258 msg = pubsub.get_message(timeout=0.1) # automatically sets block=False
1259 msg = pubsub.get_message(timeout=None) # automatically sets block=True
1260 """
1261 conn = self.connection
1262 if conn is None:
1263 raise RuntimeError(
1264 "pubsub connection not set: "
1265 "did you forget to call subscribe() or psubscribe()?"
1266 )
1268 self.check_health()
1270 def try_read():
1271 if not block:
1272 if not conn.can_read(timeout=timeout):
1273 return None
1274 read_timeout = timeout
1275 else:
1276 conn.connect()
1277 # Block indefinitely waiting for a pubsub message. timeout=None
1278 # makes the socket layer call sock.settimeout(None) for this read
1279 # (and restore the original socket_timeout afterwards), so the
1280 # configured socket_timeout does not abort the read.
1281 read_timeout = None
1282 return conn.read_response(
1283 disconnect_on_error=False, push_request=True, timeout=read_timeout
1284 )
1286 response = self._execute(conn, try_read)
1288 if self.is_health_check_response(response):
1289 # ignore the health check message as user might not expect it
1290 self.health_check_response_counter -= 1
1291 return None
1292 return response
1294 def is_health_check_response(self, response) -> bool:
1295 """
1296 Check if the response is a health check response.
1297 If there are no subscriptions redis responds to PING command with a
1298 bulk response, instead of a multi-bulk with "pong" and the response.
1299 """
1300 if self.encoder.decode_responses:
1301 return (
1302 response
1303 in [
1304 self.health_check_response, # If there is a subscription
1305 self.HEALTH_CHECK_MESSAGE, # If there are no subscriptions and decode_responses=True
1306 ]
1307 )
1308 else:
1309 return (
1310 response
1311 in [
1312 self.health_check_response, # If there is a subscription
1313 self.health_check_response_b, # If there isn't a subscription and decode_responses=False
1314 ]
1315 )
1317 def check_health(self) -> None:
1318 conn = self.connection
1319 if conn is None:
1320 raise RuntimeError(
1321 "pubsub connection not set: "
1322 "did you forget to call subscribe() or psubscribe()?"
1323 )
1325 if conn.health_check_interval and time.monotonic() > conn.next_health_check:
1326 conn.send_command("PING", self.HEALTH_CHECK_MESSAGE, check_health=False)
1327 self.health_check_response_counter += 1
1329 def _normalize_keys(self, data) -> Dict:
1330 """
1331 normalize channel/pattern names to be either bytes or strings
1332 based on whether responses are automatically decoded. this saves us
1333 from coercing the value for each message coming in.
1334 """
1335 encode = self.encoder.encode
1336 decode = self.encoder.decode
1337 return {decode(encode(k)): v for k, v in data.items()}
1339 def psubscribe(
1340 self, *args: ChannelT | Subscription, **kwargs: PubSubHandler
1341 ) -> None:
1342 """
1343 Subscribe to channel patterns.
1344 Patterns supplied as keyword arguments expect a pattern name as the
1345 key and a callable as the value.
1346 ``Subscription`` objects can also be supplied positionally with an
1347 optional handler.
1348 A pattern's callable will be invoked automatically
1349 when a message is received on that pattern rather than producing a
1350 message via ``listen()``.
1351 """
1352 new_patterns = parse_pubsub_subscriptions(args, kwargs)
1353 ret_val = self.execute_command("PSUBSCRIBE", *new_patterns.keys())
1354 # update the patterns dict AFTER we send the command. we don't want to
1355 # subscribe twice to these patterns, once for the command and again
1356 # for the reconnection.
1357 new_patterns = self._normalize_keys(new_patterns)
1358 self.patterns.update(new_patterns)
1359 if not self.subscribed:
1360 # Set the subscribed_event flag to True
1361 self.subscribed_event.set()
1362 # Clear the health check counter
1363 self.health_check_response_counter = 0
1364 self.pending_unsubscribe_patterns.difference_update(new_patterns)
1365 return ret_val
1367 def punsubscribe(self, *args):
1368 """
1369 Unsubscribe from the supplied patterns. If empty, unsubscribe from
1370 all patterns.
1371 """
1372 if args:
1373 args = list_or_args(args[0], args[1:])
1374 patterns = self._normalize_keys(dict.fromkeys(args))
1375 else:
1376 patterns = self.patterns
1377 self.pending_unsubscribe_patterns.update(patterns)
1378 return self.execute_command("PUNSUBSCRIBE", *args)
1380 def subscribe(
1381 self, *args: ChannelT | Subscription, **kwargs: PubSubHandler
1382 ) -> None:
1383 """
1384 Subscribe to channels.
1385 Channels supplied as keyword arguments expect
1386 a channel name as the key and a callable as the value.
1387 ``Subscription`` objects can also be supplied positionally with an
1388 optional handler.
1389 A channel's callable will be invoked automatically
1390 when a message is received on that channel rather than producing a
1391 message via ``listen()`` or ``get_message()``.
1392 """
1393 new_channels = parse_pubsub_subscriptions(args, kwargs)
1394 ret_val = self.execute_command("SUBSCRIBE", *new_channels.keys())
1395 # update the channels dict AFTER we send the command. we don't want to
1396 # subscribe twice to these channels, once for the command and again
1397 # for the reconnection.
1398 new_channels = self._normalize_keys(new_channels)
1399 self.channels.update(new_channels)
1400 if not self.subscribed:
1401 # Set the subscribed_event flag to True
1402 self.subscribed_event.set()
1403 # Clear the health check counter
1404 self.health_check_response_counter = 0
1405 self.pending_unsubscribe_channels.difference_update(new_channels)
1406 return ret_val
1408 def unsubscribe(self, *args):
1409 """
1410 Unsubscribe from the supplied channels. If empty, unsubscribe from
1411 all channels
1412 """
1413 if args:
1414 args = list_or_args(args[0], args[1:])
1415 channels = self._normalize_keys(dict.fromkeys(args))
1416 else:
1417 channels = self.channels
1418 self.pending_unsubscribe_channels.update(channels)
1419 return self.execute_command("UNSUBSCRIBE", *args)
1421 def ssubscribe(
1422 self,
1423 *args: ChannelT | Subscription,
1424 target_node: Any = None,
1425 **kwargs: PubSubHandler,
1426 ) -> None:
1427 """
1428 Subscribes the client to the specified shard channels.
1429 Channels supplied as keyword arguments expect a channel name as the key
1430 and a callable as the value.
1431 ``Subscription`` objects can also be supplied positionally
1432 with an optional handler.
1433 A channel's callable will be invoked automatically when a message
1434 is received on that channel rather than producing a message
1435 via ``listen()`` or ``get_sharded_message()``.
1436 """
1437 new_s_channels = parse_pubsub_subscriptions(args, kwargs)
1438 ret_val = self.execute_command("SSUBSCRIBE", *new_s_channels.keys())
1439 # update the s_channels dict AFTER we send the command. we don't want to
1440 # subscribe twice to these channels, once for the command and again
1441 # for the reconnection.
1442 new_s_channels = self._normalize_keys(new_s_channels)
1443 self.shard_channels.update(new_s_channels)
1444 if not self.subscribed:
1445 # Set the subscribed_event flag to True
1446 self.subscribed_event.set()
1447 # Clear the health check counter
1448 self.health_check_response_counter = 0
1449 self.pending_unsubscribe_shard_channels.difference_update(new_s_channels)
1450 return ret_val
1452 def sunsubscribe(self, *args, target_node=None):
1453 """
1454 Unsubscribe from the supplied shard_channels. If empty, unsubscribe from
1455 all shard_channels
1456 """
1457 if args:
1458 args = list_or_args(args[0], args[1:])
1459 s_channels = self._normalize_keys(dict.fromkeys(args))
1460 else:
1461 s_channels = self.shard_channels
1462 self.pending_unsubscribe_shard_channels.update(s_channels)
1463 return self.execute_command("SUNSUBSCRIBE", *args)
1465 def listen(self):
1466 "Listen for messages on channels this client has been subscribed to"
1467 while self.subscribed:
1468 response = self.handle_message(self.parse_response(block=True))
1469 if response is not None:
1470 yield response
1472 def get_message(
1473 self, ignore_subscribe_messages: bool = False, timeout: float = 0.0
1474 ):
1475 """
1476 Get the next message if one is available, otherwise None.
1478 If timeout is specified, the system will wait for `timeout` seconds
1479 before returning. Timeout should be specified as a floating point
1480 number, or None, to wait indefinitely.
1481 """
1482 if not self.subscribed:
1483 # Wait for subscription
1484 start_time = time.monotonic()
1485 if self.subscribed_event.wait(timeout) is True:
1486 # The connection was subscribed during the timeout time frame.
1487 # The timeout should be adjusted based on the time spent
1488 # waiting for the subscription
1489 time_spent = time.monotonic() - start_time
1490 timeout = max(0.0, timeout - time_spent)
1491 else:
1492 # The connection isn't subscribed to any channels or patterns,
1493 # so no messages are available
1494 return None
1496 response = self.parse_response(block=(timeout is None), timeout=timeout)
1498 if response:
1499 return self.handle_message(response, ignore_subscribe_messages)
1500 return None
1502 get_sharded_message = get_message
1504 def ping(self, message: Union[str, None] = None) -> bool:
1505 """
1506 Ping the Redis server to test connectivity.
1508 Sends a PING command to the Redis server and returns True if the server
1509 responds with "PONG".
1510 """
1511 args = ["PING", message] if message is not None else ["PING"]
1512 return self.execute_command(*args)
1514 def handle_message(self, response, ignore_subscribe_messages=False):
1515 """
1516 Parses a pub/sub message. If the channel or pattern was subscribed to
1517 with a message handler, the handler is invoked instead of a parsed
1518 message being returned.
1519 """
1520 if response is None:
1521 return None
1522 if isinstance(response, bytes):
1523 response = [b"pong", response] if response != b"PONG" else [b"pong", b""]
1525 message_type = str_if_bytes(response[0])
1526 if message_type == "pmessage":
1527 message = {
1528 "type": message_type,
1529 "pattern": response[1],
1530 "channel": response[2],
1531 "data": response[3],
1532 }
1533 elif message_type == "pong":
1534 message = {
1535 "type": message_type,
1536 "pattern": None,
1537 "channel": None,
1538 "data": response[1],
1539 }
1540 else:
1541 message = {
1542 "type": message_type,
1543 "pattern": None,
1544 "channel": response[1],
1545 "data": response[2],
1546 }
1548 if message_type in ["message", "pmessage"]:
1549 channel = str_if_bytes(message["channel"])
1550 record_pubsub_message(
1551 direction=PubSubDirection.RECEIVE,
1552 channel=channel,
1553 )
1554 elif message_type == "smessage":
1555 channel = str_if_bytes(message["channel"])
1556 record_pubsub_message(
1557 direction=PubSubDirection.RECEIVE,
1558 channel=channel,
1559 sharded=True,
1560 )
1562 # if this is an unsubscribe message, remove it from memory
1563 if message_type in self.UNSUBSCRIBE_MESSAGE_TYPES:
1564 if message_type == "punsubscribe":
1565 pattern = response[1]
1566 if pattern in self.pending_unsubscribe_patterns:
1567 self.pending_unsubscribe_patterns.remove(pattern)
1568 self.patterns.pop(pattern, None)
1569 elif message_type == "sunsubscribe":
1570 s_channel = response[1]
1571 if s_channel in self.pending_unsubscribe_shard_channels:
1572 self.pending_unsubscribe_shard_channels.remove(s_channel)
1573 self.shard_channels.pop(s_channel, None)
1574 else:
1575 channel = response[1]
1576 if channel in self.pending_unsubscribe_channels:
1577 self.pending_unsubscribe_channels.remove(channel)
1578 self.channels.pop(channel, None)
1579 if not self.channels and not self.patterns and not self.shard_channels:
1580 # There are no subscriptions anymore, set subscribed_event flag
1581 # to false
1582 self.subscribed_event.clear()
1584 if message_type in self.PUBLISH_MESSAGE_TYPES:
1585 # if there's a message handler, invoke it
1586 if message_type == "pmessage":
1587 handler = self.patterns.get(message["pattern"], None)
1588 elif message_type == "smessage":
1589 handler = self.shard_channels.get(message["channel"], None)
1590 else:
1591 handler = self.channels.get(message["channel"], None)
1592 if handler:
1593 handler(message)
1594 return None
1595 elif message_type != "pong":
1596 # this is a subscribe/unsubscribe message. ignore if we don't
1597 # want them
1598 if ignore_subscribe_messages or self.ignore_subscribe_messages:
1599 return None
1601 return message
1603 def run_in_thread(
1604 self,
1605 sleep_time: float = 0.0,
1606 daemon: bool = False,
1607 exception_handler: Optional[Callable] = None,
1608 pubsub=None,
1609 sharded_pubsub: bool = False,
1610 ) -> "PubSubWorkerThread":
1611 for channel, handler in self.channels.items():
1612 if handler is None:
1613 raise PubSubError(f"Channel: '{channel}' has no handler registered")
1614 for pattern, handler in self.patterns.items():
1615 if handler is None:
1616 raise PubSubError(f"Pattern: '{pattern}' has no handler registered")
1617 for s_channel, handler in self.shard_channels.items():
1618 if handler is None:
1619 raise PubSubError(
1620 f"Shard Channel: '{s_channel}' has no handler registered"
1621 )
1623 pubsub = self if pubsub is None else pubsub
1624 thread = PubSubWorkerThread(
1625 pubsub,
1626 sleep_time,
1627 daemon=daemon,
1628 exception_handler=exception_handler,
1629 sharded_pubsub=sharded_pubsub,
1630 )
1631 thread.start()
1632 return thread
1635class PubSubWorkerThread(threading.Thread):
1636 def __init__(
1637 self,
1638 pubsub,
1639 sleep_time: float,
1640 daemon: bool = False,
1641 exception_handler: Union[
1642 Callable[[Exception, "PubSub", "PubSubWorkerThread"], None], None
1643 ] = None,
1644 sharded_pubsub: bool = False,
1645 ):
1646 super().__init__()
1647 self.daemon = daemon
1648 self.pubsub = pubsub
1649 self.sleep_time = sleep_time
1650 self.exception_handler = exception_handler
1651 self.sharded_pubsub = sharded_pubsub
1652 self._running = threading.Event()
1654 def run(self) -> None:
1655 if self._running.is_set():
1656 return
1657 self._running.set()
1658 pubsub = self.pubsub
1659 sleep_time = self.sleep_time
1660 while self._running.is_set():
1661 try:
1662 if not self.sharded_pubsub:
1663 pubsub.get_message(
1664 ignore_subscribe_messages=True, timeout=sleep_time
1665 )
1666 else:
1667 pubsub.get_sharded_message(
1668 ignore_subscribe_messages=True, timeout=sleep_time
1669 )
1670 except BaseException as e:
1671 if self.exception_handler is None:
1672 raise
1673 self.exception_handler(e, pubsub, self)
1674 pubsub.close()
1676 def stop(self) -> None:
1677 # trip the flag so the run loop exits. the run loop will
1678 # close the pubsub connection, which disconnects the socket
1679 # and returns the connection to the pool.
1680 self._running.clear()
1683class Pipeline(Redis):
1684 """
1685 Pipelines provide a way to transmit multiple commands to the Redis server
1686 in one transmission. This is convenient for batch processing, such as
1687 saving all the values in a list to Redis.
1689 All commands executed within a pipeline(when running in transactional mode,
1690 which is the default behavior) are wrapped with MULTI and EXEC
1691 calls. This guarantees all commands executed in the pipeline will be
1692 executed atomically.
1694 Any command raising an exception does *not* halt the execution of
1695 subsequent commands in the pipeline. Instead, the exception is caught
1696 and its instance is placed into the response list returned by execute().
1697 Code iterating over the response list should be able to deal with an
1698 instance of an exception as a potential value. In general, these will be
1699 ResponseError exceptions, such as those raised when issuing a command
1700 on a key of a different datatype.
1701 """
1703 UNWATCH_COMMANDS = {"DISCARD", "EXEC", "UNWATCH"}
1705 def __init__(
1706 self,
1707 connection_pool: ConnectionPool,
1708 response_callbacks,
1709 transaction,
1710 shard_hint,
1711 ):
1712 self.connection_pool = connection_pool
1713 self.connection: Optional[Connection] = None
1714 self.response_callbacks = response_callbacks
1715 self.transaction = transaction
1716 self.shard_hint = shard_hint
1717 self.watching = False
1718 self.command_stack = []
1719 self.scripts: Set[Script] = set()
1720 self.explicit_transaction = False
1722 def __enter__(self) -> "Pipeline":
1723 return self
1725 def __exit__(self, exc_type, exc_value, traceback):
1726 self.reset()
1728 def __del__(self):
1729 try:
1730 self.reset()
1731 except Exception:
1732 pass
1734 def __len__(self) -> int:
1735 return len(self.command_stack)
1737 def __bool__(self) -> bool:
1738 """Pipeline instances should always evaluate to True"""
1739 return True
1741 def reset(self) -> None:
1742 self.command_stack = []
1743 self.scripts = set()
1744 # make sure to reset the connection state in the event that we were
1745 # watching something
1746 if self.watching and self.connection:
1747 try:
1748 # call this manually since our unwatch or
1749 # immediate_execute_command methods can call reset()
1750 self.connection.send_command("UNWATCH")
1751 self.connection.read_response()
1752 except ConnectionError:
1753 # disconnect will also remove any previous WATCHes
1754 self.connection.disconnect()
1755 # clean up the other instance attributes
1756 self.watching = False
1757 self.explicit_transaction = False
1759 # we can safely return the connection to the pool here since we're
1760 # sure we're no longer WATCHing anything
1761 if self.connection:
1762 self.connection_pool.release(self.connection)
1763 self.connection = None
1765 def close(self) -> None:
1766 """Close the pipeline"""
1767 self.reset()
1769 def multi(self) -> None:
1770 """
1771 Start a transactional block of the pipeline after WATCH commands
1772 are issued. End the transactional block with `execute`.
1773 """
1774 if self.explicit_transaction:
1775 raise RedisError("Cannot issue nested calls to MULTI")
1776 if self.command_stack:
1777 raise RedisError(
1778 "Commands without an initial WATCH have already been issued"
1779 )
1780 self.explicit_transaction = True
1782 def execute_command(self, *args, **kwargs):
1783 if (self.watching or args[0] == "WATCH") and not self.explicit_transaction:
1784 return self.immediate_execute_command(*args, **kwargs)
1785 return self.pipeline_execute_command(*args, **kwargs)
1787 def _disconnect_reset_raise_on_watching(
1788 self,
1789 conn: AbstractConnection,
1790 error: Exception,
1791 failure_count: Optional[int] = None,
1792 start_time: Optional[float] = None,
1793 command_name: Optional[str] = None,
1794 ) -> None:
1795 """
1796 Close the connection reset watching state and
1797 raise an exception if we were watching.
1799 The supported exceptions are already checked in the
1800 retry object so we don't need to do it here.
1802 After we disconnect the connection, it will try to reconnect and
1803 do a health check as part of the send_command logic(on connection level).
1804 """
1805 if error and failure_count <= conn.retry.get_retries():
1806 record_operation_duration(
1807 command_name=command_name,
1808 duration_seconds=time.monotonic() - start_time,
1809 server_address=getattr(conn, "host", None),
1810 server_port=getattr(conn, "port", None),
1811 db_namespace=str(conn.db),
1812 error=error,
1813 retry_attempts=failure_count,
1814 )
1815 conn.disconnect()
1817 # if we were already watching a variable, the watch is no longer
1818 # valid since this connection has died. raise a WatchError, which
1819 # indicates the user should retry this transaction.
1820 if self.watching:
1821 self.reset()
1822 raise WatchError(
1823 f"A {type(error).__name__} occurred while watching one or more keys"
1824 )
1826 def immediate_execute_command(self, *args, **options):
1827 """
1828 Execute a command immediately, but don't auto-retry on the supported
1829 errors for retry if we're already WATCHing a variable.
1830 Used when issuing WATCH or subsequent commands retrieving their values but before
1831 MULTI is called.
1832 """
1833 command_name = args[0]
1834 conn = self.connection
1835 # if this is the first call, we need a connection
1836 if not conn:
1837 conn = self.connection_pool.get_connection()
1838 self.connection = conn
1840 # Start timing for observability
1841 start_time = time.monotonic()
1842 # Track actual retry attempts for error reporting
1843 actual_retry_attempts = [0]
1845 def failure_callback(error, failure_count):
1846 if is_debug_log_enabled():
1847 add_debug_log_for_operation_failure(conn)
1848 actual_retry_attempts[0] = failure_count
1849 self._disconnect_reset_raise_on_watching(
1850 conn, error, failure_count, start_time, command_name
1851 )
1853 try:
1854 response = conn.retry.call_with_retry(
1855 lambda: self._send_command_parse_response(
1856 conn, command_name, *args, **options
1857 ),
1858 failure_callback,
1859 with_failure_count=True,
1860 )
1862 record_operation_duration(
1863 command_name=command_name,
1864 duration_seconds=time.monotonic() - start_time,
1865 server_address=getattr(conn, "host", None),
1866 server_port=getattr(conn, "port", None),
1867 db_namespace=str(conn.db),
1868 )
1870 return response
1871 except Exception as e:
1872 record_error_count(
1873 server_address=getattr(conn, "host", None),
1874 server_port=getattr(conn, "port", None),
1875 network_peer_address=getattr(conn, "host", None),
1876 network_peer_port=getattr(conn, "port", None),
1877 error_type=e,
1878 retry_attempts=actual_retry_attempts[0],
1879 is_internal=False,
1880 )
1881 raise
1883 def pipeline_execute_command(self, *args, **options) -> "Pipeline":
1884 """
1885 Stage a command to be executed when execute() is next called
1887 Returns the current Pipeline object back so commands can be
1888 chained together, such as:
1890 pipe = pipe.set('foo', 'bar').incr('baz').decr('bang')
1892 At some other point, you can then run: pipe.execute(),
1893 which will execute all commands queued in the pipe.
1894 """
1895 self.command_stack.append((args, options))
1896 return self
1898 def _execute_transaction(
1899 self, connection: Connection, commands, raise_on_error
1900 ) -> List:
1901 cmds = chain([(("MULTI",), {})], commands, [(("EXEC",), {})])
1902 all_cmds = connection.pack_commands(
1903 [args for args, options in cmds if EMPTY_RESPONSE not in options]
1904 )
1905 connection.send_packed_command(all_cmds)
1906 errors = []
1908 # parse off the response for MULTI
1909 # NOTE: we need to handle ResponseErrors here and continue
1910 # so that we read all the additional command messages from
1911 # the socket
1912 try:
1913 self.parse_response(connection, "_")
1914 except ResponseError as e:
1915 errors.append((0, e))
1917 # and all the other commands
1918 for i, command in enumerate(commands):
1919 if EMPTY_RESPONSE in command[1]:
1920 errors.append((i, command[1][EMPTY_RESPONSE]))
1921 else:
1922 try:
1923 self.parse_response(connection, "_")
1924 except ResponseError as e:
1925 self.annotate_exception(e, i + 1, command[0])
1926 errors.append((i, e))
1928 # parse the EXEC.
1929 try:
1930 response = self.parse_response(connection, "_")
1931 except ExecAbortError:
1932 if errors:
1933 raise errors[0][1]
1934 raise
1936 # EXEC clears any watched keys
1937 self.watching = False
1939 if response is None:
1940 raise WatchError("Watched variable changed.")
1942 # put any parse errors into the response
1943 for i, e in errors:
1944 response.insert(i, e)
1946 if len(response) != len(commands):
1947 self.connection.disconnect()
1948 raise ResponseError(
1949 "Wrong number of response items from pipeline execution"
1950 )
1952 # find any errors in the response and raise if necessary
1953 if raise_on_error:
1954 self.raise_first_error(commands, response)
1956 # We have to run response callbacks manually
1957 data = []
1958 for r, cmd in zip(response, commands):
1959 if not isinstance(r, Exception):
1960 args, options = cmd
1961 # Remove keys entry, it needs only for cache.
1962 options.pop("keys", None)
1963 command_name = args[0]
1964 if command_name in self.response_callbacks:
1965 r = self.response_callbacks[command_name](r, **options)
1966 data.append(r)
1968 return data
1970 def _execute_pipeline(self, connection, commands, raise_on_error):
1971 # build up all commands into a single request to increase network perf
1972 all_cmds = connection.pack_commands([args for args, _ in commands])
1973 connection.send_packed_command(all_cmds)
1975 responses = []
1976 for args, options in commands:
1977 try:
1978 responses.append(self.parse_response(connection, args[0], **options))
1979 except ResponseError as e:
1980 responses.append(e)
1982 if raise_on_error:
1983 self.raise_first_error(commands, responses)
1985 return responses
1987 def raise_first_error(self, commands, response):
1988 for i, r in enumerate(response):
1989 if isinstance(r, ResponseError):
1990 self.annotate_exception(r, i + 1, commands[i][0])
1991 raise r
1993 def annotate_exception(self, exception, number, command):
1994 cmd = " ".join(map(safe_str, command))
1995 msg = (
1996 f"Command # {number} ({truncate_text(cmd)}) of pipeline "
1997 f"caused error: {exception.args[0]}"
1998 )
1999 exception.args = (msg,) + exception.args[1:]
2001 def parse_response(self, connection, command_name, **options):
2002 result = Redis.parse_response(self, connection, command_name, **options)
2003 if command_name in self.UNWATCH_COMMANDS:
2004 self.watching = False
2005 elif command_name == "WATCH":
2006 self.watching = True
2007 return result
2009 def load_scripts(self):
2010 # make sure all scripts that are about to be run on this pipeline exist
2011 scripts = list(self.scripts)
2012 immediate = self.immediate_execute_command
2013 shas = [s.sha for s in scripts]
2014 # we can't use the normal script_* methods because they would just
2015 # get buffered in the pipeline.
2016 exists = immediate("SCRIPT EXISTS", *shas)
2017 if not all(exists):
2018 for s, exist in zip(scripts, exists):
2019 if not exist:
2020 s.sha = immediate("SCRIPT LOAD", s.script)
2022 def _disconnect_raise_on_watching(
2023 self,
2024 conn: AbstractConnection,
2025 error: Exception,
2026 failure_count: Optional[int] = None,
2027 start_time: Optional[float] = None,
2028 command_name: Optional[str] = None,
2029 ) -> None:
2030 """
2031 Close the connection, raise an exception if we were watching.
2033 The supported exceptions are already checked in the
2034 retry object so we don't need to do it here.
2036 After we disconnect the connection, it will try to reconnect and
2037 do a health check as part of the send_command logic(on connection level).
2038 """
2039 if error and failure_count <= conn.retry.get_retries():
2040 record_operation_duration(
2041 command_name=command_name,
2042 duration_seconds=time.monotonic() - start_time,
2043 server_address=getattr(conn, "host", None),
2044 server_port=getattr(conn, "port", None),
2045 db_namespace=str(conn.db),
2046 error=error,
2047 retry_attempts=failure_count,
2048 )
2049 conn.disconnect()
2050 # if we were watching a variable, the watch is no longer valid
2051 # since this connection has died. raise a WatchError, which
2052 # indicates the user should retry this transaction.
2053 if self.watching:
2054 raise WatchError(
2055 f"A {type(error).__name__} occurred while watching one or more keys"
2056 )
2058 def execute(self, raise_on_error: bool = True) -> List[Any]:
2059 """Execute all the commands in the current pipeline"""
2060 stack = self.command_stack
2061 if not stack and not self.watching:
2062 return []
2063 if self.scripts:
2064 self.load_scripts()
2065 if self.transaction or self.explicit_transaction:
2066 execute = self._execute_transaction
2067 operation_name = "MULTI"
2068 else:
2069 execute = self._execute_pipeline
2070 operation_name = "PIPELINE"
2072 conn = self.connection
2073 if not conn:
2074 conn = self.connection_pool.get_connection()
2075 # assign to self.connection so reset() releases the connection
2076 # back to the pool after we're done
2077 self.connection = conn
2079 # Start timing for observability
2080 start_time = time.monotonic()
2081 # Track actual retry attempts for error reporting
2082 actual_retry_attempts = [0]
2084 def failure_callback(error, failure_count):
2085 if is_debug_log_enabled():
2086 add_debug_log_for_operation_failure(conn)
2087 actual_retry_attempts[0] = failure_count
2088 self._disconnect_raise_on_watching(
2089 conn, error, failure_count, start_time, operation_name
2090 )
2092 try:
2093 response = conn.retry.call_with_retry(
2094 lambda: execute(conn, stack, raise_on_error),
2095 failure_callback,
2096 with_failure_count=True,
2097 )
2099 record_operation_duration(
2100 command_name=operation_name,
2101 duration_seconds=time.monotonic() - start_time,
2102 server_address=getattr(conn, "host", None),
2103 server_port=getattr(conn, "port", None),
2104 db_namespace=str(conn.db),
2105 )
2106 return response
2107 except Exception as e:
2108 record_error_count(
2109 server_address=getattr(conn, "host", None),
2110 server_port=getattr(conn, "port", None),
2111 network_peer_address=getattr(conn, "host", None),
2112 network_peer_port=getattr(conn, "port", None),
2113 error_type=e,
2114 retry_attempts=actual_retry_attempts[0],
2115 is_internal=False,
2116 )
2117 raise
2119 finally:
2120 # in reset() the connection is disconnected before returned to the pool if
2121 # it is marked for reconnect.
2122 self.reset()
2124 def discard(self):
2125 """
2126 Flushes all previously queued commands
2127 See: https://redis.io/commands/DISCARD
2128 """
2129 self.execute_command("DISCARD")
2131 def watch(self, *names):
2132 """Watches the values at keys ``names``"""
2133 if self.explicit_transaction:
2134 raise RedisError("Cannot issue a WATCH after a MULTI")
2135 return self.execute_command("WATCH", *names)
2137 def unwatch(self) -> bool:
2138 """Unwatches all previously specified keys"""
2139 return self.watching and self.execute_command("UNWATCH") or True