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