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.disconnect()
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 if conn and conn.should_reconnect():
867 self._close_connection(conn)
868 conn.connect()
869 if self._single_connection_client:
870 self.single_connection_lock.release()
871 if not self.connection:
872 pool.release(conn)
874 def parse_response(self, connection, command_name, **options):
875 """Parses a response from the Redis server"""
876 try:
877 if NEVER_DECODE in options:
878 response = connection.read_response(disable_decoding=True)
879 options.pop(NEVER_DECODE)
880 else:
881 response = connection.read_response()
882 except ResponseError:
883 if EMPTY_RESPONSE in options:
884 return options[EMPTY_RESPONSE]
885 raise
887 if EMPTY_RESPONSE in options:
888 options.pop(EMPTY_RESPONSE)
890 # Remove keys entry, it needs only for cache.
891 options.pop("keys", None)
893 if command_name in self.response_callbacks:
894 return self.response_callbacks[command_name](response, **options)
895 return response
897 def get_cache(self) -> Optional[CacheInterface]:
898 return self.connection_pool.cache
901StrictRedis = Redis
904class Monitor:
905 """
906 Monitor is useful for handling the MONITOR command to the redis server.
907 next_command() method returns one command from monitor
908 listen() method yields commands from monitor.
909 """
911 monitor_re = re.compile(r"\[(\d+) (.*?)\] (.*)")
912 command_re = re.compile(r'"(.*?)(?<!\\)"')
914 def __init__(self, connection_pool):
915 self.connection_pool = connection_pool
916 self.connection = self.connection_pool.get_connection()
918 def __enter__(self):
919 self._start_monitor()
920 return self
922 def __exit__(self, *args):
923 self.connection.disconnect()
924 self.connection_pool.release(self.connection)
926 def next_command(self):
927 """Parse the response from a monitor command"""
928 response = self.connection.read_response()
930 if response is None:
931 return None
933 if isinstance(response, bytes):
934 response = self.connection.encoder.decode(response, force=True)
936 command_time, command_data = response.split(" ", 1)
937 m = self.monitor_re.match(command_data)
938 db_id, client_info, command = m.groups()
939 command = " ".join(self.command_re.findall(command))
940 # Redis escapes double quotes because each piece of the command
941 # string is surrounded by double quotes. We don't have that
942 # requirement so remove the escaping and leave the quote.
943 command = command.replace('\\"', '"')
945 if client_info == "lua":
946 client_address = "lua"
947 client_port = ""
948 client_type = "lua"
949 elif client_info.startswith("unix"):
950 client_address = "unix"
951 client_port = client_info[5:]
952 client_type = "unix"
953 else:
954 if client_info == "":
955 client_address = ""
956 client_port = ""
957 client_type = "unknown"
958 else:
959 # use rsplit as ipv6 addresses contain colons
960 client_address, client_port = client_info.rsplit(":", 1)
961 client_type = "tcp"
962 return {
963 "time": float(command_time),
964 "db": int(db_id),
965 "client_address": client_address,
966 "client_port": client_port,
967 "client_type": client_type,
968 "command": command,
969 }
971 def listen(self):
972 """Listen for commands coming to the server."""
973 while True:
974 yield self.next_command()
976 def _start_monitor(self):
977 self.connection.send_command("MONITOR")
978 # check that monitor returns 'OK', but don't return it to user
979 response = self.connection.read_response()
981 if not bool_ok(response):
982 raise RedisError(f"MONITOR failed: {response}")
985class PubSub:
986 """
987 PubSub provides publish, subscribe and listen support to Redis channels.
989 After subscribing to one or more channels, the listen() method will block
990 until a message arrives on one of the subscribed channels. That message
991 will be returned and it's safe to start listening again.
992 """
994 PUBLISH_MESSAGE_TYPES = ("message", "pmessage", "smessage")
995 UNSUBSCRIBE_MESSAGE_TYPES = ("unsubscribe", "punsubscribe", "sunsubscribe")
996 HEALTH_CHECK_MESSAGE = "redis-py-health-check"
998 def __init__(
999 self,
1000 connection_pool,
1001 shard_hint=None,
1002 ignore_subscribe_messages: bool = False,
1003 encoder: Optional["Encoder"] = None,
1004 push_handler_func: Union[None, Callable[[str], None]] = None,
1005 event_dispatcher: Optional["EventDispatcher"] = None,
1006 ):
1007 self.connection_pool = connection_pool
1008 self.shard_hint = shard_hint
1009 self.ignore_subscribe_messages = ignore_subscribe_messages
1010 self.connection = None
1011 self.subscribed_event = threading.Event()
1012 # we need to know the encoding options for this connection in order
1013 # to lookup channel and pattern names for callback handlers.
1014 self.encoder = encoder
1015 self.push_handler_func = push_handler_func
1016 if event_dispatcher is None:
1017 self._event_dispatcher = EventDispatcher()
1018 else:
1019 self._event_dispatcher = event_dispatcher
1021 self._lock = threading.RLock()
1022 if self.encoder is None:
1023 self.encoder = self.connection_pool.get_encoder()
1024 self.health_check_response_b = self.encoder.encode(self.HEALTH_CHECK_MESSAGE)
1025 if self.encoder.decode_responses:
1026 self.health_check_response = ["pong", self.HEALTH_CHECK_MESSAGE]
1027 else:
1028 self.health_check_response = [b"pong", self.health_check_response_b]
1029 if self.push_handler_func is None:
1030 _set_info_logger()
1031 self.reset()
1033 def __enter__(self) -> "PubSub":
1034 return self
1036 def __exit__(self, exc_type, exc_value, traceback) -> None:
1037 self.reset()
1039 def __del__(self) -> None:
1040 try:
1041 # if this object went out of scope prior to shutting down
1042 # subscriptions, close the connection manually before
1043 # returning it to the connection pool
1044 self.reset()
1045 except Exception:
1046 pass
1048 def reset(self) -> None:
1049 if self.connection:
1050 self.connection.disconnect()
1051 self.connection.deregister_connect_callback(self.on_connect)
1052 self.connection_pool.release(self.connection)
1053 self.connection = None
1054 self.health_check_response_counter = 0
1055 self.channels = {}
1056 self.pending_unsubscribe_channels = set()
1057 self.shard_channels = {}
1058 self.pending_unsubscribe_shard_channels = set()
1059 self.patterns = {}
1060 self.pending_unsubscribe_patterns = set()
1061 self.subscribed_event.clear()
1063 def close(self) -> None:
1064 self.reset()
1066 def _resubscribe(self, subscribed, subscribe_fn) -> None:
1067 # Replay handler-backed subscriptions as positional Subscription objects
1068 # so binary names never need to be decoded into keyword argument keys.
1069 subscriptions = pubsub_subscription_args(subscribed)
1070 if subscriptions:
1071 subscribe_fn(*subscriptions)
1073 def _resubscribe_shard_channels(self) -> None:
1074 self._resubscribe(self.shard_channels, self.ssubscribe)
1076 def on_connect(self, connection) -> None:
1077 "Re-subscribe to any channels and patterns previously subscribed to"
1078 self.pending_unsubscribe_channels.clear()
1079 self.pending_unsubscribe_patterns.clear()
1080 self.pending_unsubscribe_shard_channels.clear()
1081 if self.channels:
1082 self._resubscribe(self.channels, self.subscribe)
1083 if self.patterns:
1084 self._resubscribe(self.patterns, self.psubscribe)
1085 if self.shard_channels:
1086 self._resubscribe_shard_channels()
1088 @property
1089 def subscribed(self) -> bool:
1090 """Indicates if there are subscriptions to any channels or patterns"""
1091 return self.subscribed_event.is_set()
1093 def execute_command(self, *args):
1094 """Execute a publish/subscribe command"""
1096 # NOTE: don't parse the response in this function -- it could pull a
1097 # legitimate message off the stack if the connection is already
1098 # subscribed to one or more channels
1100 if self.connection is None:
1101 self.connection = self.connection_pool.get_connection()
1102 # register a callback that re-subscribes to any channels we
1103 # were listening to when we were disconnected
1104 self.connection.register_connect_callback(self.on_connect)
1105 if self.push_handler_func is not None:
1106 self.connection._parser.set_pubsub_push_handler(self.push_handler_func)
1107 self._event_dispatcher.dispatch(
1108 AfterPubSubConnectionInstantiationEvent(
1109 self.connection, self.connection_pool, ClientType.SYNC, self._lock
1110 )
1111 )
1112 connection = self.connection
1113 kwargs = {"check_health": not self.subscribed}
1114 if not self.subscribed:
1115 self.clean_health_check_responses()
1116 with self._lock:
1117 self._execute(connection, connection.send_command, *args, **kwargs)
1119 def clean_health_check_responses(self) -> None:
1120 """
1121 If any health check responses are present, clean them
1122 """
1123 ttl = 10
1124 conn = self.connection
1125 while conn and self.health_check_response_counter > 0 and ttl > 0:
1126 if self._execute(conn, conn.can_read, timeout=conn.socket_timeout):
1127 response = self._execute(conn, conn.read_response)
1128 if self.is_health_check_response(response):
1129 self.health_check_response_counter -= 1
1130 else:
1131 raise PubSubError(
1132 "A non health check response was cleaned by "
1133 "execute_command: {}".format(response)
1134 )
1135 ttl -= 1
1137 def _reconnect(
1138 self,
1139 conn,
1140 error: Optional[Exception] = None,
1141 failure_count: Optional[int] = None,
1142 start_time: Optional[float] = None,
1143 command_name: Optional[str] = None,
1144 ) -> None:
1145 """
1146 The supported exceptions are already checked in the
1147 retry object so we don't need to do it here.
1149 In this error handler we are trying to reconnect to the server.
1150 """
1151 if error and failure_count <= conn.retry.get_retries():
1152 if command_name:
1153 record_operation_duration(
1154 command_name=command_name,
1155 duration_seconds=time.monotonic() - start_time,
1156 server_address=getattr(conn, "host", None),
1157 server_port=getattr(conn, "port", None),
1158 db_namespace=str(conn.db),
1159 error=error,
1160 retry_attempts=failure_count,
1161 )
1162 conn.disconnect()
1163 conn.connect()
1165 def _execute(self, conn, command, *args, **kwargs):
1166 """
1167 Connect manually upon disconnection. If the Redis server is down,
1168 this will fail and raise a ConnectionError as desired.
1169 After reconnection, the ``on_connect`` callback should have been
1170 called by the # connection to resubscribe us to any channels and
1171 patterns we were previously listening to
1172 """
1174 if conn.should_reconnect():
1175 self._reconnect(conn)
1177 if not len(args) == 0:
1178 command_name = args[0]
1179 else:
1180 command_name = None
1182 # Start timing for observability
1183 start_time = time.monotonic()
1184 # Track actual retry attempts for error reporting
1185 actual_retry_attempts = [0]
1187 def failure_callback(error, failure_count):
1188 actual_retry_attempts[0] = failure_count
1189 self._reconnect(conn, error, failure_count, start_time, command_name)
1191 try:
1192 response = conn.retry.call_with_retry(
1193 lambda: command(*args, **kwargs),
1194 failure_callback,
1195 with_failure_count=True,
1196 )
1198 if command_name:
1199 record_operation_duration(
1200 command_name=command_name,
1201 duration_seconds=time.monotonic() - start_time,
1202 server_address=getattr(conn, "host", None),
1203 server_port=getattr(conn, "port", None),
1204 db_namespace=str(conn.db),
1205 )
1207 return response
1208 except Exception as e:
1209 record_error_count(
1210 server_address=getattr(conn, "host", None),
1211 server_port=getattr(conn, "port", None),
1212 network_peer_address=getattr(conn, "host", None),
1213 network_peer_port=getattr(conn, "port", None),
1214 error_type=e,
1215 retry_attempts=actual_retry_attempts[0],
1216 is_internal=False,
1217 )
1218 raise
1220 def parse_response(self, block=True, timeout=0):
1221 """
1222 Parse the response from a publish/subscribe command.
1224 Args:
1225 block: If True, block indefinitely until a message is available.
1226 If False, return immediately if no message is available.
1227 Default: True
1228 timeout: The timeout in seconds for reading a response when block=False.
1229 This parameter is ignored when block=True.
1230 Default: 0 (return immediately if no data available)
1232 Returns:
1233 The parsed response from the server, or None if no message is available
1234 within the timeout period (when block=False).
1236 Important:
1237 The block and timeout parameters work together:
1238 - When block=True: timeout is IGNORED, method blocks indefinitely
1239 - When block=False: timeout is USED, method returns after timeout expires
1241 Typically, you should use get_message(timeout=X) instead of calling
1242 parse_response() directly. The get_message() method automatically sets
1243 block=False when a timeout is provided, and block=True when timeout=None.
1245 Example:
1246 # Block indefinitely (timeout is ignored)
1247 response = pubsub.parse_response(block=True, timeout=0.1)
1249 # Non-blocking with 0.1 second timeout
1250 response = pubsub.parse_response(block=False, timeout=0.1)
1252 # Non-blocking, return immediately
1253 response = pubsub.parse_response(block=False, timeout=0)
1255 # Recommended: use get_message() instead
1256 msg = pubsub.get_message(timeout=0.1) # automatically sets block=False
1257 msg = pubsub.get_message(timeout=None) # automatically sets block=True
1258 """
1259 conn = self.connection
1260 if conn is None:
1261 raise RuntimeError(
1262 "pubsub connection not set: "
1263 "did you forget to call subscribe() or psubscribe()?"
1264 )
1266 self.check_health()
1268 def try_read():
1269 if not block:
1270 if not conn.can_read(timeout=timeout):
1271 return None
1272 read_timeout = timeout
1273 else:
1274 conn.connect()
1275 # Block indefinitely waiting for a pubsub message. timeout=None
1276 # makes the socket layer call sock.settimeout(None) for this read
1277 # (and restore the original socket_timeout afterwards), so the
1278 # configured socket_timeout does not abort the read.
1279 read_timeout = None
1280 return conn.read_response(
1281 disconnect_on_error=False, push_request=True, timeout=read_timeout
1282 )
1284 response = self._execute(conn, try_read)
1286 if self.is_health_check_response(response):
1287 # ignore the health check message as user might not expect it
1288 self.health_check_response_counter -= 1
1289 return None
1290 return response
1292 def is_health_check_response(self, response) -> bool:
1293 """
1294 Check if the response is a health check response.
1295 If there are no subscriptions redis responds to PING command with a
1296 bulk response, instead of a multi-bulk with "pong" and the response.
1297 """
1298 if self.encoder.decode_responses:
1299 return (
1300 response
1301 in [
1302 self.health_check_response, # If there is a subscription
1303 self.HEALTH_CHECK_MESSAGE, # If there are no subscriptions and decode_responses=True
1304 ]
1305 )
1306 else:
1307 return (
1308 response
1309 in [
1310 self.health_check_response, # If there is a subscription
1311 self.health_check_response_b, # If there isn't a subscription and decode_responses=False
1312 ]
1313 )
1315 def check_health(self) -> None:
1316 conn = self.connection
1317 if conn is None:
1318 raise RuntimeError(
1319 "pubsub connection not set: "
1320 "did you forget to call subscribe() or psubscribe()?"
1321 )
1323 if conn.health_check_interval and time.monotonic() > conn.next_health_check:
1324 conn.send_command("PING", self.HEALTH_CHECK_MESSAGE, check_health=False)
1325 self.health_check_response_counter += 1
1327 def _normalize_keys(self, data) -> Dict:
1328 """
1329 normalize channel/pattern names to be either bytes or strings
1330 based on whether responses are automatically decoded. this saves us
1331 from coercing the value for each message coming in.
1332 """
1333 encode = self.encoder.encode
1334 decode = self.encoder.decode
1335 return {decode(encode(k)): v for k, v in data.items()}
1337 def psubscribe(
1338 self, *args: ChannelT | Subscription, **kwargs: PubSubHandler
1339 ) -> None:
1340 """
1341 Subscribe to channel patterns.
1342 Patterns supplied as keyword arguments expect a pattern name as the
1343 key and a callable as the value.
1344 ``Subscription`` objects can also be supplied positionally with an
1345 optional handler.
1346 A pattern's callable will be invoked automatically
1347 when a message is received on that pattern rather than producing a
1348 message via ``listen()``.
1349 """
1350 new_patterns = parse_pubsub_subscriptions(args, kwargs)
1351 ret_val = self.execute_command("PSUBSCRIBE", *new_patterns.keys())
1352 # update the patterns dict AFTER we send the command. we don't want to
1353 # subscribe twice to these patterns, once for the command and again
1354 # for the reconnection.
1355 new_patterns = self._normalize_keys(new_patterns)
1356 self.patterns.update(new_patterns)
1357 if not self.subscribed:
1358 # Set the subscribed_event flag to True
1359 self.subscribed_event.set()
1360 # Clear the health check counter
1361 self.health_check_response_counter = 0
1362 self.pending_unsubscribe_patterns.difference_update(new_patterns)
1363 return ret_val
1365 def punsubscribe(self, *args):
1366 """
1367 Unsubscribe from the supplied patterns. If empty, unsubscribe from
1368 all patterns.
1369 """
1370 if args:
1371 args = list_or_args(args[0], args[1:])
1372 patterns = self._normalize_keys(dict.fromkeys(args))
1373 else:
1374 patterns = self.patterns
1375 self.pending_unsubscribe_patterns.update(patterns)
1376 return self.execute_command("PUNSUBSCRIBE", *args)
1378 def subscribe(
1379 self, *args: ChannelT | Subscription, **kwargs: PubSubHandler
1380 ) -> None:
1381 """
1382 Subscribe to channels.
1383 Channels supplied as keyword arguments expect
1384 a channel name as the key and a callable as the value.
1385 ``Subscription`` objects can also be supplied positionally with an
1386 optional handler.
1387 A channel's callable will be invoked automatically
1388 when a message is received on that channel rather than producing a
1389 message via ``listen()`` or ``get_message()``.
1390 """
1391 new_channels = parse_pubsub_subscriptions(args, kwargs)
1392 ret_val = self.execute_command("SUBSCRIBE", *new_channels.keys())
1393 # update the channels dict AFTER we send the command. we don't want to
1394 # subscribe twice to these channels, once for the command and again
1395 # for the reconnection.
1396 new_channels = self._normalize_keys(new_channels)
1397 self.channels.update(new_channels)
1398 if not self.subscribed:
1399 # Set the subscribed_event flag to True
1400 self.subscribed_event.set()
1401 # Clear the health check counter
1402 self.health_check_response_counter = 0
1403 self.pending_unsubscribe_channels.difference_update(new_channels)
1404 return ret_val
1406 def unsubscribe(self, *args):
1407 """
1408 Unsubscribe from the supplied channels. If empty, unsubscribe from
1409 all channels
1410 """
1411 if args:
1412 args = list_or_args(args[0], args[1:])
1413 channels = self._normalize_keys(dict.fromkeys(args))
1414 else:
1415 channels = self.channels
1416 self.pending_unsubscribe_channels.update(channels)
1417 return self.execute_command("UNSUBSCRIBE", *args)
1419 def ssubscribe(
1420 self,
1421 *args: ChannelT | Subscription,
1422 target_node: Any = None,
1423 **kwargs: PubSubHandler,
1424 ) -> None:
1425 """
1426 Subscribes the client to the specified shard channels.
1427 Channels supplied as keyword arguments expect a channel name as the key
1428 and a callable as the value.
1429 ``Subscription`` objects can also be supplied positionally
1430 with an optional handler.
1431 A channel's callable will be invoked automatically when a message
1432 is received on that channel rather than producing a message
1433 via ``listen()`` or ``get_sharded_message()``.
1434 """
1435 new_s_channels = parse_pubsub_subscriptions(args, kwargs)
1436 ret_val = self.execute_command("SSUBSCRIBE", *new_s_channels.keys())
1437 # update the s_channels dict AFTER we send the command. we don't want to
1438 # subscribe twice to these channels, once for the command and again
1439 # for the reconnection.
1440 new_s_channels = self._normalize_keys(new_s_channels)
1441 self.shard_channels.update(new_s_channels)
1442 if not self.subscribed:
1443 # Set the subscribed_event flag to True
1444 self.subscribed_event.set()
1445 # Clear the health check counter
1446 self.health_check_response_counter = 0
1447 self.pending_unsubscribe_shard_channels.difference_update(new_s_channels)
1448 return ret_val
1450 def sunsubscribe(self, *args, target_node=None):
1451 """
1452 Unsubscribe from the supplied shard_channels. If empty, unsubscribe from
1453 all shard_channels
1454 """
1455 if args:
1456 args = list_or_args(args[0], args[1:])
1457 s_channels = self._normalize_keys(dict.fromkeys(args))
1458 else:
1459 s_channels = self.shard_channels
1460 self.pending_unsubscribe_shard_channels.update(s_channels)
1461 return self.execute_command("SUNSUBSCRIBE", *args)
1463 def listen(self):
1464 "Listen for messages on channels this client has been subscribed to"
1465 while self.subscribed:
1466 response = self.handle_message(self.parse_response(block=True))
1467 if response is not None:
1468 yield response
1470 def get_message(
1471 self, ignore_subscribe_messages: bool = False, timeout: float = 0.0
1472 ):
1473 """
1474 Get the next message if one is available, otherwise None.
1476 If timeout is specified, the system will wait for `timeout` seconds
1477 before returning. Timeout should be specified as a floating point
1478 number, or None, to wait indefinitely.
1479 """
1480 if not self.subscribed:
1481 # Wait for subscription
1482 start_time = time.monotonic()
1483 if self.subscribed_event.wait(timeout) is True:
1484 # The connection was subscribed during the timeout time frame.
1485 # The timeout should be adjusted based on the time spent
1486 # waiting for the subscription
1487 time_spent = time.monotonic() - start_time
1488 timeout = max(0.0, timeout - time_spent)
1489 else:
1490 # The connection isn't subscribed to any channels or patterns,
1491 # so no messages are available
1492 return None
1494 response = self.parse_response(block=(timeout is None), timeout=timeout)
1496 if response:
1497 return self.handle_message(response, ignore_subscribe_messages)
1498 return None
1500 get_sharded_message = get_message
1502 def ping(self, message: Union[str, None] = None) -> bool:
1503 """
1504 Ping the Redis server to test connectivity.
1506 Sends a PING command to the Redis server and returns True if the server
1507 responds with "PONG".
1508 """
1509 args = ["PING", message] if message is not None else ["PING"]
1510 return self.execute_command(*args)
1512 def handle_message(self, response, ignore_subscribe_messages=False):
1513 """
1514 Parses a pub/sub message. If the channel or pattern was subscribed to
1515 with a message handler, the handler is invoked instead of a parsed
1516 message being returned.
1517 """
1518 if response is None:
1519 return None
1520 if isinstance(response, bytes):
1521 response = [b"pong", response] if response != b"PONG" else [b"pong", b""]
1523 message_type = str_if_bytes(response[0])
1524 if message_type == "pmessage":
1525 message = {
1526 "type": message_type,
1527 "pattern": response[1],
1528 "channel": response[2],
1529 "data": response[3],
1530 }
1531 elif message_type == "pong":
1532 message = {
1533 "type": message_type,
1534 "pattern": None,
1535 "channel": None,
1536 "data": response[1],
1537 }
1538 else:
1539 message = {
1540 "type": message_type,
1541 "pattern": None,
1542 "channel": response[1],
1543 "data": response[2],
1544 }
1546 if message_type in ["message", "pmessage"]:
1547 channel = str_if_bytes(message["channel"])
1548 record_pubsub_message(
1549 direction=PubSubDirection.RECEIVE,
1550 channel=channel,
1551 )
1552 elif message_type == "smessage":
1553 channel = str_if_bytes(message["channel"])
1554 record_pubsub_message(
1555 direction=PubSubDirection.RECEIVE,
1556 channel=channel,
1557 sharded=True,
1558 )
1560 # if this is an unsubscribe message, remove it from memory
1561 if message_type in self.UNSUBSCRIBE_MESSAGE_TYPES:
1562 if message_type == "punsubscribe":
1563 pattern = response[1]
1564 if pattern in self.pending_unsubscribe_patterns:
1565 self.pending_unsubscribe_patterns.remove(pattern)
1566 self.patterns.pop(pattern, None)
1567 elif message_type == "sunsubscribe":
1568 s_channel = response[1]
1569 if s_channel in self.pending_unsubscribe_shard_channels:
1570 self.pending_unsubscribe_shard_channels.remove(s_channel)
1571 self.shard_channels.pop(s_channel, None)
1572 else:
1573 channel = response[1]
1574 if channel in self.pending_unsubscribe_channels:
1575 self.pending_unsubscribe_channels.remove(channel)
1576 self.channels.pop(channel, None)
1577 if not self.channels and not self.patterns and not self.shard_channels:
1578 # There are no subscriptions anymore, set subscribed_event flag
1579 # to false
1580 self.subscribed_event.clear()
1582 if message_type in self.PUBLISH_MESSAGE_TYPES:
1583 # if there's a message handler, invoke it
1584 if message_type == "pmessage":
1585 handler = self.patterns.get(message["pattern"], None)
1586 elif message_type == "smessage":
1587 handler = self.shard_channels.get(message["channel"], None)
1588 else:
1589 handler = self.channels.get(message["channel"], None)
1590 if handler:
1591 handler(message)
1592 return None
1593 elif message_type != "pong":
1594 # this is a subscribe/unsubscribe message. ignore if we don't
1595 # want them
1596 if ignore_subscribe_messages or self.ignore_subscribe_messages:
1597 return None
1599 return message
1601 def run_in_thread(
1602 self,
1603 sleep_time: float = 0.0,
1604 daemon: bool = False,
1605 exception_handler: Optional[Callable] = None,
1606 pubsub=None,
1607 sharded_pubsub: bool = False,
1608 ) -> "PubSubWorkerThread":
1609 for channel, handler in self.channels.items():
1610 if handler is None:
1611 raise PubSubError(f"Channel: '{channel}' has no handler registered")
1612 for pattern, handler in self.patterns.items():
1613 if handler is None:
1614 raise PubSubError(f"Pattern: '{pattern}' has no handler registered")
1615 for s_channel, handler in self.shard_channels.items():
1616 if handler is None:
1617 raise PubSubError(
1618 f"Shard Channel: '{s_channel}' has no handler registered"
1619 )
1621 pubsub = self if pubsub is None else pubsub
1622 thread = PubSubWorkerThread(
1623 pubsub,
1624 sleep_time,
1625 daemon=daemon,
1626 exception_handler=exception_handler,
1627 sharded_pubsub=sharded_pubsub,
1628 )
1629 thread.start()
1630 return thread
1633class PubSubWorkerThread(threading.Thread):
1634 def __init__(
1635 self,
1636 pubsub,
1637 sleep_time: float,
1638 daemon: bool = False,
1639 exception_handler: Union[
1640 Callable[[Exception, "PubSub", "PubSubWorkerThread"], None], None
1641 ] = None,
1642 sharded_pubsub: bool = False,
1643 ):
1644 super().__init__()
1645 self.daemon = daemon
1646 self.pubsub = pubsub
1647 self.sleep_time = sleep_time
1648 self.exception_handler = exception_handler
1649 self.sharded_pubsub = sharded_pubsub
1650 self._running = threading.Event()
1652 def run(self) -> None:
1653 if self._running.is_set():
1654 return
1655 self._running.set()
1656 pubsub = self.pubsub
1657 sleep_time = self.sleep_time
1658 while self._running.is_set():
1659 try:
1660 if not self.sharded_pubsub:
1661 pubsub.get_message(
1662 ignore_subscribe_messages=True, timeout=sleep_time
1663 )
1664 else:
1665 pubsub.get_sharded_message(
1666 ignore_subscribe_messages=True, timeout=sleep_time
1667 )
1668 except BaseException as e:
1669 if self.exception_handler is None:
1670 raise
1671 self.exception_handler(e, pubsub, self)
1672 pubsub.close()
1674 def stop(self) -> None:
1675 # trip the flag so the run loop exits. the run loop will
1676 # close the pubsub connection, which disconnects the socket
1677 # and returns the connection to the pool.
1678 self._running.clear()
1681class Pipeline(Redis):
1682 """
1683 Pipelines provide a way to transmit multiple commands to the Redis server
1684 in one transmission. This is convenient for batch processing, such as
1685 saving all the values in a list to Redis.
1687 All commands executed within a pipeline(when running in transactional mode,
1688 which is the default behavior) are wrapped with MULTI and EXEC
1689 calls. This guarantees all commands executed in the pipeline will be
1690 executed atomically.
1692 Any command raising an exception does *not* halt the execution of
1693 subsequent commands in the pipeline. Instead, the exception is caught
1694 and its instance is placed into the response list returned by execute().
1695 Code iterating over the response list should be able to deal with an
1696 instance of an exception as a potential value. In general, these will be
1697 ResponseError exceptions, such as those raised when issuing a command
1698 on a key of a different datatype.
1699 """
1701 UNWATCH_COMMANDS = {"DISCARD", "EXEC", "UNWATCH"}
1703 def __init__(
1704 self,
1705 connection_pool: ConnectionPool,
1706 response_callbacks,
1707 transaction,
1708 shard_hint,
1709 ):
1710 self.connection_pool = connection_pool
1711 self.connection: Optional[Connection] = None
1712 self.response_callbacks = response_callbacks
1713 self.transaction = transaction
1714 self.shard_hint = shard_hint
1715 self.watching = False
1716 self.command_stack = []
1717 self.scripts: Set[Script] = set()
1718 self.explicit_transaction = False
1720 def __enter__(self) -> "Pipeline":
1721 return self
1723 def __exit__(self, exc_type, exc_value, traceback):
1724 self.reset()
1726 def __del__(self):
1727 try:
1728 self.reset()
1729 except Exception:
1730 pass
1732 def __len__(self) -> int:
1733 return len(self.command_stack)
1735 def __bool__(self) -> bool:
1736 """Pipeline instances should always evaluate to True"""
1737 return True
1739 def reset(self) -> None:
1740 self.command_stack = []
1741 self.scripts = set()
1742 # make sure to reset the connection state in the event that we were
1743 # watching something
1744 if self.watching and self.connection:
1745 try:
1746 # call this manually since our unwatch or
1747 # immediate_execute_command methods can call reset()
1748 self.connection.send_command("UNWATCH")
1749 self.connection.read_response()
1750 except ConnectionError:
1751 # disconnect will also remove any previous WATCHes
1752 self.connection.disconnect()
1753 # clean up the other instance attributes
1754 self.watching = False
1755 self.explicit_transaction = False
1757 # we can safely return the connection to the pool here since we're
1758 # sure we're no longer WATCHing anything
1759 if self.connection:
1760 self.connection_pool.release(self.connection)
1761 self.connection = None
1763 def close(self) -> None:
1764 """Close the pipeline"""
1765 self.reset()
1767 def multi(self) -> None:
1768 """
1769 Start a transactional block of the pipeline after WATCH commands
1770 are issued. End the transactional block with `execute`.
1771 """
1772 if self.explicit_transaction:
1773 raise RedisError("Cannot issue nested calls to MULTI")
1774 if self.command_stack:
1775 raise RedisError(
1776 "Commands without an initial WATCH have already been issued"
1777 )
1778 self.explicit_transaction = True
1780 def execute_command(self, *args, **kwargs):
1781 if (self.watching or args[0] == "WATCH") and not self.explicit_transaction:
1782 return self.immediate_execute_command(*args, **kwargs)
1783 return self.pipeline_execute_command(*args, **kwargs)
1785 def _disconnect_reset_raise_on_watching(
1786 self,
1787 conn: AbstractConnection,
1788 error: Exception,
1789 failure_count: Optional[int] = None,
1790 start_time: Optional[float] = None,
1791 command_name: Optional[str] = None,
1792 ) -> None:
1793 """
1794 Close the connection reset watching state and
1795 raise an exception if we were watching.
1797 The supported exceptions are already checked in the
1798 retry object so we don't need to do it here.
1800 After we disconnect the connection, it will try to reconnect and
1801 do a health check as part of the send_command logic(on connection level).
1802 """
1803 if error and failure_count <= conn.retry.get_retries():
1804 record_operation_duration(
1805 command_name=command_name,
1806 duration_seconds=time.monotonic() - start_time,
1807 server_address=getattr(conn, "host", None),
1808 server_port=getattr(conn, "port", None),
1809 db_namespace=str(conn.db),
1810 error=error,
1811 retry_attempts=failure_count,
1812 )
1813 conn.disconnect()
1815 # if we were already watching a variable, the watch is no longer
1816 # valid since this connection has died. raise a WatchError, which
1817 # indicates the user should retry this transaction.
1818 if self.watching:
1819 self.reset()
1820 raise WatchError(
1821 f"A {type(error).__name__} occurred while watching one or more keys"
1822 )
1824 def immediate_execute_command(self, *args, **options):
1825 """
1826 Execute a command immediately, but don't auto-retry on the supported
1827 errors for retry if we're already WATCHing a variable.
1828 Used when issuing WATCH or subsequent commands retrieving their values but before
1829 MULTI is called.
1830 """
1831 command_name = args[0]
1832 conn = self.connection
1833 # if this is the first call, we need a connection
1834 if not conn:
1835 conn = self.connection_pool.get_connection()
1836 self.connection = conn
1838 # Start timing for observability
1839 start_time = time.monotonic()
1840 # Track actual retry attempts for error reporting
1841 actual_retry_attempts = [0]
1843 def failure_callback(error, failure_count):
1844 if is_debug_log_enabled():
1845 add_debug_log_for_operation_failure(conn)
1846 actual_retry_attempts[0] = failure_count
1847 self._disconnect_reset_raise_on_watching(
1848 conn, error, failure_count, start_time, command_name
1849 )
1851 try:
1852 response = conn.retry.call_with_retry(
1853 lambda: self._send_command_parse_response(
1854 conn, command_name, *args, **options
1855 ),
1856 failure_callback,
1857 with_failure_count=True,
1858 )
1860 record_operation_duration(
1861 command_name=command_name,
1862 duration_seconds=time.monotonic() - start_time,
1863 server_address=getattr(conn, "host", None),
1864 server_port=getattr(conn, "port", None),
1865 db_namespace=str(conn.db),
1866 )
1868 return response
1869 except Exception as e:
1870 record_error_count(
1871 server_address=getattr(conn, "host", None),
1872 server_port=getattr(conn, "port", None),
1873 network_peer_address=getattr(conn, "host", None),
1874 network_peer_port=getattr(conn, "port", None),
1875 error_type=e,
1876 retry_attempts=actual_retry_attempts[0],
1877 is_internal=False,
1878 )
1879 raise
1881 def pipeline_execute_command(self, *args, **options) -> "Pipeline":
1882 """
1883 Stage a command to be executed when execute() is next called
1885 Returns the current Pipeline object back so commands can be
1886 chained together, such as:
1888 pipe = pipe.set('foo', 'bar').incr('baz').decr('bang')
1890 At some other point, you can then run: pipe.execute(),
1891 which will execute all commands queued in the pipe.
1892 """
1893 self.command_stack.append((args, options))
1894 return self
1896 def _execute_transaction(
1897 self, connection: Connection, commands, raise_on_error
1898 ) -> List:
1899 cmds = chain([(("MULTI",), {})], commands, [(("EXEC",), {})])
1900 all_cmds = connection.pack_commands(
1901 [args for args, options in cmds if EMPTY_RESPONSE not in options]
1902 )
1903 connection.send_packed_command(all_cmds)
1904 errors = []
1906 # parse off the response for MULTI
1907 # NOTE: we need to handle ResponseErrors here and continue
1908 # so that we read all the additional command messages from
1909 # the socket
1910 try:
1911 self.parse_response(connection, "_")
1912 except ResponseError as e:
1913 errors.append((0, e))
1915 # and all the other commands
1916 for i, command in enumerate(commands):
1917 if EMPTY_RESPONSE in command[1]:
1918 errors.append((i, command[1][EMPTY_RESPONSE]))
1919 else:
1920 try:
1921 self.parse_response(connection, "_")
1922 except ResponseError as e:
1923 self.annotate_exception(e, i + 1, command[0])
1924 errors.append((i, e))
1926 # parse the EXEC.
1927 try:
1928 response = self.parse_response(connection, "_")
1929 except ExecAbortError:
1930 if errors:
1931 raise errors[0][1]
1932 raise
1934 # EXEC clears any watched keys
1935 self.watching = False
1937 if response is None:
1938 raise WatchError("Watched variable changed.")
1940 # put any parse errors into the response
1941 for i, e in errors:
1942 response.insert(i, e)
1944 if len(response) != len(commands):
1945 self.connection.disconnect()
1946 raise ResponseError(
1947 "Wrong number of response items from pipeline execution"
1948 )
1950 # find any errors in the response and raise if necessary
1951 if raise_on_error:
1952 self.raise_first_error(commands, response)
1954 # We have to run response callbacks manually
1955 data = []
1956 for r, cmd in zip(response, commands):
1957 if not isinstance(r, Exception):
1958 args, options = cmd
1959 # Remove keys entry, it needs only for cache.
1960 options.pop("keys", None)
1961 command_name = args[0]
1962 if command_name in self.response_callbacks:
1963 r = self.response_callbacks[command_name](r, **options)
1964 data.append(r)
1966 return data
1968 def _execute_pipeline(self, connection, commands, raise_on_error):
1969 # build up all commands into a single request to increase network perf
1970 all_cmds = connection.pack_commands([args for args, _ in commands])
1971 connection.send_packed_command(all_cmds)
1973 responses = []
1974 for args, options in commands:
1975 try:
1976 responses.append(self.parse_response(connection, args[0], **options))
1977 except ResponseError as e:
1978 responses.append(e)
1980 if raise_on_error:
1981 self.raise_first_error(commands, responses)
1983 return responses
1985 def raise_first_error(self, commands, response):
1986 for i, r in enumerate(response):
1987 if isinstance(r, ResponseError):
1988 self.annotate_exception(r, i + 1, commands[i][0])
1989 raise r
1991 def annotate_exception(self, exception, number, command):
1992 cmd = " ".join(map(safe_str, command))
1993 msg = (
1994 f"Command # {number} ({truncate_text(cmd)}) of pipeline "
1995 f"caused error: {exception.args[0]}"
1996 )
1997 exception.args = (msg,) + exception.args[1:]
1999 def parse_response(self, connection, command_name, **options):
2000 result = Redis.parse_response(self, connection, command_name, **options)
2001 if command_name in self.UNWATCH_COMMANDS:
2002 self.watching = False
2003 elif command_name == "WATCH":
2004 self.watching = True
2005 return result
2007 def load_scripts(self):
2008 # make sure all scripts that are about to be run on this pipeline exist
2009 scripts = list(self.scripts)
2010 immediate = self.immediate_execute_command
2011 shas = [s.sha for s in scripts]
2012 # we can't use the normal script_* methods because they would just
2013 # get buffered in the pipeline.
2014 exists = immediate("SCRIPT EXISTS", *shas)
2015 if not all(exists):
2016 for s, exist in zip(scripts, exists):
2017 if not exist:
2018 s.sha = immediate("SCRIPT LOAD", s.script)
2020 def _disconnect_raise_on_watching(
2021 self,
2022 conn: AbstractConnection,
2023 error: Exception,
2024 failure_count: Optional[int] = None,
2025 start_time: Optional[float] = None,
2026 command_name: Optional[str] = None,
2027 ) -> None:
2028 """
2029 Close the connection, raise an exception if we were watching.
2031 The supported exceptions are already checked in the
2032 retry object so we don't need to do it here.
2034 After we disconnect the connection, it will try to reconnect and
2035 do a health check as part of the send_command logic(on connection level).
2036 """
2037 if error and failure_count <= conn.retry.get_retries():
2038 record_operation_duration(
2039 command_name=command_name,
2040 duration_seconds=time.monotonic() - start_time,
2041 server_address=getattr(conn, "host", None),
2042 server_port=getattr(conn, "port", None),
2043 db_namespace=str(conn.db),
2044 error=error,
2045 retry_attempts=failure_count,
2046 )
2047 conn.disconnect()
2048 # if we were watching a variable, the watch is no longer valid
2049 # since this connection has died. raise a WatchError, which
2050 # indicates the user should retry this transaction.
2051 if self.watching:
2052 raise WatchError(
2053 f"A {type(error).__name__} occurred while watching one or more keys"
2054 )
2056 def execute(self, raise_on_error: bool = True) -> List[Any]:
2057 """Execute all the commands in the current pipeline"""
2058 stack = self.command_stack
2059 if not stack and not self.watching:
2060 return []
2061 if self.scripts:
2062 self.load_scripts()
2063 if self.transaction or self.explicit_transaction:
2064 execute = self._execute_transaction
2065 operation_name = "MULTI"
2066 else:
2067 execute = self._execute_pipeline
2068 operation_name = "PIPELINE"
2070 conn = self.connection
2071 if not conn:
2072 conn = self.connection_pool.get_connection()
2073 # assign to self.connection so reset() releases the connection
2074 # back to the pool after we're done
2075 self.connection = conn
2077 # Start timing for observability
2078 start_time = time.monotonic()
2079 # Track actual retry attempts for error reporting
2080 actual_retry_attempts = [0]
2082 def failure_callback(error, failure_count):
2083 if is_debug_log_enabled():
2084 add_debug_log_for_operation_failure(conn)
2085 actual_retry_attempts[0] = failure_count
2086 self._disconnect_raise_on_watching(
2087 conn, error, failure_count, start_time, operation_name
2088 )
2090 try:
2091 response = conn.retry.call_with_retry(
2092 lambda: execute(conn, stack, raise_on_error),
2093 failure_callback,
2094 with_failure_count=True,
2095 )
2097 record_operation_duration(
2098 command_name=operation_name,
2099 duration_seconds=time.monotonic() - start_time,
2100 server_address=getattr(conn, "host", None),
2101 server_port=getattr(conn, "port", None),
2102 db_namespace=str(conn.db),
2103 )
2104 return response
2105 except Exception as e:
2106 record_error_count(
2107 server_address=getattr(conn, "host", None),
2108 server_port=getattr(conn, "port", None),
2109 network_peer_address=getattr(conn, "host", None),
2110 network_peer_port=getattr(conn, "port", None),
2111 error_type=e,
2112 retry_attempts=actual_retry_attempts[0],
2113 is_internal=False,
2114 )
2115 raise
2117 finally:
2118 # in reset() the connection is disconnected before returned to the pool if
2119 # it is marked for reconnect.
2120 self.reset()
2122 def discard(self):
2123 """
2124 Flushes all previously queued commands
2125 See: https://redis.io/commands/DISCARD
2126 """
2127 self.execute_command("DISCARD")
2129 def watch(self, *names):
2130 """Watches the values at keys ``names``"""
2131 if self.explicit_transaction:
2132 raise RedisError("Cannot issue a WATCH after a MULTI")
2133 return self.execute_command("WATCH", *names)
2135 def unwatch(self) -> bool:
2136 """Unwatches all previously specified keys"""
2137 return self.watching and self.execute_command("UNWATCH") or True