Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/redis/connection.py: 21%
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 functools
3import logging
4import os
5import socket
6import sys
7import threading
8import time
9import weakref
10from abc import ABC, abstractmethod
11from itertools import chain
12from queue import Empty, Full, LifoQueue
13from typing import (
14 Any,
15 Callable,
16 Dict,
17 Iterable,
18 List,
19 Literal,
20 Optional,
21 Type,
22 TypeVar,
23 Union,
24)
25from urllib.parse import parse_qs, unquote, urlparse
27from redis.cache import (
28 CacheConfig,
29 CacheEntry,
30 CacheEntryStatus,
31 CacheFactory,
32 CacheFactoryInterface,
33 CacheInterface,
34 CacheKey,
35 CacheProxy,
36 TrackingMode,
37)
38from redis.commands.metadata import MetadataResolver
40from . import exceptions as redis_exceptions
41from ._defaults import (
42 DEFAULT_SOCKET_CONNECT_TIMEOUT,
43 DEFAULT_SOCKET_READ_SIZE,
44 DEFAULT_SOCKET_TIMEOUT,
45 get_default_socket_keepalive_options,
46)
47from ._parsers import (
48 UNRECOVERABLE_PARSE_ERRORS,
49 BaseParser,
50 Encoder,
51 _HiredisParser,
52 _RESP2Parser,
53 _RESP3Parser,
54)
55from .auth.token import TokenInterface
56from .backoff import NoBackoff
57from .credentials import CredentialProvider, UsernamePasswordCredentialProvider
58from .driver_info import DriverInfo, resolve_driver_info
59from .event import (
60 AfterConnectionReleasedEvent,
61 EventDispatcher,
62 MaintenanceCompletedEvent,
63)
64from .exceptions import (
65 AuthenticationError,
66 AuthenticationWrongNumberOfArgsError,
67 ChildDeadlockedError,
68 ConnectionError,
69 DataError,
70 MaxConnectionsError,
71 RedisError,
72 ResponseError,
73 TimeoutError,
74)
75from .himport import HImportRegistry
76from .maint_notifications import (
77 PENDING_PUSH_NOTIFICATIONS_READ_TIMEOUT,
78 MaintenanceState,
79 MaintNotificationsConfig,
80 MaintNotificationsConnectionHandler,
81 MaintNotificationsPoolHandler,
82 OSSMaintNotificationsHandler,
83 _dispatch_maintenance_event,
84)
85from .observability.attributes import (
86 DB_CLIENT_CONNECTION_POOL_NAME,
87 DB_CLIENT_CONNECTION_STATE,
88 AttributeBuilder,
89 ConnectionState,
90 CSCReason,
91 CSCResult,
92 get_pool_name,
93)
94from .observability.metrics import CloseReason
95from .observability.recorder import (
96 init_csc_items,
97 record_connection_closed,
98 record_connection_count,
99 record_connection_create_time,
100 record_connection_timeout,
101 record_connection_wait_time,
102 record_csc_eviction,
103 record_csc_network_saved,
104 record_csc_request,
105 record_error_count,
106 register_csc_items_callback,
107)
108from .retry import Retry
109from .utils import (
110 CRYPTOGRAPHY_AVAILABLE,
111 DEFAULT_RESP_VERSION,
112 HIREDIS_AVAILABLE,
113 SENTINEL,
114 SSL_AVAILABLE,
115 check_protocol_version,
116 compare_versions,
117 deprecated_args,
118 ensure_string,
119 format_error_message,
120 str_if_bytes,
121)
123if SSL_AVAILABLE:
124 import ssl
125 from ssl import VerifyFlags
126else:
127 ssl = None
128 VerifyFlags = None
130if HIREDIS_AVAILABLE:
131 import hiredis
133SYM_STAR = b"*"
134SYM_DOLLAR = b"$"
135SYM_CRLF = b"\r\n"
136SYM_EMPTY = b""
138DefaultParser: Type[Union[_RESP2Parser, _RESP3Parser, _HiredisParser]]
139if HIREDIS_AVAILABLE:
140 DefaultParser = _HiredisParser
141else:
142 DefaultParser = _RESP2Parser
144logger = logging.getLogger(__name__)
147def add_debug_log_for_connection_failure(
148 connection: "AbstractConnection",
149 error: BaseException,
150 operation: str,
151) -> None:
152 """
153 Render the connection's live state on a failure that is about to close it.
155 Must be called *before* ``disconnect()``. ``extract_connection_details()``
156 reads the resolved ip, local port and armed read timeout off the socket, so
157 once the socket is gone it can only report ``not connected`` - which hides
158 exactly the state needed to explain the failure. In particular it is what
159 tells apart a read that ran under the original timeout from one that ran
160 under a relaxed maintenance timeout.
161 """
162 if logger.isEnabledFor(logging.DEBUG):
163 logger.debug(
164 f"{type(error).__name__} while {operation}, "
165 f"with connection: {connection}, "
166 f"details: {connection.extract_connection_details()}, "
167 f"error: {error}",
168 )
171class HiredisRespSerializer:
172 def __init__(self, fallback=None, encode=None):
173 self._fallback = fallback
174 self._encode = encode
176 def pack(self, *args: List):
177 """Pack a series of arguments into the Redis protocol"""
178 if self._fallback is not None and any(
179 isinstance(arg, memoryview) for arg in args
180 ):
181 return self._fallback(*args)
182 output = []
184 if isinstance(args[0], str):
185 args = tuple(args[0].encode().split()) + args[1:]
186 elif b" " in args[0]:
187 args = tuple(args[0].split()) + args[1:]
188 args = tuple(
189 bytes(arg)
190 if isinstance(arg, (bytearray, memoryview))
191 else self._encode(arg)
192 if self._encode is not None
193 else arg
194 for arg in args
195 )
196 try:
197 output.append(hiredis.pack_command(args))
198 except TypeError:
199 _, value, traceback = sys.exc_info()
200 raise DataError(value).with_traceback(traceback)
202 return output
205class PythonRespSerializer:
206 def __init__(self, buffer_cutoff, encode) -> None:
207 self._buffer_cutoff = buffer_cutoff
208 self.encode = encode
210 def pack(self, *args):
211 """Pack a series of arguments into the Redis protocol"""
212 output = []
213 # the client might have included 1 or more literal arguments in
214 # the command name, e.g., 'CONFIG GET'. The Redis server expects these
215 # arguments to be sent separately, so split the first argument
216 # manually. These arguments should be bytestrings so that they are
217 # not encoded.
218 if isinstance(args[0], str):
219 args = tuple(args[0].encode().split()) + args[1:]
220 elif b" " in args[0]:
221 args = tuple(args[0].split()) + args[1:]
223 buff = SYM_EMPTY.join((SYM_STAR, str(len(args)).encode(), SYM_CRLF))
225 buffer_cutoff = self._buffer_cutoff
226 for arg in map(self.encode, args):
227 # to avoid large string mallocs, chunk the command into the
228 # output list if we're sending large values or memoryviews
229 arg_length = len(arg)
230 if (
231 len(buff) > buffer_cutoff
232 or arg_length > buffer_cutoff
233 or isinstance(arg, memoryview)
234 ):
235 buff = SYM_EMPTY.join(
236 (buff, SYM_DOLLAR, str(arg_length).encode(), SYM_CRLF)
237 )
238 output.append(buff)
239 output.append(arg)
240 buff = SYM_CRLF
241 else:
242 buff = SYM_EMPTY.join(
243 (
244 buff,
245 SYM_DOLLAR,
246 str(arg_length).encode(),
247 SYM_CRLF,
248 arg,
249 SYM_CRLF,
250 )
251 )
252 output.append(buff)
253 return output
256class ConnectionInterface:
257 @abstractmethod
258 def repr_pieces(self):
259 pass
261 @abstractmethod
262 def register_connect_callback(self, callback):
263 pass
265 @abstractmethod
266 def deregister_connect_callback(self, callback):
267 pass
269 @abstractmethod
270 def set_parser(self, parser_class):
271 pass
273 @abstractmethod
274 def get_protocol(self):
275 pass
277 @abstractmethod
278 def connect(self):
279 pass
281 @abstractmethod
282 def on_connect(self):
283 pass
285 @abstractmethod
286 def disconnect(self, *args, **kwargs):
287 pass
289 @abstractmethod
290 def check_health(self):
291 pass
293 @abstractmethod
294 def send_packed_command(self, command, check_health=True):
295 pass
297 @abstractmethod
298 def send_command(self, *args, **kwargs):
299 pass
301 @abstractmethod
302 def can_read(self, timeout: float = 0) -> bool:
303 # TODO: Rename this API; it detects pending data or dirty/closed
304 # connection state, not only whether application data can be read.
305 pass
307 @abstractmethod
308 def read_response(
309 self,
310 disable_decoding=False,
311 *,
312 timeout: Union[float, object] = SENTINEL,
313 disconnect_on_error=True,
314 push_request=False,
315 ):
316 pass
318 @abstractmethod
319 def pack_command(self, *args):
320 pass
322 @abstractmethod
323 def pack_commands(self, commands):
324 pass
326 @property
327 @abstractmethod
328 def handshake_metadata(self) -> Union[Dict[bytes, bytes], Dict[str, str]]:
329 pass
331 @abstractmethod
332 def set_re_auth_token(self, token: TokenInterface):
333 pass
335 @abstractmethod
336 def re_auth(self):
337 pass
339 @abstractmethod
340 def mark_for_reconnect(self):
341 """
342 Mark the connection to be reconnected on the next command.
343 This is useful when a connection is moved to a different node.
344 """
345 pass
347 @abstractmethod
348 def should_reconnect(self):
349 """
350 Returns True if the connection should be reconnected.
351 """
352 pass
354 @abstractmethod
355 def reset_should_reconnect(self):
356 """
357 Reset the internal flag to False.
358 """
359 pass
361 @abstractmethod
362 def extract_connection_details(self) -> str:
363 pass
365 @property
366 @abstractmethod
367 def is_connected(self) -> bool:
368 """
369 Return ``True`` if the connection to the server is active.
370 """
371 pass
374class MaintNotificationsAbstractConnection:
375 """
376 Abstract class for handling maintenance notifications logic.
377 This class is expected to be used as base class together with ConnectionInterface.
379 This class is intended to be used with multiple inheritance!
381 All logic related to maintenance notifications is encapsulated in this class.
382 """
384 def __init__(
385 self,
386 maint_notifications_config: Optional[MaintNotificationsConfig],
387 maint_notifications_pool_handler: Optional[
388 MaintNotificationsPoolHandler
389 ] = None,
390 maintenance_state: "MaintenanceState" = MaintenanceState.NONE,
391 maintenance_notification_hash: Optional[int] = None,
392 orig_host_address: Optional[str] = None,
393 orig_socket_timeout: Optional[float] = None,
394 orig_socket_connect_timeout: Optional[float] = None,
395 oss_cluster_maint_notifications_handler: Optional[
396 OSSMaintNotificationsHandler
397 ] = None,
398 parser: Optional[BaseParser] = None,
399 event_dispatcher: Optional[EventDispatcher] = None,
400 ):
401 """
402 Initialize the maintenance notifications for the connection.
404 Args:
405 maint_notifications_config (MaintNotificationsConfig): The configuration for maintenance notifications.
406 maint_notifications_pool_handler (Optional[MaintNotificationsPoolHandler]): The pool handler for maintenance notifications.
407 maintenance_state (MaintenanceState): The current maintenance state of the connection.
408 maintenance_notification_hash (Optional[int]): The current maintenance notification hash of the connection.
409 orig_host_address (Optional[str]): The original host address of the connection.
410 orig_socket_timeout (Optional[float]): The original socket timeout of the connection.
411 orig_socket_connect_timeout (Optional[float]): The original socket connect timeout of the connection.
412 oss_cluster_maint_notifications_handler (Optional[OSSMaintNotificationsHandler]): The OSS cluster handler for maintenance notifications.
413 parser (Optional[BaseParser]): The parser to use for maintenance notifications.
414 If not provided, the parser from the connection is used.
415 This is useful when the parser is created after this object.
416 """
417 self.maint_notifications_config = maint_notifications_config
418 self.maintenance_state = maintenance_state
419 self.maintenance_notification_hash = maintenance_notification_hash
421 if event_dispatcher is not None:
422 self.event_dispatcher = event_dispatcher
423 else:
424 self.event_dispatcher = EventDispatcher()
426 self._configure_maintenance_notifications(
427 maint_notifications_pool_handler,
428 orig_host_address,
429 orig_socket_timeout,
430 orig_socket_connect_timeout,
431 oss_cluster_maint_notifications_handler,
432 parser,
433 )
434 self._processed_start_maint_notifications = set()
435 self._skipped_end_maint_notifications = set()
437 @abstractmethod
438 def _get_parser(self) -> BaseParser:
439 pass
441 def _get_push_notifications_parser(self) -> Union[_HiredisParser, _RESP3Parser]:
442 parser = self._get_parser()
443 if not isinstance(parser, (_HiredisParser, _RESP3Parser)):
444 raise RedisError(
445 "Maintenance notifications are only supported with hiredis and RESP3 parsers!"
446 )
447 return parser
449 def _complete_maintenance_on_disconnect(self) -> None:
450 """
451 Restore the settings a maintenance relaxed on this connection when it is
452 closed, and report the relaxation as completed - the same way the
453 completion notification would have been reported. The MOVING state is
454 left alone: it is owned by the pool-level TTL cleanup.
455 """
456 if self.maintenance_state != MaintenanceState.MAINTENANCE:
457 return
458 self.reset_tmp_settings(reset_relaxed_timeout=True)
459 self.maintenance_state = MaintenanceState.NONE
460 # reset the sets that keep track of received start maint
461 # notifications and skipped end maint notifications
462 self.reset_received_notifications()
463 if self.maint_notifications_config is not None:
464 pool_handler = self._maint_notifications_pool_handler
465 _dispatch_maintenance_event(
466 self.maint_notifications_config,
467 MaintenanceCompletedEvent(
468 connection_pool=(
469 pool_handler.pool if pool_handler is not None else None
470 ),
471 connection=self,
472 state=MaintenanceState.MAINTENANCE,
473 notification=None,
474 config=self.maint_notifications_config,
475 ),
476 )
478 def handle_pending_push_notifications(self) -> None:
479 """
480 Process the push notifications buffered on this idle connection if it
481 is still alive, and report a connection that has to be reconnected.
483 A pooled connection that has readable data while idle either has push
484 notifications waiting, or has been closed by the server - or both, when
485 the notifications announced the maintenance that closed it. The socket
486 state cannot be told apart from the pending data without reading it, so
487 the notifications are collected first and handed to their handlers only
488 once the drain has shown the socket to be alive: the server closes the
489 connections of a moved endpoint when the MOVING time-to-live expires,
490 so an alive socket means the maintenance is still in progress and the
491 announced address current. A notification buffered on a socket the
492 server has since closed describes a maintenance that is over, after
493 which the configured address is what to connect to again: it is
494 discarded, and the ``ConnectionError`` raised here makes the pool
495 reconnect before any command is sent.
497 Anything read that is not a maintenance notification or a client-side
498 cache invalidation - a reply, a null or an error reply left unread by an
499 earlier command - means the connection is dirty and is reported the same
500 way, as is a drain that does not complete within
501 ``PENDING_PUSH_NOTIFICATIONS_READ_TIMEOUT``.
502 """
503 parser = self._get_push_notifications_parser()
504 deferred: list = []
506 def defer(handler):
507 def deferred_handler(notification):
508 deferred.append((handler, notification))
510 return deferred_handler
512 handlers = (
513 (parser.node_moving_push_handler_func, parser.set_node_moving_push_handler),
514 (parser.maintenance_push_handler_func, parser.set_maintenance_push_handler),
515 (
516 parser.oss_cluster_maint_push_handler_func,
517 parser.set_oss_cluster_maint_push_handler,
518 ),
519 # Client-side cache invalidations are pushes too; the ones on a
520 # connection that has to be reconnected are covered by the cache
521 # flush its disconnect performs
522 (
523 parser.invalidation_push_handler_func,
524 parser.set_invalidation_push_handler,
525 ),
526 )
527 for handler, set_handler in handlers:
528 if handler is not None:
529 set_handler(defer(handler))
531 alive = False
532 deadline = time.monotonic() + PENDING_PUSH_NOTIFICATIONS_READ_TIMEOUT
533 try:
534 while self.can_read():
535 remaining = deadline - time.monotonic()
536 if remaining <= 0:
537 raise ConnectionError(
538 "Timed out reading pending push notifications"
539 )
540 read = len(deferred)
541 # The pool disconnects the connection it handed out - with
542 # caching the proxy, not the wrapped connection - on the error.
543 # The parsers apply the timeout to each receive of a frame; the
544 # deadline bounds the drain across frames.
545 self.read_response(
546 push_request=True,
547 timeout=remaining,
548 disconnect_on_error=False,
549 )
550 # Only a push reaches a deferred handler; anything else is a
551 # reply left unread by an earlier command
552 if len(deferred) == read:
553 raise ConnectionError("Connection has data")
554 alive = True
555 except ResponseError as e:
556 raise ConnectionError("Connection has data") from e
557 finally:
558 for handler, set_handler in handlers:
559 set_handler(handler)
560 if not alive and deferred and logger.isEnabledFor(logging.DEBUG):
561 logger.debug(
562 f"Discarding {len(deferred)} push notification(s) buffered on "
563 f"a connection that is closed or dirty, reconnecting: {self}"
564 )
566 for handler, notification in deferred:
567 try:
568 handler(notification)
569 except Exception as e:
570 logger.error(f"Error handling {notification}: {e}")
572 @abstractmethod
573 def _get_socket(self) -> Optional[socket.socket]:
574 pass
576 @abstractmethod
577 def get_protocol(self) -> Union[int, str]:
578 """
579 Returns:
580 The RESP protocol version, or ``None`` if the protocol is not specified,
581 in which case the server default will be used.
582 """
583 pass
585 @property
586 @abstractmethod
587 def host(self) -> str:
588 pass
590 @host.setter
591 @abstractmethod
592 def host(self, value: str):
593 pass
595 @property
596 @abstractmethod
597 def socket_timeout(self) -> Optional[Union[float, int]]:
598 pass
600 @socket_timeout.setter
601 @abstractmethod
602 def socket_timeout(self, value: Optional[Union[float, int]]):
603 pass
605 @property
606 @abstractmethod
607 def socket_connect_timeout(self) -> Optional[Union[float, int]]:
608 pass
610 @socket_connect_timeout.setter
611 @abstractmethod
612 def socket_connect_timeout(self, value: Optional[Union[float, int]]):
613 pass
615 @abstractmethod
616 def send_command(self, *args, **kwargs):
617 pass
619 @abstractmethod
620 def read_response(
621 self,
622 disable_decoding=False,
623 *,
624 timeout: Union[float, object] = SENTINEL,
625 disconnect_on_error=True,
626 push_request=False,
627 ):
628 pass
630 @abstractmethod
631 def disconnect(self, *args, **kwargs):
632 pass
634 @abstractmethod
635 def mark_for_reconnect(self):
636 pass
638 def _configure_maintenance_notifications(
639 self,
640 maint_notifications_pool_handler: Optional[
641 MaintNotificationsPoolHandler
642 ] = None,
643 orig_host_address=None,
644 orig_socket_timeout=None,
645 orig_socket_connect_timeout=None,
646 oss_cluster_maint_notifications_handler: Optional[
647 OSSMaintNotificationsHandler
648 ] = None,
649 parser: Optional[BaseParser] = None,
650 ):
651 """
652 Enable maintenance notifications by setting up
653 handlers and storing original connection parameters.
655 Should be used ONLY with parsers that support push notifications.
656 """
657 if (
658 not self.maint_notifications_config
659 or not self.maint_notifications_config.enabled
660 ):
661 self._maint_notifications_pool_handler = None
662 self._maint_notifications_connection_handler = None
663 self._oss_cluster_maint_notifications_handler = None
664 return
666 if not parser:
667 raise RedisError(
668 "To configure maintenance notifications, a parser must be provided!"
669 )
671 if not isinstance(parser, _HiredisParser) and not isinstance(
672 parser, _RESP3Parser
673 ):
674 raise RedisError(
675 "Maintenance notifications are only supported with hiredis and RESP3 parsers!"
676 )
678 if maint_notifications_pool_handler:
679 # Extract a reference to a new pool handler that copies all properties
680 # of the original one and has a different connection reference
681 # This is needed because when we attach the handler to the parser
682 # we need to make sure that the handler has a reference to the
683 # connection that the parser is attached to.
684 self._maint_notifications_pool_handler = (
685 maint_notifications_pool_handler.get_handler_for_connection()
686 )
687 self._maint_notifications_pool_handler.set_connection(self)
688 else:
689 self._maint_notifications_pool_handler = None
691 self._maint_notifications_connection_handler = (
692 MaintNotificationsConnectionHandler(self, self.maint_notifications_config)
693 )
695 if oss_cluster_maint_notifications_handler:
696 self._oss_cluster_maint_notifications_handler = (
697 oss_cluster_maint_notifications_handler
698 )
699 # Set up OSS cluster handler to parser
700 parser.set_oss_cluster_maint_push_handler(
701 self._oss_cluster_maint_notifications_handler.handle_notification
702 )
703 else:
704 self._oss_cluster_maint_notifications_handler = None
706 # Set up pool handler to parser if available
707 if self._maint_notifications_pool_handler:
708 parser.set_node_moving_push_handler(
709 self._maint_notifications_pool_handler.handle_notification
710 )
712 # Set up connection handler
713 parser.set_maintenance_push_handler(
714 self._maint_notifications_connection_handler.handle_notification
715 )
717 # Store original connection parameters
718 self.orig_host_address = orig_host_address if orig_host_address else self.host
719 self.orig_socket_timeout = (
720 orig_socket_timeout if orig_socket_timeout else self.socket_timeout
721 )
722 self.orig_socket_connect_timeout = (
723 orig_socket_connect_timeout
724 if orig_socket_connect_timeout
725 else self.socket_connect_timeout
726 )
728 def set_maint_notifications_pool_handler_for_connection(
729 self, maint_notifications_pool_handler: MaintNotificationsPoolHandler
730 ):
731 # Deep copy the pool handler to avoid sharing the same pool handler
732 # between multiple connections, because otherwise each connection will override
733 # the connection reference and the pool handler will only hold a reference
734 # to the last connection that was set.
735 maint_notifications_pool_handler_copy = (
736 maint_notifications_pool_handler.get_handler_for_connection()
737 )
739 maint_notifications_pool_handler_copy.set_connection(self)
740 parser = self._get_push_notifications_parser()
741 parser.set_node_moving_push_handler(
742 maint_notifications_pool_handler_copy.handle_notification
743 )
745 self._maint_notifications_pool_handler = maint_notifications_pool_handler_copy
747 # Update maintenance notification connection handler if it doesn't exist
748 if not self._maint_notifications_connection_handler:
749 self._maint_notifications_connection_handler = (
750 MaintNotificationsConnectionHandler(
751 self, maint_notifications_pool_handler.config
752 )
753 )
754 parser.set_maintenance_push_handler(
755 self._maint_notifications_connection_handler.handle_notification
756 )
757 else:
758 self._maint_notifications_connection_handler.config = (
759 maint_notifications_pool_handler.config
760 )
762 def set_maint_notifications_cluster_handler_for_connection(
763 self, oss_cluster_maint_notifications_handler: OSSMaintNotificationsHandler
764 ):
765 parser = self._get_push_notifications_parser()
766 parser.set_oss_cluster_maint_push_handler(
767 oss_cluster_maint_notifications_handler.handle_notification
768 )
769 # OSS cluster mode and pool-handler mode are mutually exclusive. Clear
770 # any node-moving/pool handler a default (RESP3 "auto") pool wired in
771 # __init__ so this existing connection is not configured with both.
772 parser.set_node_moving_push_handler(None)
773 self._maint_notifications_pool_handler = None
775 self._oss_cluster_maint_notifications_handler = (
776 oss_cluster_maint_notifications_handler
777 )
779 # Update maintenance notification connection handler if it doesn't exist
780 if not self._maint_notifications_connection_handler:
781 self._maint_notifications_connection_handler = (
782 MaintNotificationsConnectionHandler(
783 self, oss_cluster_maint_notifications_handler.config
784 )
785 )
786 parser.set_maintenance_push_handler(
787 self._maint_notifications_connection_handler.handle_notification
788 )
789 else:
790 self._maint_notifications_connection_handler.config = (
791 oss_cluster_maint_notifications_handler.config
792 )
794 def _should_enable_maint_notifications(self) -> bool:
795 # Maintenance notifications are sent only if RESP3 is active
796 # and maintenance notifications are enabled
797 # and we have a host to determine the endpoint type from.
798 host = getattr(self, "host", None)
799 return bool(
800 check_protocol_version(self.get_protocol(), 3)
801 and self.maint_notifications_config
802 and self.maint_notifications_config.enabled
803 and self._maint_notifications_connection_handler
804 and host is not None
805 )
807 def activate_maint_notifications_handling_if_enabled(self, check_health=True):
808 # When the maint_notifications_config enabled mode is "auto",
809 # we just log a warning if the handshake fails
810 # When the mode is enabled=True, we raise an exception in case of failure
811 if self._should_enable_maint_notifications():
812 self._enable_maintenance_notifications(
813 maint_notifications_config=self.maint_notifications_config,
814 check_health=check_health,
815 )
817 def _enable_maintenance_notifications(
818 self, maint_notifications_config: MaintNotificationsConfig, check_health=True
819 ):
820 # Kept for callers that enable maintenance notifications outside of the
821 # connection handshake. During on_connect the send and the response
822 # handling are split (see _send_maint_notifications_command /
823 # _handle_maint_notifications_response) so the reply can be pipelined
824 # with the rest of the handshake.
825 self._send_maint_notifications_command(
826 maint_notifications_config, check_health=check_health
827 )
828 self._handle_maint_notifications_response(maint_notifications_config)
830 def _maint_notifications_command_args(
831 self, maint_notifications_config: MaintNotificationsConfig
832 ):
833 host = getattr(self, "host", None)
834 if host is None:
835 raise ValueError(
836 "Cannot enable maintenance notifications for connection"
837 " object that doesn't have a host attribute."
838 )
839 endpoint_type = maint_notifications_config.get_endpoint_type(host, self)
840 return (
841 "CLIENT",
842 "MAINT_NOTIFICATIONS",
843 "ON",
844 "moving-endpoint-type",
845 endpoint_type.value,
846 )
848 def _send_maint_notifications_command(
849 self, maint_notifications_config: MaintNotificationsConfig, check_health=True
850 ):
851 self.send_command(
852 *self._maint_notifications_command_args(maint_notifications_config),
853 check_health=check_health,
854 )
856 def _add_maint_notifications_to_handshake(self, deferred_reads, check_health=True):
857 # If maintenance notifications are enabled for this connection, send the
858 # CLIENT MAINT_NOTIFICATIONS command with the rest of the handshake tail and
859 # defer reading its reply (appended to deferred_reads), rather than paying its
860 # own round-trip. When enabled == "auto" a failure is logged and swallowed;
861 # when enabled is True it raises.
862 if not self._should_enable_maint_notifications():
863 return
864 maint_notifications_config = self.maint_notifications_config
865 self._send_maint_notifications_command(
866 maint_notifications_config, check_health=check_health
867 )
868 deferred_reads.append(
869 lambda: self._handle_maint_notifications_response(
870 maint_notifications_config
871 )
872 )
874 def _handle_maint_notifications_response(
875 self, maint_notifications_config: MaintNotificationsConfig
876 ):
877 try:
878 response = self.read_response()
879 if not response or str_if_bytes(response) != "OK":
880 raise ResponseError(
881 "The server doesn't support maintenance notifications"
882 )
883 except Exception as e:
884 if (
885 isinstance(e, ResponseError)
886 and maint_notifications_config.enabled == "auto"
887 ):
888 # Log warning but don't fail the connection
889 logger.debug(f"Failed to enable maintenance notifications: {e}")
890 else:
891 raise
893 def _read_ok_or_raise(self, error_message):
894 # Read one handshake reply and require it to be "OK".
895 if str_if_bytes(self.read_response()) != "OK":
896 raise ConnectionError(error_message)
898 def _read_optional_setinfo(self):
899 # Read one CLIENT SETINFO reply. Older servers may not support the command,
900 # so a ResponseError is swallowed instead of failing the connection.
901 try:
902 self.read_response()
903 except ResponseError:
904 pass
906 def get_resolved_ip(self) -> Optional[str]:
907 """
908 Extract the resolved IP address from an
909 established connection or resolve it from the host.
911 First tries to get the actual IP from the socket (most accurate),
912 then falls back to DNS resolution if needed.
914 Returns:
915 str: The resolved IP address, or None if it cannot be determined
916 """
918 # Method 1: Try to get the actual IP from the established socket connection
919 # This is most accurate as it shows the exact IP being used
920 try:
921 conn_socket = self._get_socket()
922 if conn_socket is not None:
923 peer_addr = conn_socket.getpeername()
924 # For TCP sockets, peer_addr is typically a (host, port) tuple.
925 # AF_UNIX sockets report a path string instead, and indexing it
926 # would yield the first character of the path.
927 if isinstance(peer_addr, tuple) and peer_addr:
928 # Return just the host part
929 return peer_addr[0]
930 except (AttributeError, OSError):
931 # Socket might not be connected or getpeername() might fail
932 pass
934 # Method 2: Fallback to DNS resolution of the host
935 # This is less accurate but works when socket is not available
936 try:
937 host = getattr(self, "host", "localhost")
938 port = getattr(self, "port", 6379)
939 if host:
940 # Use getaddrinfo to resolve the hostname to IP
941 # This mimics what the connection would do during _connect()
942 addr_info = socket.getaddrinfo(
943 host, port, socket.AF_UNSPEC, socket.SOCK_STREAM
944 )
945 if addr_info:
946 # Return the IP from the first result
947 # addr_info[0] is (family, socktype, proto, canonname, sockaddr)
948 # sockaddr[0] is the IP address
949 return str(addr_info[0][4][0])
950 except (AttributeError, OSError, socket.gaierror):
951 # DNS resolution might fail
952 pass
954 return None
956 @property
957 def maintenance_state(self) -> MaintenanceState:
958 return self._maintenance_state
960 @maintenance_state.setter
961 def maintenance_state(self, state: "MaintenanceState"):
962 self._maintenance_state = state
964 def add_maint_start_notification(self, id: int):
965 self._processed_start_maint_notifications.add(id)
967 def get_processed_start_notifications(self) -> set:
968 return self._processed_start_maint_notifications
970 def add_skipped_end_notification(self, id: int):
971 self._skipped_end_maint_notifications.add(id)
973 def get_skipped_end_notifications(self) -> set:
974 return self._skipped_end_maint_notifications
976 def reset_received_notifications(self):
977 self._processed_start_maint_notifications.clear()
978 self._skipped_end_maint_notifications.clear()
980 def getpeername(self):
981 """
982 Returns the peer name of the connection.
983 """
984 conn_socket = self._get_socket()
985 if conn_socket:
986 return conn_socket.getpeername()[0]
987 return None
989 def update_current_socket_timeout(self, relaxed_timeout: Optional[float] = None):
990 conn_socket = self._get_socket()
991 if conn_socket:
992 timeout = relaxed_timeout if relaxed_timeout != -1 else self.socket_timeout
993 # if the current timeout is 0 it means we are in the middle of a can_read call
994 # in this case we don't want to change the timeout because the operation
995 # is non-blocking and should return immediately
996 # Changing the state from non-blocking to blocking in the middle of a read operation
997 # will lead to a deadlock
998 if conn_socket.gettimeout() != 0:
999 conn_socket.settimeout(timeout)
1000 # Deliberately outside the guard above. The parser caches this value
1001 # to restore after a per-call timeout override, so while a can_read
1002 # probe holds the socket at 0 the cache *is* the pending restore
1003 # value: writing it is how the new timeout gets armed without
1004 # flipping the socket back to blocking mid-read. Skipping it here
1005 # instead would let the probe restore the pre-relaxation timeout and
1006 # lose a maintenance relaxation for the life of the connection.
1007 self.update_parser_timeout(timeout)
1009 def update_parser_timeout(self, timeout: Optional[float] = None):
1010 parser = self._get_parser()
1011 if parser and parser._buffer:
1012 # Both parsers cache this value to restore after a per-call timeout
1013 # override, so both must receive exactly what was armed on the
1014 # socket - including None, which means "block indefinitely".
1015 if isinstance(parser, _RESP3Parser):
1016 parser._buffer.socket_timeout = timeout
1017 elif isinstance(parser, _HiredisParser):
1018 parser._socket_timeout = timeout
1020 def set_tmp_settings(
1021 self,
1022 tmp_host_address: Optional[Union[str, object]] = SENTINEL,
1023 tmp_relaxed_timeout: Optional[float] = -1,
1024 ):
1025 """
1026 SENTINEL keeps the host unchanged. -1 keeps the relaxed timeout unchanged.
1027 """
1028 if tmp_host_address and tmp_host_address != SENTINEL:
1029 self.host = str(tmp_host_address)
1030 if tmp_relaxed_timeout != -1:
1031 self.socket_timeout = tmp_relaxed_timeout
1032 self.socket_connect_timeout = tmp_relaxed_timeout
1034 def reset_tmp_settings(
1035 self,
1036 reset_host_address: bool = False,
1037 reset_relaxed_timeout: bool = False,
1038 ):
1039 if reset_host_address:
1040 self.host = self.orig_host_address
1041 if reset_relaxed_timeout:
1042 self.socket_timeout = self.orig_socket_timeout
1043 self.socket_connect_timeout = self.orig_socket_connect_timeout
1046class AbstractConnection(MaintNotificationsAbstractConnection, ConnectionInterface):
1047 "Manages communication to and from a Redis server"
1049 @deprecated_args(
1050 args_to_warn=["lib_name", "lib_version"],
1051 reason="Use 'driver_info' parameter instead. "
1052 "lib_name and lib_version will be removed in a future version.",
1053 )
1054 def __init__(
1055 self,
1056 db: int = 0,
1057 password: Optional[str] = None,
1058 socket_timeout: Optional[float] = DEFAULT_SOCKET_TIMEOUT,
1059 socket_connect_timeout: Optional[float] = DEFAULT_SOCKET_CONNECT_TIMEOUT,
1060 retry_on_timeout: bool = False,
1061 retry_on_error: Union[Iterable[Type[Exception]], object] = SENTINEL,
1062 encoding: str = "utf-8",
1063 encoding_errors: str = "strict",
1064 decode_responses: bool = False,
1065 parser_class=DefaultParser,
1066 socket_read_size: int = DEFAULT_SOCKET_READ_SIZE,
1067 health_check_interval: int = 0,
1068 client_name: Optional[str] = None,
1069 lib_name: Union[Optional[str], object] = SENTINEL,
1070 lib_version: Union[Optional[str], object] = SENTINEL,
1071 driver_info: Union[Optional[DriverInfo], object] = SENTINEL,
1072 username: Optional[str] = None,
1073 retry: Union[Any, None] = None,
1074 redis_connect_func: Optional[Callable[[], None]] = None,
1075 credential_provider: Optional[CredentialProvider] = None,
1076 protocol: Optional[int] = None,
1077 legacy_responses: bool = True,
1078 command_packer: Optional[Callable[[], None]] = None,
1079 event_dispatcher: Optional[EventDispatcher] = None,
1080 maint_notifications_config: Optional[MaintNotificationsConfig] = None,
1081 maint_notifications_pool_handler: Optional[
1082 MaintNotificationsPoolHandler
1083 ] = None,
1084 maintenance_state: "MaintenanceState" = MaintenanceState.NONE,
1085 maintenance_notification_hash: Optional[int] = None,
1086 orig_host_address: Optional[str] = None,
1087 orig_socket_timeout: Optional[float] = None,
1088 orig_socket_connect_timeout: Optional[float] = None,
1089 oss_cluster_maint_notifications_handler: Optional[
1090 OSSMaintNotificationsHandler
1091 ] = None,
1092 himport_registry: HImportRegistry | None = None,
1093 ):
1094 """
1095 Initialize a new Connection.
1097 To specify a retry policy for specific errors, first set
1098 `retry_on_error` to a list of the error/s to retry on, then set
1099 `retry` to a valid `Retry` object.
1100 To retry on TimeoutError, `retry_on_timeout` can also be set to `True`.
1102 Parameters
1103 ----------
1104 driver_info : DriverInfo, optional
1105 Driver metadata for CLIENT SETINFO. If provided, lib_name and lib_version
1106 are ignored. If not provided, a DriverInfo will be created from lib_name
1107 and lib_version. Explicit None disables CLIENT SETINFO.
1108 lib_name : str, optional
1109 **Deprecated.** Use driver_info instead. Library name for CLIENT SETINFO.
1110 lib_version : str, optional
1111 **Deprecated.** Use driver_info instead. Library version for CLIENT SETINFO.
1112 """
1113 if (username or password) and credential_provider is not None:
1114 raise DataError(
1115 "'username' and 'password' cannot be passed along with 'credential_"
1116 "provider'. Please provide only one of the following arguments: \n"
1117 "1. 'password' and (optional) 'username'\n"
1118 "2. 'credential_provider'"
1119 )
1120 if event_dispatcher is None:
1121 self._event_dispatcher = EventDispatcher()
1122 else:
1123 self._event_dispatcher = event_dispatcher
1124 self.pid = os.getpid()
1125 self.db = db
1126 self.client_name = client_name
1128 # Handle driver_info: if provided, use it; otherwise create from lib_name/lib_version.
1129 self.driver_info = resolve_driver_info(driver_info, lib_name, lib_version)
1131 self.credential_provider = credential_provider
1132 self.password = password
1133 self.username = username
1134 self._socket_timeout = socket_timeout
1135 if socket_connect_timeout is None:
1136 socket_connect_timeout = socket_timeout
1137 self._socket_connect_timeout = socket_connect_timeout
1138 self.retry_on_timeout = retry_on_timeout
1139 if retry_on_error is SENTINEL:
1140 retry_on_errors_list = []
1141 else:
1142 retry_on_errors_list = list(retry_on_error)
1143 if retry_on_timeout:
1144 # Add TimeoutError to the errors list to retry on
1145 retry_on_errors_list.append(TimeoutError)
1146 self.retry_on_error = retry_on_errors_list
1147 if retry or self.retry_on_error:
1148 if retry is None:
1149 self.retry = Retry(NoBackoff(), 1)
1150 else:
1151 # deep-copy the Retry object as it is mutable
1152 self.retry = copy.deepcopy(retry)
1153 if self.retry_on_error:
1154 # Update the retry's supported errors with the specified errors
1155 self.retry.update_supported_errors(self.retry_on_error)
1156 else:
1157 self.retry = Retry(NoBackoff(), 0)
1158 self.health_check_interval = health_check_interval
1159 self.next_health_check = 0
1160 self.redis_connect_func = redis_connect_func
1161 self.encoder = Encoder(encoding, encoding_errors, decode_responses)
1162 self.handshake_metadata = None
1163 self._sock = None
1164 self._socket_read_size = socket_read_size
1165 self._connect_callbacks = []
1166 self._buffer_cutoff = 6000
1167 self._re_auth_token: Optional[TokenInterface] = None
1168 try:
1169 p = int(protocol)
1170 except TypeError:
1171 p = DEFAULT_RESP_VERSION
1172 except ValueError:
1173 raise ConnectionError("protocol must be an integer")
1174 else:
1175 if p < 2 or p > 3:
1176 raise ConnectionError("protocol must be either 2 or 3")
1177 self.protocol = p
1178 self.legacy_responses = legacy_responses
1179 if self.protocol == 3 and parser_class == _RESP2Parser:
1180 # If the protocol is 3 but the parser is RESP2, change it to RESP3
1181 # This is needed because the parser might be set before the protocol
1182 # or might be provided as a kwarg to the constructor
1183 # We need to react on discrepancy only for RESP2 and RESP3
1184 # as hiredis supports both
1185 parser_class = _RESP3Parser
1186 self.set_parser(parser_class)
1188 self._command_packer = self._construct_command_packer(command_packer)
1189 self._should_reconnect = False
1191 # HIMPORT client-side state. `himport_registry` is the shared client-level
1192 # registry (empty if unconfigured) and persists across reconnects.
1193 self.himport_registry = himport_registry
1194 self._reset_himport_state()
1196 # Set up maintenance notifications
1197 MaintNotificationsAbstractConnection.__init__(
1198 self,
1199 maint_notifications_config,
1200 maint_notifications_pool_handler,
1201 maintenance_state,
1202 maintenance_notification_hash,
1203 orig_host_address,
1204 orig_socket_timeout,
1205 orig_socket_connect_timeout,
1206 oss_cluster_maint_notifications_handler,
1207 self._parser,
1208 event_dispatcher=self._event_dispatcher,
1209 )
1211 def __repr__(self):
1212 repr_args = ",".join([f"{k}={v}" for k, v in self.repr_pieces()])
1213 return f"<{self.__class__.__module__}.{self.__class__.__name__}({repr_args})>"
1215 @abstractmethod
1216 def repr_pieces(self):
1217 pass
1219 def __del__(self):
1220 try:
1221 self.disconnect()
1222 except Exception:
1223 pass
1225 @property
1226 def is_connected(self) -> bool:
1227 return self._sock is not None
1229 def _construct_command_packer(self, packer):
1230 if packer is not None:
1231 return packer
1232 elif HIREDIS_AVAILABLE:
1233 fallback = PythonRespSerializer(self._buffer_cutoff, self.encoder.encode)
1234 return HiredisRespSerializer(fallback.pack, self.encoder.encode)
1235 else:
1236 return PythonRespSerializer(self._buffer_cutoff, self.encoder.encode)
1238 def register_connect_callback(self, callback):
1239 """
1240 Register a callback to be called when the connection is established either
1241 initially or reconnected. This allows listeners to issue commands that
1242 are ephemeral to the connection, for example pub/sub subscription or
1243 key tracking. The callback must be a _method_ and will be kept as
1244 a weak reference.
1245 """
1246 wm = weakref.WeakMethod(callback)
1247 if wm not in self._connect_callbacks:
1248 self._connect_callbacks.append(wm)
1250 def deregister_connect_callback(self, callback):
1251 """
1252 De-register a previously registered callback. It will no-longer receive
1253 notifications on connection events. Calling this is not required when the
1254 listener goes away, since the callbacks are kept as weak methods.
1255 """
1256 try:
1257 self._connect_callbacks.remove(weakref.WeakMethod(callback))
1258 except ValueError:
1259 pass
1261 def set_parser(self, parser_class):
1262 """
1263 Creates a new instance of parser_class with socket size:
1264 _socket_read_size and assigns it to the parser for the connection
1265 :param parser_class: The required parser class
1266 """
1267 self._parser = parser_class(socket_read_size=self._socket_read_size)
1269 def _get_parser(self) -> Union[_HiredisParser, _RESP3Parser, _RESP2Parser]:
1270 return self._parser
1272 def connect(self):
1273 "Connects to the Redis server if not already connected"
1274 # try once the socket connect with the handshake, retry the whole
1275 # connect/handshake flow based on retry policy
1276 self.retry.call_with_retry(
1277 lambda: self.connect_check_health(
1278 check_health=True, retry_socket_connect=False
1279 ),
1280 lambda error: self.disconnect(error),
1281 )
1283 def connect_check_health(
1284 self, check_health: bool = True, retry_socket_connect: bool = True
1285 ):
1286 if self._sock:
1287 return
1288 # Track actual retry attempts for error reporting
1289 actual_retry_attempts = [0]
1291 def failure_callback(error, failure_count):
1292 actual_retry_attempts[0] = failure_count
1293 self.disconnect(error=error, failure_count=failure_count)
1295 try:
1296 if retry_socket_connect:
1297 sock = self.retry.call_with_retry(
1298 self._connect,
1299 failure_callback,
1300 with_failure_count=True,
1301 )
1302 else:
1303 sock = self._connect()
1304 except socket.timeout:
1305 e = TimeoutError("Timeout connecting to server")
1306 record_error_count(
1307 server_address=self.host,
1308 server_port=self.port,
1309 network_peer_address=self.host,
1310 network_peer_port=self.port,
1311 error_type=e,
1312 retry_attempts=actual_retry_attempts[0],
1313 )
1314 raise e
1315 except OSError as e:
1316 e = ConnectionError(self._error_message(e))
1317 record_error_count(
1318 server_address=getattr(self, "host", None),
1319 server_port=getattr(self, "port", None),
1320 network_peer_address=getattr(self, "host", None),
1321 network_peer_port=getattr(self, "port", None),
1322 error_type=e,
1323 retry_attempts=actual_retry_attempts[0],
1324 )
1325 raise e
1327 self._sock = sock
1328 try:
1329 if self.redis_connect_func is None:
1330 # Use the default on_connect function
1331 self.on_connect_check_health(check_health=check_health)
1332 else:
1333 # Use the passed function redis_connect_func
1334 self.redis_connect_func(self)
1335 except RedisError:
1336 # clean up after any error in on_connect
1337 self.disconnect()
1338 raise
1340 # run any user callbacks. right now the only internal callback
1341 # is for pubsub channel/pattern resubscription
1342 # first, remove any dead weakrefs
1343 self._connect_callbacks = [ref for ref in self._connect_callbacks if ref()]
1344 for ref in self._connect_callbacks:
1345 callback = ref()
1346 if callback:
1347 callback(self)
1349 @abstractmethod
1350 def _connect(self):
1351 pass
1353 @abstractmethod
1354 def _host_error(self):
1355 pass
1357 def _error_message(self, exception):
1358 return format_error_message(self._host_error(), exception)
1360 def _reset_himport_state(self):
1361 # A fresh server session has no prepared HIMPORT fieldsets, so the next
1362 # himport_set must re-prepare on this connection. ``_himport_prepared`` maps
1363 # fieldset name -> the version prepared on the server; ``_himport_reconciled
1364 # _revision`` is the registry revision this connection last reconciled discards
1365 # against. Both are reset on connect/disconnect since the session is gone.
1366 self._himport_prepared: dict[str, int] = {}
1367 self._himport_reconciled_revision: int = 0
1369 def on_connect(self):
1370 self.on_connect_check_health(check_health=True)
1372 def on_connect_check_health(self, check_health: bool = True):
1373 "Initialize the connection, authenticate and select a database"
1374 # A fresh socket is a new server session: no prepared HIMPORT fieldsets.
1375 self._reset_himport_state()
1376 self._parser.on_connect(self)
1377 parser = self._parser
1379 auth_args = None
1380 # if credential provider or username and/or password are set, authenticate
1381 if self.credential_provider or (self.username or self.password):
1382 cred_provider = (
1383 self.credential_provider
1384 or UsernamePasswordCredentialProvider(self.username, self.password)
1385 )
1386 auth_args = cred_provider.get_credentials()
1388 # if resp version is specified and we have auth args,
1389 # we need to send them via HELLO
1390 if auth_args and check_protocol_version(self.protocol, 3):
1391 if isinstance(self._parser, _RESP2Parser):
1392 self.set_parser(_RESP3Parser)
1393 # update cluster exception classes
1394 self._parser.EXCEPTION_CLASSES = parser.EXCEPTION_CLASSES
1395 self._parser.on_connect(self)
1396 if len(auth_args) == 1:
1397 auth_args = ["default", auth_args[0]]
1398 # avoid checking health here -- PING will fail if we try
1399 # to check the health prior to the AUTH
1400 self.send_command(
1401 "HELLO", self.protocol, "AUTH", *auth_args, check_health=False
1402 )
1403 self.handshake_metadata = self.read_response()
1404 # if response.get(b"proto") != self.protocol and response.get(
1405 # "proto"
1406 # ) != self.protocol:
1407 # raise ConnectionError("Invalid RESP version")
1408 elif auth_args:
1409 # avoid checking health here -- PING will fail if we try
1410 # to check the health prior to the AUTH
1411 self.send_command("AUTH", *auth_args, check_health=False)
1413 try:
1414 auth_response = self.read_response()
1415 except AuthenticationWrongNumberOfArgsError:
1416 # a username and password were specified but the Redis
1417 # server seems to be < 6.0.0 which expects a single password
1418 # arg. retry auth with just the password.
1419 # https://github.com/andymccurdy/redis-py/issues/1274
1420 self.send_command("AUTH", auth_args[-1], check_health=False)
1421 auth_response = self.read_response()
1423 if str_if_bytes(auth_response) != "OK":
1424 raise AuthenticationError("Invalid Username or Password")
1426 # if resp version is specified, switch to it
1427 elif check_protocol_version(self.protocol, 3):
1428 if isinstance(self._parser, _RESP2Parser):
1429 self.set_parser(_RESP3Parser)
1430 # update cluster exception classes
1431 self._parser.EXCEPTION_CLASSES = parser.EXCEPTION_CLASSES
1432 self._parser.on_connect(self)
1433 self.send_command("HELLO", self.protocol, check_health=check_health)
1434 self.handshake_metadata = self.read_response()
1435 if (
1436 self.handshake_metadata.get(b"proto") != self.protocol
1437 and self.handshake_metadata.get("proto") != self.protocol
1438 ):
1439 raise ConnectionError("Invalid RESP version")
1441 # The tail of the handshake (optional CLIENT MAINT_NOTIFICATIONS, then
1442 # CLIENT SETNAME / SETINFO / SELECT) does not affect control flow -- the replies
1443 # are only validated or discarded. So we optimize the flow: send every command
1444 # first (without blocking on a reply between them), then read the replies back in
1445 # send order. All requests are on the wire before we block on the first read, so
1446 # the whole tail costs a single round-trip instead of one per command. AUTH/HELLO
1447 # stays a separate round-trip above because its reply drives the RESP2->RESP3
1448 # parser upgrade, the pre-6.0 AUTH retry, and proto validation.
1449 #
1450 # deferred_reads holds one zero-arg handler per command sent below; each reads
1451 # exactly one reply (in send order) and validates it. Per-command check_health
1452 # reproduces the original behavior: at most one health PING/PONG fires before the
1453 # first tail command (only when no HELLO/AUTH ran, e.g. RESP2 no-auth), and it is
1454 # self-contained so it never desyncs the deferred replies.
1455 #
1456 # TODO: apply the same optimization to the async handshake in
1457 # redis/asyncio/connection.py, where not all tail commands are handled this way
1458 # yet (e.g. maintenance notifications are still issued separately there).
1459 deferred_reads = []
1461 # Maintenance notifications (RESP3-only, opt-in) go first when enabled, so their
1462 # reply is read together with the rest of the tail.
1463 self._add_maint_notifications_to_handshake(deferred_reads, check_health)
1465 # if a client_name is given, set it
1466 if self.client_name:
1467 self.send_command(
1468 "CLIENT", "SETNAME", self.client_name, check_health=check_health
1469 )
1470 deferred_reads.append(
1471 functools.partial(self._read_ok_or_raise, "Error setting client name")
1472 )
1474 # Set the library name and version from driver_info. Older servers may not
1475 # support CLIENT SETINFO, so any ResponseError to these replies is swallowed.
1476 if self.driver_info and self.driver_info.formatted_name:
1477 self.send_command(
1478 "CLIENT",
1479 "SETINFO",
1480 "LIB-NAME",
1481 self.driver_info.formatted_name,
1482 check_health=check_health,
1483 )
1484 deferred_reads.append(self._read_optional_setinfo)
1486 if self.driver_info and self.driver_info.lib_version:
1487 self.send_command(
1488 "CLIENT",
1489 "SETINFO",
1490 "LIB-VER",
1491 self.driver_info.lib_version,
1492 check_health=check_health,
1493 )
1494 deferred_reads.append(self._read_optional_setinfo)
1496 # if a database is specified, switch to it
1497 if self.db:
1498 self.send_command("SELECT", self.db, check_health=check_health)
1499 deferred_reads.append(
1500 functools.partial(self._read_ok_or_raise, "Invalid Database")
1501 )
1503 # Read the deferred replies in the order the commands were sent.
1504 for read_and_validate_response in deferred_reads:
1505 read_and_validate_response()
1507 def disconnect(self, *args, **kwargs):
1508 "Disconnects from the Redis server"
1509 # The server session is gone, so any HIMPORT fieldsets prepared on this
1510 # socket no longer exist; reset the tracking.
1511 self._reset_himport_state()
1512 # Likewise the relaxation a maintenance applied to it: restored and
1513 # reported here, ahead of a close that may find no socket to close
1514 self._complete_maintenance_on_disconnect()
1515 self._parser.on_disconnect()
1517 conn_sock = self._sock
1518 self._sock = None
1519 # reset the reconnect flag
1520 self.reset_should_reconnect()
1522 if conn_sock is None:
1523 return
1525 if os.getpid() == self.pid:
1526 try:
1527 conn_sock.shutdown(socket.SHUT_RDWR)
1528 except (OSError, TypeError):
1529 pass
1531 try:
1532 conn_sock.close()
1533 except OSError:
1534 pass
1536 error = kwargs.get("error")
1537 failure_count = kwargs.get("failure_count")
1538 health_check_failed = kwargs.get("health_check_failed")
1540 if error:
1541 if health_check_failed:
1542 close_reason = CloseReason.HEALTHCHECK_FAILED
1543 else:
1544 close_reason = CloseReason.ERROR
1546 if failure_count is not None and failure_count > self.retry.get_retries():
1547 record_error_count(
1548 server_address=self.host,
1549 server_port=self.port,
1550 network_peer_address=self.host,
1551 network_peer_port=self.port,
1552 error_type=error,
1553 retry_attempts=failure_count,
1554 )
1556 record_connection_closed(
1557 close_reason=close_reason,
1558 error_type=error,
1559 )
1560 else:
1561 record_connection_closed(
1562 close_reason=CloseReason.APPLICATION_CLOSE,
1563 )
1565 def mark_for_reconnect(self):
1566 self._should_reconnect = True
1568 def should_reconnect(self):
1569 return self._should_reconnect
1571 def reset_should_reconnect(self):
1572 self._should_reconnect = False
1574 def _send_ping(self):
1575 """Send PING, expect PONG in return"""
1576 self.send_command("PING", check_health=False)
1577 if str_if_bytes(self.read_response()) != "PONG":
1578 raise ConnectionError("Bad response from PING health check")
1580 def _ping_failed(self, error, failure_count):
1581 """Function to call when PING fails"""
1582 self.disconnect(
1583 error=error, failure_count=failure_count, health_check_failed=True
1584 )
1586 def check_health(self):
1587 """Check the health of the connection with a PING/PONG"""
1588 if self.health_check_interval and time.monotonic() > self.next_health_check:
1589 self.retry.call_with_retry(
1590 self._send_ping,
1591 self._ping_failed,
1592 with_failure_count=True,
1593 )
1595 def send_packed_command(self, command, check_health=True):
1596 """Send an already packed command to the Redis server"""
1597 if not self._sock:
1598 self.connect_check_health(check_health=False)
1599 # guard against health check recursion
1600 if check_health:
1601 self.check_health()
1602 try:
1603 if isinstance(command, str):
1604 command = [command]
1605 for item in command:
1606 self._sock.sendall(item)
1607 except socket.timeout as e:
1608 add_debug_log_for_connection_failure(self, e, "writing command")
1609 self.disconnect()
1610 raise TimeoutError("Timeout writing to socket")
1611 except OSError as e:
1612 add_debug_log_for_connection_failure(self, e, "writing command")
1613 self.disconnect()
1614 if len(e.args) == 1:
1615 errno, errmsg = "UNKNOWN", e.args[0]
1616 else:
1617 errno = e.args[0]
1618 errmsg = e.args[1]
1619 raise ConnectionError(f"Error {errno} while writing to socket. {errmsg}.")
1620 except BaseException as e:
1621 # BaseExceptions can be raised when a socket send operation is not
1622 # finished, e.g. due to a timeout. Ideally, a caller could then re-try
1623 # to send un-sent data. However, the send_packed_command() API
1624 # does not support it so there is no point in keeping the connection open.
1625 add_debug_log_for_connection_failure(self, e, "writing command")
1626 self.disconnect()
1627 raise
1629 def send_command(self, *args, **kwargs):
1630 """Pack and send a command to the Redis server"""
1631 self.send_packed_command(
1632 self._command_packer.pack(*args),
1633 check_health=kwargs.get("check_health", True),
1634 )
1636 def can_read(self, timeout: float = 0) -> bool:
1637 """Poll the socket to see if there's data that can be read."""
1638 # TODO: Rename this API; it detects pending data or dirty/closed
1639 # connection state, not only whether application data can be read.
1640 sock = self._sock
1641 if not sock:
1642 self.connect()
1644 host_error = self._host_error()
1646 try:
1647 return self._parser.can_read(timeout)
1649 except OSError as e:
1650 self.disconnect()
1651 raise ConnectionError(f"Error while reading from {host_error}: {e.args}")
1653 def read_response(
1654 self,
1655 disable_decoding=False,
1656 *,
1657 timeout: Union[float, object] = SENTINEL,
1658 disconnect_on_error=True,
1659 push_request=False,
1660 ):
1661 """Read the response from a previously sent command"""
1663 host_error = self._host_error()
1665 try:
1666 if self.protocol in ["3", 3]:
1667 response = self._parser.read_response(
1668 disable_decoding=disable_decoding,
1669 push_request=push_request,
1670 timeout=timeout,
1671 )
1672 else:
1673 response = self._parser.read_response(
1674 disable_decoding=disable_decoding, timeout=timeout
1675 )
1676 except socket.timeout as e:
1677 if disconnect_on_error:
1678 add_debug_log_for_connection_failure(self, e, "reading response")
1679 self.disconnect()
1680 raise TimeoutError(f"Timeout reading from {host_error}")
1681 except TimeoutError as e:
1682 # The parsers raise redis.exceptions.TimeoutError, which is not an
1683 # OSError, so without this branch it would fall through to
1684 # BaseException and keep the parser's undecorated message. Re-raise
1685 # it with the host, matching what the async stack already reports.
1686 if disconnect_on_error:
1687 add_debug_log_for_connection_failure(self, e, "reading response")
1688 self.disconnect()
1689 raise TimeoutError(f"Timeout reading from {host_error}")
1690 except OSError as e:
1691 if disconnect_on_error:
1692 add_debug_log_for_connection_failure(self, e, "reading response")
1693 self.disconnect()
1694 raise ConnectionError(f"Error while reading from {host_error} : {e.args}")
1695 except UNRECOVERABLE_PARSE_ERRORS as e:
1696 # The parser failed partway through a reply, after the read cursor
1697 # had already passed bytes it cannot re-interpret. The rewind and
1698 # disconnect_on_error=False exist so an *interrupted* read can be
1699 # re-parsed from the start (#2510, #2695) - an in-band ResponseError
1700 # is returned as a value and purged, never rewound. Re-parsing here
1701 # just reproduces the same failure, so honouring
1702 # disconnect_on_error=False would make every later read fail
1703 # identically, forever. Drop the connection regardless. See #4291.
1704 add_debug_log_for_connection_failure(self, e, "reading response")
1705 self.disconnect()
1706 raise
1707 except BaseException as e:
1708 # Also by default close in case of BaseException. A lot of code
1709 # relies on this behaviour when doing Command/Response pairs.
1710 # See #1128.
1711 if disconnect_on_error:
1712 add_debug_log_for_connection_failure(self, e, "reading response")
1713 self.disconnect()
1714 raise
1716 if self.health_check_interval:
1717 self.next_health_check = time.monotonic() + self.health_check_interval
1719 if isinstance(response, ResponseError):
1720 try:
1721 raise response
1722 finally:
1723 del response # avoid creating ref cycles
1724 return response
1726 def pack_command(self, *args):
1727 """Pack a series of arguments into the Redis protocol"""
1728 return self._command_packer.pack(*args)
1730 def pack_commands(self, commands):
1731 """Pack multiple commands into the Redis protocol"""
1732 output = []
1733 pieces = []
1734 buffer_length = 0
1735 buffer_cutoff = self._buffer_cutoff
1737 for cmd in commands:
1738 for chunk in self._command_packer.pack(*cmd):
1739 chunklen = len(chunk)
1740 if (
1741 buffer_length > buffer_cutoff
1742 or chunklen > buffer_cutoff
1743 or isinstance(chunk, memoryview)
1744 ):
1745 if pieces:
1746 output.append(SYM_EMPTY.join(pieces))
1747 buffer_length = 0
1748 pieces = []
1750 if chunklen > buffer_cutoff or isinstance(chunk, memoryview):
1751 output.append(chunk)
1752 else:
1753 pieces.append(chunk)
1754 buffer_length += chunklen
1756 if pieces:
1757 output.append(SYM_EMPTY.join(pieces))
1758 return output
1760 def get_protocol(self) -> Union[int, str]:
1761 return self.protocol
1763 @property
1764 def handshake_metadata(self) -> Union[Dict[bytes, bytes], Dict[str, str]]:
1765 return self._handshake_metadata
1767 @handshake_metadata.setter
1768 def handshake_metadata(self, value: Union[Dict[bytes, bytes], Dict[str, str]]):
1769 self._handshake_metadata = value
1771 def set_re_auth_token(self, token: TokenInterface):
1772 self._re_auth_token = token
1774 def re_auth(self):
1775 if self._re_auth_token is not None:
1776 self.send_command(
1777 "AUTH",
1778 self._re_auth_token.try_get("oid"),
1779 self._re_auth_token.get_value(),
1780 )
1781 self.read_response()
1782 self._re_auth_token = None
1784 def _get_socket(self) -> Optional[socket.socket]:
1785 return self._sock
1787 @property
1788 def socket_timeout(self) -> Optional[Union[float, int]]:
1789 return self._socket_timeout
1791 @socket_timeout.setter
1792 def socket_timeout(self, value: Optional[Union[float, int]]):
1793 self._socket_timeout = value
1795 @property
1796 def socket_connect_timeout(self) -> Optional[Union[float, int]]:
1797 return self._socket_connect_timeout
1799 @socket_connect_timeout.setter
1800 def socket_connect_timeout(self, value: Optional[Union[float, int]]):
1801 self._socket_connect_timeout = value
1803 def extract_connection_details(self) -> str:
1804 """
1805 Render the connection's identity, maintenance state and effective timeouts.
1807 This is what the debug logs use to explain a failed or timed out command:
1809 - ``host`` vs ``orig host`` says whether the connection still points at the
1810 node being moved away from, or has already been repointed at the new one.
1811 - ``state`` says whether maintenance handling touched this connection at
1812 all, so an unaffected node's connections are distinguishable.
1813 - ``socket_timeout`` vs ``active read timeout`` says which timeout the read
1814 actually ran under. They diverge for a command that was already in flight
1815 when the relaxed timeout was applied.
1816 """
1817 if self._sock is None:
1818 return "not connected"
1820 socket_address = None
1821 active_read_timeout = None
1822 try:
1823 socket_name = self._sock.getsockname()
1824 # AF_UNIX sockets report a path string rather than a (host, port) tuple
1825 if isinstance(socket_name, tuple) and len(socket_name) > 1:
1826 socket_address = socket_name[1]
1827 # The timeout armed on the socket is what a blocking read is really
1828 # using, which can lag self.socket_timeout for an in-flight command.
1829 active_read_timeout = self._sock.gettimeout()
1830 except (AttributeError, OSError):
1831 pass
1833 state = getattr(self.maintenance_state, "value", self.maintenance_state)
1834 return (
1835 f"connected to ip {self.get_resolved_ip()}, "
1836 f"local socket port: {socket_address}, "
1837 f"host: {self._host_error()} "
1838 f"(orig: {getattr(self, 'orig_host_address', None)}), "
1839 f"state: {state}, "
1840 f"socket_timeout: {self.socket_timeout} "
1841 f"(orig: {getattr(self, 'orig_socket_timeout', None)}), "
1842 f"active read timeout: {active_read_timeout}, "
1843 f"should_reconnect: {self.should_reconnect()}, "
1844 f"notification_hash: {self.maintenance_notification_hash}"
1845 )
1848class Connection(AbstractConnection):
1849 "Manages TCP communication to and from a Redis server"
1851 def __init__(
1852 self,
1853 host="localhost",
1854 port=6379,
1855 socket_keepalive=True,
1856 socket_keepalive_options=SENTINEL,
1857 socket_type=0,
1858 **kwargs,
1859 ):
1860 """
1861 Initialize a TCP connection.
1863 Parameters
1864 ----------
1865 socket_keepalive : bool
1866 If `True`, TCP keepalive is enabled for TCP socket connections.
1867 socket_keepalive_options : Mapping[int, int | bytes] | object | None
1868 Mapping of TCP keepalive socket option constants to values, for
1869 example `{socket.TCP_KEEPIDLE: 30}`. If left unspecified, redis-py
1870 uses TCP keepalive defaults when `socket_keepalive` is enabled:
1871 idle 30 seconds, interval 5 seconds, and 3 probes. Platform-specific
1872 options that are not available are skipped. Pass `None` or `{}` to
1873 avoid setting additional TCP keepalive options.
1874 """
1875 self._host = host
1876 # bool subclasses int; port=True would become privileged port 1.
1877 # Numeric strings stay valid. Callers still pass "6379".
1878 if isinstance(port, bool):
1879 raise TypeError("port must be an integer, not bool")
1880 if isinstance(port, str):
1881 try:
1882 port = int(port)
1883 except ValueError:
1884 raise TypeError("port must be an integer, not str") from None
1885 elif not isinstance(port, int):
1886 raise TypeError(f"port must be an integer, not {type(port).__name__}")
1887 if not 0 <= port <= 65535:
1888 raise ValueError(f"port must be in 0..65535, got {port}")
1889 self.port = port
1890 self.socket_keepalive = socket_keepalive
1891 if socket_keepalive_options is SENTINEL:
1892 socket_keepalive_options = get_default_socket_keepalive_options()
1893 self.socket_keepalive_options = socket_keepalive_options or {}
1894 self.socket_type = socket_type
1895 super().__init__(**kwargs)
1897 def repr_pieces(self):
1898 pieces = [("host", self.host), ("port", self.port), ("db", self.db)]
1899 if self.client_name:
1900 pieces.append(("client_name", self.client_name))
1901 return pieces
1903 def _connect(self):
1904 "Create a TCP socket connection"
1905 # we want to mimic what socket.create_connection does to support
1906 # ipv4/ipv6, but we want to set options prior to calling
1907 # socket.connect()
1909 # Last caught connection error.
1910 # Re-thrown if we are unable to connect to any of the options returned
1911 # by getaddrinfo.
1912 # Note that we must clear this variable before returning - otherwise,
1913 # a caught err's traceback points to this frame, which points to err.
1914 # Clearing this lets refcounting reclaim the exception immediately
1915 # without deferring to the python garbage collector.
1916 err = None
1918 for res in socket.getaddrinfo(
1919 self.host, self.port, self.socket_type, socket.SOCK_STREAM
1920 ):
1921 family, socktype, proto, canonname, socket_address = res
1922 sock = None
1923 try:
1924 sock = socket.socket(family, socktype, proto)
1925 # TCP_NODELAY
1926 sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
1928 # TCP_KEEPALIVE
1929 if self.socket_keepalive:
1930 sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
1931 for k, v in self.socket_keepalive_options.items():
1932 sock.setsockopt(socket.IPPROTO_TCP, k, v)
1934 # set the socket_connect_timeout before we connect
1935 sock.settimeout(self.socket_connect_timeout)
1937 # connect
1938 sock.connect(socket_address)
1940 # set the socket_timeout now that we're connected
1941 sock.settimeout(self.socket_timeout)
1943 # If a previous connection attempt failed, clear the error
1944 err = None
1946 return sock
1948 except OSError as _:
1949 err = _
1950 if sock is not None:
1951 try:
1952 sock.shutdown(socket.SHUT_RDWR) # ensure a clean close
1953 except OSError:
1954 pass
1955 sock.close()
1957 if err is not None:
1958 try:
1959 raise err
1960 finally:
1961 # Ensure we clear local references to caught exceptions
1962 err = None
1963 raise OSError("socket.getaddrinfo returned an empty list")
1965 def _host_error(self):
1966 return f"{self.host}:{self.port}"
1968 @property
1969 def host(self) -> str:
1970 return self._host
1972 @host.setter
1973 def host(self, value: str):
1974 self._host = value
1977# Distinguishes "no locally-served reply is waiting" from a cached value that happens to be
1978# falsy. ``None`` cannot do the job: it is a legitimate reply shape, and the cache refuses to
1979# store it only by convention.
1980_NO_PENDING_HIT = object()
1982# A command name that does not start with one of these cannot be ``CLIENT`` or ``RESET``,
1983# so ``_cache_owned_command`` rejects nearly every command before it allocates anything.
1984_CLIENT_FIRST_CHARS = frozenset(("c", "C", b"c", b"C"))
1985_RESET_FIRST_CHARS = frozenset(("r", "R", b"r", b"R"))
1986# ``pack`` splits the command name on whitespace, so a name padded with it still reaches
1987# the server as ``CLIENT ...`` or ``RESET``. It encodes a ``str`` name before splitting,
1988# so ASCII whitespace is the only kind that splits.
1989_WHITESPACE_CHARS = frozenset(
1990 (*" \t\n\r\x0b\x0c", *(bytes((c,)) for c in b" \t\n\r\x0b\x0c"))
1991)
1993# The ``CLIENT`` subcommands the cache owns. ``TRACKINGINFO`` and ``GETREDIR`` only read the
1994# tracking state, so they are not here.
1995_CACHE_OWNED_CLIENT_SUBCOMMANDS = frozenset(("CACHING", "TRACKING"))
1998def _cache_owned_command(args) -> str | None:
1999 """
2000 The name of the command in ``args`` if it is one the cache owns - ``CLIENT CACHING``,
2001 ``CLIENT TRACKING`` or ``RESET`` - else ``None``. Matches every spelling
2002 ``pack_command`` accepts: ``"CLIENT CACHING"`` as one argument or ``"CLIENT"``,
2003 ``"CACHING"`` as two, as ``str``, ``bytes``, ``bytearray`` or ``memoryview``, in any
2004 case, padded with whitespace or not.
2005 """
2006 command = args[0]
2007 if not isinstance(command, (str, bytes)):
2008 if not isinstance(command, (bytearray, memoryview)):
2009 return None
2010 command = bytes(command)
2011 first_char = command[:1]
2012 if first_char in _RESET_FIRST_CHARS:
2013 # ``RESET`` takes no arguments, so only a five-character name, or one followed by
2014 # whitespace, can match. The check keeps the many other ``R`` commands (``RPOP``,
2015 # ``RENAME``, ...) allocation-free.
2016 if len(command) != 5 and command[5:6] not in _WHITESPACE_CHARS:
2017 return None
2018 elif first_char not in _CLIENT_FIRST_CHARS and first_char not in _WHITESPACE_CHARS:
2019 return None
2020 words = str_if_bytes(command).upper().split()
2021 if not words:
2022 return None
2023 if words[0] == "RESET":
2024 return "RESET"
2025 if words[0] != "CLIENT":
2026 return None
2027 if len(words) == 1 and len(args) > 1:
2028 subcommand = args[1]
2029 if isinstance(subcommand, (bytearray, memoryview)):
2030 subcommand = bytes(subcommand)
2031 if isinstance(subcommand, (str, bytes)):
2032 words.append(str_if_bytes(subcommand).upper())
2033 if len(words) > 1 and words[1] in _CACHE_OWNED_CLIENT_SUBCOMMANDS:
2034 return f"CLIENT {words[1]}"
2035 return None
2038class CacheProxyConnection(MaintNotificationsAbstractConnection, ConnectionInterface):
2039 DUMMY_CACHE_VALUE = b"foo"
2040 MIN_ALLOWED_VERSION = "7.4.0"
2041 DEFAULT_SERVER_NAME = "redis"
2043 def __init__(
2044 self,
2045 conn: ConnectionInterface,
2046 cache: CacheInterface,
2047 pool_lock: threading.RLock,
2048 ):
2049 self.pid = os.getpid()
2050 self._conn = conn
2051 self.retry = self._conn.retry
2052 self.host = self._conn.host
2053 self.port = self._conn.port
2054 self.db = self._conn.db
2055 self._event_dispatcher = self._conn._event_dispatcher
2056 self.credential_provider = conn.credential_provider
2057 self._pool_lock = pool_lock
2058 self._cache = cache
2059 self._cache_lock = threading.RLock()
2060 self._current_command_cache_key = None
2061 # Read once: the configuration cannot change after the pool built the cache, and the
2062 # mode is a connection-setup property anyway - the server refuses to switch a live
2063 # connection between OPTIN and OPTOUT, so a configuration change applies to new
2064 # connections only.
2065 self._tracking_mode = cache.config.get_tracking_mode()
2066 # A CLIENT CACHING ``+OK`` is queued ahead of the next data reply.
2067 self._pending_caching_reply = False
2068 # The next command is an ASK-redirected attempt, which must not be paired.
2069 self._skip_next_caching = False
2070 # Distinguishes a reconnect from the first connect in the tracking callback.
2071 self._connected_once = False
2072 # A reply ``send_command`` resolved from the cache, waiting for ``read_response`` to
2073 # hand it back. Set means nothing was written, so there is no reply on the socket.
2074 self._pending_cache_hit = _NO_PENDING_HIT
2075 self.register_connect_callback(self._enable_tracking_callback)
2077 if isinstance(self._conn, MaintNotificationsAbstractConnection):
2078 MaintNotificationsAbstractConnection.__init__(
2079 self,
2080 self._conn.maint_notifications_config,
2081 self._conn._maint_notifications_pool_handler,
2082 self._conn.maintenance_state,
2083 self._conn.maintenance_notification_hash,
2084 self._conn.host,
2085 self._conn.socket_timeout,
2086 self._conn.socket_connect_timeout,
2087 self._conn._oss_cluster_maint_notifications_handler,
2088 self._conn._get_parser(),
2089 event_dispatcher=self._conn.event_dispatcher,
2090 )
2092 def repr_pieces(self):
2093 return self._conn.repr_pieces()
2095 @property
2096 def is_connected(self) -> bool:
2097 return self._conn.is_connected
2099 def register_connect_callback(self, callback):
2100 self._conn.register_connect_callback(callback)
2102 def deregister_connect_callback(self, callback):
2103 self._conn.deregister_connect_callback(callback)
2105 def set_parser(self, parser_class):
2106 self._conn.set_parser(parser_class)
2108 def set_maint_notifications_pool_handler_for_connection(
2109 self, maint_notifications_pool_handler
2110 ):
2111 if isinstance(self._conn, MaintNotificationsAbstractConnection):
2112 self._conn.set_maint_notifications_pool_handler_for_connection(
2113 maint_notifications_pool_handler
2114 )
2116 def set_maint_notifications_cluster_handler_for_connection(
2117 self, oss_cluster_maint_notifications_handler
2118 ):
2119 if isinstance(self._conn, MaintNotificationsAbstractConnection):
2120 self._conn.set_maint_notifications_cluster_handler_for_connection(
2121 oss_cluster_maint_notifications_handler
2122 )
2124 def get_protocol(self):
2125 return self._conn.get_protocol()
2127 def connect(self):
2128 self._conn.connect()
2130 server_name = self._conn.handshake_metadata.get(b"server", None)
2131 if server_name is None:
2132 server_name = self._conn.handshake_metadata.get("server", None)
2133 server_ver = self._conn.handshake_metadata.get(b"version", None)
2134 if server_ver is None:
2135 server_ver = self._conn.handshake_metadata.get("version", None)
2136 if server_ver is None or server_name is None:
2137 raise ConnectionError("Cannot retrieve information about server version")
2139 server_ver = ensure_string(server_ver)
2140 server_name = ensure_string(server_name)
2142 if (
2143 server_name != self.DEFAULT_SERVER_NAME
2144 or compare_versions(server_ver, self.MIN_ALLOWED_VERSION) == 1
2145 ):
2146 raise ConnectionError(
2147 "To maximize compatibility with all Redis products, client-side caching is supported by Redis 7.4 or later" # noqa: E501
2148 )
2150 def on_connect(self):
2151 self._conn.on_connect()
2153 def disconnect(self, *args, **kwargs):
2154 with self._cache_lock:
2155 self._cache.flush()
2156 # Both flags describe an exchange on the socket that is about to die with it. A stale
2157 # ``_pending_caching_reply`` would make the next ``read_response`` swallow a reply;
2158 # a stale ``_skip_next_caching`` would leak the ASK suppression onto an unrelated
2159 # later command.
2160 self._pending_caching_reply = False
2161 self._skip_next_caching = False
2162 self._pending_cache_hit = _NO_PENDING_HIT
2163 # The maintenance handlers are bound to this proxy, so the relaxation
2164 # a maintenance applied is reported with the proxy as its source
2165 if isinstance(self._conn, MaintNotificationsAbstractConnection):
2166 self._complete_maintenance_on_disconnect()
2167 self._conn.disconnect(*args, **kwargs)
2169 def check_health(self):
2170 self._conn.check_health()
2172 def send_packed_command(self, command, check_health=True):
2173 # TODO: Investigate if it's possible to unpack command
2174 # or extract keys from packed command
2175 # Pre-packed commands are not individually cacheable, so make sure the
2176 # next read_response does not try to cache their reply under a stale key.
2177 self._current_command_cache_key = None
2178 # A pre-packed write carries no CLIENT CACHING command of its own, so a flag left
2179 # over from an earlier send must not make the next read_response swallow a reply -
2180 # nor may a stale locally-served reply be handed back in place of this write's.
2181 # The server consumes a pending ASKING on this write, so the ASK suppression must
2182 # not carry over to a later command either.
2183 self._pending_caching_reply = False
2184 self._skip_next_caching = False
2185 self._pending_cache_hit = _NO_PENDING_HIT
2186 self._conn.send_packed_command(command)
2188 def send_command(self, *args, **kwargs):
2189 # Before anything else, so a refused command leaves no state behind.
2190 self._refuse_cache_owned_command(args)
2191 self._process_pending_invalidations()
2192 self._pending_caching_reply = False
2193 self._pending_cache_hit = _NO_PENDING_HIT
2195 if self._skip_next_caching:
2196 # An ASK-redirected attempt.
2197 # ``ASKING`` like ``CLIENT CACHING`` needs to be immediatelly before the next command
2198 # so we can't have both additions to the command we are sending,
2199 # and pairing here would strip the ASK allowance and the read would
2200 # be redirected again.
2201 # Cache nothing either: the reply belongs to a migrating
2202 # slot, and the attempt is tracked by default under optout anyway.
2203 self._skip_next_caching = False
2204 self._current_command_cache_key = None
2205 self._conn.send_command(*args, **kwargs)
2206 return
2208 command = args[0]
2209 if isinstance(command, str) and command.upper() == "ASKING":
2210 # The cluster executor sends ASKING as its own command on this connection right
2211 # before the redirected attempt, so observing it here is all the notice needed -
2212 # no kwarg threading, and the flag mirrors the server's own one-shot semantics.
2213 self._skip_next_caching = True
2215 # Eligibility, keyability and intent are three separate questions, and all three must
2216 # be answered yes before a reply may be stored. The command metadata answers the
2217 # first; the presence of ``keys`` answers the second, because a command's key
2218 # positions are supplied by its command method rather than derived here. A cacheable
2219 # command whose invocation carries no key list is therefore a gap in what this client
2220 # has been taught, not an error: send it normally and cache nothing. The configured
2221 # tracking mode and predicate answer the third.
2222 keys = kwargs.get("keys")
2223 store = False
2225 if keys is not None:
2226 # Materialized once: ``keys`` may be a one-shot iterable, and it is read again
2227 # below to build the ``CacheKey`` the reverse index is keyed by.
2228 keys = tuple(keys)
2230 # An empty key list - ``mget([])``, ``exists()`` - is treated as no key list: there
2231 # is nothing to track, and the predicate is promised never to see an empty tuple.
2232 if keys:
2233 with self._cache_lock:
2234 # Eligibility is asked with a throwaway empty-keys ``CacheKey`` because
2235 # ``DefaultCache.is_cachable`` looks only at ``key.command``.
2236 eligible = self._cache.is_cachable(
2237 CacheKey(command=command, redis_keys=(), redis_args=())
2238 )
2240 # Intent is asked separately, after the real keys are known, and outside the
2241 # cache lock: it invokes application code and touches no cache state. Folding it
2242 # into ``is_cachable`` would also invoke it from ``DefaultCache.set`` and from
2243 # the eligibility probe above, both with an empty key tuple.
2244 store = eligible and self._cache.config.should_cache(command, keys)
2246 if not store:
2247 self._current_command_cache_key = None
2249 # One path for keyless and ineligible commands alike, so an eligible read whose
2250 # command method never plumbed ``keys=`` still gets its optout ``NO``. That is the
2251 # correct direction: the server tracks it regardless, and we will not store it.
2252 #
2253 # Trackability is asked only here, and never affects storage. A ``NO`` in front of
2254 # a command the server would not track - a write, a keyless read - is consumed
2255 # with no effect, so the only cost of getting it wrong is a wasted command; the
2256 # check fails closed, which skips the ``NO`` rather than risking a stored reply.
2257 if self._tracking_mode is TrackingMode.OPTOUT and (
2258 self._cache.config.is_trackable_read(command)
2259 ):
2260 self._send_with_caching(b"NO", args, kwargs)
2261 else:
2262 self._conn.send_command(*args, **kwargs)
2263 return
2265 # Creates cache key.
2266 self._current_command_cache_key = CacheKey(
2267 command=command, redis_keys=keys, redis_args=args
2268 )
2270 with self._cache_lock:
2271 # We have to trigger invalidation processing in case if
2272 # it was cached by another connection to avoid
2273 # queueing invalidations in stale connections.
2274 #
2275 # Only an entry we might serve is drained. An IN_PROGRESS one is never served and
2276 # is overwritten by our own placeholder below, so draining cannot change the
2277 # result - while its ``connection_ref`` socket may hold the owner's reply pair,
2278 # or another command's reply if the placeholder was stranded, and the drain
2279 # (which returns non-push replies too) would consume either one.
2280 entry = self._cache.get(self._current_command_cache_key)
2281 if entry is not None and entry.status != CacheEntryStatus.IN_PROGRESS:
2282 with self._pool_lock:
2283 self._drain_invalidations(entry.connection_ref)
2285 # Re-check: the entry may have been invalidated during the drain, or filled
2286 # in by the connection that was fetching it.
2287 entry = self._cache.get(self._current_command_cache_key)
2289 # Whether this read is served locally is decided here, once, and the value is
2290 # carried to ``read_response`` - which must not re-derive it from cache state.
2291 # Two things can happen to the entry between the two calls, and re-deriving
2292 # desynchronises the connection under both:
2293 #
2294 # - another connection resolves this very entry object and flips it to VALID with
2295 # *its* reply, so a re-derived check would serve that value and never read the
2296 # reply we did send;
2297 # - an invalidation removes it, so a re-derived check would find nothing and read
2298 # a reply for a command we never sent.
2299 #
2300 # An IN_PROGRESS entry is somebody else's fetch in flight and carries no value to
2301 # serve, so it is not a hit: fall through and send. Returning without sending
2302 # would leave ``read_response`` reading a reply that was never requested.
2303 if entry is not None and entry.status != CacheEntryStatus.IN_PROGRESS:
2304 self._pending_cache_hit = copy.deepcopy(entry.cache_value)
2305 self._current_command_cache_key = None
2306 return
2308 # Set temporary entry value to prevent
2309 # race condition from another connection.
2310 self._cache.set(
2311 CacheEntry(
2312 cache_key=self._current_command_cache_key,
2313 cache_value=self.DUMMY_CACHE_VALUE,
2314 status=CacheEntryStatus.IN_PROGRESS,
2315 connection_ref=self._conn,
2316 )
2317 )
2319 # Send command over socket only if it's allowed
2320 # read-only command that not yet cached. Under optin the server remembers nothing
2321 # unless ``CLIENT CACHING YES`` comes right before the read, so this - the one place
2322 # that decided to store the reply - is where the pair is sent. Under plain and optout
2323 # the read is tracked by default and goes out alone.
2324 if self._tracking_mode is TrackingMode.OPTIN:
2325 self._send_with_caching(b"YES", args, kwargs)
2326 else:
2327 self._conn.send_command(*args, **kwargs)
2329 def _send_with_caching(self, decision: bytes, args, kwargs) -> None:
2330 """
2331 Write ``CLIENT CACHING YES|NO`` and its read to the socket as one write.
2333 One write is the entire pairing guarantee: the server consumes the CACHING flag on the
2334 next command that is not a ``CLIENT`` subcommand, so anything the library slipped in
2335 between would take the flag instead of the read - under optin leaving the read
2336 untracked, under optout exempting the wrong command. Health-checking is left to
2337 ``send_packed_command``, which PINGs before it writes anything, so the PING cannot
2338 land inside the pair.
2340 Calls the wrapped connection's ``send_packed_command`` rather than this class's own
2341 override, which clears ``_current_command_cache_key`` and would discard the
2342 IN_PROGRESS placeholder the caller just set.
2344 Args:
2345 decision: ``b"YES"`` under optin, ``b"NO"`` under optout.
2346 args: The paired command, as ``send_command`` received it.
2347 kwargs: The paired command's keyword arguments, read for ``check_health``.
2348 """
2349 packed = self._conn.pack_commands([(b"CLIENT", b"CACHING", decision), args])
2350 self._conn.send_packed_command(
2351 packed, check_health=kwargs.get("check_health", True)
2352 )
2353 self._pending_caching_reply = True
2355 def can_read(self, timeout: float = 0) -> bool:
2356 # TODO: Rename this API; it detects pending data or dirty/closed
2357 # connection state, not only whether application data can be read.
2358 return self._conn.can_read(timeout)
2360 def read_response(
2361 self,
2362 disable_decoding=False,
2363 *,
2364 timeout: Union[float, object] = SENTINEL,
2365 disconnect_on_error=True,
2366 push_request=False,
2367 ):
2368 if self._pending_cache_hit is not _NO_PENDING_HIT:
2369 # ``send_command`` resolved this read from the cache and wrote nothing, so there
2370 # is no reply on the socket and the cache is deliberately not consulted again -
2371 # see the note there on why re-deriving the decision desynchronises the
2372 # connection.
2373 response = self._pending_cache_hit
2374 self._pending_cache_hit = _NO_PENDING_HIT
2375 record_csc_request(
2376 result=CSCResult.HIT,
2377 )
2378 record_csc_network_saved(
2379 bytes_saved=len(response) if hasattr(response, "__len__") else 0,
2380 )
2381 return response
2383 if self._current_command_cache_key is not None:
2384 # A key is still set, so ``send_command`` decided to store this reply and sent the
2385 # command: by construction a miss.
2386 record_csc_request(
2387 result=CSCResult.MISS,
2388 )
2390 try:
2391 if self._pending_caching_reply:
2392 # Consumed before the wire read below, so an interleaved invalidation cannot
2393 # be mistaken for either reply of the pair. Inside the ``try`` so a failure
2394 # here drops the placeholder exactly as a failed data read does.
2395 self._pending_caching_reply = False
2396 self._read_caching_reply(
2397 timeout=timeout, disconnect_on_error=disconnect_on_error
2398 )
2400 response = self._conn.read_response(
2401 disable_decoding=disable_decoding,
2402 timeout=timeout,
2403 disconnect_on_error=disconnect_on_error,
2404 push_request=push_request,
2405 )
2406 except BaseException:
2407 # The placeholder ``send_command`` staked is resolved only by the read that just
2408 # failed, so it has to go with it. Left behind, it sits in the pool-wide cache
2409 # until an invalidation or a disconnect happens to clear it: nothing else will,
2410 # because this command never completed and so the server is tracking nothing for
2411 # it. A ``WRONGTYPE`` or ``NOPERM`` reply is enough to strand one, as is the
2412 # ``-ASK`` on the first attempt of a redirected read.
2413 self._drop_own_placeholder()
2414 raise
2416 with self._cache_lock:
2417 # Prevent not-allowed command from caching.
2418 if self._current_command_cache_key is None:
2419 return response
2420 # If response is None prevent from caching. Only our own placeholder is
2421 # dropped: the entry may by now belong to another connection.
2422 if response is None:
2423 self._drop_own_placeholder()
2424 return response
2426 cache_entry = self._cache.get(self._current_command_cache_key)
2428 # Promote only the placeholder this connection staked. It may have been
2429 # invalidated in the meantime, or replaced by another connection's placeholder
2430 # for the same key - promoting that one would bind our reply to its
2431 # ``connection_ref``, and a later hit would drain that connection's own reply.
2432 if (
2433 cache_entry is not None
2434 and cache_entry.status == CacheEntryStatus.IN_PROGRESS
2435 and cache_entry.connection_ref is self._conn
2436 ):
2437 cache_entry.status = CacheEntryStatus.VALID
2438 cache_entry.cache_value = response
2439 self._cache.set(cache_entry)
2441 self._current_command_cache_key = None
2443 return response
2445 def _drop_own_placeholder(self) -> None:
2446 """
2447 Drop the ``IN_PROGRESS`` placeholder this connection staked, if it is still ours.
2449 Called when the read that would have resolved the placeholder fails. The cache is
2450 shared by the whole pool while each proxy's lock is its own, so the entry under this
2451 key may no longer be the one this connection put there, and the delete is scoped to
2452 the two conditions that prove it is:
2454 - still ``IN_PROGRESS`` - another connection's successful read can resolve this very
2455 entry to ``VALID`` in place, and deleting it would discard a correctly stored reply;
2456 - ``connection_ref`` is this connection - another connection's ``send_command`` for
2457 the same key replaces the entry with its own placeholder, and deleting it would
2458 evict a fetch that is still in flight.
2460 Clears ``_current_command_cache_key`` either way: the read it pointed at is over.
2461 """
2462 key = self._current_command_cache_key
2463 if key is None:
2464 return
2466 with self._cache_lock:
2467 entry = self._cache.get(key)
2468 if (
2469 entry is not None
2470 and entry.status == CacheEntryStatus.IN_PROGRESS
2471 and entry.connection_ref is self._conn
2472 ):
2473 self._cache.delete_by_cache_keys([key])
2475 self._current_command_cache_key = None
2477 def _read_caching_reply(self, *, timeout, disconnect_on_error) -> None:
2478 """
2479 Consume the ``+OK`` of a ``CLIENT CACHING`` command sent ahead of a read.
2481 Called inside ``read_response``'s ``try``, whose handlers clean up after any failure
2482 here the same way they do after a failed data read. A ``ConnectionError`` fails both
2483 commands together - never the read alone, whose reply would otherwise be stored
2484 without the tracking that makes it safe. The wrapped socket is closed, but the cache
2485 is flushed only when the client's error handling then disconnects this connection,
2486 or when the next connect re-enables tracking.
2488 Args:
2489 timeout: The caller's bound for this read.
2490 disconnect_on_error: Passed through to the wrapped connection.
2492 Raises:
2493 ResponseError: Re-raised after the paired read's reply has been drained.
2494 ConnectionError: If the server answered something other than ``OK``.
2495 """
2496 try:
2497 reply = self._conn.read_response(
2498 timeout=timeout, disconnect_on_error=disconnect_on_error
2499 )
2500 except ResponseError:
2501 # The paired read executed on the server regardless. Leaving its reply on the
2502 # socket would return this connection to the pool one reply out of sync, and the
2503 # next borrower would read our answer. The placeholder is dropped by the caller's
2504 # ``except BaseException``, as for any other failed read.
2505 self._conn.read_response(
2506 timeout=timeout, disconnect_on_error=disconnect_on_error
2507 )
2508 raise
2510 if str_if_bytes(reply) != "OK":
2511 raise ConnectionError(f"Unexpected CLIENT CACHING reply: {reply!r}")
2513 def pack_command(self, *args):
2514 return self._conn.pack_command(*args)
2516 def pack_commands(self, commands):
2517 # Every pipeline and transaction packs through here, so the guard covers them as
2518 # well. The cache's own commands are unaffected: ``_send_with_caching`` packs
2519 # through the wrapped connection, and the tracking handshake is sent on it.
2520 for args in commands:
2521 self._refuse_cache_owned_command(args)
2522 return self._conn.pack_commands(commands)
2524 @staticmethod
2525 def _refuse_cache_owned_command(args) -> None:
2526 """
2527 Refuse a user-sent ``CLIENT CACHING``, ``CLIENT TRACKING`` or ``RESET`` on a
2528 connection the cache manages.
2530 Each of them changes the tracking state the cache relies on for invalidations:
2532 - ``CLIENT CACHING`` sets a flag the server applies to the next command on the
2533 socket, and a pooled connection promises nothing about which command that is.
2534 Under ``optout`` a stray ``NO`` would leave the next cached read untracked while
2535 its reply is still stored, so it would never be invalidated. The cache sends it
2536 itself, paired with the read it belongs to, whenever the tracking mode needs it.
2537 - ``CLIENT TRACKING`` turns tracking off or reconfigures it, so the cache would keep
2538 storing replies that no invalidation reaches. The cache enables tracking itself
2539 when the connection is established.
2540 - ``RESET`` clears tracking along with the rest of the connection state.
2542 Raises:
2543 RedisError: If ``args`` is one of these commands.
2544 """
2545 command = _cache_owned_command(args)
2546 if command is None:
2547 return
2548 if command == "CLIENT CACHING":
2549 reason = "the cache sends it itself, paired with the read it applies to"
2550 elif command == "CLIENT TRACKING":
2551 reason = (
2552 "the cache enables tracking itself, and changing it would stop "
2553 "invalidations for cached replies"
2554 )
2555 else:
2556 reason = "it turns off the tracking the cache relies on for invalidations"
2557 raise RedisError(
2558 f"{command} cannot be sent on a connection with client-side caching "
2559 f"enabled: {reason}"
2560 )
2562 # HIMPORT state lives on the wrapped connection (HIMPORT is never cacheable);
2563 # delegate so callers treat the proxy like a plain connection and never need
2564 # to know a proxy is in play.
2565 @property
2566 def himport_registry(self):
2567 return self._conn.himport_registry
2569 @property
2570 def _himport_prepared(self):
2571 return self._conn._himport_prepared
2573 @_himport_prepared.setter
2574 def _himport_prepared(self, value):
2575 # Delegate reassignment to the wrapped connection, mirroring
2576 # ``_himport_reconciled_revision``. Production code only mutates the dict
2577 # in place, but ``_reset_himport_state`` (and any future caller) reassigns
2578 # it, and a getter-only property here would raise ``AttributeError`` only
2579 # when client-side caching is enabled -- a caching-specific latent trap.
2580 self._conn._himport_prepared = value
2582 @property
2583 def _himport_reconciled_revision(self):
2584 return self._conn._himport_reconciled_revision
2586 @_himport_reconciled_revision.setter
2587 def _himport_reconciled_revision(self, value):
2588 self._conn._himport_reconciled_revision = value
2590 @property
2591 def handshake_metadata(self) -> Union[Dict[bytes, bytes], Dict[str, str]]:
2592 return self._conn.handshake_metadata
2594 def set_re_auth_token(self, token: TokenInterface):
2595 self._conn.set_re_auth_token(token)
2597 def re_auth(self):
2598 self._conn.re_auth()
2600 def mark_for_reconnect(self):
2601 self._conn.mark_for_reconnect()
2603 def should_reconnect(self):
2604 return self._conn.should_reconnect()
2606 def reset_should_reconnect(self):
2607 self._conn.reset_should_reconnect()
2609 @property
2610 def host(self) -> str:
2611 return self._conn.host
2613 @host.setter
2614 def host(self, value: str):
2615 self._conn.host = value
2617 @property
2618 def socket_timeout(self) -> Optional[Union[float, int]]:
2619 return self._conn.socket_timeout
2621 @socket_timeout.setter
2622 def socket_timeout(self, value: Optional[Union[float, int]]):
2623 self._conn.socket_timeout = value
2625 @property
2626 def socket_connect_timeout(self) -> Optional[Union[float, int]]:
2627 return self._conn.socket_connect_timeout
2629 @socket_connect_timeout.setter
2630 def socket_connect_timeout(self, value: Optional[Union[float, int]]):
2631 self._conn.socket_connect_timeout = value
2633 @property
2634 def _maint_notifications_connection_handler(
2635 self,
2636 ) -> Optional[MaintNotificationsConnectionHandler]:
2637 if isinstance(self._conn, MaintNotificationsAbstractConnection):
2638 return self._conn._maint_notifications_connection_handler
2640 @_maint_notifications_connection_handler.setter
2641 def _maint_notifications_connection_handler(
2642 self, value: Optional[MaintNotificationsConnectionHandler]
2643 ):
2644 self._conn._maint_notifications_connection_handler = value
2646 def _get_socket(self) -> Optional[socket.socket]:
2647 if isinstance(self._conn, MaintNotificationsAbstractConnection):
2648 return self._conn._get_socket()
2649 else:
2650 raise NotImplementedError(
2651 "Maintenance notifications are not supported by this connection type"
2652 )
2654 def _get_parser(self) -> BaseParser:
2655 return self._conn._get_parser()
2657 def _get_maint_notifications_connection_instance(
2658 self, connection
2659 ) -> MaintNotificationsAbstractConnection:
2660 """
2661 Validate that connection instance supports maintenance notifications.
2662 With this helper method we ensure that we are working
2663 with the correct connection type.
2664 After twe validate that connection instance supports maintenance notifications
2665 we can safely return the connection instance
2666 as MaintNotificationsAbstractConnection.
2667 """
2668 if not isinstance(connection, MaintNotificationsAbstractConnection):
2669 raise NotImplementedError(
2670 "Maintenance notifications are not supported by this connection type"
2671 )
2672 else:
2673 return connection
2675 @property
2676 def maintenance_state(self) -> MaintenanceState:
2677 con = self._get_maint_notifications_connection_instance(self._conn)
2678 return con.maintenance_state
2680 @maintenance_state.setter
2681 def maintenance_state(self, state: MaintenanceState):
2682 con = self._get_maint_notifications_connection_instance(self._conn)
2683 con.maintenance_state = state
2685 def getpeername(self):
2686 con = self._get_maint_notifications_connection_instance(self._conn)
2687 return con.getpeername()
2689 def get_resolved_ip(self):
2690 con = self._get_maint_notifications_connection_instance(self._conn)
2691 return con.get_resolved_ip()
2693 def update_current_socket_timeout(self, relaxed_timeout: Optional[float] = None):
2694 con = self._get_maint_notifications_connection_instance(self._conn)
2695 con.update_current_socket_timeout(relaxed_timeout)
2697 def set_tmp_settings(
2698 self,
2699 tmp_host_address: Optional[str] = None,
2700 tmp_relaxed_timeout: Optional[float] = -1,
2701 ):
2702 con = self._get_maint_notifications_connection_instance(self._conn)
2703 con.set_tmp_settings(tmp_host_address, tmp_relaxed_timeout)
2705 def reset_tmp_settings(
2706 self,
2707 reset_host_address: bool = False,
2708 reset_relaxed_timeout: bool = False,
2709 ):
2710 con = self._get_maint_notifications_connection_instance(self._conn)
2711 con.reset_tmp_settings(reset_host_address, reset_relaxed_timeout)
2713 def _connect(self):
2714 self._conn._connect()
2716 def _host_error(self):
2717 return self._conn._host_error()
2719 def _enable_tracking_callback(self, conn: ConnectionInterface) -> None:
2720 if self._connected_once:
2721 # The server destroys a connection's tracking state on disconnect, so every entry
2722 # cached through the previous session has lost its invalidation channel. Paths
2723 # that reconnect silently inside the send path never call ``disconnect``, so the
2724 # flush cannot live there alone.
2725 #
2726 # Gated on a previous connect because this callback also fires on a first connect
2727 # and the cache is pool-shared: an unconditional flush would wipe other
2728 # connections' entries every time the pool grows. Flushing the whole shared cache
2729 # on reconnect is what ``disconnect`` already does, so the semantics do not
2730 # change - only the coverage does.
2731 #
2732 # The one entry kept is this connection's own in-flight placeholder. Every
2733 # reconnect that fires this callback happens inside the send path, before the
2734 # command is written, so the reply it is waiting for comes from the new session -
2735 # which this callback is about to enable tracking for. Dropping it would cost the
2736 # fill and gain nothing.
2737 with self._cache_lock:
2738 own_placeholder = None
2739 if self._current_command_cache_key is not None:
2740 entry = self._cache.get(self._current_command_cache_key)
2741 if (
2742 entry is not None
2743 and entry.status == CacheEntryStatus.IN_PROGRESS
2744 and entry.connection_ref is self._conn
2745 ):
2746 own_placeholder = entry
2747 self._cache.flush()
2748 if own_placeholder is not None:
2749 self._cache.set(own_placeholder)
2751 self._pending_caching_reply = False
2753 # The mode is sent in the handshake because that is what the server requires:
2754 # switching OPTIN <-> OPTOUT on a live connection is an error, so a configuration
2755 # change applies to new connections only.
2756 args = ["CLIENT", "TRACKING", "ON"]
2757 if self._tracking_mode is TrackingMode.OPTIN:
2758 args.append("OPTIN")
2759 elif self._tracking_mode is TrackingMode.OPTOUT:
2760 args.append("OPTOUT")
2762 conn.send_command(*args)
2763 conn.read_response()
2764 # Set only once tracking is on: a handshake that fails before this point cached
2765 # nothing, so the next connect has no session to flush for.
2766 self._connected_once = True
2767 conn._parser.set_invalidation_push_handler(self._on_invalidation_callback)
2769 def _drain_invalidations(self, conn: ConnectionInterface):
2770 while conn.can_read():
2771 try:
2772 conn.read_response(
2773 push_request=True, timeout=0, disconnect_on_error=False
2774 )
2775 except TimeoutError:
2776 break
2777 except UNRECOVERABLE_PARSE_ERRORS:
2778 # Invalidation replies are read straight off the raw
2779 # connection, so the disconnect a parse failure now forces
2780 # bypasses this proxy's disconnect() and its cache flush. The
2781 # next connect() opens a fresh CLIENT TRACKING session that the
2782 # server has no invalidation state for, so entries cached under
2783 # the old session would be served as if still tracked. Flush
2784 # them here for the same reason disconnect() does.
2785 with self._cache_lock:
2786 self._cache.flush()
2787 raise
2789 def _process_pending_invalidations(self):
2790 self._drain_invalidations(self._conn)
2792 def _on_invalidation_callback(self, data: List[Union[str, Optional[List[bytes]]]]):
2793 with self._cache_lock:
2794 # Flush cache when DB flushed on server-side
2795 if data[1] is None:
2796 self._cache.flush()
2797 else:
2798 keys_deleted = self._cache.delete_by_redis_keys(data[1])
2800 if len(keys_deleted) > 0:
2801 record_csc_eviction(
2802 count=len(keys_deleted),
2803 reason=CSCReason.INVALIDATION,
2804 )
2806 def extract_connection_details(self) -> str:
2807 return self._conn.extract_connection_details()
2810class SSLConnection(Connection):
2811 """Manages SSL connections to and from the Redis server(s).
2812 This class extends the Connection class, adding SSL functionality, and making
2813 use of ssl.SSLContext (https://docs.python.org/3/library/ssl.html#ssl.SSLContext)
2814 """ # noqa
2816 def __init__(
2817 self,
2818 ssl_keyfile=None,
2819 ssl_certfile=None,
2820 ssl_cert_reqs="required",
2821 ssl_include_verify_flags: Optional[List["VerifyFlags"]] = None,
2822 ssl_exclude_verify_flags: Optional[List["VerifyFlags"]] = None,
2823 ssl_ca_certs=None,
2824 ssl_ca_data=None,
2825 ssl_check_hostname=True,
2826 ssl_ca_path=None,
2827 ssl_password=None,
2828 ssl_validate_ocsp=False,
2829 ssl_validate_ocsp_stapled=False,
2830 ssl_ocsp_context=None,
2831 ssl_ocsp_expected_cert=None,
2832 ssl_min_version=None,
2833 ssl_ciphers=None,
2834 **kwargs,
2835 ):
2836 """Constructor
2838 Args:
2839 ssl_keyfile: Path to an ssl private key. Defaults to None.
2840 ssl_certfile: Path to an ssl certificate. Defaults to None.
2841 ssl_cert_reqs: The string value for the SSLContext.verify_mode (none, optional, required),
2842 or an ssl.VerifyMode. Defaults to "required".
2843 ssl_include_verify_flags: A list of flags to be included in the SSLContext.verify_flags. Defaults to None.
2844 ssl_exclude_verify_flags: A list of flags to be excluded from the SSLContext.verify_flags. Defaults to None.
2845 ssl_ca_certs: The path to a file of concatenated CA certificates in PEM format. Defaults to None.
2846 ssl_ca_data: Either an ASCII string of one or more PEM-encoded certificates or a bytes-like object of DER-encoded certificates.
2847 ssl_check_hostname: If set, match the hostname during the SSL handshake. Defaults to True.
2848 ssl_ca_path: The path to a directory containing several CA certificates in PEM format. Defaults to None.
2849 ssl_password: Password for unlocking an encrypted private key. Defaults to None.
2851 ssl_validate_ocsp: If set, perform a full ocsp validation (i.e not a stapled verification)
2852 ssl_validate_ocsp_stapled: If set, perform a validation on a stapled ocsp response
2853 ssl_ocsp_context: A fully initialized OpenSSL.SSL.Context object to be used in verifying the ssl_ocsp_expected_cert
2854 ssl_ocsp_expected_cert: A PEM armoured string containing the expected certificate to be returned from the ocsp verification service.
2855 ssl_min_version: The lowest supported SSL version. It affects the supported SSL versions of the SSLContext. None leaves the default provided by ssl module.
2856 ssl_ciphers: A string listing the ciphers that are allowed to be used. Defaults to None, which means that the default ciphers are used. See https://docs.python.org/3/library/ssl.html#ssl.SSLContext.set_ciphers for more information.
2858 Raises:
2859 RedisError
2860 """ # noqa
2861 if not SSL_AVAILABLE:
2862 raise RedisError("Python wasn't built with SSL support")
2864 self.keyfile = ssl_keyfile
2865 self.certfile = ssl_certfile
2866 if ssl_cert_reqs is None:
2867 ssl_cert_reqs = ssl.CERT_NONE
2868 elif isinstance(ssl_cert_reqs, str):
2869 CERT_REQS = { # noqa: N806
2870 "none": ssl.CERT_NONE,
2871 "optional": ssl.CERT_OPTIONAL,
2872 "required": ssl.CERT_REQUIRED,
2873 }
2874 if ssl_cert_reqs not in CERT_REQS:
2875 raise RedisError(
2876 f"Invalid SSL Certificate Requirements Flag: {ssl_cert_reqs}"
2877 )
2878 ssl_cert_reqs = CERT_REQS[ssl_cert_reqs]
2879 self.cert_reqs = ssl_cert_reqs
2880 self.ssl_include_verify_flags = ssl_include_verify_flags
2881 self.ssl_exclude_verify_flags = ssl_exclude_verify_flags
2882 self.ca_certs = ssl_ca_certs
2883 self.ca_data = ssl_ca_data
2884 self.ca_path = ssl_ca_path
2885 self.check_hostname = (
2886 ssl_check_hostname if self.cert_reqs != ssl.CERT_NONE else False
2887 )
2888 self.certificate_password = ssl_password
2889 self.ssl_validate_ocsp = ssl_validate_ocsp
2890 self.ssl_validate_ocsp_stapled = ssl_validate_ocsp_stapled
2891 self.ssl_ocsp_context = ssl_ocsp_context
2892 self.ssl_ocsp_expected_cert = ssl_ocsp_expected_cert
2893 self.ssl_min_version = ssl_min_version
2894 self.ssl_ciphers = ssl_ciphers
2895 super().__init__(**kwargs)
2897 def _connect(self):
2898 """
2899 Wrap the socket with SSL support, handling potential errors.
2900 """
2901 sock = super()._connect()
2902 try:
2903 return self._wrap_socket_with_ssl(sock)
2904 except (OSError, RedisError):
2905 sock.close()
2906 raise
2908 def _wrap_socket_with_ssl(self, sock):
2909 """
2910 Wraps the socket with SSL support.
2912 Args:
2913 sock: The plain socket to wrap with SSL.
2915 Returns:
2916 An SSL wrapped socket.
2917 """
2918 context = ssl.create_default_context()
2919 context.check_hostname = self.check_hostname
2920 context.verify_mode = self.cert_reqs
2921 if self.ssl_include_verify_flags:
2922 for flag in self.ssl_include_verify_flags:
2923 context.verify_flags |= flag
2924 if self.ssl_exclude_verify_flags:
2925 for flag in self.ssl_exclude_verify_flags:
2926 context.verify_flags &= ~flag
2927 if self.certfile or self.keyfile:
2928 context.load_cert_chain(
2929 certfile=self.certfile,
2930 keyfile=self.keyfile,
2931 password=self.certificate_password,
2932 )
2933 if (
2934 self.ca_certs is not None
2935 or self.ca_path is not None
2936 or self.ca_data is not None
2937 ):
2938 context.load_verify_locations(
2939 cafile=self.ca_certs, capath=self.ca_path, cadata=self.ca_data
2940 )
2941 if self.ssl_min_version is not None:
2942 context.minimum_version = self.ssl_min_version
2943 if self.ssl_ciphers:
2944 context.set_ciphers(self.ssl_ciphers)
2945 if self.ssl_validate_ocsp is True and CRYPTOGRAPHY_AVAILABLE is False:
2946 raise RedisError("cryptography is not installed.")
2948 if self.ssl_validate_ocsp_stapled and self.ssl_validate_ocsp:
2949 raise RedisError(
2950 "Either an OCSP staple or pure OCSP connection must be validated "
2951 "- not both."
2952 )
2954 sslsock = context.wrap_socket(sock, server_hostname=self.host)
2956 # validation for the stapled case
2957 if self.ssl_validate_ocsp_stapled:
2958 import OpenSSL
2960 from .ocsp import ocsp_staple_verifier
2962 # if a context is provided use it - otherwise, a basic context
2963 if self.ssl_ocsp_context is None:
2964 staple_ctx = OpenSSL.SSL.Context(OpenSSL.SSL.SSLv23_METHOD)
2965 staple_ctx.use_certificate_file(self.certfile)
2966 staple_ctx.use_privatekey_file(self.keyfile)
2967 else:
2968 staple_ctx = self.ssl_ocsp_context
2970 staple_ctx.set_ocsp_client_callback(
2971 ocsp_staple_verifier, self.ssl_ocsp_expected_cert
2972 )
2974 # need another socket
2975 con = OpenSSL.SSL.Connection(staple_ctx, socket.socket())
2976 con.request_ocsp()
2977 con.connect((self.host, self.port))
2978 con.do_handshake()
2979 con.shutdown()
2980 return sslsock
2982 # pure ocsp validation
2983 if self.ssl_validate_ocsp is True and CRYPTOGRAPHY_AVAILABLE:
2984 from .ocsp import OCSPVerifier
2986 o = OCSPVerifier(sslsock, self.host, self.port, self.ca_certs)
2987 if o.is_valid():
2988 return sslsock
2989 else:
2990 raise ConnectionError("ocsp validation error")
2991 return sslsock
2994class UnixDomainSocketConnection(AbstractConnection):
2995 "Manages UDS communication to and from a Redis server"
2997 def __init__(self, path="", socket_timeout=DEFAULT_SOCKET_TIMEOUT, **kwargs):
2998 super().__init__(**kwargs)
2999 self.path = path
3000 self.socket_timeout = socket_timeout
3002 def repr_pieces(self):
3003 pieces = [("path", self.path), ("db", self.db)]
3004 if self.client_name:
3005 pieces.append(("client_name", self.client_name))
3006 return pieces
3008 def _connect(self):
3009 "Create a Unix domain socket connection"
3010 sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
3011 sock.settimeout(self.socket_connect_timeout)
3012 try:
3013 sock.connect(self.path)
3014 except OSError:
3015 # Prevent ResourceWarnings for unclosed sockets.
3016 try:
3017 sock.shutdown(socket.SHUT_RDWR) # ensure a clean close
3018 except OSError:
3019 pass
3020 sock.close()
3021 raise
3022 sock.settimeout(self.socket_timeout)
3023 return sock
3025 def _host_error(self):
3026 return self.path
3029FALSE_STRINGS = ("0", "F", "FALSE", "N", "NO", "OFF")
3032def to_bool(value):
3033 if value is None or value == "":
3034 return None
3035 if isinstance(value, str) and value.upper() in FALSE_STRINGS:
3036 return False
3037 return bool(value)
3040def parse_ssl_verify_flags(value):
3041 # flags are passed in as a string representation of a list,
3042 # e.g. VERIFY_X509_STRICT, VERIFY_X509_PARTIAL_CHAIN
3043 verify_flags_str = value.replace("[", "").replace("]", "")
3045 verify_flags = []
3046 for flag in verify_flags_str.split(","):
3047 flag = flag.strip()
3048 if not hasattr(VerifyFlags, flag):
3049 raise ValueError(f"Invalid ssl verify flag: {flag}")
3050 verify_flags.append(getattr(VerifyFlags, flag))
3051 return verify_flags
3054def parse_retry_on_error(value):
3055 # exception class names are passed as a comma-separated list,
3056 # e.g. ConnectionError,TimeoutError
3057 retry_on_error = []
3058 for name in value.replace("[", "").replace("]", "").split(","):
3059 name = name.strip()
3060 if not name:
3061 raise ValueError("Empty retry_on_error entry")
3062 exc = getattr(redis_exceptions, name, None)
3063 if not (isinstance(exc, type) and issubclass(exc, Exception)):
3064 raise ValueError(f"Unknown redis exception {name!r}")
3065 retry_on_error.append(exc)
3066 return retry_on_error
3069URL_QUERY_ARGUMENT_PARSERS = {
3070 "db": int,
3071 "socket_timeout": float,
3072 "socket_connect_timeout": float,
3073 "socket_read_size": int,
3074 "socket_keepalive": to_bool,
3075 "retry_on_timeout": to_bool,
3076 "retry_on_error": parse_retry_on_error,
3077 "max_connections": int,
3078 "health_check_interval": int,
3079 "ssl_check_hostname": to_bool,
3080 "ssl_include_verify_flags": parse_ssl_verify_flags,
3081 "ssl_exclude_verify_flags": parse_ssl_verify_flags,
3082 "ssl_min_version": int,
3083 "timeout": float,
3084 "protocol": int,
3085 "legacy_responses": to_bool,
3086}
3089def parse_url(url):
3090 # Scheme names are case-insensitive (RFC 3986), so normalize before the
3091 # prefix check; the "://" is required so a URL like "redis:foo" (which
3092 # urlparse would still report as the "redis" scheme) is rejected.
3093 if not url.lower().startswith(("redis://", "rediss://", "unix://")):
3094 raise ValueError(
3095 "Redis URL must specify one of the following "
3096 "schemes (redis://, rediss://, unix://)"
3097 )
3099 url = urlparse(url)
3100 kwargs = {}
3102 for name, value in parse_qs(url.query).items():
3103 if value and len(value) > 0:
3104 # parse_qs() already percent-decodes query values, so use the value
3105 # as-is; unquoting again here would double-decode (e.g. "%2520" ->
3106 # "%20" -> " "). See issue #4208.
3107 value = value[0]
3108 parser = URL_QUERY_ARGUMENT_PARSERS.get(name)
3109 if parser:
3110 try:
3111 kwargs[name] = parser(value)
3112 except (TypeError, ValueError):
3113 raise ValueError(f"Invalid value for '{name}' in connection URL.")
3114 else:
3115 kwargs[name] = value
3117 if url.username:
3118 kwargs["username"] = unquote(url.username)
3119 if url.password:
3120 kwargs["password"] = unquote(url.password)
3122 # We only support redis://, rediss:// and unix:// schemes.
3123 if url.scheme == "unix":
3124 if url.path:
3125 kwargs["path"] = unquote(url.path)
3126 kwargs["connection_class"] = UnixDomainSocketConnection
3128 else: # implied: url.scheme in ("redis", "rediss"):
3129 if url.hostname:
3130 kwargs["host"] = unquote(url.hostname)
3131 if url.port is not None:
3132 kwargs["port"] = int(url.port)
3134 # If there's a path argument, use it as the db argument if a
3135 # querystring value wasn't specified
3136 if url.path and "db" not in kwargs:
3137 try:
3138 kwargs["db"] = int(unquote(url.path).replace("/", ""))
3139 except (AttributeError, ValueError):
3140 pass
3142 if url.scheme == "rediss":
3143 kwargs["connection_class"] = SSLConnection
3145 return kwargs
3148_CP = TypeVar("_CP", bound="ConnectionPool")
3151class ConnectionPoolInterface(ABC):
3152 @abstractmethod
3153 def get_protocol(self):
3154 pass
3156 @abstractmethod
3157 def reset(self):
3158 pass
3160 @abstractmethod
3161 @deprecated_args(
3162 args_to_warn=["*"],
3163 reason="Use get_connection() without args instead",
3164 version="5.3.0",
3165 )
3166 def get_connection(
3167 self, command_name: Optional[str], *keys, **options
3168 ) -> ConnectionInterface:
3169 pass
3171 @abstractmethod
3172 def get_encoder(self):
3173 pass
3175 @abstractmethod
3176 def release(self, connection: ConnectionInterface):
3177 pass
3179 @abstractmethod
3180 def disconnect(self, inuse_connections: bool = True):
3181 pass
3183 @abstractmethod
3184 def close(self):
3185 pass
3187 @abstractmethod
3188 def set_retry(self, retry: Retry):
3189 pass
3191 @abstractmethod
3192 def re_auth_callback(self, token: TokenInterface):
3193 pass
3195 @abstractmethod
3196 def get_connection_count(self) -> list[tuple[int, dict]]:
3197 """
3198 Returns a connection count (both idle and in use).
3199 """
3200 pass
3203class MaintNotificationsAbstractConnectionPool:
3204 """
3205 Abstract class for handling maintenance notifications logic.
3206 This class is mixed into the ConnectionPool classes.
3208 This class is not intended to be used directly!
3210 All logic related to maintenance notifications and
3211 connection pool handling is encapsulated in this class.
3212 """
3214 def __init__(
3215 self,
3216 maint_notifications_config: Optional[MaintNotificationsConfig] = None,
3217 oss_cluster_maint_notifications_handler: Optional[
3218 OSSMaintNotificationsHandler
3219 ] = None,
3220 **kwargs,
3221 ):
3222 # Initialize maintenance notifications
3223 is_protocol_supported = check_protocol_version(kwargs.get("protocol"), 3)
3225 if maint_notifications_config is None and is_protocol_supported:
3226 maint_notifications_config = MaintNotificationsConfig()
3228 if maint_notifications_config and maint_notifications_config.enabled:
3229 if not is_protocol_supported:
3230 raise RedisError(
3231 "Maintenance notifications handlers on connection are only supported with RESP version 3"
3232 )
3234 self._event_dispatcher = kwargs.get("event_dispatcher", None)
3235 if self._event_dispatcher is None:
3236 self._event_dispatcher = EventDispatcher()
3238 self._maint_notifications_pool_handler = MaintNotificationsPoolHandler(
3239 self, maint_notifications_config
3240 )
3241 if oss_cluster_maint_notifications_handler:
3242 self._oss_cluster_maint_notifications_handler = (
3243 oss_cluster_maint_notifications_handler
3244 )
3245 self._update_connection_kwargs_for_maint_notifications(
3246 oss_cluster_maint_notifications_handler=self._oss_cluster_maint_notifications_handler
3247 )
3248 self._maint_notifications_pool_handler = None
3249 else:
3250 self._oss_cluster_maint_notifications_handler = None
3251 self._maint_notifications_pool_handler = MaintNotificationsPoolHandler(
3252 self, maint_notifications_config
3253 )
3255 self._update_connection_kwargs_for_maint_notifications(
3256 maint_notifications_pool_handler=self._maint_notifications_pool_handler
3257 )
3258 else:
3259 self._maint_notifications_pool_handler = None
3260 self._oss_cluster_maint_notifications_handler = None
3262 @property
3263 @abstractmethod
3264 def connection_kwargs(self) -> Dict[str, Any]:
3265 pass
3267 @connection_kwargs.setter
3268 @abstractmethod
3269 def connection_kwargs(self, value: Dict[str, Any]):
3270 pass
3272 @abstractmethod
3273 def _get_pool_lock(self) -> threading.RLock:
3274 pass
3276 @abstractmethod
3277 def _get_free_connections(self) -> Iterable["MaintNotificationsAbstractConnection"]:
3278 pass
3280 @abstractmethod
3281 def _get_in_use_connections(
3282 self,
3283 ) -> Iterable["MaintNotificationsAbstractConnection"]:
3284 pass
3286 def maint_notifications_enabled(self):
3287 """
3288 Returns:
3289 True if the maintenance notifications are enabled, False otherwise.
3290 The maintenance notifications config is stored in the pool handler.
3291 If the pool handler is not set, the maintenance notifications are not enabled.
3292 """
3293 if self._oss_cluster_maint_notifications_handler:
3294 maint_notifications_config = (
3295 self._oss_cluster_maint_notifications_handler.config
3296 )
3297 else:
3298 maint_notifications_config = (
3299 self._maint_notifications_pool_handler.config
3300 if self._maint_notifications_pool_handler
3301 else None
3302 )
3304 return maint_notifications_config and maint_notifications_config.enabled
3306 def update_maint_notifications_config(
3307 self,
3308 maint_notifications_config: MaintNotificationsConfig,
3309 oss_cluster_maint_notifications_handler: Optional[
3310 OSSMaintNotificationsHandler
3311 ] = None,
3312 ):
3313 """
3314 Updates the maintenance notifications configuration.
3315 This method should be called only if the pool was created
3316 without enabling the maintenance notifications and
3317 in a later point in time maintenance notifications
3318 are requested to be enabled.
3319 """
3320 if (
3321 self.maint_notifications_enabled()
3322 and not maint_notifications_config.enabled
3323 ):
3324 raise ValueError(
3325 "Cannot disable maintenance notifications after enabling them"
3326 )
3327 if oss_cluster_maint_notifications_handler:
3328 self._oss_cluster_maint_notifications_handler = (
3329 oss_cluster_maint_notifications_handler
3330 )
3331 # OSS cluster mode and pool-handler mode are mutually exclusive
3332 # (see __init__). A pool created with the default RESP3 "auto"
3333 # config wires a pool handler before this method runs; clear it so
3334 # new and existing connections are not configured with both handlers.
3335 self._maint_notifications_pool_handler = None
3336 else:
3337 # first update pool settings
3338 if self._oss_cluster_maint_notifications_handler:
3339 # Pool already in OSS cluster mode; update the OSS handler config
3340 # instead of creating a mutually-exclusive pool handler (which
3341 # would be silently ignored because the OSS handler wins priority
3342 # in both update helpers below).
3343 self._oss_cluster_maint_notifications_handler.config = (
3344 maint_notifications_config
3345 )
3346 elif not self._maint_notifications_pool_handler:
3347 self._maint_notifications_pool_handler = MaintNotificationsPoolHandler(
3348 self, maint_notifications_config
3349 )
3350 else:
3351 self._maint_notifications_pool_handler.config = (
3352 maint_notifications_config
3353 )
3355 # then update connection kwargs and existing connections
3356 self._update_connection_kwargs_for_maint_notifications(
3357 maint_notifications_pool_handler=self._maint_notifications_pool_handler,
3358 oss_cluster_maint_notifications_handler=self._oss_cluster_maint_notifications_handler,
3359 )
3360 self._update_maint_notifications_configs_for_connections(
3361 maint_notifications_pool_handler=self._maint_notifications_pool_handler,
3362 oss_cluster_maint_notifications_handler=self._oss_cluster_maint_notifications_handler,
3363 )
3365 def _update_connection_kwargs_for_maint_notifications(
3366 self,
3367 maint_notifications_pool_handler: Optional[
3368 MaintNotificationsPoolHandler
3369 ] = None,
3370 oss_cluster_maint_notifications_handler: Optional[
3371 OSSMaintNotificationsHandler
3372 ] = None,
3373 ):
3374 """
3375 Update the connection kwargs for all future connections.
3376 """
3377 if not self.maint_notifications_enabled():
3378 return
3379 if maint_notifications_pool_handler:
3380 self.connection_kwargs.update(
3381 {
3382 "maint_notifications_pool_handler": maint_notifications_pool_handler,
3383 "maint_notifications_config": maint_notifications_pool_handler.config,
3384 }
3385 )
3386 if oss_cluster_maint_notifications_handler:
3387 self.connection_kwargs.update(
3388 {
3389 "oss_cluster_maint_notifications_handler": oss_cluster_maint_notifications_handler,
3390 "maint_notifications_config": oss_cluster_maint_notifications_handler.config,
3391 }
3392 )
3393 # OSS cluster mode and pool-handler mode are mutually exclusive.
3394 # Drop any pool handler a default (RESP3 "auto") pool creation may
3395 # have wired so future connections are not configured with both.
3396 self.connection_kwargs.pop("maint_notifications_pool_handler", None)
3398 # Store original connection parameters for maintenance notifications.
3399 if self.connection_kwargs.get("orig_host_address", None) is None:
3400 # If orig_host_address is None it means we haven't
3401 # configured the original values yet
3402 self.connection_kwargs.update(
3403 {
3404 "orig_host_address": self.connection_kwargs.get("host"),
3405 "orig_socket_timeout": self.connection_kwargs.get(
3406 "socket_timeout", DEFAULT_SOCKET_TIMEOUT
3407 ),
3408 "orig_socket_connect_timeout": self.connection_kwargs.get(
3409 "socket_connect_timeout", DEFAULT_SOCKET_CONNECT_TIMEOUT
3410 ),
3411 }
3412 )
3414 def _update_maint_notifications_configs_for_connections(
3415 self,
3416 maint_notifications_pool_handler: Optional[
3417 MaintNotificationsPoolHandler
3418 ] = None,
3419 oss_cluster_maint_notifications_handler: Optional[
3420 OSSMaintNotificationsHandler
3421 ] = None,
3422 ):
3423 """Update the maintenance notifications config for all connections in the pool."""
3424 with self._get_pool_lock():
3425 for conn in self._get_free_connections():
3426 if oss_cluster_maint_notifications_handler:
3427 # set cluster handler for conn
3428 conn.set_maint_notifications_cluster_handler_for_connection(
3429 oss_cluster_maint_notifications_handler
3430 )
3431 conn.maint_notifications_config = (
3432 oss_cluster_maint_notifications_handler.config
3433 )
3434 elif maint_notifications_pool_handler:
3435 conn.set_maint_notifications_pool_handler_for_connection(
3436 maint_notifications_pool_handler
3437 )
3438 conn.maint_notifications_config = (
3439 maint_notifications_pool_handler.config
3440 )
3441 else:
3442 raise ValueError(
3443 "Either maint_notifications_pool_handler or oss_cluster_maint_notifications_handler must be set"
3444 )
3445 conn.disconnect()
3446 for conn in self._get_in_use_connections():
3447 if oss_cluster_maint_notifications_handler:
3448 # Use set_maint_notifications_cluster_handler_for_connection
3449 # (not _configure_maintenance_notifications) so the parser is
3450 # obtained from the connection itself. _configure_* requires a
3451 # parser argument and would raise here; it would also reset the
3452 # connection's orig_* settings, which is wrong for an in-use
3453 # (active) connection. This mirrors the idle-connection branch
3454 # above and the pool-handler branches.
3455 conn.set_maint_notifications_cluster_handler_for_connection(
3456 oss_cluster_maint_notifications_handler
3457 )
3458 conn.maint_notifications_config = (
3459 oss_cluster_maint_notifications_handler.config
3460 )
3461 elif maint_notifications_pool_handler:
3462 conn.set_maint_notifications_pool_handler_for_connection(
3463 maint_notifications_pool_handler
3464 )
3465 conn.maint_notifications_config = (
3466 maint_notifications_pool_handler.config
3467 )
3468 else:
3469 raise ValueError(
3470 "Either maint_notifications_pool_handler or oss_cluster_maint_notifications_handler must be set"
3471 )
3472 if logger.isEnabledFor(logging.DEBUG):
3473 logger.debug(
3474 "Marking active connection for reconnect after config update "
3475 f"config update: {conn}, {conn.extract_connection_details()}"
3476 )
3477 conn.mark_for_reconnect()
3479 def _should_update_connection(
3480 self,
3481 conn: "MaintNotificationsAbstractConnection",
3482 matching_pattern: Literal[
3483 "connected_address", "configured_address", "notification_hash"
3484 ] = "connected_address",
3485 matching_address: Optional[str] = None,
3486 matching_notification_hash: Optional[int] = None,
3487 ) -> bool:
3488 """
3489 Check if the connection should be updated based on the matching criteria.
3490 """
3491 if matching_pattern == "connected_address":
3492 if matching_address and conn.getpeername() != matching_address:
3493 return False
3494 elif matching_pattern == "configured_address":
3495 if matching_address and conn.host != matching_address:
3496 return False
3497 elif matching_pattern == "notification_hash":
3498 if (
3499 matching_notification_hash is not None
3500 and conn.maintenance_notification_hash != matching_notification_hash
3501 ):
3502 return False
3503 return True
3505 def update_connection_settings(
3506 self,
3507 conn: "MaintNotificationsAbstractConnection",
3508 state: Optional["MaintenanceState"] = None,
3509 maintenance_notification_hash: Optional[int] = None,
3510 host_address: Optional[str] = None,
3511 relaxed_timeout: Optional[float] = None,
3512 update_notification_hash: bool = False,
3513 reset_host_address: bool = False,
3514 reset_relaxed_timeout: bool = False,
3515 ):
3516 """
3517 Update the settings for a single connection.
3518 """
3519 if state:
3520 self._complete_maintenance_taken_over_by_handoff(conn, state)
3521 conn.maintenance_state = state
3523 if update_notification_hash:
3524 # update the notification hash only if requested
3525 conn.maintenance_notification_hash = maintenance_notification_hash
3527 if host_address is not None:
3528 conn.set_tmp_settings(tmp_host_address=host_address)
3530 if relaxed_timeout is not None:
3531 conn.set_tmp_settings(tmp_relaxed_timeout=relaxed_timeout)
3533 if reset_relaxed_timeout or reset_host_address:
3534 conn.reset_tmp_settings(
3535 reset_host_address=reset_host_address,
3536 reset_relaxed_timeout=reset_relaxed_timeout,
3537 )
3539 conn.update_current_socket_timeout(relaxed_timeout)
3541 def _complete_maintenance_taken_over_by_handoff(
3542 self, conn: "MaintNotificationsAbstractConnection", state: "MaintenanceState"
3543 ) -> None:
3544 """
3545 Report the MAINTENANCE relaxation of a connection a handoff is taking
3546 over as completed. The connection's completion notification is skipped
3547 while it is MOVING, and the handoff's own cleanup restores it without
3548 looking at what it superseded, so this is the only point at which the
3549 pair opened by the start event can be closed.
3550 """
3551 if (
3552 state is not MaintenanceState.MOVING
3553 or conn.maintenance_state is not MaintenanceState.MAINTENANCE
3554 or conn.maint_notifications_config is None
3555 ):
3556 return
3557 _dispatch_maintenance_event(
3558 conn.maint_notifications_config,
3559 MaintenanceCompletedEvent(
3560 connection_pool=self,
3561 connection=conn,
3562 state=MaintenanceState.MAINTENANCE,
3563 notification=None,
3564 config=conn.maint_notifications_config,
3565 ),
3566 )
3568 def update_connections_settings(
3569 self,
3570 state: Optional["MaintenanceState"] = None,
3571 maintenance_notification_hash: Optional[int] = None,
3572 host_address: Optional[str] = None,
3573 relaxed_timeout: Optional[float] = None,
3574 matching_address: Optional[str] = None,
3575 matching_notification_hash: Optional[int] = None,
3576 matching_pattern: Literal[
3577 "connected_address", "configured_address", "notification_hash"
3578 ] = "connected_address",
3579 update_notification_hash: bool = False,
3580 reset_host_address: bool = False,
3581 reset_relaxed_timeout: bool = False,
3582 include_free_connections: bool = True,
3583 ):
3584 """
3585 Update the settings for all matching connections in the pool.
3587 This method does not create new connections.
3588 This method does not affect the connection kwargs.
3590 :param state: The maintenance state to set for the connection.
3591 :param maintenance_notification_hash: The hash of the maintenance notification
3592 to set for the connection.
3593 :param host_address: The host address to set for the connection.
3594 :param relaxed_timeout: The relaxed timeout to set for the connection.
3595 :param matching_address: The address to match for the connection.
3596 :param matching_notification_hash: The notification hash to match for the connection.
3597 :param matching_pattern: The pattern to match for the connection.
3598 :param update_notification_hash: Whether to update the notification hash for the connection.
3599 :param reset_host_address: Whether to reset the host address to the original address.
3600 :param reset_relaxed_timeout: Whether to reset the relaxed timeout to the original timeout.
3601 :param include_free_connections: Whether to include free/available connections.
3602 """
3603 with self._get_pool_lock():
3604 for conn in self._get_in_use_connections():
3605 if self._should_update_connection(
3606 conn,
3607 matching_pattern,
3608 matching_address,
3609 matching_notification_hash,
3610 ):
3611 self.update_connection_settings(
3612 conn,
3613 state=state,
3614 maintenance_notification_hash=maintenance_notification_hash,
3615 host_address=host_address,
3616 relaxed_timeout=relaxed_timeout,
3617 update_notification_hash=update_notification_hash,
3618 reset_host_address=reset_host_address,
3619 reset_relaxed_timeout=reset_relaxed_timeout,
3620 )
3622 if include_free_connections:
3623 for conn in self._get_free_connections():
3624 if self._should_update_connection(
3625 conn,
3626 matching_pattern,
3627 matching_address,
3628 matching_notification_hash,
3629 ):
3630 self.update_connection_settings(
3631 conn,
3632 state=state,
3633 maintenance_notification_hash=maintenance_notification_hash,
3634 host_address=host_address,
3635 relaxed_timeout=relaxed_timeout,
3636 update_notification_hash=update_notification_hash,
3637 reset_host_address=reset_host_address,
3638 reset_relaxed_timeout=reset_relaxed_timeout,
3639 )
3641 def update_connection_kwargs(
3642 self,
3643 **kwargs,
3644 ):
3645 """
3646 Update the connection kwargs for all future connections.
3648 This method updates the connection kwargs for all future connections created by the pool.
3649 Existing connections are not affected.
3650 """
3651 self.connection_kwargs.update(kwargs)
3653 def update_active_connections_for_reconnect(
3654 self,
3655 moving_address_src: Optional[str] = None,
3656 ):
3657 """
3658 Mark all active connections for reconnect.
3659 This is used when a cluster node is migrated to a different address.
3661 :param moving_address_src: The address of the node that is being moved.
3662 """
3663 debug = logger.isEnabledFor(logging.DEBUG)
3664 with self._get_pool_lock():
3665 for conn in self._get_in_use_connections():
3666 if self._should_update_connection(
3667 conn, "connected_address", moving_address_src
3668 ):
3669 if debug:
3670 logger.debug(
3671 f"Marking active connection for reconnect: {conn}, "
3672 f"{conn.extract_connection_details()}"
3673 )
3674 conn.mark_for_reconnect()
3676 def disconnect_free_connections(
3677 self,
3678 moving_address_src: Optional[str] = None,
3679 ):
3680 """
3681 Disconnect all free/available connections.
3682 This is used when a cluster node is migrated to a different address.
3684 :param moving_address_src: The address of the node that is being moved.
3685 """
3686 debug = logger.isEnabledFor(logging.DEBUG)
3687 with self._get_pool_lock():
3688 for conn in self._get_free_connections():
3689 if self._should_update_connection(
3690 conn, "connected_address", moving_address_src
3691 ):
3692 if debug:
3693 logger.debug(
3694 f"Disconnecting free connection: {conn}, "
3695 f"{conn.extract_connection_details()}"
3696 )
3697 conn.disconnect()
3700class ConnectionPool(MaintNotificationsAbstractConnectionPool, ConnectionPoolInterface):
3701 """
3702 Create a connection pool. ``If max_connections`` is set, then this
3703 object raises :py:class:`~redis.exceptions.ConnectionError` when the pool's
3704 limit is reached.
3706 By default, TCP connections are created unless ``connection_class``
3707 is specified. Use class:`.UnixDomainSocketConnection` for
3708 unix sockets.
3709 :py:class:`~redis.SSLConnection` can be used for SSL enabled connections.
3711 If ``maint_notifications_config`` is provided, the connection pool will support
3712 maintenance notifications.
3713 Maintenance notifications are supported only with RESP3.
3714 If the ``maint_notifications_config`` is not provided but the ``protocol`` is 3,
3715 the maintenance notifications will be enabled by default.
3717 If ``metadata_resolver`` is provided, it decides which commands are eligible for
3718 client-side caching - see `redis.commands.metadata.MetadataResolver`. It is handed to the
3719 cache configuration, so one resolver can serve cache eligibility and cluster routing
3720 alike. Defaults to the static command metadata this library ships, which the cache
3721 configuration resolves through on its own. A resolver built from a live ``COMMAND`` reply
3722 is a snapshot of the server it was read from, so give each pool its own rather than
3723 sharing one across pools on different servers.
3725 Any additional keyword arguments are passed to the constructor of
3726 ``connection_class``.
3727 """
3729 @classmethod
3730 def from_url(cls: Type[_CP], url: str, **kwargs) -> _CP:
3731 """
3732 Return a connection pool configured from the given URL.
3734 For example::
3736 redis://[[username]:[password]]@localhost:6379/0
3737 rediss://[[username]:[password]]@localhost:6379/0
3738 unix://[username@]/path/to/socket.sock?db=0[&password=password]
3740 Three URL schemes are supported:
3742 - `redis://` creates a TCP socket connection. See more at:
3743 <https://www.iana.org/assignments/uri-schemes/prov/redis>
3744 - `rediss://` creates a SSL wrapped TCP socket connection. See more at:
3745 <https://www.iana.org/assignments/uri-schemes/prov/rediss>
3746 - ``unix://``: creates a Unix Domain Socket connection.
3748 The username, password, hostname and path are passed through
3749 urllib.parse.unquote in order to replace any percent-encoded values
3750 with their corresponding characters. Querystring values are decoded
3751 by urllib.parse.parse_qs and are not unquoted again.
3753 There are several ways to specify a database number. The first value
3754 found will be used:
3756 1. A ``db`` querystring option, e.g. redis://localhost?db=0
3757 2. If using the redis:// or rediss:// schemes, the path argument
3758 of the url, e.g. redis://localhost/0
3759 3. A ``db`` keyword argument to this function.
3761 If none of these options are specified, the default db=0 is used.
3763 All querystring options are cast to their appropriate Python types.
3764 Boolean arguments can be specified with string values "True"/"False"
3765 or "Yes"/"No". Values that cannot be properly cast cause a
3766 ``ValueError`` to be raised. Once parsed, the querystring arguments
3767 and keyword arguments are passed to the ``ConnectionPool``'s
3768 class initializer. In the case of conflicting arguments, querystring
3769 arguments always win.
3770 """
3771 url_options = parse_url(url)
3773 if "connection_class" in kwargs:
3774 url_options["connection_class"] = kwargs["connection_class"]
3776 kwargs.update(url_options)
3777 return cls(**kwargs)
3779 def __init__(
3780 self,
3781 connection_class=Connection,
3782 max_connections: Optional[int] = None,
3783 cache_factory: Optional[CacheFactoryInterface] = None,
3784 maint_notifications_config: Optional[MaintNotificationsConfig] = None,
3785 metadata_resolver: Optional[MetadataResolver] = None,
3786 **connection_kwargs,
3787 ):
3788 max_connections = max_connections or 100
3789 if not isinstance(max_connections, int) or max_connections < 0:
3790 raise ValueError('"max_connections" must be a positive integer')
3792 self.connection_class = connection_class
3793 self._connection_kwargs = connection_kwargs
3794 self.max_connections = max_connections
3795 self.cache = None
3796 self._cache_factory = cache_factory
3797 self.metadata_resolver = metadata_resolver
3799 try:
3800 supports_maint_notifications = issubclass(
3801 connection_class, MaintNotificationsAbstractConnection
3802 )
3803 is_unix_domain_socket_connection = issubclass(
3804 connection_class, UnixDomainSocketConnection
3805 )
3806 except TypeError:
3807 supports_maint_notifications = False
3808 is_unix_domain_socket_connection = False
3810 if is_unix_domain_socket_connection or not supports_maint_notifications:
3811 if (
3812 maint_notifications_config
3813 and maint_notifications_config.enabled is True
3814 ):
3815 raise RedisError(
3816 "Maintenance notifications are not supported with "
3817 f"{connection_class}"
3818 )
3819 maint_notifications_config = MaintNotificationsConfig(enabled=False)
3821 self._event_dispatcher = self._connection_kwargs.get("event_dispatcher", None)
3822 if self._event_dispatcher is None:
3823 self._event_dispatcher = EventDispatcher()
3825 if connection_kwargs.get("cache_config") or connection_kwargs.get("cache"):
3826 if not check_protocol_version(self._connection_kwargs.get("protocol"), 3):
3827 raise RedisError("Client caching is only supported with RESP version 3")
3829 cache = self._connection_kwargs.get("cache")
3830 cache_config = self._connection_kwargs.get("cache_config")
3832 # Hand the client-level metadata resolver to the cache configuration, which is
3833 # where command eligibility is decided, so routing and caching read the same
3834 # records. Skipped when no resolver was configured, which leaves ``CacheConfig``
3835 # on the static default it builds for itself.
3836 #
3837 # Done on a copy rather than on the caller's object: a ``CacheConfig`` carries
3838 # only sizing and eviction settings, so reusing one across clients is reasonable,
3839 # and writing the resolver into it would give every one of them whichever
3840 # resolver was injected last. Every other setting is read-only after
3841 # construction, so the copy diverges from the caller's object in nothing but the
3842 # resolver - and a later ``set_metadata_resolver`` on the original does not reach
3843 # this pool, which is the intended direction: a pool decides eligibility by the
3844 # resolver its client was built with. Note that the ``cache=`` /
3845 # ``cache_factory=`` path below cannot copy, and so does write into the caller's
3846 # configuration; ``CacheConfig.set_metadata_resolver`` documents the difference.
3847 #
3848 # Guarded by ``isinstance`` rather than done through
3849 # ``CacheConfigurationInterface``: that ABC is public and implemented by third
3850 # parties, so a custom configuration keeps its own eligibility logic.
3851 if (
3852 metadata_resolver is not None
3853 and cache is None
3854 and self._cache_factory is None
3855 and isinstance(cache_config, CacheConfig)
3856 ):
3857 cache_config = copy.copy(cache_config)
3858 cache_config.set_metadata_resolver(metadata_resolver)
3860 if cache is not None:
3861 if not isinstance(cache, CacheInterface):
3862 raise ValueError("Cache must implement CacheInterface")
3864 self.cache = cache
3865 else:
3866 if self._cache_factory is not None:
3867 self.cache = CacheProxy(self._cache_factory.get_cache())
3868 else:
3869 self.cache = CacheFactory(cache_config).get_cache()
3871 # A caller who supplied a whole cache - ``cache=`` or ``cache_factory=`` - owns
3872 # the configuration inside it, and it cannot be swapped without rebuilding the
3873 # cache, so the resolver is set on it in place. This is the one path that writes
3874 # into the caller's configuration rather than into a copy of it, because a cache
3875 # reads its configuration on every lookup and ``CacheInterface`` exposes no way to
3876 # hand it a different one. Sharing one cache object across clients already shares
3877 # its entries, which couples them far more tightly than its eligibility does, so
3878 # the asymmetry with the ``cache_config=`` copy above is documented on
3879 # ``CacheConfig.set_metadata_resolver`` rather than removed.
3880 if metadata_resolver is not None and (
3881 cache is not None or self._cache_factory is not None
3882 ):
3883 own_config = self.cache.config
3884 if isinstance(own_config, CacheConfig):
3885 own_config.set_metadata_resolver(metadata_resolver)
3887 init_csc_items()
3888 register_csc_items_callback(
3889 callback=lambda: self.cache.size,
3890 pool_name=get_pool_name(self),
3891 )
3893 connection_kwargs.pop("cache", None)
3894 connection_kwargs.pop("cache_config", None)
3896 # Resolve the HIMPORT registry. A pre-built ``himport_registry`` (shared, e.g.
3897 # from the cluster client) takes precedence; otherwise build a fresh empty one.
3898 # A registry always exists so runtime ``himport_prepare`` mutates a single object
3899 # every connection already shares. The object stays in ``connection_kwargs`` so
3900 # it reaches every connection. It is injected unconditionally (like other
3901 # auto-added pool kwargs), so a custom ``connection_class`` must accept
3902 # ``**kwargs`` (or a ``himport_registry`` parameter), as built-ins do.
3903 himport_registry = connection_kwargs.get("himport_registry")
3904 if himport_registry is None:
3905 himport_registry = HImportRegistry()
3906 connection_kwargs["himport_registry"] = himport_registry
3907 self.himport_registry = himport_registry
3909 # a lock to protect the critical section in _checkpid().
3910 # this lock is acquired when the process id changes, such as
3911 # after a fork. during this time, multiple threads in the child
3912 # process could attempt to acquire this lock. the first thread
3913 # to acquire the lock will reset the data structures and lock
3914 # object of this pool. subsequent threads acquiring this lock
3915 # will notice the first thread already did the work and simply
3916 # release the lock.
3918 self._fork_lock = threading.RLock()
3919 self._lock = threading.RLock()
3921 # Generate unique pool ID for observability (matches go-redis behavior)
3922 import secrets
3924 self._pool_id = secrets.token_hex(4)
3926 MaintNotificationsAbstractConnectionPool.__init__(
3927 self,
3928 maint_notifications_config=maint_notifications_config,
3929 **connection_kwargs,
3930 )
3932 self.reset()
3934 # Keys that should be redacted in __repr__ to avoid exposing sensitive information
3935 SENSITIVE_REPR_KEYS = frozenset(
3936 {
3937 "password",
3938 "username",
3939 "ssl_password",
3940 "credential_provider",
3941 }
3942 )
3944 # Internal plumbing kwargs omitted from __repr__ (not user-facing config).
3945 OMIT_REPR_KEYS = frozenset({"himport_registry"})
3947 def __repr__(self) -> str:
3948 conn_kwargs = ",".join(
3949 [
3950 f"{k}={'<REDACTED>' if k in self.SENSITIVE_REPR_KEYS else v}"
3951 for k, v in self.connection_kwargs.items()
3952 if k not in self.OMIT_REPR_KEYS
3953 ]
3954 )
3955 return (
3956 f"<{self.__class__.__module__}.{self.__class__.__name__}"
3957 f"(<{self.connection_class.__module__}.{self.connection_class.__name__}"
3958 f"({conn_kwargs})>)>"
3959 )
3961 @property
3962 def connection_kwargs(self) -> Dict[str, Any]:
3963 return self._connection_kwargs
3965 @connection_kwargs.setter
3966 def connection_kwargs(self, value: Dict[str, Any]):
3967 self._connection_kwargs = value
3969 def get_protocol(self):
3970 """
3971 Returns:
3972 The RESP protocol version, or ``None`` if the protocol is not specified,
3973 in which case the server default will be used.
3974 """
3975 return self.connection_kwargs.get("protocol", None)
3977 def reset(self) -> None:
3978 # Record metrics for connections being removed before clearing
3979 # (only if attributes exist - they won't during __init__)
3980 if hasattr(self, "_available_connections") and hasattr(
3981 self, "_in_use_connections"
3982 ):
3983 with self._lock:
3984 idle_count = len(self._available_connections)
3985 in_use_count = len(self._in_use_connections)
3986 if idle_count > 0 or in_use_count > 0:
3987 pool_name = get_pool_name(self)
3988 if idle_count > 0:
3989 record_connection_count(
3990 pool_name=pool_name,
3991 connection_state=ConnectionState.IDLE,
3992 counter=-idle_count,
3993 )
3994 if in_use_count > 0:
3995 record_connection_count(
3996 pool_name=pool_name,
3997 connection_state=ConnectionState.USED,
3998 counter=-in_use_count,
3999 )
4001 self._created_connections = 0
4002 self._available_connections = []
4003 self._in_use_connections = set()
4005 # this must be the last operation in this method. while reset() is
4006 # called when holding _fork_lock, other threads in this process
4007 # can call _checkpid() which compares self.pid and os.getpid() without
4008 # holding any lock (for performance reasons). keeping this assignment
4009 # as the last operation ensures that those other threads will also
4010 # notice a pid difference and block waiting for the first thread to
4011 # release _fork_lock. when each of these threads eventually acquire
4012 # _fork_lock, they will notice that another thread already called
4013 # reset() and they will immediately release _fork_lock and continue on.
4014 self.pid = os.getpid()
4016 def __del__(self) -> None:
4017 """Clean up connection pool and record metrics when garbage collected."""
4018 try:
4019 if not hasattr(self, "_available_connections") or not hasattr(
4020 self, "_in_use_connections"
4021 ):
4022 return
4023 # Record metrics for all connections being removed
4024 idle_count = len(self._available_connections)
4025 in_use_count = len(self._in_use_connections)
4026 if idle_count > 0 or in_use_count > 0:
4027 pool_name = get_pool_name(self)
4028 if idle_count > 0:
4029 record_connection_count(
4030 pool_name=pool_name,
4031 connection_state=ConnectionState.IDLE,
4032 counter=-idle_count,
4033 )
4034 if in_use_count > 0:
4035 record_connection_count(
4036 pool_name=pool_name,
4037 connection_state=ConnectionState.USED,
4038 counter=-in_use_count,
4039 )
4040 except Exception:
4041 pass
4043 def _checkpid(self) -> None:
4044 # _checkpid() attempts to keep ConnectionPool fork-safe on modern
4045 # systems. this is called by all ConnectionPool methods that
4046 # manipulate the pool's state such as get_connection() and release().
4047 #
4048 # _checkpid() determines whether the process has forked by comparing
4049 # the current process id to the process id saved on the ConnectionPool
4050 # instance. if these values are the same, _checkpid() simply returns.
4051 #
4052 # when the process ids differ, _checkpid() assumes that the process
4053 # has forked and that we're now running in the child process. the child
4054 # process cannot use the parent's file descriptors (e.g., sockets).
4055 # therefore, when _checkpid() sees the process id change, it calls
4056 # reset() in order to reinitialize the child's ConnectionPool. this
4057 # will cause the child to make all new connection objects.
4058 #
4059 # _checkpid() is protected by self._fork_lock to ensure that multiple
4060 # threads in the child process do not call reset() multiple times.
4061 #
4062 # there is an extremely small chance this could fail in the following
4063 # scenario:
4064 # 1. process A calls _checkpid() for the first time and acquires
4065 # self._fork_lock.
4066 # 2. while holding self._fork_lock, process A forks (the fork()
4067 # could happen in a different thread owned by process A)
4068 # 3. process B (the forked child process) inherits the
4069 # ConnectionPool's state from the parent. that state includes
4070 # a locked _fork_lock. process B will not be notified when
4071 # process A releases the _fork_lock and will thus never be
4072 # able to acquire the _fork_lock.
4073 #
4074 # to mitigate this possible deadlock, _checkpid() will only wait 5
4075 # seconds to acquire _fork_lock. if _fork_lock cannot be acquired in
4076 # that time it is assumed that the child is deadlocked and a
4077 # redis.ChildDeadlockedError error is raised.
4078 if self.pid != os.getpid():
4079 acquired = self._fork_lock.acquire(timeout=5)
4080 if not acquired:
4081 raise ChildDeadlockedError
4082 # reset() the instance for the new process if another thread
4083 # hasn't already done so
4084 try:
4085 if self.pid != os.getpid():
4086 self.reset()
4087 finally:
4088 self._fork_lock.release()
4090 @deprecated_args(
4091 args_to_warn=["*"],
4092 reason="Use get_connection() without args instead",
4093 version="5.3.0",
4094 )
4095 def get_connection(self, command_name=None, *keys, **options) -> "Connection":
4096 "Get a connection from the pool"
4098 # Start timing for observability
4099 self._checkpid()
4100 is_created = False
4102 with self._lock:
4103 try:
4104 connection = self._available_connections.pop()
4105 except IndexError:
4106 # Start timing for observability
4107 start_time_created = time.monotonic()
4109 connection = self.make_connection()
4110 is_created = True
4111 self._in_use_connections.add(connection)
4113 # Record state transition for observability
4114 pool_name = get_pool_name(self)
4115 if is_created:
4116 # New connection created and acquired: just USED +1
4117 # (connection was never idle in the pool)
4118 record_connection_count(
4119 pool_name=pool_name,
4120 connection_state=ConnectionState.USED,
4121 counter=1,
4122 )
4123 else:
4124 # Existing connection acquired from pool: IDLE -> USED
4125 record_connection_count(
4126 pool_name=pool_name,
4127 connection_state=ConnectionState.IDLE,
4128 counter=-1,
4129 )
4130 record_connection_count(
4131 pool_name=pool_name,
4132 connection_state=ConnectionState.USED,
4133 counter=1,
4134 )
4136 try:
4137 self.ensure_connection(connection)
4138 except BaseException:
4139 # release the connection back to the pool so that we don't
4140 # leak it
4141 self.release(connection)
4142 raise
4144 if is_created:
4145 record_connection_create_time(
4146 connection_pool=self,
4147 duration_seconds=time.monotonic() - start_time_created,
4148 )
4150 return connection
4152 def ensure_connection(self, connection: "Connection") -> None:
4153 """Ensure that the connection object is connected and valid"""
4154 connection.connect()
4155 # connections that the pool provides should be ready to send
4156 # a command. if not, the connection was either returned to the
4157 # pool before all data has been read or the socket has been
4158 # closed. either way, reconnect and verify everything is good.
4159 # With maintenance notifications enabled the pending data may be
4160 # push notifications that have to be applied rather than a reason
4161 # to reconnect; reading them is what tells the two apart, and it
4162 # raises for a socket the server has closed in the meantime.
4163 try:
4164 if connection.can_read():
4165 if self.maint_notifications_enabled():
4166 connection.handle_pending_push_notifications()
4167 elif self.cache is None:
4168 raise ConnectionError("Connection has data")
4169 except (ConnectionError, TimeoutError, OSError):
4170 connection.disconnect()
4171 connection.connect()
4172 if (
4173 connection.can_read()
4174 and self.cache is None
4175 and not self.maint_notifications_enabled()
4176 ):
4177 raise ConnectionError("Connection not ready")
4179 def get_encoder(self) -> Encoder:
4180 "Return an encoder based on encoding settings"
4181 kwargs = self.connection_kwargs
4182 return Encoder(
4183 encoding=kwargs.get("encoding", "utf-8"),
4184 encoding_errors=kwargs.get("encoding_errors", "strict"),
4185 decode_responses=kwargs.get("decode_responses", False),
4186 )
4188 def make_connection(self) -> "ConnectionInterface":
4189 "Create a new connection"
4190 if self._created_connections >= self.max_connections:
4191 raise MaxConnectionsError("Too many connections")
4193 kwargs = dict(self.connection_kwargs)
4195 # Create the connection first, then record metrics only on success
4196 if self.cache is not None:
4197 connection = CacheProxyConnection(
4198 self.connection_class(**kwargs), self.cache, self._lock
4199 )
4200 else:
4201 connection = self.connection_class(**kwargs)
4203 # Only a connection that exists can be released, so count it here.
4204 self._created_connections += 1
4206 return connection
4208 def release(self, connection: "Connection") -> None:
4209 "Releases the connection back to the pool"
4210 self._checkpid()
4211 with self._lock:
4212 try:
4213 self._in_use_connections.remove(connection)
4214 except KeyError:
4215 # Gracefully fail when a connection is returned to this pool
4216 # that the pool doesn't actually own
4217 return
4219 if self.owns_connection(connection):
4220 if connection.should_reconnect():
4221 if logger.isEnabledFor(logging.DEBUG):
4222 logger.debug(
4223 "Disconnecting released connection marked for reconnect: "
4224 f"{connection}, {connection.extract_connection_details()}"
4225 )
4226 connection.disconnect()
4227 self._available_connections.append(connection)
4228 self._event_dispatcher.dispatch(
4229 AfterConnectionReleasedEvent(connection)
4230 )
4232 # Record state transition: USED -> IDLE
4233 pool_name = get_pool_name(self)
4234 record_connection_count(
4235 pool_name=pool_name,
4236 connection_state=ConnectionState.USED,
4237 counter=-1,
4238 )
4239 record_connection_count(
4240 pool_name=pool_name,
4241 connection_state=ConnectionState.IDLE,
4242 counter=1,
4243 )
4244 else:
4245 # Pool doesn't own this connection, do not add it back
4246 # to the pool.
4247 connection.disconnect()
4248 # Subclasses such as SentinelConnectionPool can override
4249 # owns_connection() with a comparison different from local PID
4250 # ownership. When such a subclass rejects a connection, also require
4251 # connection.pid == self.pid before reclaiming its slot.
4252 if connection.pid == self.pid:
4253 self._created_connections -= 1
4254 # The connection was checked out by this process, so
4255 # decrement USED, which was counted in get_connection().
4256 # Connections inherited across a fork are skipped here,
4257 # their accounting is handled by reset()/__del__.
4258 record_connection_count(
4259 pool_name=get_pool_name(self),
4260 connection_state=ConnectionState.USED,
4261 counter=-1,
4262 )
4263 return
4265 def owns_connection(self, connection: "Connection") -> int:
4266 return connection.pid == self.pid
4268 def disconnect(self, inuse_connections: bool = True) -> None:
4269 """
4270 Disconnects connections in the pool
4272 If ``inuse_connections`` is True, disconnect connections that are
4273 currently in use, potentially by other threads. Otherwise only disconnect
4274 connections that are idle in the pool.
4275 """
4276 self._checkpid()
4277 with self._lock:
4278 if inuse_connections:
4279 connections = chain(
4280 self._available_connections, self._in_use_connections
4281 )
4282 else:
4283 connections = self._available_connections
4285 for connection in connections:
4286 connection.disconnect()
4288 def close(self) -> None:
4289 """Close the pool, disconnecting all connections"""
4290 self.disconnect()
4292 def __enter__(self: _CP) -> _CP:
4293 return self
4295 def __exit__(self, exc_type, exc_value, traceback) -> None:
4296 self.close()
4298 def set_retry(self, retry: Retry) -> None:
4299 self.connection_kwargs.update({"retry": retry})
4300 for conn in self._available_connections:
4301 conn.retry = retry
4302 for conn in self._in_use_connections:
4303 conn.retry = retry
4305 def re_auth_callback(self, token: TokenInterface):
4306 with self._lock:
4307 for conn in self._available_connections:
4308 conn.retry.call_with_retry(
4309 lambda: conn.send_command(
4310 "AUTH", token.try_get("oid"), token.get_value()
4311 ),
4312 lambda error: self._mock(error),
4313 )
4314 conn.retry.call_with_retry(
4315 lambda: conn.read_response(), lambda error: self._mock(error)
4316 )
4317 for conn in self._in_use_connections:
4318 conn.set_re_auth_token(token)
4320 def _get_pool_lock(self):
4321 return self._lock
4323 def _get_free_connections(self):
4324 with self._lock:
4325 return list(self._available_connections)
4327 def _get_in_use_connections(self):
4328 with self._lock:
4329 return set(self._in_use_connections)
4331 def _mock(self, error: RedisError):
4332 """
4333 Dummy functions, needs to be passed as error callback to retry object.
4334 :param error:
4335 :return:
4336 """
4337 pass
4339 def get_connection_count(self) -> List[tuple[int, dict]]:
4340 from redis.observability.attributes import get_pool_name
4342 attributes = AttributeBuilder.build_base_attributes()
4343 attributes[DB_CLIENT_CONNECTION_POOL_NAME] = get_pool_name(self)
4344 free_connections_attributes = attributes.copy()
4345 in_use_connections_attributes = attributes.copy()
4347 free_connections_attributes[DB_CLIENT_CONNECTION_STATE] = (
4348 ConnectionState.IDLE.value
4349 )
4350 in_use_connections_attributes[DB_CLIENT_CONNECTION_STATE] = (
4351 ConnectionState.USED.value
4352 )
4354 return [
4355 (len(self._get_free_connections()), free_connections_attributes),
4356 (len(self._get_in_use_connections()), in_use_connections_attributes),
4357 ]
4360class BlockingConnectionPool(ConnectionPool):
4361 """
4362 Thread-safe blocking connection pool::
4364 >>> from redis.client import Redis
4365 >>> client = Redis(connection_pool=BlockingConnectionPool())
4367 It performs the same function as the default
4368 :py:class:`~redis.ConnectionPool` implementation, in that,
4369 it maintains a pool of reusable connections that can be shared by
4370 multiple redis clients (safely across threads if required).
4372 The difference is that, in the event that a client tries to get a
4373 connection from the pool when all of connections are in use, rather than
4374 raising a :py:class:`~redis.ConnectionError` (as the default
4375 :py:class:`~redis.ConnectionPool` implementation does), it
4376 makes the client wait ("blocks") for a specified number of seconds until
4377 a connection becomes available.
4379 Use ``max_connections`` to increase / decrease the pool size::
4381 >>> pool = BlockingConnectionPool(max_connections=10)
4383 Use ``timeout`` to tell it either how many seconds to wait for a connection
4384 to become available, or to block forever:
4386 >>> # Block forever.
4387 >>> pool = BlockingConnectionPool(timeout=None)
4389 >>> # Raise a ``ConnectionError`` after five seconds if a connection is
4390 >>> # not available.
4391 >>> pool = BlockingConnectionPool(timeout=5)
4392 """
4394 def __init__(
4395 self,
4396 max_connections=50,
4397 timeout=20,
4398 connection_class=Connection,
4399 queue_class=LifoQueue,
4400 **connection_kwargs,
4401 ):
4402 self.queue_class = queue_class
4403 self.timeout = timeout
4404 self._in_maintenance = False
4405 self._locked = False
4406 super().__init__(
4407 connection_class=connection_class,
4408 max_connections=max_connections,
4409 **connection_kwargs,
4410 )
4412 def reset(self):
4413 # Create and fill up a thread safe queue with ``None`` values.
4414 try:
4415 if self._in_maintenance:
4416 self._lock.acquire()
4417 self._locked = True
4419 # Record metrics for connections being removed before clearing
4420 # Note: Access pool.queue directly to avoid deadlock since we may
4421 # already hold self._lock (which is non-reentrant)
4422 if (
4423 hasattr(self, "_connections")
4424 and self._connections
4425 and hasattr(self, "pool")
4426 ):
4427 with self._lock:
4428 connections_in_queue = {conn for conn in self.pool.queue if conn}
4429 idle_count = len(connections_in_queue)
4430 in_use_count = len(self._connections) - idle_count
4431 if idle_count > 0 or in_use_count > 0:
4432 pool_name = get_pool_name(self)
4433 if idle_count > 0:
4434 record_connection_count(
4435 pool_name=pool_name,
4436 connection_state=ConnectionState.IDLE,
4437 counter=-idle_count,
4438 )
4439 if in_use_count > 0:
4440 record_connection_count(
4441 pool_name=pool_name,
4442 connection_state=ConnectionState.USED,
4443 counter=-in_use_count,
4444 )
4446 self.pool = self.queue_class(self.max_connections)
4447 while True:
4448 try:
4449 self.pool.put_nowait(None)
4450 except Full:
4451 break
4453 # Keep a list of actual connection instances so that we can
4454 # disconnect them later.
4455 self._connections = []
4456 finally:
4457 if self._locked:
4458 try:
4459 self._lock.release()
4460 except Exception:
4461 pass
4462 self._locked = False
4464 # this must be the last operation in this method. while reset() is
4465 # called when holding _fork_lock, other threads in this process
4466 # can call _checkpid() which compares self.pid and os.getpid() without
4467 # holding any lock (for performance reasons). keeping this assignment
4468 # as the last operation ensures that those other threads will also
4469 # notice a pid difference and block waiting for the first thread to
4470 # release _fork_lock. when each of these threads eventually acquire
4471 # _fork_lock, they will notice that another thread already called
4472 # reset() and they will immediately release _fork_lock and continue on.
4473 self.pid = os.getpid()
4475 def __del__(self) -> None:
4476 """Clean up connection pool and record metrics when garbage collected."""
4477 try:
4478 # Note: Access pool.queue directly to avoid potential deadlock
4479 # if GC runs while the lock is held by the same thread
4480 if (
4481 hasattr(self, "_connections")
4482 and self._connections
4483 and hasattr(self, "pool")
4484 ):
4485 connections_in_queue = {conn for conn in self.pool.queue if conn}
4486 idle_count = len(connections_in_queue)
4487 in_use_count = len(self._connections) - idle_count
4488 if idle_count > 0 or in_use_count > 0:
4489 pool_name = get_pool_name(self)
4490 if idle_count > 0:
4491 record_connection_count(
4492 pool_name=pool_name,
4493 connection_state=ConnectionState.IDLE,
4494 counter=-idle_count,
4495 )
4496 if in_use_count > 0:
4497 record_connection_count(
4498 pool_name=pool_name,
4499 connection_state=ConnectionState.USED,
4500 counter=-in_use_count,
4501 )
4502 except Exception:
4503 pass
4505 def make_connection(self):
4506 "Make a fresh connection."
4507 try:
4508 if self._in_maintenance:
4509 self._lock.acquire()
4510 self._locked = True
4512 if self.cache is not None:
4513 connection = CacheProxyConnection(
4514 self.connection_class(**self.connection_kwargs),
4515 self.cache,
4516 self._lock,
4517 )
4518 else:
4519 connection = self.connection_class(**self.connection_kwargs)
4520 self._connections.append(connection)
4522 return connection
4523 finally:
4524 if self._locked:
4525 try:
4526 self._lock.release()
4527 except Exception:
4528 pass
4529 self._locked = False
4531 @deprecated_args(
4532 args_to_warn=["*"],
4533 reason="Use get_connection() without args instead",
4534 version="5.3.0",
4535 )
4536 def get_connection(self, command_name=None, *keys, **options):
4537 """
4538 Get a connection, blocking for ``self.timeout`` until a connection
4539 is available from the pool.
4541 If the connection returned is ``None`` then creates a new connection.
4542 Because we use a last-in first-out queue, the existing connections
4543 (having been returned to the pool after the initial ``None`` values
4544 were added) will be returned before ``None`` values. This means we only
4545 create new connections when we need to, i.e.: the actual number of
4546 connections will only increase in response to demand.
4547 """
4548 start_time_acquired = time.monotonic()
4549 # Make sure we haven't changed process.
4550 self._checkpid()
4551 is_created = False
4553 # Try and get a connection from the pool. If one isn't available within
4554 # self.timeout then raise a ``ConnectionError``.
4555 connection = None
4556 try:
4557 if self._in_maintenance:
4558 self._lock.acquire()
4559 self._locked = True
4560 try:
4561 connection = self.pool.get(block=True, timeout=self.timeout)
4562 except Empty:
4563 record_connection_timeout(pool_name=get_pool_name(self))
4564 # Note that this is not caught by the redis client and will be
4565 # raised unless handled by application code. If you want never to
4566 raise ConnectionError("No connection available.")
4568 # If the ``connection`` is actually ``None`` then that's a cue to make
4569 # a new connection to add to the pool.
4570 if connection is None:
4571 # Start timing for observability
4572 start_time_created = time.monotonic()
4573 try:
4574 connection = self.make_connection()
4575 except BaseException:
4576 # Hand the slot back, or the pool shrinks on every failure.
4577 try:
4578 self.pool.put_nowait(None)
4579 except Full:
4580 pass
4581 raise
4582 is_created = True
4583 finally:
4584 if self._locked:
4585 try:
4586 self._lock.release()
4587 except Exception:
4588 pass
4589 self._locked = False
4591 # Record state transition for observability
4592 pool_name = get_pool_name(self)
4593 if is_created:
4594 # New connection created and acquired: just USED +1
4595 # (connection was never idle in the pool)
4596 record_connection_count(
4597 pool_name=pool_name,
4598 connection_state=ConnectionState.USED,
4599 counter=1,
4600 )
4601 else:
4602 # Existing connection acquired from pool: IDLE -> USED
4603 record_connection_count(
4604 pool_name=pool_name,
4605 connection_state=ConnectionState.IDLE,
4606 counter=-1,
4607 )
4608 record_connection_count(
4609 pool_name=pool_name,
4610 connection_state=ConnectionState.USED,
4611 counter=1,
4612 )
4614 try:
4615 self.ensure_connection(connection)
4616 except BaseException:
4617 # release the connection back to the pool so that we don't leak it
4618 self.release(connection)
4619 raise
4621 if is_created:
4622 record_connection_create_time(
4623 connection_pool=self,
4624 duration_seconds=time.monotonic() - start_time_created,
4625 )
4627 record_connection_wait_time(
4628 pool_name=pool_name,
4629 duration_seconds=time.monotonic() - start_time_acquired,
4630 )
4632 return connection
4634 def release(self, connection):
4635 "Releases the connection back to the pool."
4636 # Make sure we haven't changed process.
4637 self._checkpid()
4639 try:
4640 if self._in_maintenance:
4641 self._lock.acquire()
4642 self._locked = True
4643 if not self.owns_connection(connection):
4644 # pool doesn't own this connection. do not add it back
4645 # to the pool. instead add a None value which is a placeholder
4646 # that will cause the pool to recreate the connection if
4647 # its needed.
4648 connection.disconnect()
4649 self.pool.put_nowait(None)
4650 # Subclasses can override owns_connection() with a comparison
4651 # different from local PID ownership. When such a subclass
4652 # rejects a connection checked out by this process, stop
4653 # tracking it and decrement USED, which was counted in
4654 # get_connection(). Connections inherited across a fork are
4655 # skipped here, their accounting is handled by reset()/__del__.
4656 if connection.pid == self.pid:
4657 try:
4658 self._connections.remove(connection)
4659 except ValueError:
4660 pass
4661 record_connection_count(
4662 pool_name=get_pool_name(self),
4663 connection_state=ConnectionState.USED,
4664 counter=-1,
4665 )
4666 return
4667 if connection.should_reconnect():
4668 if logger.isEnabledFor(logging.DEBUG):
4669 logger.debug(
4670 "Disconnecting released connection marked for reconnect: "
4671 f"{connection}, {connection.extract_connection_details()}"
4672 )
4673 connection.disconnect()
4674 # Put the connection back into the pool.
4675 pool_name = get_pool_name(self)
4676 try:
4677 self.pool.put_nowait(connection)
4679 # Record state transition: USED -> IDLE
4680 record_connection_count(
4681 pool_name=pool_name,
4682 connection_state=ConnectionState.USED,
4683 counter=-1,
4684 )
4685 record_connection_count(
4686 pool_name=pool_name,
4687 connection_state=ConnectionState.IDLE,
4688 counter=1,
4689 )
4690 except Full:
4691 pass
4692 finally:
4693 if self._locked:
4694 try:
4695 self._lock.release()
4696 except Exception:
4697 pass
4698 self._locked = False
4700 def disconnect(self, inuse_connections: bool = True):
4701 """
4702 Disconnects either all connections in the pool or just the free connections.
4703 """
4704 self._checkpid()
4705 try:
4706 if self._in_maintenance:
4707 self._lock.acquire()
4708 self._locked = True
4710 if inuse_connections:
4711 connections = self._connections
4712 else:
4713 connections = self._get_free_connections()
4715 for connection in connections:
4716 connection.disconnect()
4717 finally:
4718 if self._locked:
4719 try:
4720 self._lock.release()
4721 except Exception:
4722 pass
4723 self._locked = False
4725 def _get_free_connections(self):
4726 with self._lock:
4727 return {conn for conn in self.pool.queue if conn}
4729 def _get_in_use_connections(self):
4730 with self._lock:
4731 # free connections
4732 connections_in_queue = {conn for conn in self.pool.queue if conn}
4733 # in self._connections we keep all created connections
4734 # so the ones that are not in the queue are the in use ones
4735 return {
4736 conn for conn in self._connections if conn not in connections_in_queue
4737 }
4739 def set_in_maintenance(self, in_maintenance: bool):
4740 """
4741 Sets a flag that this Blocking ConnectionPool is in maintenance mode.
4743 This is used to prevent new connections from being created while we are in maintenance mode.
4744 The pool will be in maintenance mode only when we are processing a MOVING notification.
4745 """
4746 self._in_maintenance = in_maintenance