Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/redis/connection.py: 23%
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)
37from redis.commands.metadata import MetadataResolver
39from . import exceptions as redis_exceptions
40from ._defaults import (
41 DEFAULT_SOCKET_CONNECT_TIMEOUT,
42 DEFAULT_SOCKET_READ_SIZE,
43 DEFAULT_SOCKET_TIMEOUT,
44 get_default_socket_keepalive_options,
45)
46from ._parsers import BaseParser, Encoder, _HiredisParser, _RESP2Parser, _RESP3Parser
47from .auth.token import TokenInterface
48from .backoff import NoBackoff
49from .credentials import CredentialProvider, UsernamePasswordCredentialProvider
50from .driver_info import DriverInfo, resolve_driver_info
51from .event import AfterConnectionReleasedEvent, EventDispatcher
52from .exceptions import (
53 AuthenticationError,
54 AuthenticationWrongNumberOfArgsError,
55 ChildDeadlockedError,
56 ConnectionError,
57 DataError,
58 MaxConnectionsError,
59 RedisError,
60 ResponseError,
61 TimeoutError,
62)
63from .himport import HImportRegistry
64from .maint_notifications import (
65 MaintenanceState,
66 MaintNotificationsConfig,
67 MaintNotificationsConnectionHandler,
68 MaintNotificationsPoolHandler,
69 OSSMaintNotificationsHandler,
70)
71from .observability.attributes import (
72 DB_CLIENT_CONNECTION_POOL_NAME,
73 DB_CLIENT_CONNECTION_STATE,
74 AttributeBuilder,
75 ConnectionState,
76 CSCReason,
77 CSCResult,
78 get_pool_name,
79)
80from .observability.metrics import CloseReason
81from .observability.recorder import (
82 init_csc_items,
83 record_connection_closed,
84 record_connection_count,
85 record_connection_create_time,
86 record_connection_wait_time,
87 record_csc_eviction,
88 record_csc_network_saved,
89 record_csc_request,
90 record_error_count,
91 register_csc_items_callback,
92)
93from .retry import Retry
94from .utils import (
95 CRYPTOGRAPHY_AVAILABLE,
96 DEFAULT_RESP_VERSION,
97 HIREDIS_AVAILABLE,
98 SENTINEL,
99 SSL_AVAILABLE,
100 check_protocol_version,
101 compare_versions,
102 deprecated_args,
103 ensure_string,
104 format_error_message,
105 str_if_bytes,
106)
108if SSL_AVAILABLE:
109 import ssl
110 from ssl import VerifyFlags
111else:
112 ssl = None
113 VerifyFlags = None
115if HIREDIS_AVAILABLE:
116 import hiredis
118SYM_STAR = b"*"
119SYM_DOLLAR = b"$"
120SYM_CRLF = b"\r\n"
121SYM_EMPTY = b""
123DefaultParser: Type[Union[_RESP2Parser, _RESP3Parser, _HiredisParser]]
124if HIREDIS_AVAILABLE:
125 DefaultParser = _HiredisParser
126else:
127 DefaultParser = _RESP2Parser
129logger = logging.getLogger(__name__)
132def add_debug_log_for_connection_failure(
133 connection: "AbstractConnection",
134 error: BaseException,
135 operation: str,
136) -> None:
137 """
138 Render the connection's live state on a failure that is about to close it.
140 Must be called *before* ``disconnect()``. ``extract_connection_details()``
141 reads the resolved ip, local port and armed read timeout off the socket, so
142 once the socket is gone it can only report ``not connected`` - which hides
143 exactly the state needed to explain the failure. In particular it is what
144 tells apart a read that ran under the original timeout from one that ran
145 under a relaxed maintenance timeout.
146 """
147 if logger.isEnabledFor(logging.DEBUG):
148 logger.debug(
149 f"{type(error).__name__} while {operation}, "
150 f"with connection: {connection}, "
151 f"details: {connection.extract_connection_details()}, "
152 f"error: {error}",
153 )
156class HiredisRespSerializer:
157 def pack(self, *args: List):
158 """Pack a series of arguments into the Redis protocol"""
159 output = []
161 if isinstance(args[0], str):
162 args = tuple(args[0].encode().split()) + args[1:]
163 elif b" " in args[0]:
164 args = tuple(args[0].split()) + args[1:]
165 args = tuple(
166 bytes(arg) if isinstance(arg, (bytearray, memoryview)) else arg
167 for arg in args
168 )
169 try:
170 output.append(hiredis.pack_command(args))
171 except TypeError:
172 _, value, traceback = sys.exc_info()
173 raise DataError(value).with_traceback(traceback)
175 return output
178class PythonRespSerializer:
179 def __init__(self, buffer_cutoff, encode) -> None:
180 self._buffer_cutoff = buffer_cutoff
181 self.encode = encode
183 def pack(self, *args):
184 """Pack a series of arguments into the Redis protocol"""
185 output = []
186 # the client might have included 1 or more literal arguments in
187 # the command name, e.g., 'CONFIG GET'. The Redis server expects these
188 # arguments to be sent separately, so split the first argument
189 # manually. These arguments should be bytestrings so that they are
190 # not encoded.
191 if isinstance(args[0], str):
192 args = tuple(args[0].encode().split()) + args[1:]
193 elif b" " in args[0]:
194 args = tuple(args[0].split()) + args[1:]
196 buff = SYM_EMPTY.join((SYM_STAR, str(len(args)).encode(), SYM_CRLF))
198 buffer_cutoff = self._buffer_cutoff
199 for arg in map(self.encode, args):
200 # to avoid large string mallocs, chunk the command into the
201 # output list if we're sending large values or memoryviews
202 arg_length = len(arg)
203 if (
204 len(buff) > buffer_cutoff
205 or arg_length > buffer_cutoff
206 or isinstance(arg, memoryview)
207 ):
208 buff = SYM_EMPTY.join(
209 (buff, SYM_DOLLAR, str(arg_length).encode(), SYM_CRLF)
210 )
211 output.append(buff)
212 output.append(arg)
213 buff = SYM_CRLF
214 else:
215 buff = SYM_EMPTY.join(
216 (
217 buff,
218 SYM_DOLLAR,
219 str(arg_length).encode(),
220 SYM_CRLF,
221 arg,
222 SYM_CRLF,
223 )
224 )
225 output.append(buff)
226 return output
229class ConnectionInterface:
230 @abstractmethod
231 def repr_pieces(self):
232 pass
234 @abstractmethod
235 def register_connect_callback(self, callback):
236 pass
238 @abstractmethod
239 def deregister_connect_callback(self, callback):
240 pass
242 @abstractmethod
243 def set_parser(self, parser_class):
244 pass
246 @abstractmethod
247 def get_protocol(self):
248 pass
250 @abstractmethod
251 def connect(self):
252 pass
254 @abstractmethod
255 def on_connect(self):
256 pass
258 @abstractmethod
259 def disconnect(self, *args, **kwargs):
260 pass
262 @abstractmethod
263 def check_health(self):
264 pass
266 @abstractmethod
267 def send_packed_command(self, command, check_health=True):
268 pass
270 @abstractmethod
271 def send_command(self, *args, **kwargs):
272 pass
274 @abstractmethod
275 def can_read(self, timeout: float = 0) -> bool:
276 # TODO: Rename this API; it detects pending data or dirty/closed
277 # connection state, not only whether application data can be read.
278 pass
280 @abstractmethod
281 def read_response(
282 self,
283 disable_decoding=False,
284 *,
285 timeout: Union[float, object] = SENTINEL,
286 disconnect_on_error=True,
287 push_request=False,
288 ):
289 pass
291 @abstractmethod
292 def pack_command(self, *args):
293 pass
295 @abstractmethod
296 def pack_commands(self, commands):
297 pass
299 @property
300 @abstractmethod
301 def handshake_metadata(self) -> Union[Dict[bytes, bytes], Dict[str, str]]:
302 pass
304 @abstractmethod
305 def set_re_auth_token(self, token: TokenInterface):
306 pass
308 @abstractmethod
309 def re_auth(self):
310 pass
312 @abstractmethod
313 def mark_for_reconnect(self):
314 """
315 Mark the connection to be reconnected on the next command.
316 This is useful when a connection is moved to a different node.
317 """
318 pass
320 @abstractmethod
321 def should_reconnect(self):
322 """
323 Returns True if the connection should be reconnected.
324 """
325 pass
327 @abstractmethod
328 def reset_should_reconnect(self):
329 """
330 Reset the internal flag to False.
331 """
332 pass
334 @abstractmethod
335 def extract_connection_details(self) -> str:
336 pass
338 @property
339 @abstractmethod
340 def is_connected(self) -> bool:
341 """
342 Return ``True`` if the connection to the server is active.
343 """
344 pass
347class MaintNotificationsAbstractConnection:
348 """
349 Abstract class for handling maintenance notifications logic.
350 This class is expected to be used as base class together with ConnectionInterface.
352 This class is intended to be used with multiple inheritance!
354 All logic related to maintenance notifications is encapsulated in this class.
355 """
357 def __init__(
358 self,
359 maint_notifications_config: Optional[MaintNotificationsConfig],
360 maint_notifications_pool_handler: Optional[
361 MaintNotificationsPoolHandler
362 ] = None,
363 maintenance_state: "MaintenanceState" = MaintenanceState.NONE,
364 maintenance_notification_hash: Optional[int] = None,
365 orig_host_address: Optional[str] = None,
366 orig_socket_timeout: Optional[float] = None,
367 orig_socket_connect_timeout: Optional[float] = None,
368 oss_cluster_maint_notifications_handler: Optional[
369 OSSMaintNotificationsHandler
370 ] = None,
371 parser: Optional[BaseParser] = None,
372 event_dispatcher: Optional[EventDispatcher] = None,
373 ):
374 """
375 Initialize the maintenance notifications for the connection.
377 Args:
378 maint_notifications_config (MaintNotificationsConfig): The configuration for maintenance notifications.
379 maint_notifications_pool_handler (Optional[MaintNotificationsPoolHandler]): The pool handler for maintenance notifications.
380 maintenance_state (MaintenanceState): The current maintenance state of the connection.
381 maintenance_notification_hash (Optional[int]): The current maintenance notification hash of the connection.
382 orig_host_address (Optional[str]): The original host address of the connection.
383 orig_socket_timeout (Optional[float]): The original socket timeout of the connection.
384 orig_socket_connect_timeout (Optional[float]): The original socket connect timeout of the connection.
385 oss_cluster_maint_notifications_handler (Optional[OSSMaintNotificationsHandler]): The OSS cluster handler for maintenance notifications.
386 parser (Optional[BaseParser]): The parser to use for maintenance notifications.
387 If not provided, the parser from the connection is used.
388 This is useful when the parser is created after this object.
389 """
390 self.maint_notifications_config = maint_notifications_config
391 self.maintenance_state = maintenance_state
392 self.maintenance_notification_hash = maintenance_notification_hash
394 if event_dispatcher is not None:
395 self.event_dispatcher = event_dispatcher
396 else:
397 self.event_dispatcher = EventDispatcher()
399 self._configure_maintenance_notifications(
400 maint_notifications_pool_handler,
401 orig_host_address,
402 orig_socket_timeout,
403 orig_socket_connect_timeout,
404 oss_cluster_maint_notifications_handler,
405 parser,
406 )
407 self._processed_start_maint_notifications = set()
408 self._skipped_end_maint_notifications = set()
410 @abstractmethod
411 def _get_parser(self) -> BaseParser:
412 pass
414 def _get_push_notifications_parser(self) -> Union[_HiredisParser, _RESP3Parser]:
415 parser = self._get_parser()
416 if not isinstance(parser, (_HiredisParser, _RESP3Parser)):
417 raise RedisError(
418 "Maintenance notifications are only supported with hiredis and RESP3 parsers!"
419 )
420 return parser
422 @abstractmethod
423 def _get_socket(self) -> Optional[socket.socket]:
424 pass
426 @abstractmethod
427 def get_protocol(self) -> Union[int, str]:
428 """
429 Returns:
430 The RESP protocol version, or ``None`` if the protocol is not specified,
431 in which case the server default will be used.
432 """
433 pass
435 @property
436 @abstractmethod
437 def host(self) -> str:
438 pass
440 @host.setter
441 @abstractmethod
442 def host(self, value: str):
443 pass
445 @property
446 @abstractmethod
447 def socket_timeout(self) -> Optional[Union[float, int]]:
448 pass
450 @socket_timeout.setter
451 @abstractmethod
452 def socket_timeout(self, value: Optional[Union[float, int]]):
453 pass
455 @property
456 @abstractmethod
457 def socket_connect_timeout(self) -> Optional[Union[float, int]]:
458 pass
460 @socket_connect_timeout.setter
461 @abstractmethod
462 def socket_connect_timeout(self, value: Optional[Union[float, int]]):
463 pass
465 @abstractmethod
466 def send_command(self, *args, **kwargs):
467 pass
469 @abstractmethod
470 def read_response(
471 self,
472 disable_decoding=False,
473 *,
474 timeout: Union[float, object] = SENTINEL,
475 disconnect_on_error=True,
476 push_request=False,
477 ):
478 pass
480 @abstractmethod
481 def disconnect(self, *args, **kwargs):
482 pass
484 @abstractmethod
485 def mark_for_reconnect(self):
486 pass
488 def _configure_maintenance_notifications(
489 self,
490 maint_notifications_pool_handler: Optional[
491 MaintNotificationsPoolHandler
492 ] = None,
493 orig_host_address=None,
494 orig_socket_timeout=None,
495 orig_socket_connect_timeout=None,
496 oss_cluster_maint_notifications_handler: Optional[
497 OSSMaintNotificationsHandler
498 ] = None,
499 parser: Optional[BaseParser] = None,
500 ):
501 """
502 Enable maintenance notifications by setting up
503 handlers and storing original connection parameters.
505 Should be used ONLY with parsers that support push notifications.
506 """
507 if (
508 not self.maint_notifications_config
509 or not self.maint_notifications_config.enabled
510 ):
511 self._maint_notifications_pool_handler = None
512 self._maint_notifications_connection_handler = None
513 self._oss_cluster_maint_notifications_handler = None
514 return
516 if not parser:
517 raise RedisError(
518 "To configure maintenance notifications, a parser must be provided!"
519 )
521 if not isinstance(parser, _HiredisParser) and not isinstance(
522 parser, _RESP3Parser
523 ):
524 raise RedisError(
525 "Maintenance notifications are only supported with hiredis and RESP3 parsers!"
526 )
528 if maint_notifications_pool_handler:
529 # Extract a reference to a new pool handler that copies all properties
530 # of the original one and has a different connection reference
531 # This is needed because when we attach the handler to the parser
532 # we need to make sure that the handler has a reference to the
533 # connection that the parser is attached to.
534 self._maint_notifications_pool_handler = (
535 maint_notifications_pool_handler.get_handler_for_connection()
536 )
537 self._maint_notifications_pool_handler.set_connection(self)
538 else:
539 self._maint_notifications_pool_handler = None
541 self._maint_notifications_connection_handler = (
542 MaintNotificationsConnectionHandler(self, self.maint_notifications_config)
543 )
545 if oss_cluster_maint_notifications_handler:
546 self._oss_cluster_maint_notifications_handler = (
547 oss_cluster_maint_notifications_handler
548 )
549 # Set up OSS cluster handler to parser
550 parser.set_oss_cluster_maint_push_handler(
551 self._oss_cluster_maint_notifications_handler.handle_notification
552 )
553 else:
554 self._oss_cluster_maint_notifications_handler = None
556 # Set up pool handler to parser if available
557 if self._maint_notifications_pool_handler:
558 parser.set_node_moving_push_handler(
559 self._maint_notifications_pool_handler.handle_notification
560 )
562 # Set up connection handler
563 parser.set_maintenance_push_handler(
564 self._maint_notifications_connection_handler.handle_notification
565 )
567 # Store original connection parameters
568 self.orig_host_address = orig_host_address if orig_host_address else self.host
569 self.orig_socket_timeout = (
570 orig_socket_timeout if orig_socket_timeout else self.socket_timeout
571 )
572 self.orig_socket_connect_timeout = (
573 orig_socket_connect_timeout
574 if orig_socket_connect_timeout
575 else self.socket_connect_timeout
576 )
578 def set_maint_notifications_pool_handler_for_connection(
579 self, maint_notifications_pool_handler: MaintNotificationsPoolHandler
580 ):
581 # Deep copy the pool handler to avoid sharing the same pool handler
582 # between multiple connections, because otherwise each connection will override
583 # the connection reference and the pool handler will only hold a reference
584 # to the last connection that was set.
585 maint_notifications_pool_handler_copy = (
586 maint_notifications_pool_handler.get_handler_for_connection()
587 )
589 maint_notifications_pool_handler_copy.set_connection(self)
590 parser = self._get_push_notifications_parser()
591 parser.set_node_moving_push_handler(
592 maint_notifications_pool_handler_copy.handle_notification
593 )
595 self._maint_notifications_pool_handler = maint_notifications_pool_handler_copy
597 # Update maintenance notification connection handler if it doesn't exist
598 if not self._maint_notifications_connection_handler:
599 self._maint_notifications_connection_handler = (
600 MaintNotificationsConnectionHandler(
601 self, maint_notifications_pool_handler.config
602 )
603 )
604 parser.set_maintenance_push_handler(
605 self._maint_notifications_connection_handler.handle_notification
606 )
607 else:
608 self._maint_notifications_connection_handler.config = (
609 maint_notifications_pool_handler.config
610 )
612 def set_maint_notifications_cluster_handler_for_connection(
613 self, oss_cluster_maint_notifications_handler: OSSMaintNotificationsHandler
614 ):
615 parser = self._get_push_notifications_parser()
616 parser.set_oss_cluster_maint_push_handler(
617 oss_cluster_maint_notifications_handler.handle_notification
618 )
619 # OSS cluster mode and pool-handler mode are mutually exclusive. Clear
620 # any node-moving/pool handler a default (RESP3 "auto") pool wired in
621 # __init__ so this existing connection is not configured with both.
622 parser.set_node_moving_push_handler(None)
623 self._maint_notifications_pool_handler = None
625 self._oss_cluster_maint_notifications_handler = (
626 oss_cluster_maint_notifications_handler
627 )
629 # Update maintenance notification connection handler if it doesn't exist
630 if not self._maint_notifications_connection_handler:
631 self._maint_notifications_connection_handler = (
632 MaintNotificationsConnectionHandler(
633 self, oss_cluster_maint_notifications_handler.config
634 )
635 )
636 parser.set_maintenance_push_handler(
637 self._maint_notifications_connection_handler.handle_notification
638 )
639 else:
640 self._maint_notifications_connection_handler.config = (
641 oss_cluster_maint_notifications_handler.config
642 )
644 def _should_enable_maint_notifications(self) -> bool:
645 # Maintenance notifications are sent only if RESP3 is active
646 # and maintenance notifications are enabled
647 # and we have a host to determine the endpoint type from.
648 host = getattr(self, "host", None)
649 return bool(
650 check_protocol_version(self.get_protocol(), 3)
651 and self.maint_notifications_config
652 and self.maint_notifications_config.enabled
653 and self._maint_notifications_connection_handler
654 and host is not None
655 )
657 def activate_maint_notifications_handling_if_enabled(self, check_health=True):
658 # When the maint_notifications_config enabled mode is "auto",
659 # we just log a warning if the handshake fails
660 # When the mode is enabled=True, we raise an exception in case of failure
661 if self._should_enable_maint_notifications():
662 self._enable_maintenance_notifications(
663 maint_notifications_config=self.maint_notifications_config,
664 check_health=check_health,
665 )
667 def _enable_maintenance_notifications(
668 self, maint_notifications_config: MaintNotificationsConfig, check_health=True
669 ):
670 # Kept for callers that enable maintenance notifications outside of the
671 # connection handshake. During on_connect the send and the response
672 # handling are split (see _send_maint_notifications_command /
673 # _handle_maint_notifications_response) so the reply can be pipelined
674 # with the rest of the handshake.
675 self._send_maint_notifications_command(
676 maint_notifications_config, check_health=check_health
677 )
678 self._handle_maint_notifications_response(maint_notifications_config)
680 def _maint_notifications_command_args(
681 self, maint_notifications_config: MaintNotificationsConfig
682 ):
683 host = getattr(self, "host", None)
684 if host is None:
685 raise ValueError(
686 "Cannot enable maintenance notifications for connection"
687 " object that doesn't have a host attribute."
688 )
689 endpoint_type = maint_notifications_config.get_endpoint_type(host, self)
690 return (
691 "CLIENT",
692 "MAINT_NOTIFICATIONS",
693 "ON",
694 "moving-endpoint-type",
695 endpoint_type.value,
696 )
698 def _send_maint_notifications_command(
699 self, maint_notifications_config: MaintNotificationsConfig, check_health=True
700 ):
701 self.send_command(
702 *self._maint_notifications_command_args(maint_notifications_config),
703 check_health=check_health,
704 )
706 def _add_maint_notifications_to_handshake(self, deferred_reads, check_health=True):
707 # If maintenance notifications are enabled for this connection, send the
708 # CLIENT MAINT_NOTIFICATIONS command with the rest of the handshake tail and
709 # defer reading its reply (appended to deferred_reads), rather than paying its
710 # own round-trip. When enabled == "auto" a failure is logged and swallowed;
711 # when enabled is True it raises.
712 if not self._should_enable_maint_notifications():
713 return
714 maint_notifications_config = self.maint_notifications_config
715 self._send_maint_notifications_command(
716 maint_notifications_config, check_health=check_health
717 )
718 deferred_reads.append(
719 lambda: self._handle_maint_notifications_response(
720 maint_notifications_config
721 )
722 )
724 def _handle_maint_notifications_response(
725 self, maint_notifications_config: MaintNotificationsConfig
726 ):
727 try:
728 response = self.read_response()
729 if not response or str_if_bytes(response) != "OK":
730 raise ResponseError(
731 "The server doesn't support maintenance notifications"
732 )
733 except Exception as e:
734 if (
735 isinstance(e, ResponseError)
736 and maint_notifications_config.enabled == "auto"
737 ):
738 # Log warning but don't fail the connection
739 logger.debug(f"Failed to enable maintenance notifications: {e}")
740 else:
741 raise
743 def _read_ok_or_raise(self, error_message):
744 # Read one handshake reply and require it to be "OK".
745 if str_if_bytes(self.read_response()) != "OK":
746 raise ConnectionError(error_message)
748 def _read_optional_setinfo(self):
749 # Read one CLIENT SETINFO reply. Older servers may not support the command,
750 # so a ResponseError is swallowed instead of failing the connection.
751 try:
752 self.read_response()
753 except ResponseError:
754 pass
756 def get_resolved_ip(self) -> Optional[str]:
757 """
758 Extract the resolved IP address from an
759 established connection or resolve it from the host.
761 First tries to get the actual IP from the socket (most accurate),
762 then falls back to DNS resolution if needed.
764 Returns:
765 str: The resolved IP address, or None if it cannot be determined
766 """
768 # Method 1: Try to get the actual IP from the established socket connection
769 # This is most accurate as it shows the exact IP being used
770 try:
771 conn_socket = self._get_socket()
772 if conn_socket is not None:
773 peer_addr = conn_socket.getpeername()
774 # For TCP sockets, peer_addr is typically a (host, port) tuple.
775 # AF_UNIX sockets report a path string instead, and indexing it
776 # would yield the first character of the path.
777 if isinstance(peer_addr, tuple) and peer_addr:
778 # Return just the host part
779 return peer_addr[0]
780 except (AttributeError, OSError):
781 # Socket might not be connected or getpeername() might fail
782 pass
784 # Method 2: Fallback to DNS resolution of the host
785 # This is less accurate but works when socket is not available
786 try:
787 host = getattr(self, "host", "localhost")
788 port = getattr(self, "port", 6379)
789 if host:
790 # Use getaddrinfo to resolve the hostname to IP
791 # This mimics what the connection would do during _connect()
792 addr_info = socket.getaddrinfo(
793 host, port, socket.AF_UNSPEC, socket.SOCK_STREAM
794 )
795 if addr_info:
796 # Return the IP from the first result
797 # addr_info[0] is (family, socktype, proto, canonname, sockaddr)
798 # sockaddr[0] is the IP address
799 return str(addr_info[0][4][0])
800 except (AttributeError, OSError, socket.gaierror):
801 # DNS resolution might fail
802 pass
804 return None
806 @property
807 def maintenance_state(self) -> MaintenanceState:
808 return self._maintenance_state
810 @maintenance_state.setter
811 def maintenance_state(self, state: "MaintenanceState"):
812 self._maintenance_state = state
814 def add_maint_start_notification(self, id: int):
815 self._processed_start_maint_notifications.add(id)
817 def get_processed_start_notifications(self) -> set:
818 return self._processed_start_maint_notifications
820 def add_skipped_end_notification(self, id: int):
821 self._skipped_end_maint_notifications.add(id)
823 def get_skipped_end_notifications(self) -> set:
824 return self._skipped_end_maint_notifications
826 def reset_received_notifications(self):
827 self._processed_start_maint_notifications.clear()
828 self._skipped_end_maint_notifications.clear()
830 def getpeername(self):
831 """
832 Returns the peer name of the connection.
833 """
834 conn_socket = self._get_socket()
835 if conn_socket:
836 return conn_socket.getpeername()[0]
837 return None
839 def update_current_socket_timeout(self, relaxed_timeout: Optional[float] = None):
840 conn_socket = self._get_socket()
841 if conn_socket:
842 timeout = relaxed_timeout if relaxed_timeout != -1 else self.socket_timeout
843 # if the current timeout is 0 it means we are in the middle of a can_read call
844 # in this case we don't want to change the timeout because the operation
845 # is non-blocking and should return immediately
846 # Changing the state from non-blocking to blocking in the middle of a read operation
847 # will lead to a deadlock
848 if conn_socket.gettimeout() != 0:
849 conn_socket.settimeout(timeout)
850 # Deliberately outside the guard above. The parser caches this value
851 # to restore after a per-call timeout override, so while a can_read
852 # probe holds the socket at 0 the cache *is* the pending restore
853 # value: writing it is how the new timeout gets armed without
854 # flipping the socket back to blocking mid-read. Skipping it here
855 # instead would let the probe restore the pre-relaxation timeout and
856 # lose a maintenance relaxation for the life of the connection.
857 self.update_parser_timeout(timeout)
859 def update_parser_timeout(self, timeout: Optional[float] = None):
860 parser = self._get_parser()
861 if parser and parser._buffer:
862 # Both parsers cache this value to restore after a per-call timeout
863 # override, so both must receive exactly what was armed on the
864 # socket - including None, which means "block indefinitely".
865 if isinstance(parser, _RESP3Parser):
866 parser._buffer.socket_timeout = timeout
867 elif isinstance(parser, _HiredisParser):
868 parser._socket_timeout = timeout
870 def set_tmp_settings(
871 self,
872 tmp_host_address: Optional[Union[str, object]] = SENTINEL,
873 tmp_relaxed_timeout: Optional[float] = -1,
874 ):
875 """
876 SENTINEL keeps the host unchanged. -1 keeps the relaxed timeout unchanged.
877 """
878 if tmp_host_address and tmp_host_address != SENTINEL:
879 self.host = str(tmp_host_address)
880 if tmp_relaxed_timeout != -1:
881 self.socket_timeout = tmp_relaxed_timeout
882 self.socket_connect_timeout = tmp_relaxed_timeout
884 def reset_tmp_settings(
885 self,
886 reset_host_address: bool = False,
887 reset_relaxed_timeout: bool = False,
888 ):
889 if reset_host_address:
890 self.host = self.orig_host_address
891 if reset_relaxed_timeout:
892 self.socket_timeout = self.orig_socket_timeout
893 self.socket_connect_timeout = self.orig_socket_connect_timeout
896class AbstractConnection(MaintNotificationsAbstractConnection, ConnectionInterface):
897 "Manages communication to and from a Redis server"
899 @deprecated_args(
900 args_to_warn=["lib_name", "lib_version"],
901 reason="Use 'driver_info' parameter instead. "
902 "lib_name and lib_version will be removed in a future version.",
903 )
904 def __init__(
905 self,
906 db: int = 0,
907 password: Optional[str] = None,
908 socket_timeout: Optional[float] = DEFAULT_SOCKET_TIMEOUT,
909 socket_connect_timeout: Optional[float] = DEFAULT_SOCKET_CONNECT_TIMEOUT,
910 retry_on_timeout: bool = False,
911 retry_on_error: Union[Iterable[Type[Exception]], object] = SENTINEL,
912 encoding: str = "utf-8",
913 encoding_errors: str = "strict",
914 decode_responses: bool = False,
915 parser_class=DefaultParser,
916 socket_read_size: int = DEFAULT_SOCKET_READ_SIZE,
917 health_check_interval: int = 0,
918 client_name: Optional[str] = None,
919 lib_name: Union[Optional[str], object] = SENTINEL,
920 lib_version: Union[Optional[str], object] = SENTINEL,
921 driver_info: Union[Optional[DriverInfo], object] = SENTINEL,
922 username: Optional[str] = None,
923 retry: Union[Any, None] = None,
924 redis_connect_func: Optional[Callable[[], None]] = None,
925 credential_provider: Optional[CredentialProvider] = None,
926 protocol: Optional[int] = None,
927 legacy_responses: bool = True,
928 command_packer: Optional[Callable[[], None]] = None,
929 event_dispatcher: Optional[EventDispatcher] = None,
930 maint_notifications_config: Optional[MaintNotificationsConfig] = None,
931 maint_notifications_pool_handler: Optional[
932 MaintNotificationsPoolHandler
933 ] = None,
934 maintenance_state: "MaintenanceState" = MaintenanceState.NONE,
935 maintenance_notification_hash: Optional[int] = None,
936 orig_host_address: Optional[str] = None,
937 orig_socket_timeout: Optional[float] = None,
938 orig_socket_connect_timeout: Optional[float] = None,
939 oss_cluster_maint_notifications_handler: Optional[
940 OSSMaintNotificationsHandler
941 ] = None,
942 himport_registry: HImportRegistry | None = None,
943 ):
944 """
945 Initialize a new Connection.
947 To specify a retry policy for specific errors, first set
948 `retry_on_error` to a list of the error/s to retry on, then set
949 `retry` to a valid `Retry` object.
950 To retry on TimeoutError, `retry_on_timeout` can also be set to `True`.
952 Parameters
953 ----------
954 driver_info : DriverInfo, optional
955 Driver metadata for CLIENT SETINFO. If provided, lib_name and lib_version
956 are ignored. If not provided, a DriverInfo will be created from lib_name
957 and lib_version. Explicit None disables CLIENT SETINFO.
958 lib_name : str, optional
959 **Deprecated.** Use driver_info instead. Library name for CLIENT SETINFO.
960 lib_version : str, optional
961 **Deprecated.** Use driver_info instead. Library version for CLIENT SETINFO.
962 """
963 if (username or password) and credential_provider is not None:
964 raise DataError(
965 "'username' and 'password' cannot be passed along with 'credential_"
966 "provider'. Please provide only one of the following arguments: \n"
967 "1. 'password' and (optional) 'username'\n"
968 "2. 'credential_provider'"
969 )
970 if event_dispatcher is None:
971 self._event_dispatcher = EventDispatcher()
972 else:
973 self._event_dispatcher = event_dispatcher
974 self.pid = os.getpid()
975 self.db = db
976 self.client_name = client_name
978 # Handle driver_info: if provided, use it; otherwise create from lib_name/lib_version.
979 self.driver_info = resolve_driver_info(driver_info, lib_name, lib_version)
981 self.credential_provider = credential_provider
982 self.password = password
983 self.username = username
984 self._socket_timeout = socket_timeout
985 if socket_connect_timeout is None:
986 socket_connect_timeout = socket_timeout
987 self._socket_connect_timeout = socket_connect_timeout
988 self.retry_on_timeout = retry_on_timeout
989 if retry_on_error is SENTINEL:
990 retry_on_errors_list = []
991 else:
992 retry_on_errors_list = list(retry_on_error)
993 if retry_on_timeout:
994 # Add TimeoutError to the errors list to retry on
995 retry_on_errors_list.append(TimeoutError)
996 self.retry_on_error = retry_on_errors_list
997 if retry or self.retry_on_error:
998 if retry is None:
999 self.retry = Retry(NoBackoff(), 1)
1000 else:
1001 # deep-copy the Retry object as it is mutable
1002 self.retry = copy.deepcopy(retry)
1003 if self.retry_on_error:
1004 # Update the retry's supported errors with the specified errors
1005 self.retry.update_supported_errors(self.retry_on_error)
1006 else:
1007 self.retry = Retry(NoBackoff(), 0)
1008 self.health_check_interval = health_check_interval
1009 self.next_health_check = 0
1010 self.redis_connect_func = redis_connect_func
1011 self.encoder = Encoder(encoding, encoding_errors, decode_responses)
1012 self.handshake_metadata = None
1013 self._sock = None
1014 self._socket_read_size = socket_read_size
1015 self._connect_callbacks = []
1016 self._buffer_cutoff = 6000
1017 self._re_auth_token: Optional[TokenInterface] = None
1018 try:
1019 p = int(protocol)
1020 except TypeError:
1021 p = DEFAULT_RESP_VERSION
1022 except ValueError:
1023 raise ConnectionError("protocol must be an integer")
1024 else:
1025 if p < 2 or p > 3:
1026 raise ConnectionError("protocol must be either 2 or 3")
1027 self.protocol = p
1028 self.legacy_responses = legacy_responses
1029 if self.protocol == 3 and parser_class == _RESP2Parser:
1030 # If the protocol is 3 but the parser is RESP2, change it to RESP3
1031 # This is needed because the parser might be set before the protocol
1032 # or might be provided as a kwarg to the constructor
1033 # We need to react on discrepancy only for RESP2 and RESP3
1034 # as hiredis supports both
1035 parser_class = _RESP3Parser
1036 self.set_parser(parser_class)
1038 self._command_packer = self._construct_command_packer(command_packer)
1039 self._should_reconnect = False
1041 # HIMPORT client-side state. `himport_registry` is the shared client-level
1042 # registry (empty if unconfigured) and persists across reconnects.
1043 self.himport_registry = himport_registry
1044 self._reset_himport_state()
1046 # Set up maintenance notifications
1047 MaintNotificationsAbstractConnection.__init__(
1048 self,
1049 maint_notifications_config,
1050 maint_notifications_pool_handler,
1051 maintenance_state,
1052 maintenance_notification_hash,
1053 orig_host_address,
1054 orig_socket_timeout,
1055 orig_socket_connect_timeout,
1056 oss_cluster_maint_notifications_handler,
1057 self._parser,
1058 event_dispatcher=self._event_dispatcher,
1059 )
1061 def __repr__(self):
1062 repr_args = ",".join([f"{k}={v}" for k, v in self.repr_pieces()])
1063 return f"<{self.__class__.__module__}.{self.__class__.__name__}({repr_args})>"
1065 @abstractmethod
1066 def repr_pieces(self):
1067 pass
1069 def __del__(self):
1070 try:
1071 self.disconnect()
1072 except Exception:
1073 pass
1075 @property
1076 def is_connected(self) -> bool:
1077 return self._sock is not None
1079 def _construct_command_packer(self, packer):
1080 if packer is not None:
1081 return packer
1082 elif HIREDIS_AVAILABLE:
1083 return HiredisRespSerializer()
1084 else:
1085 return PythonRespSerializer(self._buffer_cutoff, self.encoder.encode)
1087 def register_connect_callback(self, callback):
1088 """
1089 Register a callback to be called when the connection is established either
1090 initially or reconnected. This allows listeners to issue commands that
1091 are ephemeral to the connection, for example pub/sub subscription or
1092 key tracking. The callback must be a _method_ and will be kept as
1093 a weak reference.
1094 """
1095 wm = weakref.WeakMethod(callback)
1096 if wm not in self._connect_callbacks:
1097 self._connect_callbacks.append(wm)
1099 def deregister_connect_callback(self, callback):
1100 """
1101 De-register a previously registered callback. It will no-longer receive
1102 notifications on connection events. Calling this is not required when the
1103 listener goes away, since the callbacks are kept as weak methods.
1104 """
1105 try:
1106 self._connect_callbacks.remove(weakref.WeakMethod(callback))
1107 except ValueError:
1108 pass
1110 def set_parser(self, parser_class):
1111 """
1112 Creates a new instance of parser_class with socket size:
1113 _socket_read_size and assigns it to the parser for the connection
1114 :param parser_class: The required parser class
1115 """
1116 self._parser = parser_class(socket_read_size=self._socket_read_size)
1118 def _get_parser(self) -> Union[_HiredisParser, _RESP3Parser, _RESP2Parser]:
1119 return self._parser
1121 def connect(self):
1122 "Connects to the Redis server if not already connected"
1123 # try once the socket connect with the handshake, retry the whole
1124 # connect/handshake flow based on retry policy
1125 self.retry.call_with_retry(
1126 lambda: self.connect_check_health(
1127 check_health=True, retry_socket_connect=False
1128 ),
1129 lambda error: self.disconnect(error),
1130 )
1132 def connect_check_health(
1133 self, check_health: bool = True, retry_socket_connect: bool = True
1134 ):
1135 if self._sock:
1136 return
1137 # Track actual retry attempts for error reporting
1138 actual_retry_attempts = [0]
1140 def failure_callback(error, failure_count):
1141 actual_retry_attempts[0] = failure_count
1142 self.disconnect(error=error, failure_count=failure_count)
1144 try:
1145 if retry_socket_connect:
1146 sock = self.retry.call_with_retry(
1147 self._connect,
1148 failure_callback,
1149 with_failure_count=True,
1150 )
1151 else:
1152 sock = self._connect()
1153 except socket.timeout:
1154 e = TimeoutError("Timeout connecting to server")
1155 record_error_count(
1156 server_address=self.host,
1157 server_port=self.port,
1158 network_peer_address=self.host,
1159 network_peer_port=self.port,
1160 error_type=e,
1161 retry_attempts=actual_retry_attempts[0],
1162 )
1163 raise e
1164 except OSError as e:
1165 e = ConnectionError(self._error_message(e))
1166 record_error_count(
1167 server_address=getattr(self, "host", None),
1168 server_port=getattr(self, "port", None),
1169 network_peer_address=getattr(self, "host", None),
1170 network_peer_port=getattr(self, "port", None),
1171 error_type=e,
1172 retry_attempts=actual_retry_attempts[0],
1173 )
1174 raise e
1176 self._sock = sock
1177 try:
1178 if self.redis_connect_func is None:
1179 # Use the default on_connect function
1180 self.on_connect_check_health(check_health=check_health)
1181 else:
1182 # Use the passed function redis_connect_func
1183 self.redis_connect_func(self)
1184 except RedisError:
1185 # clean up after any error in on_connect
1186 self.disconnect()
1187 raise
1189 # run any user callbacks. right now the only internal callback
1190 # is for pubsub channel/pattern resubscription
1191 # first, remove any dead weakrefs
1192 self._connect_callbacks = [ref for ref in self._connect_callbacks if ref()]
1193 for ref in self._connect_callbacks:
1194 callback = ref()
1195 if callback:
1196 callback(self)
1198 @abstractmethod
1199 def _connect(self):
1200 pass
1202 @abstractmethod
1203 def _host_error(self):
1204 pass
1206 def _error_message(self, exception):
1207 return format_error_message(self._host_error(), exception)
1209 def _reset_himport_state(self):
1210 # A fresh server session has no prepared HIMPORT fieldsets, so the next
1211 # himport_set must re-prepare on this connection. ``_himport_prepared`` maps
1212 # fieldset name -> the version prepared on the server; ``_himport_reconciled
1213 # _revision`` is the registry revision this connection last reconciled discards
1214 # against. Both are reset on connect/disconnect since the session is gone.
1215 self._himport_prepared: dict[str, int] = {}
1216 self._himport_reconciled_revision: int = 0
1218 def on_connect(self):
1219 self.on_connect_check_health(check_health=True)
1221 def on_connect_check_health(self, check_health: bool = True):
1222 "Initialize the connection, authenticate and select a database"
1223 # A fresh socket is a new server session: no prepared HIMPORT fieldsets.
1224 self._reset_himport_state()
1225 self._parser.on_connect(self)
1226 parser = self._parser
1228 auth_args = None
1229 # if credential provider or username and/or password are set, authenticate
1230 if self.credential_provider or (self.username or self.password):
1231 cred_provider = (
1232 self.credential_provider
1233 or UsernamePasswordCredentialProvider(self.username, self.password)
1234 )
1235 auth_args = cred_provider.get_credentials()
1237 # if resp version is specified and we have auth args,
1238 # we need to send them via HELLO
1239 if auth_args and check_protocol_version(self.protocol, 3):
1240 if isinstance(self._parser, _RESP2Parser):
1241 self.set_parser(_RESP3Parser)
1242 # update cluster exception classes
1243 self._parser.EXCEPTION_CLASSES = parser.EXCEPTION_CLASSES
1244 self._parser.on_connect(self)
1245 if len(auth_args) == 1:
1246 auth_args = ["default", auth_args[0]]
1247 # avoid checking health here -- PING will fail if we try
1248 # to check the health prior to the AUTH
1249 self.send_command(
1250 "HELLO", self.protocol, "AUTH", *auth_args, check_health=False
1251 )
1252 self.handshake_metadata = self.read_response()
1253 # if response.get(b"proto") != self.protocol and response.get(
1254 # "proto"
1255 # ) != self.protocol:
1256 # raise ConnectionError("Invalid RESP version")
1257 elif auth_args:
1258 # avoid checking health here -- PING will fail if we try
1259 # to check the health prior to the AUTH
1260 self.send_command("AUTH", *auth_args, check_health=False)
1262 try:
1263 auth_response = self.read_response()
1264 except AuthenticationWrongNumberOfArgsError:
1265 # a username and password were specified but the Redis
1266 # server seems to be < 6.0.0 which expects a single password
1267 # arg. retry auth with just the password.
1268 # https://github.com/andymccurdy/redis-py/issues/1274
1269 self.send_command("AUTH", auth_args[-1], check_health=False)
1270 auth_response = self.read_response()
1272 if str_if_bytes(auth_response) != "OK":
1273 raise AuthenticationError("Invalid Username or Password")
1275 # if resp version is specified, switch to it
1276 elif check_protocol_version(self.protocol, 3):
1277 if isinstance(self._parser, _RESP2Parser):
1278 self.set_parser(_RESP3Parser)
1279 # update cluster exception classes
1280 self._parser.EXCEPTION_CLASSES = parser.EXCEPTION_CLASSES
1281 self._parser.on_connect(self)
1282 self.send_command("HELLO", self.protocol, check_health=check_health)
1283 self.handshake_metadata = self.read_response()
1284 if (
1285 self.handshake_metadata.get(b"proto") != self.protocol
1286 and self.handshake_metadata.get("proto") != self.protocol
1287 ):
1288 raise ConnectionError("Invalid RESP version")
1290 # The tail of the handshake (optional CLIENT MAINT_NOTIFICATIONS, then
1291 # CLIENT SETNAME / SETINFO / SELECT) does not affect control flow -- the replies
1292 # are only validated or discarded. So we optimize the flow: send every command
1293 # first (without blocking on a reply between them), then read the replies back in
1294 # send order. All requests are on the wire before we block on the first read, so
1295 # the whole tail costs a single round-trip instead of one per command. AUTH/HELLO
1296 # stays a separate round-trip above because its reply drives the RESP2->RESP3
1297 # parser upgrade, the pre-6.0 AUTH retry, and proto validation.
1298 #
1299 # deferred_reads holds one zero-arg handler per command sent below; each reads
1300 # exactly one reply (in send order) and validates it. Per-command check_health
1301 # reproduces the original behavior: at most one health PING/PONG fires before the
1302 # first tail command (only when no HELLO/AUTH ran, e.g. RESP2 no-auth), and it is
1303 # self-contained so it never desyncs the deferred replies.
1304 #
1305 # TODO: apply the same optimization to the async handshake in
1306 # redis/asyncio/connection.py, where not all tail commands are handled this way
1307 # yet (e.g. maintenance notifications are still issued separately there).
1308 deferred_reads = []
1310 # Maintenance notifications (RESP3-only, opt-in) go first when enabled, so their
1311 # reply is read together with the rest of the tail.
1312 self._add_maint_notifications_to_handshake(deferred_reads, check_health)
1314 # if a client_name is given, set it
1315 if self.client_name:
1316 self.send_command(
1317 "CLIENT", "SETNAME", self.client_name, check_health=check_health
1318 )
1319 deferred_reads.append(
1320 functools.partial(self._read_ok_or_raise, "Error setting client name")
1321 )
1323 # Set the library name and version from driver_info. Older servers may not
1324 # support CLIENT SETINFO, so any ResponseError to these replies is swallowed.
1325 if self.driver_info and self.driver_info.formatted_name:
1326 self.send_command(
1327 "CLIENT",
1328 "SETINFO",
1329 "LIB-NAME",
1330 self.driver_info.formatted_name,
1331 check_health=check_health,
1332 )
1333 deferred_reads.append(self._read_optional_setinfo)
1335 if self.driver_info and self.driver_info.lib_version:
1336 self.send_command(
1337 "CLIENT",
1338 "SETINFO",
1339 "LIB-VER",
1340 self.driver_info.lib_version,
1341 check_health=check_health,
1342 )
1343 deferred_reads.append(self._read_optional_setinfo)
1345 # if a database is specified, switch to it
1346 if self.db:
1347 self.send_command("SELECT", self.db, check_health=check_health)
1348 deferred_reads.append(
1349 functools.partial(self._read_ok_or_raise, "Invalid Database")
1350 )
1352 # Read the deferred replies in the order the commands were sent.
1353 for read_and_validate_response in deferred_reads:
1354 read_and_validate_response()
1356 def disconnect(self, *args, **kwargs):
1357 "Disconnects from the Redis server"
1358 # The server session is gone, so any HIMPORT fieldsets prepared on this
1359 # socket no longer exist; reset the tracking.
1360 self._reset_himport_state()
1361 self._parser.on_disconnect()
1363 conn_sock = self._sock
1364 self._sock = None
1365 # reset the reconnect flag
1366 self.reset_should_reconnect()
1368 if conn_sock is None:
1369 return
1371 if os.getpid() == self.pid:
1372 try:
1373 conn_sock.shutdown(socket.SHUT_RDWR)
1374 except (OSError, TypeError):
1375 pass
1377 try:
1378 conn_sock.close()
1379 except OSError:
1380 pass
1382 error = kwargs.get("error")
1383 failure_count = kwargs.get("failure_count")
1384 health_check_failed = kwargs.get("health_check_failed")
1386 if error:
1387 if health_check_failed:
1388 close_reason = CloseReason.HEALTHCHECK_FAILED
1389 else:
1390 close_reason = CloseReason.ERROR
1392 if failure_count is not None and failure_count > self.retry.get_retries():
1393 record_error_count(
1394 server_address=self.host,
1395 server_port=self.port,
1396 network_peer_address=self.host,
1397 network_peer_port=self.port,
1398 error_type=error,
1399 retry_attempts=failure_count,
1400 )
1402 record_connection_closed(
1403 close_reason=close_reason,
1404 error_type=error,
1405 )
1406 else:
1407 record_connection_closed(
1408 close_reason=CloseReason.APPLICATION_CLOSE,
1409 )
1411 if self.maintenance_state == MaintenanceState.MAINTENANCE:
1412 # this block will be executed only if the connection was in maintenance state
1413 # and the connection was closed.
1414 # The state change won't be applied on connections that are in Moving state
1415 # because their state and configurations will be handled when the moving ttl expires.
1416 self.reset_tmp_settings(reset_relaxed_timeout=True)
1417 self.maintenance_state = MaintenanceState.NONE
1418 # reset the sets that keep track of received start maint
1419 # notifications and skipped end maint notifications
1420 self.reset_received_notifications()
1422 def mark_for_reconnect(self):
1423 self._should_reconnect = True
1425 def should_reconnect(self):
1426 return self._should_reconnect
1428 def reset_should_reconnect(self):
1429 self._should_reconnect = False
1431 def _send_ping(self):
1432 """Send PING, expect PONG in return"""
1433 self.send_command("PING", check_health=False)
1434 if str_if_bytes(self.read_response()) != "PONG":
1435 raise ConnectionError("Bad response from PING health check")
1437 def _ping_failed(self, error, failure_count):
1438 """Function to call when PING fails"""
1439 self.disconnect(
1440 error=error, failure_count=failure_count, health_check_failed=True
1441 )
1443 def check_health(self):
1444 """Check the health of the connection with a PING/PONG"""
1445 if self.health_check_interval and time.monotonic() > self.next_health_check:
1446 self.retry.call_with_retry(
1447 self._send_ping,
1448 self._ping_failed,
1449 with_failure_count=True,
1450 )
1452 def send_packed_command(self, command, check_health=True):
1453 """Send an already packed command to the Redis server"""
1454 if not self._sock:
1455 self.connect_check_health(check_health=False)
1456 # guard against health check recursion
1457 if check_health:
1458 self.check_health()
1459 try:
1460 if isinstance(command, str):
1461 command = [command]
1462 for item in command:
1463 self._sock.sendall(item)
1464 except socket.timeout as e:
1465 add_debug_log_for_connection_failure(self, e, "writing command")
1466 self.disconnect()
1467 raise TimeoutError("Timeout writing to socket")
1468 except OSError as e:
1469 add_debug_log_for_connection_failure(self, e, "writing command")
1470 self.disconnect()
1471 if len(e.args) == 1:
1472 errno, errmsg = "UNKNOWN", e.args[0]
1473 else:
1474 errno = e.args[0]
1475 errmsg = e.args[1]
1476 raise ConnectionError(f"Error {errno} while writing to socket. {errmsg}.")
1477 except BaseException as e:
1478 # BaseExceptions can be raised when a socket send operation is not
1479 # finished, e.g. due to a timeout. Ideally, a caller could then re-try
1480 # to send un-sent data. However, the send_packed_command() API
1481 # does not support it so there is no point in keeping the connection open.
1482 add_debug_log_for_connection_failure(self, e, "writing command")
1483 self.disconnect()
1484 raise
1486 def send_command(self, *args, **kwargs):
1487 """Pack and send a command to the Redis server"""
1488 self.send_packed_command(
1489 self._command_packer.pack(*args),
1490 check_health=kwargs.get("check_health", True),
1491 )
1493 def can_read(self, timeout: float = 0) -> bool:
1494 """Poll the socket to see if there's data that can be read."""
1495 # TODO: Rename this API; it detects pending data or dirty/closed
1496 # connection state, not only whether application data can be read.
1497 sock = self._sock
1498 if not sock:
1499 self.connect()
1501 host_error = self._host_error()
1503 try:
1504 return self._parser.can_read(timeout)
1506 except OSError as e:
1507 self.disconnect()
1508 raise ConnectionError(f"Error while reading from {host_error}: {e.args}")
1510 def read_response(
1511 self,
1512 disable_decoding=False,
1513 *,
1514 timeout: Union[float, object] = SENTINEL,
1515 disconnect_on_error=True,
1516 push_request=False,
1517 ):
1518 """Read the response from a previously sent command"""
1520 host_error = self._host_error()
1522 try:
1523 if self.protocol in ["3", 3]:
1524 response = self._parser.read_response(
1525 disable_decoding=disable_decoding,
1526 push_request=push_request,
1527 timeout=timeout,
1528 )
1529 else:
1530 response = self._parser.read_response(
1531 disable_decoding=disable_decoding, timeout=timeout
1532 )
1533 except socket.timeout as e:
1534 if disconnect_on_error:
1535 add_debug_log_for_connection_failure(self, e, "reading response")
1536 self.disconnect()
1537 raise TimeoutError(f"Timeout reading from {host_error}")
1538 except TimeoutError as e:
1539 # The parsers raise redis.exceptions.TimeoutError, which is not an
1540 # OSError, so without this branch it would fall through to
1541 # BaseException and keep the parser's undecorated message. Re-raise
1542 # it with the host, matching what the async stack already reports.
1543 if disconnect_on_error:
1544 add_debug_log_for_connection_failure(self, e, "reading response")
1545 self.disconnect()
1546 raise TimeoutError(f"Timeout reading from {host_error}")
1547 except OSError as e:
1548 if disconnect_on_error:
1549 add_debug_log_for_connection_failure(self, e, "reading response")
1550 self.disconnect()
1551 raise ConnectionError(f"Error while reading from {host_error} : {e.args}")
1552 except BaseException as e:
1553 # Also by default close in case of BaseException. A lot of code
1554 # relies on this behaviour when doing Command/Response pairs.
1555 # See #1128.
1556 if disconnect_on_error:
1557 add_debug_log_for_connection_failure(self, e, "reading response")
1558 self.disconnect()
1559 raise
1561 if self.health_check_interval:
1562 self.next_health_check = time.monotonic() + self.health_check_interval
1564 if isinstance(response, ResponseError):
1565 try:
1566 raise response
1567 finally:
1568 del response # avoid creating ref cycles
1569 return response
1571 def pack_command(self, *args):
1572 """Pack a series of arguments into the Redis protocol"""
1573 return self._command_packer.pack(*args)
1575 def pack_commands(self, commands):
1576 """Pack multiple commands into the Redis protocol"""
1577 output = []
1578 pieces = []
1579 buffer_length = 0
1580 buffer_cutoff = self._buffer_cutoff
1582 for cmd in commands:
1583 for chunk in self._command_packer.pack(*cmd):
1584 chunklen = len(chunk)
1585 if (
1586 buffer_length > buffer_cutoff
1587 or chunklen > buffer_cutoff
1588 or isinstance(chunk, memoryview)
1589 ):
1590 if pieces:
1591 output.append(SYM_EMPTY.join(pieces))
1592 buffer_length = 0
1593 pieces = []
1595 if chunklen > buffer_cutoff or isinstance(chunk, memoryview):
1596 output.append(chunk)
1597 else:
1598 pieces.append(chunk)
1599 buffer_length += chunklen
1601 if pieces:
1602 output.append(SYM_EMPTY.join(pieces))
1603 return output
1605 def get_protocol(self) -> Union[int, str]:
1606 return self.protocol
1608 @property
1609 def handshake_metadata(self) -> Union[Dict[bytes, bytes], Dict[str, str]]:
1610 return self._handshake_metadata
1612 @handshake_metadata.setter
1613 def handshake_metadata(self, value: Union[Dict[bytes, bytes], Dict[str, str]]):
1614 self._handshake_metadata = value
1616 def set_re_auth_token(self, token: TokenInterface):
1617 self._re_auth_token = token
1619 def re_auth(self):
1620 if self._re_auth_token is not None:
1621 self.send_command(
1622 "AUTH",
1623 self._re_auth_token.try_get("oid"),
1624 self._re_auth_token.get_value(),
1625 )
1626 self.read_response()
1627 self._re_auth_token = None
1629 def _get_socket(self) -> Optional[socket.socket]:
1630 return self._sock
1632 @property
1633 def socket_timeout(self) -> Optional[Union[float, int]]:
1634 return self._socket_timeout
1636 @socket_timeout.setter
1637 def socket_timeout(self, value: Optional[Union[float, int]]):
1638 self._socket_timeout = value
1640 @property
1641 def socket_connect_timeout(self) -> Optional[Union[float, int]]:
1642 return self._socket_connect_timeout
1644 @socket_connect_timeout.setter
1645 def socket_connect_timeout(self, value: Optional[Union[float, int]]):
1646 self._socket_connect_timeout = value
1648 def extract_connection_details(self) -> str:
1649 """
1650 Render the connection's identity, maintenance state and effective timeouts.
1652 This is what the debug logs use to explain a failed or timed out command:
1654 - ``host`` vs ``orig host`` says whether the connection still points at the
1655 node being moved away from, or has already been repointed at the new one.
1656 - ``state`` says whether maintenance handling touched this connection at
1657 all, so an unaffected node's connections are distinguishable.
1658 - ``socket_timeout`` vs ``active read timeout`` says which timeout the read
1659 actually ran under. They diverge for a command that was already in flight
1660 when the relaxed timeout was applied.
1661 """
1662 if self._sock is None:
1663 return "not connected"
1665 socket_address = None
1666 active_read_timeout = None
1667 try:
1668 socket_name = self._sock.getsockname()
1669 # AF_UNIX sockets report a path string rather than a (host, port) tuple
1670 if isinstance(socket_name, tuple) and len(socket_name) > 1:
1671 socket_address = socket_name[1]
1672 # The timeout armed on the socket is what a blocking read is really
1673 # using, which can lag self.socket_timeout for an in-flight command.
1674 active_read_timeout = self._sock.gettimeout()
1675 except (AttributeError, OSError):
1676 pass
1678 state = getattr(self.maintenance_state, "value", self.maintenance_state)
1679 return (
1680 f"connected to ip {self.get_resolved_ip()}, "
1681 f"local socket port: {socket_address}, "
1682 f"host: {self._host_error()} "
1683 f"(orig: {getattr(self, 'orig_host_address', None)}), "
1684 f"state: {state}, "
1685 f"socket_timeout: {self.socket_timeout} "
1686 f"(orig: {getattr(self, 'orig_socket_timeout', None)}), "
1687 f"active read timeout: {active_read_timeout}, "
1688 f"should_reconnect: {self.should_reconnect()}, "
1689 f"notification_hash: {self.maintenance_notification_hash}"
1690 )
1693class Connection(AbstractConnection):
1694 "Manages TCP communication to and from a Redis server"
1696 def __init__(
1697 self,
1698 host="localhost",
1699 port=6379,
1700 socket_keepalive=True,
1701 socket_keepalive_options=SENTINEL,
1702 socket_type=0,
1703 **kwargs,
1704 ):
1705 """
1706 Initialize a TCP connection.
1708 Parameters
1709 ----------
1710 socket_keepalive : bool
1711 If `True`, TCP keepalive is enabled for TCP socket connections.
1712 socket_keepalive_options : Mapping[int, int | bytes] | object | None
1713 Mapping of TCP keepalive socket option constants to values, for
1714 example `{socket.TCP_KEEPIDLE: 30}`. If left unspecified, redis-py
1715 uses TCP keepalive defaults when `socket_keepalive` is enabled:
1716 idle 30 seconds, interval 5 seconds, and 3 probes. Platform-specific
1717 options that are not available are skipped. Pass `None` or `{}` to
1718 avoid setting additional TCP keepalive options.
1719 """
1720 self._host = host
1721 # bool subclasses int; port=True would become privileged port 1.
1722 # Numeric strings stay valid. Callers still pass "6379".
1723 if isinstance(port, bool):
1724 raise TypeError("port must be an integer, not bool")
1725 if isinstance(port, str):
1726 try:
1727 port = int(port)
1728 except ValueError:
1729 raise TypeError("port must be an integer, not str") from None
1730 elif not isinstance(port, int):
1731 raise TypeError(f"port must be an integer, not {type(port).__name__}")
1732 if not 0 <= port <= 65535:
1733 raise ValueError(f"port must be in 0..65535, got {port}")
1734 self.port = port
1735 self.socket_keepalive = socket_keepalive
1736 if socket_keepalive_options is SENTINEL:
1737 socket_keepalive_options = get_default_socket_keepalive_options()
1738 self.socket_keepalive_options = socket_keepalive_options or {}
1739 self.socket_type = socket_type
1740 super().__init__(**kwargs)
1742 def repr_pieces(self):
1743 pieces = [("host", self.host), ("port", self.port), ("db", self.db)]
1744 if self.client_name:
1745 pieces.append(("client_name", self.client_name))
1746 return pieces
1748 def _connect(self):
1749 "Create a TCP socket connection"
1750 # we want to mimic what socket.create_connection does to support
1751 # ipv4/ipv6, but we want to set options prior to calling
1752 # socket.connect()
1754 # Last caught connection error.
1755 # Re-thrown if we are unable to connect to any of the options returned
1756 # by getaddrinfo.
1757 # Note that we must clear this variable before returning - otherwise,
1758 # a caught err's traceback points to this frame, which points to err.
1759 # Clearing this lets refcounting reclaim the exception immediately
1760 # without deferring to the python garbage collector.
1761 err = None
1763 for res in socket.getaddrinfo(
1764 self.host, self.port, self.socket_type, socket.SOCK_STREAM
1765 ):
1766 family, socktype, proto, canonname, socket_address = res
1767 sock = None
1768 try:
1769 sock = socket.socket(family, socktype, proto)
1770 # TCP_NODELAY
1771 sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
1773 # TCP_KEEPALIVE
1774 if self.socket_keepalive:
1775 sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
1776 for k, v in self.socket_keepalive_options.items():
1777 sock.setsockopt(socket.IPPROTO_TCP, k, v)
1779 # set the socket_connect_timeout before we connect
1780 sock.settimeout(self.socket_connect_timeout)
1782 # connect
1783 sock.connect(socket_address)
1785 # set the socket_timeout now that we're connected
1786 sock.settimeout(self.socket_timeout)
1788 # If a previous connection attempt failed, clear the error
1789 err = None
1791 return sock
1793 except OSError as _:
1794 err = _
1795 if sock is not None:
1796 try:
1797 sock.shutdown(socket.SHUT_RDWR) # ensure a clean close
1798 except OSError:
1799 pass
1800 sock.close()
1802 if err is not None:
1803 try:
1804 raise err
1805 finally:
1806 # Ensure we clear local references to caught exceptions
1807 err = None
1808 raise OSError("socket.getaddrinfo returned an empty list")
1810 def _host_error(self):
1811 return f"{self.host}:{self.port}"
1813 @property
1814 def host(self) -> str:
1815 return self._host
1817 @host.setter
1818 def host(self, value: str):
1819 self._host = value
1822class CacheProxyConnection(MaintNotificationsAbstractConnection, ConnectionInterface):
1823 DUMMY_CACHE_VALUE = b"foo"
1824 MIN_ALLOWED_VERSION = "7.4.0"
1825 DEFAULT_SERVER_NAME = "redis"
1827 def __init__(
1828 self,
1829 conn: ConnectionInterface,
1830 cache: CacheInterface,
1831 pool_lock: threading.RLock,
1832 ):
1833 self.pid = os.getpid()
1834 self._conn = conn
1835 self.retry = self._conn.retry
1836 self.host = self._conn.host
1837 self.port = self._conn.port
1838 self.db = self._conn.db
1839 self._event_dispatcher = self._conn._event_dispatcher
1840 self.credential_provider = conn.credential_provider
1841 self._pool_lock = pool_lock
1842 self._cache = cache
1843 self._cache_lock = threading.RLock()
1844 self._current_command_cache_key = None
1845 self._current_options = None
1846 self.register_connect_callback(self._enable_tracking_callback)
1848 if isinstance(self._conn, MaintNotificationsAbstractConnection):
1849 MaintNotificationsAbstractConnection.__init__(
1850 self,
1851 self._conn.maint_notifications_config,
1852 self._conn._maint_notifications_pool_handler,
1853 self._conn.maintenance_state,
1854 self._conn.maintenance_notification_hash,
1855 self._conn.host,
1856 self._conn.socket_timeout,
1857 self._conn.socket_connect_timeout,
1858 self._conn._oss_cluster_maint_notifications_handler,
1859 self._conn._get_parser(),
1860 event_dispatcher=self._conn.event_dispatcher,
1861 )
1863 def repr_pieces(self):
1864 return self._conn.repr_pieces()
1866 @property
1867 def is_connected(self) -> bool:
1868 return self._conn.is_connected
1870 def register_connect_callback(self, callback):
1871 self._conn.register_connect_callback(callback)
1873 def deregister_connect_callback(self, callback):
1874 self._conn.deregister_connect_callback(callback)
1876 def set_parser(self, parser_class):
1877 self._conn.set_parser(parser_class)
1879 def set_maint_notifications_pool_handler_for_connection(
1880 self, maint_notifications_pool_handler
1881 ):
1882 if isinstance(self._conn, MaintNotificationsAbstractConnection):
1883 self._conn.set_maint_notifications_pool_handler_for_connection(
1884 maint_notifications_pool_handler
1885 )
1887 def set_maint_notifications_cluster_handler_for_connection(
1888 self, oss_cluster_maint_notifications_handler
1889 ):
1890 if isinstance(self._conn, MaintNotificationsAbstractConnection):
1891 self._conn.set_maint_notifications_cluster_handler_for_connection(
1892 oss_cluster_maint_notifications_handler
1893 )
1895 def get_protocol(self):
1896 return self._conn.get_protocol()
1898 def connect(self):
1899 self._conn.connect()
1901 server_name = self._conn.handshake_metadata.get(b"server", None)
1902 if server_name is None:
1903 server_name = self._conn.handshake_metadata.get("server", None)
1904 server_ver = self._conn.handshake_metadata.get(b"version", None)
1905 if server_ver is None:
1906 server_ver = self._conn.handshake_metadata.get("version", None)
1907 if server_ver is None or server_name is None:
1908 raise ConnectionError("Cannot retrieve information about server version")
1910 server_ver = ensure_string(server_ver)
1911 server_name = ensure_string(server_name)
1913 if (
1914 server_name != self.DEFAULT_SERVER_NAME
1915 or compare_versions(server_ver, self.MIN_ALLOWED_VERSION) == 1
1916 ):
1917 raise ConnectionError(
1918 "To maximize compatibility with all Redis products, client-side caching is supported by Redis 7.4 or later" # noqa: E501
1919 )
1921 def on_connect(self):
1922 self._conn.on_connect()
1924 def disconnect(self, *args, **kwargs):
1925 with self._cache_lock:
1926 self._cache.flush()
1927 self._conn.disconnect(*args, **kwargs)
1929 def check_health(self):
1930 self._conn.check_health()
1932 def send_packed_command(self, command, check_health=True):
1933 # TODO: Investigate if it's possible to unpack command
1934 # or extract keys from packed command
1935 # Pre-packed commands are not individually cacheable, so make sure the
1936 # next read_response does not try to cache their reply under a stale key.
1937 self._current_command_cache_key = None
1938 self._conn.send_packed_command(command)
1940 def send_command(self, *args, **kwargs):
1941 self._process_pending_invalidations()
1943 with self._cache_lock:
1944 # Command is write command or not allowed
1945 # to be cached.
1946 if not self._cache.is_cachable(
1947 CacheKey(command=args[0], redis_keys=(), redis_args=())
1948 ):
1949 self._current_command_cache_key = None
1950 self._conn.send_command(*args, **kwargs)
1951 return
1953 # Eligibility and keyability are two separate questions, and both must be answered
1954 # yes before a reply may be stored. The command metadata answers the first; the
1955 # presence of ``keys`` answers the second, because a command's key positions are
1956 # supplied by its command method rather than derived here. A cacheable command
1957 # whose invocation carries no key list is therefore a gap in what this client has
1958 # been taught, not an error: send it normally and cache nothing.
1959 keys = kwargs.get("keys")
1960 if keys is None:
1961 self._current_command_cache_key = None
1962 self._conn.send_command(*args, **kwargs)
1963 return
1965 # Creates cache key.
1966 self._current_command_cache_key = CacheKey(
1967 command=args[0], redis_keys=tuple(keys), redis_args=args
1968 )
1970 with self._cache_lock:
1971 # We have to trigger invalidation processing in case if
1972 # it was cached by another connection to avoid
1973 # queueing invalidations in stale connections.
1974 if self._cache.get(self._current_command_cache_key):
1975 entry = self._cache.get(self._current_command_cache_key)
1977 with self._pool_lock:
1978 while entry.connection_ref.can_read():
1979 try:
1980 entry.connection_ref.read_response(
1981 push_request=True,
1982 timeout=0,
1983 disconnect_on_error=False,
1984 )
1985 except TimeoutError:
1986 break
1988 # Re-check: if the entry was invalidated during the drain,
1989 # fall through to send the command over the network.
1990 if self._cache.get(self._current_command_cache_key):
1991 return
1993 # Set temporary entry value to prevent
1994 # race condition from another connection.
1995 self._cache.set(
1996 CacheEntry(
1997 cache_key=self._current_command_cache_key,
1998 cache_value=self.DUMMY_CACHE_VALUE,
1999 status=CacheEntryStatus.IN_PROGRESS,
2000 connection_ref=self._conn,
2001 )
2002 )
2004 # Send command over socket only if it's allowed
2005 # read-only command that not yet cached.
2006 self._conn.send_command(*args, **kwargs)
2008 def can_read(self, timeout: float = 0) -> bool:
2009 # TODO: Rename this API; it detects pending data or dirty/closed
2010 # connection state, not only whether application data can be read.
2011 return self._conn.can_read(timeout)
2013 def read_response(
2014 self,
2015 disable_decoding=False,
2016 *,
2017 timeout: Union[float, object] = SENTINEL,
2018 disconnect_on_error=True,
2019 push_request=False,
2020 ):
2021 with self._cache_lock:
2022 # Check if command response exists in a cache and it's not in progress.
2023 if self._current_command_cache_key is not None:
2024 if (
2025 self._cache.get(self._current_command_cache_key) is not None
2026 and self._cache.get(self._current_command_cache_key).status
2027 != CacheEntryStatus.IN_PROGRESS
2028 ):
2029 res = copy.deepcopy(
2030 self._cache.get(self._current_command_cache_key).cache_value
2031 )
2032 self._current_command_cache_key = None
2033 record_csc_request(
2034 result=CSCResult.HIT,
2035 )
2036 record_csc_network_saved(
2037 bytes_saved=len(res) if hasattr(res, "__len__") else 0,
2038 )
2039 return res
2040 record_csc_request(
2041 result=CSCResult.MISS,
2042 )
2044 response = self._conn.read_response(
2045 disable_decoding=disable_decoding,
2046 timeout=timeout,
2047 disconnect_on_error=disconnect_on_error,
2048 push_request=push_request,
2049 )
2051 with self._cache_lock:
2052 # Prevent not-allowed command from caching.
2053 if self._current_command_cache_key is None:
2054 return response
2055 # If response is None prevent from caching.
2056 if response is None:
2057 self._cache.delete_by_cache_keys([self._current_command_cache_key])
2058 return response
2060 cache_entry = self._cache.get(self._current_command_cache_key)
2062 # Cache only responses that still valid
2063 # and wasn't invalidated by another connection in meantime.
2064 if cache_entry is not None:
2065 cache_entry.status = CacheEntryStatus.VALID
2066 cache_entry.cache_value = response
2067 self._cache.set(cache_entry)
2069 self._current_command_cache_key = None
2071 return response
2073 def pack_command(self, *args):
2074 return self._conn.pack_command(*args)
2076 def pack_commands(self, commands):
2077 return self._conn.pack_commands(commands)
2079 # HIMPORT state lives on the wrapped connection (HIMPORT is never cacheable);
2080 # delegate so callers treat the proxy like a plain connection and never need
2081 # to know a proxy is in play.
2082 @property
2083 def himport_registry(self):
2084 return self._conn.himport_registry
2086 @property
2087 def _himport_prepared(self):
2088 return self._conn._himport_prepared
2090 @_himport_prepared.setter
2091 def _himport_prepared(self, value):
2092 # Delegate reassignment to the wrapped connection, mirroring
2093 # ``_himport_reconciled_revision``. Production code only mutates the dict
2094 # in place, but ``_reset_himport_state`` (and any future caller) reassigns
2095 # it, and a getter-only property here would raise ``AttributeError`` only
2096 # when client-side caching is enabled -- a caching-specific latent trap.
2097 self._conn._himport_prepared = value
2099 @property
2100 def _himport_reconciled_revision(self):
2101 return self._conn._himport_reconciled_revision
2103 @_himport_reconciled_revision.setter
2104 def _himport_reconciled_revision(self, value):
2105 self._conn._himport_reconciled_revision = value
2107 @property
2108 def handshake_metadata(self) -> Union[Dict[bytes, bytes], Dict[str, str]]:
2109 return self._conn.handshake_metadata
2111 def set_re_auth_token(self, token: TokenInterface):
2112 self._conn.set_re_auth_token(token)
2114 def re_auth(self):
2115 self._conn.re_auth()
2117 def mark_for_reconnect(self):
2118 self._conn.mark_for_reconnect()
2120 def should_reconnect(self):
2121 return self._conn.should_reconnect()
2123 def reset_should_reconnect(self):
2124 self._conn.reset_should_reconnect()
2126 @property
2127 def host(self) -> str:
2128 return self._conn.host
2130 @host.setter
2131 def host(self, value: str):
2132 self._conn.host = value
2134 @property
2135 def socket_timeout(self) -> Optional[Union[float, int]]:
2136 return self._conn.socket_timeout
2138 @socket_timeout.setter
2139 def socket_timeout(self, value: Optional[Union[float, int]]):
2140 self._conn.socket_timeout = value
2142 @property
2143 def socket_connect_timeout(self) -> Optional[Union[float, int]]:
2144 return self._conn.socket_connect_timeout
2146 @socket_connect_timeout.setter
2147 def socket_connect_timeout(self, value: Optional[Union[float, int]]):
2148 self._conn.socket_connect_timeout = value
2150 @property
2151 def _maint_notifications_connection_handler(
2152 self,
2153 ) -> Optional[MaintNotificationsConnectionHandler]:
2154 if isinstance(self._conn, MaintNotificationsAbstractConnection):
2155 return self._conn._maint_notifications_connection_handler
2157 @_maint_notifications_connection_handler.setter
2158 def _maint_notifications_connection_handler(
2159 self, value: Optional[MaintNotificationsConnectionHandler]
2160 ):
2161 self._conn._maint_notifications_connection_handler = value
2163 def _get_socket(self) -> Optional[socket.socket]:
2164 if isinstance(self._conn, MaintNotificationsAbstractConnection):
2165 return self._conn._get_socket()
2166 else:
2167 raise NotImplementedError(
2168 "Maintenance notifications are not supported by this connection type"
2169 )
2171 def _get_maint_notifications_connection_instance(
2172 self, connection
2173 ) -> MaintNotificationsAbstractConnection:
2174 """
2175 Validate that connection instance supports maintenance notifications.
2176 With this helper method we ensure that we are working
2177 with the correct connection type.
2178 After twe validate that connection instance supports maintenance notifications
2179 we can safely return the connection instance
2180 as MaintNotificationsAbstractConnection.
2181 """
2182 if not isinstance(connection, MaintNotificationsAbstractConnection):
2183 raise NotImplementedError(
2184 "Maintenance notifications are not supported by this connection type"
2185 )
2186 else:
2187 return connection
2189 @property
2190 def maintenance_state(self) -> MaintenanceState:
2191 con = self._get_maint_notifications_connection_instance(self._conn)
2192 return con.maintenance_state
2194 @maintenance_state.setter
2195 def maintenance_state(self, state: MaintenanceState):
2196 con = self._get_maint_notifications_connection_instance(self._conn)
2197 con.maintenance_state = state
2199 def getpeername(self):
2200 con = self._get_maint_notifications_connection_instance(self._conn)
2201 return con.getpeername()
2203 def get_resolved_ip(self):
2204 con = self._get_maint_notifications_connection_instance(self._conn)
2205 return con.get_resolved_ip()
2207 def update_current_socket_timeout(self, relaxed_timeout: Optional[float] = None):
2208 con = self._get_maint_notifications_connection_instance(self._conn)
2209 con.update_current_socket_timeout(relaxed_timeout)
2211 def set_tmp_settings(
2212 self,
2213 tmp_host_address: Optional[str] = None,
2214 tmp_relaxed_timeout: Optional[float] = -1,
2215 ):
2216 con = self._get_maint_notifications_connection_instance(self._conn)
2217 con.set_tmp_settings(tmp_host_address, tmp_relaxed_timeout)
2219 def reset_tmp_settings(
2220 self,
2221 reset_host_address: bool = False,
2222 reset_relaxed_timeout: bool = False,
2223 ):
2224 con = self._get_maint_notifications_connection_instance(self._conn)
2225 con.reset_tmp_settings(reset_host_address, reset_relaxed_timeout)
2227 def _connect(self):
2228 self._conn._connect()
2230 def _host_error(self):
2231 return self._conn._host_error()
2233 def _enable_tracking_callback(self, conn: ConnectionInterface) -> None:
2234 conn.send_command("CLIENT", "TRACKING", "ON")
2235 conn.read_response()
2236 conn._parser.set_invalidation_push_handler(self._on_invalidation_callback)
2238 def _process_pending_invalidations(self):
2239 while self.can_read():
2240 try:
2241 self._conn.read_response(
2242 push_request=True, timeout=0, disconnect_on_error=False
2243 )
2244 except TimeoutError:
2245 break
2247 def _on_invalidation_callback(self, data: List[Union[str, Optional[List[bytes]]]]):
2248 with self._cache_lock:
2249 # Flush cache when DB flushed on server-side
2250 if data[1] is None:
2251 self._cache.flush()
2252 else:
2253 keys_deleted = self._cache.delete_by_redis_keys(data[1])
2255 if len(keys_deleted) > 0:
2256 record_csc_eviction(
2257 count=len(keys_deleted),
2258 reason=CSCReason.INVALIDATION,
2259 )
2261 def extract_connection_details(self) -> str:
2262 return self._conn.extract_connection_details()
2265class SSLConnection(Connection):
2266 """Manages SSL connections to and from the Redis server(s).
2267 This class extends the Connection class, adding SSL functionality, and making
2268 use of ssl.SSLContext (https://docs.python.org/3/library/ssl.html#ssl.SSLContext)
2269 """ # noqa
2271 def __init__(
2272 self,
2273 ssl_keyfile=None,
2274 ssl_certfile=None,
2275 ssl_cert_reqs="required",
2276 ssl_include_verify_flags: Optional[List["VerifyFlags"]] = None,
2277 ssl_exclude_verify_flags: Optional[List["VerifyFlags"]] = None,
2278 ssl_ca_certs=None,
2279 ssl_ca_data=None,
2280 ssl_check_hostname=True,
2281 ssl_ca_path=None,
2282 ssl_password=None,
2283 ssl_validate_ocsp=False,
2284 ssl_validate_ocsp_stapled=False,
2285 ssl_ocsp_context=None,
2286 ssl_ocsp_expected_cert=None,
2287 ssl_min_version=None,
2288 ssl_ciphers=None,
2289 **kwargs,
2290 ):
2291 """Constructor
2293 Args:
2294 ssl_keyfile: Path to an ssl private key. Defaults to None.
2295 ssl_certfile: Path to an ssl certificate. Defaults to None.
2296 ssl_cert_reqs: The string value for the SSLContext.verify_mode (none, optional, required),
2297 or an ssl.VerifyMode. Defaults to "required".
2298 ssl_include_verify_flags: A list of flags to be included in the SSLContext.verify_flags. Defaults to None.
2299 ssl_exclude_verify_flags: A list of flags to be excluded from the SSLContext.verify_flags. Defaults to None.
2300 ssl_ca_certs: The path to a file of concatenated CA certificates in PEM format. Defaults to None.
2301 ssl_ca_data: Either an ASCII string of one or more PEM-encoded certificates or a bytes-like object of DER-encoded certificates.
2302 ssl_check_hostname: If set, match the hostname during the SSL handshake. Defaults to True.
2303 ssl_ca_path: The path to a directory containing several CA certificates in PEM format. Defaults to None.
2304 ssl_password: Password for unlocking an encrypted private key. Defaults to None.
2306 ssl_validate_ocsp: If set, perform a full ocsp validation (i.e not a stapled verification)
2307 ssl_validate_ocsp_stapled: If set, perform a validation on a stapled ocsp response
2308 ssl_ocsp_context: A fully initialized OpenSSL.SSL.Context object to be used in verifying the ssl_ocsp_expected_cert
2309 ssl_ocsp_expected_cert: A PEM armoured string containing the expected certificate to be returned from the ocsp verification service.
2310 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.
2311 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.
2313 Raises:
2314 RedisError
2315 """ # noqa
2316 if not SSL_AVAILABLE:
2317 raise RedisError("Python wasn't built with SSL support")
2319 self.keyfile = ssl_keyfile
2320 self.certfile = ssl_certfile
2321 if ssl_cert_reqs is None:
2322 ssl_cert_reqs = ssl.CERT_NONE
2323 elif isinstance(ssl_cert_reqs, str):
2324 CERT_REQS = { # noqa: N806
2325 "none": ssl.CERT_NONE,
2326 "optional": ssl.CERT_OPTIONAL,
2327 "required": ssl.CERT_REQUIRED,
2328 }
2329 if ssl_cert_reqs not in CERT_REQS:
2330 raise RedisError(
2331 f"Invalid SSL Certificate Requirements Flag: {ssl_cert_reqs}"
2332 )
2333 ssl_cert_reqs = CERT_REQS[ssl_cert_reqs]
2334 self.cert_reqs = ssl_cert_reqs
2335 self.ssl_include_verify_flags = ssl_include_verify_flags
2336 self.ssl_exclude_verify_flags = ssl_exclude_verify_flags
2337 self.ca_certs = ssl_ca_certs
2338 self.ca_data = ssl_ca_data
2339 self.ca_path = ssl_ca_path
2340 self.check_hostname = (
2341 ssl_check_hostname if self.cert_reqs != ssl.CERT_NONE else False
2342 )
2343 self.certificate_password = ssl_password
2344 self.ssl_validate_ocsp = ssl_validate_ocsp
2345 self.ssl_validate_ocsp_stapled = ssl_validate_ocsp_stapled
2346 self.ssl_ocsp_context = ssl_ocsp_context
2347 self.ssl_ocsp_expected_cert = ssl_ocsp_expected_cert
2348 self.ssl_min_version = ssl_min_version
2349 self.ssl_ciphers = ssl_ciphers
2350 super().__init__(**kwargs)
2352 def _connect(self):
2353 """
2354 Wrap the socket with SSL support, handling potential errors.
2355 """
2356 sock = super()._connect()
2357 try:
2358 return self._wrap_socket_with_ssl(sock)
2359 except (OSError, RedisError):
2360 sock.close()
2361 raise
2363 def _wrap_socket_with_ssl(self, sock):
2364 """
2365 Wraps the socket with SSL support.
2367 Args:
2368 sock: The plain socket to wrap with SSL.
2370 Returns:
2371 An SSL wrapped socket.
2372 """
2373 context = ssl.create_default_context()
2374 context.check_hostname = self.check_hostname
2375 context.verify_mode = self.cert_reqs
2376 if self.ssl_include_verify_flags:
2377 for flag in self.ssl_include_verify_flags:
2378 context.verify_flags |= flag
2379 if self.ssl_exclude_verify_flags:
2380 for flag in self.ssl_exclude_verify_flags:
2381 context.verify_flags &= ~flag
2382 if self.certfile or self.keyfile:
2383 context.load_cert_chain(
2384 certfile=self.certfile,
2385 keyfile=self.keyfile,
2386 password=self.certificate_password,
2387 )
2388 if (
2389 self.ca_certs is not None
2390 or self.ca_path is not None
2391 or self.ca_data is not None
2392 ):
2393 context.load_verify_locations(
2394 cafile=self.ca_certs, capath=self.ca_path, cadata=self.ca_data
2395 )
2396 if self.ssl_min_version is not None:
2397 context.minimum_version = self.ssl_min_version
2398 if self.ssl_ciphers:
2399 context.set_ciphers(self.ssl_ciphers)
2400 if self.ssl_validate_ocsp is True and CRYPTOGRAPHY_AVAILABLE is False:
2401 raise RedisError("cryptography is not installed.")
2403 if self.ssl_validate_ocsp_stapled and self.ssl_validate_ocsp:
2404 raise RedisError(
2405 "Either an OCSP staple or pure OCSP connection must be validated "
2406 "- not both."
2407 )
2409 sslsock = context.wrap_socket(sock, server_hostname=self.host)
2411 # validation for the stapled case
2412 if self.ssl_validate_ocsp_stapled:
2413 import OpenSSL
2415 from .ocsp import ocsp_staple_verifier
2417 # if a context is provided use it - otherwise, a basic context
2418 if self.ssl_ocsp_context is None:
2419 staple_ctx = OpenSSL.SSL.Context(OpenSSL.SSL.SSLv23_METHOD)
2420 staple_ctx.use_certificate_file(self.certfile)
2421 staple_ctx.use_privatekey_file(self.keyfile)
2422 else:
2423 staple_ctx = self.ssl_ocsp_context
2425 staple_ctx.set_ocsp_client_callback(
2426 ocsp_staple_verifier, self.ssl_ocsp_expected_cert
2427 )
2429 # need another socket
2430 con = OpenSSL.SSL.Connection(staple_ctx, socket.socket())
2431 con.request_ocsp()
2432 con.connect((self.host, self.port))
2433 con.do_handshake()
2434 con.shutdown()
2435 return sslsock
2437 # pure ocsp validation
2438 if self.ssl_validate_ocsp is True and CRYPTOGRAPHY_AVAILABLE:
2439 from .ocsp import OCSPVerifier
2441 o = OCSPVerifier(sslsock, self.host, self.port, self.ca_certs)
2442 if o.is_valid():
2443 return sslsock
2444 else:
2445 raise ConnectionError("ocsp validation error")
2446 return sslsock
2449class UnixDomainSocketConnection(AbstractConnection):
2450 "Manages UDS communication to and from a Redis server"
2452 def __init__(self, path="", socket_timeout=DEFAULT_SOCKET_TIMEOUT, **kwargs):
2453 super().__init__(**kwargs)
2454 self.path = path
2455 self.socket_timeout = socket_timeout
2457 def repr_pieces(self):
2458 pieces = [("path", self.path), ("db", self.db)]
2459 if self.client_name:
2460 pieces.append(("client_name", self.client_name))
2461 return pieces
2463 def _connect(self):
2464 "Create a Unix domain socket connection"
2465 sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
2466 sock.settimeout(self.socket_connect_timeout)
2467 try:
2468 sock.connect(self.path)
2469 except OSError:
2470 # Prevent ResourceWarnings for unclosed sockets.
2471 try:
2472 sock.shutdown(socket.SHUT_RDWR) # ensure a clean close
2473 except OSError:
2474 pass
2475 sock.close()
2476 raise
2477 sock.settimeout(self.socket_timeout)
2478 return sock
2480 def _host_error(self):
2481 return self.path
2484FALSE_STRINGS = ("0", "F", "FALSE", "N", "NO")
2487def to_bool(value):
2488 if value is None or value == "":
2489 return None
2490 if isinstance(value, str) and value.upper() in FALSE_STRINGS:
2491 return False
2492 return bool(value)
2495def parse_ssl_verify_flags(value):
2496 # flags are passed in as a string representation of a list,
2497 # e.g. VERIFY_X509_STRICT, VERIFY_X509_PARTIAL_CHAIN
2498 verify_flags_str = value.replace("[", "").replace("]", "")
2500 verify_flags = []
2501 for flag in verify_flags_str.split(","):
2502 flag = flag.strip()
2503 if not hasattr(VerifyFlags, flag):
2504 raise ValueError(f"Invalid ssl verify flag: {flag}")
2505 verify_flags.append(getattr(VerifyFlags, flag))
2506 return verify_flags
2509def parse_retry_on_error(value):
2510 # exception class names are passed as a comma-separated list,
2511 # e.g. ConnectionError,TimeoutError
2512 retry_on_error = []
2513 for name in value.replace("[", "").replace("]", "").split(","):
2514 name = name.strip()
2515 if not name:
2516 raise ValueError("Empty retry_on_error entry")
2517 exc = getattr(redis_exceptions, name, None)
2518 if not (isinstance(exc, type) and issubclass(exc, Exception)):
2519 raise ValueError(f"Unknown redis exception {name!r}")
2520 retry_on_error.append(exc)
2521 return retry_on_error
2524URL_QUERY_ARGUMENT_PARSERS = {
2525 "db": int,
2526 "socket_timeout": float,
2527 "socket_connect_timeout": float,
2528 "socket_read_size": int,
2529 "socket_keepalive": to_bool,
2530 "retry_on_timeout": to_bool,
2531 "retry_on_error": parse_retry_on_error,
2532 "max_connections": int,
2533 "health_check_interval": int,
2534 "ssl_check_hostname": to_bool,
2535 "ssl_include_verify_flags": parse_ssl_verify_flags,
2536 "ssl_exclude_verify_flags": parse_ssl_verify_flags,
2537 "ssl_min_version": int,
2538 "timeout": float,
2539 "protocol": int,
2540 "legacy_responses": to_bool,
2541}
2544def parse_url(url):
2545 # Scheme names are case-insensitive (RFC 3986), so normalize before the
2546 # prefix check; the "://" is required so a URL like "redis:foo" (which
2547 # urlparse would still report as the "redis" scheme) is rejected.
2548 if not url.lower().startswith(("redis://", "rediss://", "unix://")):
2549 raise ValueError(
2550 "Redis URL must specify one of the following "
2551 "schemes (redis://, rediss://, unix://)"
2552 )
2554 url = urlparse(url)
2555 kwargs = {}
2557 for name, value in parse_qs(url.query).items():
2558 if value and len(value) > 0:
2559 # parse_qs() already percent-decodes query values, so use the value
2560 # as-is; unquoting again here would double-decode (e.g. "%2520" ->
2561 # "%20" -> " "). See issue #4208.
2562 value = value[0]
2563 parser = URL_QUERY_ARGUMENT_PARSERS.get(name)
2564 if parser:
2565 try:
2566 kwargs[name] = parser(value)
2567 except (TypeError, ValueError):
2568 raise ValueError(f"Invalid value for '{name}' in connection URL.")
2569 else:
2570 kwargs[name] = value
2572 if url.username:
2573 kwargs["username"] = unquote(url.username)
2574 if url.password:
2575 kwargs["password"] = unquote(url.password)
2577 # We only support redis://, rediss:// and unix:// schemes.
2578 if url.scheme == "unix":
2579 if url.path:
2580 kwargs["path"] = unquote(url.path)
2581 kwargs["connection_class"] = UnixDomainSocketConnection
2583 else: # implied: url.scheme in ("redis", "rediss"):
2584 if url.hostname:
2585 kwargs["host"] = unquote(url.hostname)
2586 if url.port is not None:
2587 kwargs["port"] = int(url.port)
2589 # If there's a path argument, use it as the db argument if a
2590 # querystring value wasn't specified
2591 if url.path and "db" not in kwargs:
2592 try:
2593 kwargs["db"] = int(unquote(url.path).replace("/", ""))
2594 except (AttributeError, ValueError):
2595 pass
2597 if url.scheme == "rediss":
2598 kwargs["connection_class"] = SSLConnection
2600 return kwargs
2603_CP = TypeVar("_CP", bound="ConnectionPool")
2606class ConnectionPoolInterface(ABC):
2607 @abstractmethod
2608 def get_protocol(self):
2609 pass
2611 @abstractmethod
2612 def reset(self):
2613 pass
2615 @abstractmethod
2616 @deprecated_args(
2617 args_to_warn=["*"],
2618 reason="Use get_connection() without args instead",
2619 version="5.3.0",
2620 )
2621 def get_connection(
2622 self, command_name: Optional[str], *keys, **options
2623 ) -> ConnectionInterface:
2624 pass
2626 @abstractmethod
2627 def get_encoder(self):
2628 pass
2630 @abstractmethod
2631 def release(self, connection: ConnectionInterface):
2632 pass
2634 @abstractmethod
2635 def disconnect(self, inuse_connections: bool = True):
2636 pass
2638 @abstractmethod
2639 def close(self):
2640 pass
2642 @abstractmethod
2643 def set_retry(self, retry: Retry):
2644 pass
2646 @abstractmethod
2647 def re_auth_callback(self, token: TokenInterface):
2648 pass
2650 @abstractmethod
2651 def get_connection_count(self) -> list[tuple[int, dict]]:
2652 """
2653 Returns a connection count (both idle and in use).
2654 """
2655 pass
2658class MaintNotificationsAbstractConnectionPool:
2659 """
2660 Abstract class for handling maintenance notifications logic.
2661 This class is mixed into the ConnectionPool classes.
2663 This class is not intended to be used directly!
2665 All logic related to maintenance notifications and
2666 connection pool handling is encapsulated in this class.
2667 """
2669 def __init__(
2670 self,
2671 maint_notifications_config: Optional[MaintNotificationsConfig] = None,
2672 oss_cluster_maint_notifications_handler: Optional[
2673 OSSMaintNotificationsHandler
2674 ] = None,
2675 **kwargs,
2676 ):
2677 # Initialize maintenance notifications
2678 is_protocol_supported = check_protocol_version(kwargs.get("protocol"), 3)
2680 if maint_notifications_config is None and is_protocol_supported:
2681 maint_notifications_config = MaintNotificationsConfig()
2683 if maint_notifications_config and maint_notifications_config.enabled:
2684 if not is_protocol_supported:
2685 raise RedisError(
2686 "Maintenance notifications handlers on connection are only supported with RESP version 3"
2687 )
2689 self._event_dispatcher = kwargs.get("event_dispatcher", None)
2690 if self._event_dispatcher is None:
2691 self._event_dispatcher = EventDispatcher()
2693 self._maint_notifications_pool_handler = MaintNotificationsPoolHandler(
2694 self, maint_notifications_config
2695 )
2696 if oss_cluster_maint_notifications_handler:
2697 self._oss_cluster_maint_notifications_handler = (
2698 oss_cluster_maint_notifications_handler
2699 )
2700 self._update_connection_kwargs_for_maint_notifications(
2701 oss_cluster_maint_notifications_handler=self._oss_cluster_maint_notifications_handler
2702 )
2703 self._maint_notifications_pool_handler = None
2704 else:
2705 self._oss_cluster_maint_notifications_handler = None
2706 self._maint_notifications_pool_handler = MaintNotificationsPoolHandler(
2707 self, maint_notifications_config
2708 )
2710 self._update_connection_kwargs_for_maint_notifications(
2711 maint_notifications_pool_handler=self._maint_notifications_pool_handler
2712 )
2713 else:
2714 self._maint_notifications_pool_handler = None
2715 self._oss_cluster_maint_notifications_handler = None
2717 @property
2718 @abstractmethod
2719 def connection_kwargs(self) -> Dict[str, Any]:
2720 pass
2722 @connection_kwargs.setter
2723 @abstractmethod
2724 def connection_kwargs(self, value: Dict[str, Any]):
2725 pass
2727 @abstractmethod
2728 def _get_pool_lock(self) -> threading.RLock:
2729 pass
2731 @abstractmethod
2732 def _get_free_connections(self) -> Iterable["MaintNotificationsAbstractConnection"]:
2733 pass
2735 @abstractmethod
2736 def _get_in_use_connections(
2737 self,
2738 ) -> Iterable["MaintNotificationsAbstractConnection"]:
2739 pass
2741 def maint_notifications_enabled(self):
2742 """
2743 Returns:
2744 True if the maintenance notifications are enabled, False otherwise.
2745 The maintenance notifications config is stored in the pool handler.
2746 If the pool handler is not set, the maintenance notifications are not enabled.
2747 """
2748 if self._oss_cluster_maint_notifications_handler:
2749 maint_notifications_config = (
2750 self._oss_cluster_maint_notifications_handler.config
2751 )
2752 else:
2753 maint_notifications_config = (
2754 self._maint_notifications_pool_handler.config
2755 if self._maint_notifications_pool_handler
2756 else None
2757 )
2759 return maint_notifications_config and maint_notifications_config.enabled
2761 def update_maint_notifications_config(
2762 self,
2763 maint_notifications_config: MaintNotificationsConfig,
2764 oss_cluster_maint_notifications_handler: Optional[
2765 OSSMaintNotificationsHandler
2766 ] = None,
2767 ):
2768 """
2769 Updates the maintenance notifications configuration.
2770 This method should be called only if the pool was created
2771 without enabling the maintenance notifications and
2772 in a later point in time maintenance notifications
2773 are requested to be enabled.
2774 """
2775 if (
2776 self.maint_notifications_enabled()
2777 and not maint_notifications_config.enabled
2778 ):
2779 raise ValueError(
2780 "Cannot disable maintenance notifications after enabling them"
2781 )
2782 if oss_cluster_maint_notifications_handler:
2783 self._oss_cluster_maint_notifications_handler = (
2784 oss_cluster_maint_notifications_handler
2785 )
2786 # OSS cluster mode and pool-handler mode are mutually exclusive
2787 # (see __init__). A pool created with the default RESP3 "auto"
2788 # config wires a pool handler before this method runs; clear it so
2789 # new and existing connections are not configured with both handlers.
2790 self._maint_notifications_pool_handler = None
2791 else:
2792 # first update pool settings
2793 if self._oss_cluster_maint_notifications_handler:
2794 # Pool already in OSS cluster mode; update the OSS handler config
2795 # instead of creating a mutually-exclusive pool handler (which
2796 # would be silently ignored because the OSS handler wins priority
2797 # in both update helpers below).
2798 self._oss_cluster_maint_notifications_handler.config = (
2799 maint_notifications_config
2800 )
2801 elif not self._maint_notifications_pool_handler:
2802 self._maint_notifications_pool_handler = MaintNotificationsPoolHandler(
2803 self, maint_notifications_config
2804 )
2805 else:
2806 self._maint_notifications_pool_handler.config = (
2807 maint_notifications_config
2808 )
2810 # then update connection kwargs and existing connections
2811 self._update_connection_kwargs_for_maint_notifications(
2812 maint_notifications_pool_handler=self._maint_notifications_pool_handler,
2813 oss_cluster_maint_notifications_handler=self._oss_cluster_maint_notifications_handler,
2814 )
2815 self._update_maint_notifications_configs_for_connections(
2816 maint_notifications_pool_handler=self._maint_notifications_pool_handler,
2817 oss_cluster_maint_notifications_handler=self._oss_cluster_maint_notifications_handler,
2818 )
2820 def _update_connection_kwargs_for_maint_notifications(
2821 self,
2822 maint_notifications_pool_handler: Optional[
2823 MaintNotificationsPoolHandler
2824 ] = None,
2825 oss_cluster_maint_notifications_handler: Optional[
2826 OSSMaintNotificationsHandler
2827 ] = None,
2828 ):
2829 """
2830 Update the connection kwargs for all future connections.
2831 """
2832 if not self.maint_notifications_enabled():
2833 return
2834 if maint_notifications_pool_handler:
2835 self.connection_kwargs.update(
2836 {
2837 "maint_notifications_pool_handler": maint_notifications_pool_handler,
2838 "maint_notifications_config": maint_notifications_pool_handler.config,
2839 }
2840 )
2841 if oss_cluster_maint_notifications_handler:
2842 self.connection_kwargs.update(
2843 {
2844 "oss_cluster_maint_notifications_handler": oss_cluster_maint_notifications_handler,
2845 "maint_notifications_config": oss_cluster_maint_notifications_handler.config,
2846 }
2847 )
2848 # OSS cluster mode and pool-handler mode are mutually exclusive.
2849 # Drop any pool handler a default (RESP3 "auto") pool creation may
2850 # have wired so future connections are not configured with both.
2851 self.connection_kwargs.pop("maint_notifications_pool_handler", None)
2853 # Store original connection parameters for maintenance notifications.
2854 if self.connection_kwargs.get("orig_host_address", None) is None:
2855 # If orig_host_address is None it means we haven't
2856 # configured the original values yet
2857 self.connection_kwargs.update(
2858 {
2859 "orig_host_address": self.connection_kwargs.get("host"),
2860 "orig_socket_timeout": self.connection_kwargs.get(
2861 "socket_timeout", DEFAULT_SOCKET_TIMEOUT
2862 ),
2863 "orig_socket_connect_timeout": self.connection_kwargs.get(
2864 "socket_connect_timeout", DEFAULT_SOCKET_CONNECT_TIMEOUT
2865 ),
2866 }
2867 )
2869 def _update_maint_notifications_configs_for_connections(
2870 self,
2871 maint_notifications_pool_handler: Optional[
2872 MaintNotificationsPoolHandler
2873 ] = None,
2874 oss_cluster_maint_notifications_handler: Optional[
2875 OSSMaintNotificationsHandler
2876 ] = None,
2877 ):
2878 """Update the maintenance notifications config for all connections in the pool."""
2879 with self._get_pool_lock():
2880 for conn in self._get_free_connections():
2881 if oss_cluster_maint_notifications_handler:
2882 # set cluster handler for conn
2883 conn.set_maint_notifications_cluster_handler_for_connection(
2884 oss_cluster_maint_notifications_handler
2885 )
2886 conn.maint_notifications_config = (
2887 oss_cluster_maint_notifications_handler.config
2888 )
2889 elif maint_notifications_pool_handler:
2890 conn.set_maint_notifications_pool_handler_for_connection(
2891 maint_notifications_pool_handler
2892 )
2893 conn.maint_notifications_config = (
2894 maint_notifications_pool_handler.config
2895 )
2896 else:
2897 raise ValueError(
2898 "Either maint_notifications_pool_handler or oss_cluster_maint_notifications_handler must be set"
2899 )
2900 conn.disconnect()
2901 for conn in self._get_in_use_connections():
2902 if oss_cluster_maint_notifications_handler:
2903 # Use set_maint_notifications_cluster_handler_for_connection
2904 # (not _configure_maintenance_notifications) so the parser is
2905 # obtained from the connection itself. _configure_* requires a
2906 # parser argument and would raise here; it would also reset the
2907 # connection's orig_* settings, which is wrong for an in-use
2908 # (active) connection. This mirrors the idle-connection branch
2909 # above and the pool-handler branches.
2910 conn.set_maint_notifications_cluster_handler_for_connection(
2911 oss_cluster_maint_notifications_handler
2912 )
2913 conn.maint_notifications_config = (
2914 oss_cluster_maint_notifications_handler.config
2915 )
2916 elif maint_notifications_pool_handler:
2917 conn.set_maint_notifications_pool_handler_for_connection(
2918 maint_notifications_pool_handler
2919 )
2920 conn.maint_notifications_config = (
2921 maint_notifications_pool_handler.config
2922 )
2923 else:
2924 raise ValueError(
2925 "Either maint_notifications_pool_handler or oss_cluster_maint_notifications_handler must be set"
2926 )
2927 if logger.isEnabledFor(logging.DEBUG):
2928 logger.debug(
2929 "Marking active connection for reconnect after config update "
2930 f"config update: {conn}, {conn.extract_connection_details()}"
2931 )
2932 conn.mark_for_reconnect()
2934 def _should_update_connection(
2935 self,
2936 conn: "MaintNotificationsAbstractConnection",
2937 matching_pattern: Literal[
2938 "connected_address", "configured_address", "notification_hash"
2939 ] = "connected_address",
2940 matching_address: Optional[str] = None,
2941 matching_notification_hash: Optional[int] = None,
2942 ) -> bool:
2943 """
2944 Check if the connection should be updated based on the matching criteria.
2945 """
2946 if matching_pattern == "connected_address":
2947 if matching_address and conn.getpeername() != matching_address:
2948 return False
2949 elif matching_pattern == "configured_address":
2950 if matching_address and conn.host != matching_address:
2951 return False
2952 elif matching_pattern == "notification_hash":
2953 if (
2954 matching_notification_hash is not None
2955 and conn.maintenance_notification_hash != matching_notification_hash
2956 ):
2957 return False
2958 return True
2960 def update_connection_settings(
2961 self,
2962 conn: "MaintNotificationsAbstractConnection",
2963 state: Optional["MaintenanceState"] = None,
2964 maintenance_notification_hash: Optional[int] = None,
2965 host_address: Optional[str] = None,
2966 relaxed_timeout: Optional[float] = None,
2967 update_notification_hash: bool = False,
2968 reset_host_address: bool = False,
2969 reset_relaxed_timeout: bool = False,
2970 ):
2971 """
2972 Update the settings for a single connection.
2973 """
2974 if state:
2975 conn.maintenance_state = state
2977 if update_notification_hash:
2978 # update the notification hash only if requested
2979 conn.maintenance_notification_hash = maintenance_notification_hash
2981 if host_address is not None:
2982 conn.set_tmp_settings(tmp_host_address=host_address)
2984 if relaxed_timeout is not None:
2985 conn.set_tmp_settings(tmp_relaxed_timeout=relaxed_timeout)
2987 if reset_relaxed_timeout or reset_host_address:
2988 conn.reset_tmp_settings(
2989 reset_host_address=reset_host_address,
2990 reset_relaxed_timeout=reset_relaxed_timeout,
2991 )
2993 conn.update_current_socket_timeout(relaxed_timeout)
2995 def update_connections_settings(
2996 self,
2997 state: Optional["MaintenanceState"] = None,
2998 maintenance_notification_hash: Optional[int] = None,
2999 host_address: Optional[str] = None,
3000 relaxed_timeout: Optional[float] = None,
3001 matching_address: Optional[str] = None,
3002 matching_notification_hash: Optional[int] = None,
3003 matching_pattern: Literal[
3004 "connected_address", "configured_address", "notification_hash"
3005 ] = "connected_address",
3006 update_notification_hash: bool = False,
3007 reset_host_address: bool = False,
3008 reset_relaxed_timeout: bool = False,
3009 include_free_connections: bool = True,
3010 ):
3011 """
3012 Update the settings for all matching connections in the pool.
3014 This method does not create new connections.
3015 This method does not affect the connection kwargs.
3017 :param state: The maintenance state to set for the connection.
3018 :param maintenance_notification_hash: The hash of the maintenance notification
3019 to set for the connection.
3020 :param host_address: The host address to set for the connection.
3021 :param relaxed_timeout: The relaxed timeout to set for the connection.
3022 :param matching_address: The address to match for the connection.
3023 :param matching_notification_hash: The notification hash to match for the connection.
3024 :param matching_pattern: The pattern to match for the connection.
3025 :param update_notification_hash: Whether to update the notification hash for the connection.
3026 :param reset_host_address: Whether to reset the host address to the original address.
3027 :param reset_relaxed_timeout: Whether to reset the relaxed timeout to the original timeout.
3028 :param include_free_connections: Whether to include free/available connections.
3029 """
3030 with self._get_pool_lock():
3031 for conn in self._get_in_use_connections():
3032 if self._should_update_connection(
3033 conn,
3034 matching_pattern,
3035 matching_address,
3036 matching_notification_hash,
3037 ):
3038 self.update_connection_settings(
3039 conn,
3040 state=state,
3041 maintenance_notification_hash=maintenance_notification_hash,
3042 host_address=host_address,
3043 relaxed_timeout=relaxed_timeout,
3044 update_notification_hash=update_notification_hash,
3045 reset_host_address=reset_host_address,
3046 reset_relaxed_timeout=reset_relaxed_timeout,
3047 )
3049 if include_free_connections:
3050 for conn in self._get_free_connections():
3051 if self._should_update_connection(
3052 conn,
3053 matching_pattern,
3054 matching_address,
3055 matching_notification_hash,
3056 ):
3057 self.update_connection_settings(
3058 conn,
3059 state=state,
3060 maintenance_notification_hash=maintenance_notification_hash,
3061 host_address=host_address,
3062 relaxed_timeout=relaxed_timeout,
3063 update_notification_hash=update_notification_hash,
3064 reset_host_address=reset_host_address,
3065 reset_relaxed_timeout=reset_relaxed_timeout,
3066 )
3068 def update_connection_kwargs(
3069 self,
3070 **kwargs,
3071 ):
3072 """
3073 Update the connection kwargs for all future connections.
3075 This method updates the connection kwargs for all future connections created by the pool.
3076 Existing connections are not affected.
3077 """
3078 self.connection_kwargs.update(kwargs)
3080 def update_active_connections_for_reconnect(
3081 self,
3082 moving_address_src: Optional[str] = None,
3083 ):
3084 """
3085 Mark all active connections for reconnect.
3086 This is used when a cluster node is migrated to a different address.
3088 :param moving_address_src: The address of the node that is being moved.
3089 """
3090 debug = logger.isEnabledFor(logging.DEBUG)
3091 with self._get_pool_lock():
3092 for conn in self._get_in_use_connections():
3093 if self._should_update_connection(
3094 conn, "connected_address", moving_address_src
3095 ):
3096 if debug:
3097 logger.debug(
3098 f"Marking active connection for reconnect: {conn}, "
3099 f"{conn.extract_connection_details()}"
3100 )
3101 conn.mark_for_reconnect()
3103 def disconnect_free_connections(
3104 self,
3105 moving_address_src: Optional[str] = None,
3106 ):
3107 """
3108 Disconnect all free/available connections.
3109 This is used when a cluster node is migrated to a different address.
3111 :param moving_address_src: The address of the node that is being moved.
3112 """
3113 debug = logger.isEnabledFor(logging.DEBUG)
3114 with self._get_pool_lock():
3115 for conn in self._get_free_connections():
3116 if self._should_update_connection(
3117 conn, "connected_address", moving_address_src
3118 ):
3119 if debug:
3120 logger.debug(
3121 f"Disconnecting free connection: {conn}, "
3122 f"{conn.extract_connection_details()}"
3123 )
3124 conn.disconnect()
3127class ConnectionPool(MaintNotificationsAbstractConnectionPool, ConnectionPoolInterface):
3128 """
3129 Create a connection pool. ``If max_connections`` is set, then this
3130 object raises :py:class:`~redis.exceptions.ConnectionError` when the pool's
3131 limit is reached.
3133 By default, TCP connections are created unless ``connection_class``
3134 is specified. Use class:`.UnixDomainSocketConnection` for
3135 unix sockets.
3136 :py:class:`~redis.SSLConnection` can be used for SSL enabled connections.
3138 If ``maint_notifications_config`` is provided, the connection pool will support
3139 maintenance notifications.
3140 Maintenance notifications are supported only with RESP3.
3141 If the ``maint_notifications_config`` is not provided but the ``protocol`` is 3,
3142 the maintenance notifications will be enabled by default.
3144 If ``metadata_resolver`` is provided, it decides which commands are eligible for
3145 client-side caching - see `redis.commands.metadata.MetadataResolver`. It is handed to the
3146 cache configuration, so one resolver can serve cache eligibility and cluster routing
3147 alike. Defaults to the static command metadata this library ships, which the cache
3148 configuration resolves through on its own. A resolver built from a live ``COMMAND`` reply
3149 is a snapshot of the server it was read from, so give each pool its own rather than
3150 sharing one across pools on different servers.
3152 Any additional keyword arguments are passed to the constructor of
3153 ``connection_class``.
3154 """
3156 @classmethod
3157 def from_url(cls: Type[_CP], url: str, **kwargs) -> _CP:
3158 """
3159 Return a connection pool configured from the given URL.
3161 For example::
3163 redis://[[username]:[password]]@localhost:6379/0
3164 rediss://[[username]:[password]]@localhost:6379/0
3165 unix://[username@]/path/to/socket.sock?db=0[&password=password]
3167 Three URL schemes are supported:
3169 - `redis://` creates a TCP socket connection. See more at:
3170 <https://www.iana.org/assignments/uri-schemes/prov/redis>
3171 - `rediss://` creates a SSL wrapped TCP socket connection. See more at:
3172 <https://www.iana.org/assignments/uri-schemes/prov/rediss>
3173 - ``unix://``: creates a Unix Domain Socket connection.
3175 The username, password, hostname and path are passed through
3176 urllib.parse.unquote in order to replace any percent-encoded values
3177 with their corresponding characters. Querystring values are decoded
3178 by urllib.parse.parse_qs and are not unquoted again.
3180 There are several ways to specify a database number. The first value
3181 found will be used:
3183 1. A ``db`` querystring option, e.g. redis://localhost?db=0
3184 2. If using the redis:// or rediss:// schemes, the path argument
3185 of the url, e.g. redis://localhost/0
3186 3. A ``db`` keyword argument to this function.
3188 If none of these options are specified, the default db=0 is used.
3190 All querystring options are cast to their appropriate Python types.
3191 Boolean arguments can be specified with string values "True"/"False"
3192 or "Yes"/"No". Values that cannot be properly cast cause a
3193 ``ValueError`` to be raised. Once parsed, the querystring arguments
3194 and keyword arguments are passed to the ``ConnectionPool``'s
3195 class initializer. In the case of conflicting arguments, querystring
3196 arguments always win.
3197 """
3198 url_options = parse_url(url)
3200 if "connection_class" in kwargs:
3201 url_options["connection_class"] = kwargs["connection_class"]
3203 kwargs.update(url_options)
3204 return cls(**kwargs)
3206 def __init__(
3207 self,
3208 connection_class=Connection,
3209 max_connections: Optional[int] = None,
3210 cache_factory: Optional[CacheFactoryInterface] = None,
3211 maint_notifications_config: Optional[MaintNotificationsConfig] = None,
3212 metadata_resolver: Optional[MetadataResolver] = None,
3213 **connection_kwargs,
3214 ):
3215 max_connections = max_connections or 100
3216 if not isinstance(max_connections, int) or max_connections < 0:
3217 raise ValueError('"max_connections" must be a positive integer')
3219 self.connection_class = connection_class
3220 self._connection_kwargs = connection_kwargs
3221 self.max_connections = max_connections
3222 self.cache = None
3223 self._cache_factory = cache_factory
3224 self.metadata_resolver = metadata_resolver
3226 try:
3227 supports_maint_notifications = issubclass(
3228 connection_class, MaintNotificationsAbstractConnection
3229 )
3230 is_unix_domain_socket_connection = issubclass(
3231 connection_class, UnixDomainSocketConnection
3232 )
3233 except TypeError:
3234 supports_maint_notifications = False
3235 is_unix_domain_socket_connection = False
3237 if is_unix_domain_socket_connection or not supports_maint_notifications:
3238 if (
3239 maint_notifications_config
3240 and maint_notifications_config.enabled is True
3241 ):
3242 raise RedisError(
3243 "Maintenance notifications are not supported with "
3244 f"{connection_class}"
3245 )
3246 maint_notifications_config = MaintNotificationsConfig(enabled=False)
3248 self._event_dispatcher = self._connection_kwargs.get("event_dispatcher", None)
3249 if self._event_dispatcher is None:
3250 self._event_dispatcher = EventDispatcher()
3252 if connection_kwargs.get("cache_config") or connection_kwargs.get("cache"):
3253 if not check_protocol_version(self._connection_kwargs.get("protocol"), 3):
3254 raise RedisError("Client caching is only supported with RESP version 3")
3256 cache = self._connection_kwargs.get("cache")
3257 cache_config = self._connection_kwargs.get("cache_config")
3259 # Hand the client-level metadata resolver to the cache configuration, which is
3260 # where command eligibility is decided, so routing and caching read the same
3261 # records. Skipped when no resolver was configured, which leaves ``CacheConfig``
3262 # on the static default it builds for itself.
3263 #
3264 # Done on a copy rather than on the caller's object: a ``CacheConfig`` carries
3265 # only sizing and eviction settings, so reusing one across clients is reasonable,
3266 # and writing the resolver into it would give every one of them whichever
3267 # resolver was injected last. Every other setting is read-only after
3268 # construction, so the copy diverges from the caller's object in nothing but the
3269 # resolver - and a later ``set_metadata_resolver`` on the original does not reach
3270 # this pool, which is the intended direction: a pool decides eligibility by the
3271 # resolver its client was built with. Note that the ``cache=`` /
3272 # ``cache_factory=`` path below cannot copy, and so does write into the caller's
3273 # configuration; ``CacheConfig.set_metadata_resolver`` documents the difference.
3274 #
3275 # Guarded by ``isinstance`` rather than done through
3276 # ``CacheConfigurationInterface``: that ABC is public and implemented by third
3277 # parties, so a custom configuration keeps its own eligibility logic.
3278 if (
3279 metadata_resolver is not None
3280 and cache is None
3281 and self._cache_factory is None
3282 and isinstance(cache_config, CacheConfig)
3283 ):
3284 cache_config = copy.copy(cache_config)
3285 cache_config.set_metadata_resolver(metadata_resolver)
3287 if cache is not None:
3288 if not isinstance(cache, CacheInterface):
3289 raise ValueError("Cache must implement CacheInterface")
3291 self.cache = cache
3292 else:
3293 if self._cache_factory is not None:
3294 self.cache = CacheProxy(self._cache_factory.get_cache())
3295 else:
3296 self.cache = CacheFactory(cache_config).get_cache()
3298 # A caller who supplied a whole cache - ``cache=`` or ``cache_factory=`` - owns
3299 # the configuration inside it, and it cannot be swapped without rebuilding the
3300 # cache, so the resolver is set on it in place. This is the one path that writes
3301 # into the caller's configuration rather than into a copy of it, because a cache
3302 # reads its configuration on every lookup and ``CacheInterface`` exposes no way to
3303 # hand it a different one. Sharing one cache object across clients already shares
3304 # its entries, which couples them far more tightly than its eligibility does, so
3305 # the asymmetry with the ``cache_config=`` copy above is documented on
3306 # ``CacheConfig.set_metadata_resolver`` rather than removed.
3307 if metadata_resolver is not None and (
3308 cache is not None or self._cache_factory is not None
3309 ):
3310 own_config = self.cache.config
3311 if isinstance(own_config, CacheConfig):
3312 own_config.set_metadata_resolver(metadata_resolver)
3314 init_csc_items()
3315 register_csc_items_callback(
3316 callback=lambda: self.cache.size,
3317 pool_name=get_pool_name(self),
3318 )
3320 connection_kwargs.pop("cache", None)
3321 connection_kwargs.pop("cache_config", None)
3323 # Resolve the HIMPORT registry. A pre-built ``himport_registry`` (shared, e.g.
3324 # from the cluster client) takes precedence; otherwise build a fresh empty one.
3325 # A registry always exists so runtime ``himport_prepare`` mutates a single object
3326 # every connection already shares. The object stays in ``connection_kwargs`` so
3327 # it reaches every connection. It is injected unconditionally (like other
3328 # auto-added pool kwargs), so a custom ``connection_class`` must accept
3329 # ``**kwargs`` (or a ``himport_registry`` parameter), as built-ins do.
3330 himport_registry = connection_kwargs.get("himport_registry")
3331 if himport_registry is None:
3332 himport_registry = HImportRegistry()
3333 connection_kwargs["himport_registry"] = himport_registry
3334 self.himport_registry = himport_registry
3336 # a lock to protect the critical section in _checkpid().
3337 # this lock is acquired when the process id changes, such as
3338 # after a fork. during this time, multiple threads in the child
3339 # process could attempt to acquire this lock. the first thread
3340 # to acquire the lock will reset the data structures and lock
3341 # object of this pool. subsequent threads acquiring this lock
3342 # will notice the first thread already did the work and simply
3343 # release the lock.
3345 self._fork_lock = threading.RLock()
3346 self._lock = threading.RLock()
3348 # Generate unique pool ID for observability (matches go-redis behavior)
3349 import secrets
3351 self._pool_id = secrets.token_hex(4)
3353 MaintNotificationsAbstractConnectionPool.__init__(
3354 self,
3355 maint_notifications_config=maint_notifications_config,
3356 **connection_kwargs,
3357 )
3359 self.reset()
3361 # Keys that should be redacted in __repr__ to avoid exposing sensitive information
3362 SENSITIVE_REPR_KEYS = frozenset(
3363 {
3364 "password",
3365 "username",
3366 "ssl_password",
3367 "credential_provider",
3368 }
3369 )
3371 # Internal plumbing kwargs omitted from __repr__ (not user-facing config).
3372 OMIT_REPR_KEYS = frozenset({"himport_registry"})
3374 def __repr__(self) -> str:
3375 conn_kwargs = ",".join(
3376 [
3377 f"{k}={'<REDACTED>' if k in self.SENSITIVE_REPR_KEYS else v}"
3378 for k, v in self.connection_kwargs.items()
3379 if k not in self.OMIT_REPR_KEYS
3380 ]
3381 )
3382 return (
3383 f"<{self.__class__.__module__}.{self.__class__.__name__}"
3384 f"(<{self.connection_class.__module__}.{self.connection_class.__name__}"
3385 f"({conn_kwargs})>)>"
3386 )
3388 @property
3389 def connection_kwargs(self) -> Dict[str, Any]:
3390 return self._connection_kwargs
3392 @connection_kwargs.setter
3393 def connection_kwargs(self, value: Dict[str, Any]):
3394 self._connection_kwargs = value
3396 def get_protocol(self):
3397 """
3398 Returns:
3399 The RESP protocol version, or ``None`` if the protocol is not specified,
3400 in which case the server default will be used.
3401 """
3402 return self.connection_kwargs.get("protocol", None)
3404 def reset(self) -> None:
3405 # Record metrics for connections being removed before clearing
3406 # (only if attributes exist - they won't during __init__)
3407 if hasattr(self, "_available_connections") and hasattr(
3408 self, "_in_use_connections"
3409 ):
3410 with self._lock:
3411 idle_count = len(self._available_connections)
3412 in_use_count = len(self._in_use_connections)
3413 if idle_count > 0 or in_use_count > 0:
3414 pool_name = get_pool_name(self)
3415 if idle_count > 0:
3416 record_connection_count(
3417 pool_name=pool_name,
3418 connection_state=ConnectionState.IDLE,
3419 counter=-idle_count,
3420 )
3421 if in_use_count > 0:
3422 record_connection_count(
3423 pool_name=pool_name,
3424 connection_state=ConnectionState.USED,
3425 counter=-in_use_count,
3426 )
3428 self._created_connections = 0
3429 self._available_connections = []
3430 self._in_use_connections = set()
3432 # this must be the last operation in this method. while reset() is
3433 # called when holding _fork_lock, other threads in this process
3434 # can call _checkpid() which compares self.pid and os.getpid() without
3435 # holding any lock (for performance reasons). keeping this assignment
3436 # as the last operation ensures that those other threads will also
3437 # notice a pid difference and block waiting for the first thread to
3438 # release _fork_lock. when each of these threads eventually acquire
3439 # _fork_lock, they will notice that another thread already called
3440 # reset() and they will immediately release _fork_lock and continue on.
3441 self.pid = os.getpid()
3443 def __del__(self) -> None:
3444 """Clean up connection pool and record metrics when garbage collected."""
3445 try:
3446 if not hasattr(self, "_available_connections") or not hasattr(
3447 self, "_in_use_connections"
3448 ):
3449 return
3450 # Record metrics for all connections being removed
3451 idle_count = len(self._available_connections)
3452 in_use_count = len(self._in_use_connections)
3453 if idle_count > 0 or in_use_count > 0:
3454 pool_name = get_pool_name(self)
3455 if idle_count > 0:
3456 record_connection_count(
3457 pool_name=pool_name,
3458 connection_state=ConnectionState.IDLE,
3459 counter=-idle_count,
3460 )
3461 if in_use_count > 0:
3462 record_connection_count(
3463 pool_name=pool_name,
3464 connection_state=ConnectionState.USED,
3465 counter=-in_use_count,
3466 )
3467 except Exception:
3468 pass
3470 def _checkpid(self) -> None:
3471 # _checkpid() attempts to keep ConnectionPool fork-safe on modern
3472 # systems. this is called by all ConnectionPool methods that
3473 # manipulate the pool's state such as get_connection() and release().
3474 #
3475 # _checkpid() determines whether the process has forked by comparing
3476 # the current process id to the process id saved on the ConnectionPool
3477 # instance. if these values are the same, _checkpid() simply returns.
3478 #
3479 # when the process ids differ, _checkpid() assumes that the process
3480 # has forked and that we're now running in the child process. the child
3481 # process cannot use the parent's file descriptors (e.g., sockets).
3482 # therefore, when _checkpid() sees the process id change, it calls
3483 # reset() in order to reinitialize the child's ConnectionPool. this
3484 # will cause the child to make all new connection objects.
3485 #
3486 # _checkpid() is protected by self._fork_lock to ensure that multiple
3487 # threads in the child process do not call reset() multiple times.
3488 #
3489 # there is an extremely small chance this could fail in the following
3490 # scenario:
3491 # 1. process A calls _checkpid() for the first time and acquires
3492 # self._fork_lock.
3493 # 2. while holding self._fork_lock, process A forks (the fork()
3494 # could happen in a different thread owned by process A)
3495 # 3. process B (the forked child process) inherits the
3496 # ConnectionPool's state from the parent. that state includes
3497 # a locked _fork_lock. process B will not be notified when
3498 # process A releases the _fork_lock and will thus never be
3499 # able to acquire the _fork_lock.
3500 #
3501 # to mitigate this possible deadlock, _checkpid() will only wait 5
3502 # seconds to acquire _fork_lock. if _fork_lock cannot be acquired in
3503 # that time it is assumed that the child is deadlocked and a
3504 # redis.ChildDeadlockedError error is raised.
3505 if self.pid != os.getpid():
3506 acquired = self._fork_lock.acquire(timeout=5)
3507 if not acquired:
3508 raise ChildDeadlockedError
3509 # reset() the instance for the new process if another thread
3510 # hasn't already done so
3511 try:
3512 if self.pid != os.getpid():
3513 self.reset()
3514 finally:
3515 self._fork_lock.release()
3517 @deprecated_args(
3518 args_to_warn=["*"],
3519 reason="Use get_connection() without args instead",
3520 version="5.3.0",
3521 )
3522 def get_connection(self, command_name=None, *keys, **options) -> "Connection":
3523 "Get a connection from the pool"
3525 # Start timing for observability
3526 self._checkpid()
3527 is_created = False
3529 with self._lock:
3530 try:
3531 connection = self._available_connections.pop()
3532 except IndexError:
3533 # Start timing for observability
3534 start_time_created = time.monotonic()
3536 connection = self.make_connection()
3537 is_created = True
3538 self._in_use_connections.add(connection)
3540 # Record state transition: IDLE -> USED
3541 # (make_connection already recorded IDLE +1 for new connections)
3542 # This ensures counters stay balanced if connect() fails and release() is called
3543 pool_name = get_pool_name(self)
3544 record_connection_count(
3545 pool_name=pool_name,
3546 connection_state=ConnectionState.IDLE,
3547 counter=-1,
3548 )
3549 record_connection_count(
3550 pool_name=pool_name,
3551 connection_state=ConnectionState.USED,
3552 counter=1,
3553 )
3555 try:
3556 # ensure this connection is connected to Redis
3557 connection.connect()
3558 # connections that the pool provides should be ready to send
3559 # a command. if not, the connection was either returned to the
3560 # pool before all data has been read or the socket has been
3561 # closed. either way, reconnect and verify everything is good.
3562 try:
3563 if (
3564 connection.can_read()
3565 and self.cache is None
3566 and not self.maint_notifications_enabled()
3567 ):
3568 raise ConnectionError("Connection has data")
3569 except (ConnectionError, TimeoutError, OSError):
3570 connection.disconnect()
3571 connection.connect()
3572 if (
3573 connection.can_read()
3574 and self.cache is None
3575 and not self.maint_notifications_enabled()
3576 ):
3577 raise ConnectionError("Connection not ready")
3578 except BaseException:
3579 # release the connection back to the pool so that we don't
3580 # leak it
3581 self.release(connection)
3582 raise
3584 if is_created:
3585 record_connection_create_time(
3586 connection_pool=self,
3587 duration_seconds=time.monotonic() - start_time_created,
3588 )
3590 return connection
3592 def get_encoder(self) -> Encoder:
3593 "Return an encoder based on encoding settings"
3594 kwargs = self.connection_kwargs
3595 return Encoder(
3596 encoding=kwargs.get("encoding", "utf-8"),
3597 encoding_errors=kwargs.get("encoding_errors", "strict"),
3598 decode_responses=kwargs.get("decode_responses", False),
3599 )
3601 def make_connection(self) -> "ConnectionInterface":
3602 "Create a new connection"
3603 if self._created_connections >= self.max_connections:
3604 raise MaxConnectionsError("Too many connections")
3605 self._created_connections += 1
3607 kwargs = dict(self.connection_kwargs)
3609 # Create the connection first, then record metrics only on success
3610 if self.cache is not None:
3611 connection = CacheProxyConnection(
3612 self.connection_class(**kwargs), self.cache, self._lock
3613 )
3614 else:
3615 connection = self.connection_class(**kwargs)
3617 # Record new connection created (starts as IDLE) - only after successful construction
3618 record_connection_count(
3619 pool_name=get_pool_name(self),
3620 connection_state=ConnectionState.IDLE,
3621 counter=1,
3622 )
3624 return connection
3626 def release(self, connection: "Connection") -> None:
3627 "Releases the connection back to the pool"
3628 self._checkpid()
3629 with self._lock:
3630 try:
3631 self._in_use_connections.remove(connection)
3632 except KeyError:
3633 # Gracefully fail when a connection is returned to this pool
3634 # that the pool doesn't actually own
3635 return
3637 if self.owns_connection(connection):
3638 if connection.should_reconnect():
3639 if logger.isEnabledFor(logging.DEBUG):
3640 logger.debug(
3641 "Disconnecting released connection marked for reconnect: "
3642 f"{connection}, {connection.extract_connection_details()}"
3643 )
3644 connection.disconnect()
3645 self._available_connections.append(connection)
3646 self._event_dispatcher.dispatch(
3647 AfterConnectionReleasedEvent(connection)
3648 )
3650 # Record state transition: USED -> IDLE
3651 pool_name = get_pool_name(self)
3652 record_connection_count(
3653 pool_name=pool_name,
3654 connection_state=ConnectionState.USED,
3655 counter=-1,
3656 )
3657 record_connection_count(
3658 pool_name=pool_name,
3659 connection_state=ConnectionState.IDLE,
3660 counter=1,
3661 )
3662 else:
3663 # Pool doesn't own this connection, do not add it back
3664 # to the pool.
3665 # Still need to decrement USED since it was counted in get_connection()
3666 connection.disconnect()
3667 # Subclasses such as SentinelConnectionPool can override
3668 # owns_connection() with a comparison different from local PID
3669 # ownership. When such a subclass rejects a connection, also require
3670 # connection.pid == self.pid before reclaiming its slot.
3671 if connection.pid == self.pid:
3672 self._created_connections -= 1
3673 record_connection_count(
3674 pool_name="unknown_pool",
3675 connection_state=ConnectionState.USED,
3676 counter=-1,
3677 )
3678 return
3680 def owns_connection(self, connection: "Connection") -> int:
3681 return connection.pid == self.pid
3683 def disconnect(self, inuse_connections: bool = True) -> None:
3684 """
3685 Disconnects connections in the pool
3687 If ``inuse_connections`` is True, disconnect connections that are
3688 currently in use, potentially by other threads. Otherwise only disconnect
3689 connections that are idle in the pool.
3690 """
3691 self._checkpid()
3692 with self._lock:
3693 if inuse_connections:
3694 connections = chain(
3695 self._available_connections, self._in_use_connections
3696 )
3697 else:
3698 connections = self._available_connections
3700 for connection in connections:
3701 connection.disconnect()
3703 def close(self) -> None:
3704 """Close the pool, disconnecting all connections"""
3705 self.disconnect()
3707 def __enter__(self: _CP) -> _CP:
3708 return self
3710 def __exit__(self, exc_type, exc_value, traceback) -> None:
3711 self.close()
3713 def set_retry(self, retry: Retry) -> None:
3714 self.connection_kwargs.update({"retry": retry})
3715 for conn in self._available_connections:
3716 conn.retry = retry
3717 for conn in self._in_use_connections:
3718 conn.retry = retry
3720 def re_auth_callback(self, token: TokenInterface):
3721 with self._lock:
3722 for conn in self._available_connections:
3723 conn.retry.call_with_retry(
3724 lambda: conn.send_command(
3725 "AUTH", token.try_get("oid"), token.get_value()
3726 ),
3727 lambda error: self._mock(error),
3728 )
3729 conn.retry.call_with_retry(
3730 lambda: conn.read_response(), lambda error: self._mock(error)
3731 )
3732 for conn in self._in_use_connections:
3733 conn.set_re_auth_token(token)
3735 def _get_pool_lock(self):
3736 return self._lock
3738 def _get_free_connections(self):
3739 with self._lock:
3740 return list(self._available_connections)
3742 def _get_in_use_connections(self):
3743 with self._lock:
3744 return set(self._in_use_connections)
3746 def _mock(self, error: RedisError):
3747 """
3748 Dummy functions, needs to be passed as error callback to retry object.
3749 :param error:
3750 :return:
3751 """
3752 pass
3754 def get_connection_count(self) -> List[tuple[int, dict]]:
3755 from redis.observability.attributes import get_pool_name
3757 attributes = AttributeBuilder.build_base_attributes()
3758 attributes[DB_CLIENT_CONNECTION_POOL_NAME] = get_pool_name(self)
3759 free_connections_attributes = attributes.copy()
3760 in_use_connections_attributes = attributes.copy()
3762 free_connections_attributes[DB_CLIENT_CONNECTION_STATE] = (
3763 ConnectionState.IDLE.value
3764 )
3765 in_use_connections_attributes[DB_CLIENT_CONNECTION_STATE] = (
3766 ConnectionState.USED.value
3767 )
3769 return [
3770 (len(self._get_free_connections()), free_connections_attributes),
3771 (len(self._get_in_use_connections()), in_use_connections_attributes),
3772 ]
3775class BlockingConnectionPool(ConnectionPool):
3776 """
3777 Thread-safe blocking connection pool::
3779 >>> from redis.client import Redis
3780 >>> client = Redis(connection_pool=BlockingConnectionPool())
3782 It performs the same function as the default
3783 :py:class:`~redis.ConnectionPool` implementation, in that,
3784 it maintains a pool of reusable connections that can be shared by
3785 multiple redis clients (safely across threads if required).
3787 The difference is that, in the event that a client tries to get a
3788 connection from the pool when all of connections are in use, rather than
3789 raising a :py:class:`~redis.ConnectionError` (as the default
3790 :py:class:`~redis.ConnectionPool` implementation does), it
3791 makes the client wait ("blocks") for a specified number of seconds until
3792 a connection becomes available.
3794 Use ``max_connections`` to increase / decrease the pool size::
3796 >>> pool = BlockingConnectionPool(max_connections=10)
3798 Use ``timeout`` to tell it either how many seconds to wait for a connection
3799 to become available, or to block forever:
3801 >>> # Block forever.
3802 >>> pool = BlockingConnectionPool(timeout=None)
3804 >>> # Raise a ``ConnectionError`` after five seconds if a connection is
3805 >>> # not available.
3806 >>> pool = BlockingConnectionPool(timeout=5)
3807 """
3809 def __init__(
3810 self,
3811 max_connections=50,
3812 timeout=20,
3813 connection_class=Connection,
3814 queue_class=LifoQueue,
3815 **connection_kwargs,
3816 ):
3817 self.queue_class = queue_class
3818 self.timeout = timeout
3819 self._in_maintenance = False
3820 self._locked = False
3821 super().__init__(
3822 connection_class=connection_class,
3823 max_connections=max_connections,
3824 **connection_kwargs,
3825 )
3827 def reset(self):
3828 # Create and fill up a thread safe queue with ``None`` values.
3829 try:
3830 if self._in_maintenance:
3831 self._lock.acquire()
3832 self._locked = True
3834 # Record metrics for connections being removed before clearing
3835 # Note: Access pool.queue directly to avoid deadlock since we may
3836 # already hold self._lock (which is non-reentrant)
3837 if (
3838 hasattr(self, "_connections")
3839 and self._connections
3840 and hasattr(self, "pool")
3841 ):
3842 with self._lock:
3843 connections_in_queue = {conn for conn in self.pool.queue if conn}
3844 idle_count = len(connections_in_queue)
3845 in_use_count = len(self._connections) - idle_count
3846 if idle_count > 0 or in_use_count > 0:
3847 pool_name = get_pool_name(self)
3848 if idle_count > 0:
3849 record_connection_count(
3850 pool_name=pool_name,
3851 connection_state=ConnectionState.IDLE,
3852 counter=-idle_count,
3853 )
3854 if in_use_count > 0:
3855 record_connection_count(
3856 pool_name=pool_name,
3857 connection_state=ConnectionState.USED,
3858 counter=-in_use_count,
3859 )
3861 self.pool = self.queue_class(self.max_connections)
3862 while True:
3863 try:
3864 self.pool.put_nowait(None)
3865 except Full:
3866 break
3868 # Keep a list of actual connection instances so that we can
3869 # disconnect them later.
3870 self._connections = []
3871 finally:
3872 if self._locked:
3873 try:
3874 self._lock.release()
3875 except Exception:
3876 pass
3877 self._locked = False
3879 # this must be the last operation in this method. while reset() is
3880 # called when holding _fork_lock, other threads in this process
3881 # can call _checkpid() which compares self.pid and os.getpid() without
3882 # holding any lock (for performance reasons). keeping this assignment
3883 # as the last operation ensures that those other threads will also
3884 # notice a pid difference and block waiting for the first thread to
3885 # release _fork_lock. when each of these threads eventually acquire
3886 # _fork_lock, they will notice that another thread already called
3887 # reset() and they will immediately release _fork_lock and continue on.
3888 self.pid = os.getpid()
3890 def __del__(self) -> None:
3891 """Clean up connection pool and record metrics when garbage collected."""
3892 try:
3893 # Note: Access pool.queue directly to avoid potential deadlock
3894 # if GC runs while the lock is held by the same thread
3895 if (
3896 hasattr(self, "_connections")
3897 and self._connections
3898 and hasattr(self, "pool")
3899 ):
3900 connections_in_queue = {conn for conn in self.pool.queue if conn}
3901 idle_count = len(connections_in_queue)
3902 in_use_count = len(self._connections) - idle_count
3903 if idle_count > 0 or in_use_count > 0:
3904 pool_name = get_pool_name(self)
3905 if idle_count > 0:
3906 record_connection_count(
3907 pool_name=pool_name,
3908 connection_state=ConnectionState.IDLE,
3909 counter=-idle_count,
3910 )
3911 if in_use_count > 0:
3912 record_connection_count(
3913 pool_name=pool_name,
3914 connection_state=ConnectionState.USED,
3915 counter=-in_use_count,
3916 )
3917 except Exception:
3918 pass
3920 def make_connection(self):
3921 "Make a fresh connection."
3922 try:
3923 if self._in_maintenance:
3924 self._lock.acquire()
3925 self._locked = True
3927 if self.cache is not None:
3928 connection = CacheProxyConnection(
3929 self.connection_class(**self.connection_kwargs),
3930 self.cache,
3931 self._lock,
3932 )
3933 else:
3934 connection = self.connection_class(**self.connection_kwargs)
3935 self._connections.append(connection)
3937 # Record new connection created (starts as IDLE)
3938 record_connection_count(
3939 pool_name=get_pool_name(self),
3940 connection_state=ConnectionState.IDLE,
3941 counter=1,
3942 )
3944 return connection
3945 finally:
3946 if self._locked:
3947 try:
3948 self._lock.release()
3949 except Exception:
3950 pass
3951 self._locked = False
3953 @deprecated_args(
3954 args_to_warn=["*"],
3955 reason="Use get_connection() without args instead",
3956 version="5.3.0",
3957 )
3958 def get_connection(self, command_name=None, *keys, **options):
3959 """
3960 Get a connection, blocking for ``self.timeout`` until a connection
3961 is available from the pool.
3963 If the connection returned is ``None`` then creates a new connection.
3964 Because we use a last-in first-out queue, the existing connections
3965 (having been returned to the pool after the initial ``None`` values
3966 were added) will be returned before ``None`` values. This means we only
3967 create new connections when we need to, i.e.: the actual number of
3968 connections will only increase in response to demand.
3969 """
3970 start_time_acquired = time.monotonic()
3971 # Make sure we haven't changed process.
3972 self._checkpid()
3973 is_created = False
3975 # Try and get a connection from the pool. If one isn't available within
3976 # self.timeout then raise a ``ConnectionError``.
3977 connection = None
3978 try:
3979 if self._in_maintenance:
3980 self._lock.acquire()
3981 self._locked = True
3982 try:
3983 connection = self.pool.get(block=True, timeout=self.timeout)
3984 except Empty:
3985 # Note that this is not caught by the redis client and will be
3986 # raised unless handled by application code. If you want never to
3987 raise ConnectionError("No connection available.")
3989 # If the ``connection`` is actually ``None`` then that's a cue to make
3990 # a new connection to add to the pool.
3991 if connection is None:
3992 # Start timing for observability
3993 start_time_created = time.monotonic()
3994 connection = self.make_connection()
3995 is_created = True
3996 finally:
3997 if self._locked:
3998 try:
3999 self._lock.release()
4000 except Exception:
4001 pass
4002 self._locked = False
4004 # Record state transition: IDLE -> USED
4005 # (make_connection already recorded IDLE +1 for new connections)
4006 # This ensures counters stay balanced if connect() fails and release() is called
4007 pool_name = get_pool_name(self)
4008 record_connection_count(
4009 pool_name=pool_name,
4010 connection_state=ConnectionState.IDLE,
4011 counter=-1,
4012 )
4013 record_connection_count(
4014 pool_name=pool_name,
4015 connection_state=ConnectionState.USED,
4016 counter=1,
4017 )
4019 try:
4020 # ensure this connection is connected to Redis
4021 connection.connect()
4022 # connections that the pool provides should be ready to send
4023 # a command. if not, the connection was either returned to the
4024 # pool before all data has been read or the socket has been
4025 # closed. either way, reconnect and verify everything is good.
4026 try:
4027 if (
4028 connection.can_read()
4029 and self.cache is None
4030 and not self.maint_notifications_enabled()
4031 ):
4032 raise ConnectionError("Connection has data")
4033 except (ConnectionError, TimeoutError, OSError):
4034 connection.disconnect()
4035 connection.connect()
4036 if (
4037 connection.can_read()
4038 and self.cache is None
4039 and not self.maint_notifications_enabled()
4040 ):
4041 raise ConnectionError("Connection not ready")
4042 except BaseException:
4043 # release the connection back to the pool so that we don't leak it
4044 self.release(connection)
4045 raise
4047 if is_created:
4048 record_connection_create_time(
4049 connection_pool=self,
4050 duration_seconds=time.monotonic() - start_time_created,
4051 )
4053 record_connection_wait_time(
4054 pool_name=pool_name,
4055 duration_seconds=time.monotonic() - start_time_acquired,
4056 )
4058 return connection
4060 def release(self, connection):
4061 "Releases the connection back to the pool."
4062 # Make sure we haven't changed process.
4063 self._checkpid()
4065 try:
4066 if self._in_maintenance:
4067 self._lock.acquire()
4068 self._locked = True
4069 if not self.owns_connection(connection):
4070 # pool doesn't own this connection. do not add it back
4071 # to the pool. instead add a None value which is a placeholder
4072 # that will cause the pool to recreate the connection if
4073 # its needed.
4074 connection.disconnect()
4075 self.pool.put_nowait(None)
4076 # Still need to decrement USED since it was counted in get_connection()
4077 record_connection_count(
4078 pool_name="unknown_pool",
4079 connection_state=ConnectionState.USED,
4080 counter=-1,
4081 )
4082 return
4083 if connection.should_reconnect():
4084 if logger.isEnabledFor(logging.DEBUG):
4085 logger.debug(
4086 "Disconnecting released connection marked for reconnect: "
4087 f"{connection}, {connection.extract_connection_details()}"
4088 )
4089 connection.disconnect()
4090 # Put the connection back into the pool.
4091 pool_name = get_pool_name(self)
4092 try:
4093 self.pool.put_nowait(connection)
4095 # Record state transition: USED -> IDLE
4096 record_connection_count(
4097 pool_name=pool_name,
4098 connection_state=ConnectionState.USED,
4099 counter=-1,
4100 )
4101 record_connection_count(
4102 pool_name=pool_name,
4103 connection_state=ConnectionState.IDLE,
4104 counter=1,
4105 )
4106 except Full:
4107 pass
4108 finally:
4109 if self._locked:
4110 try:
4111 self._lock.release()
4112 except Exception:
4113 pass
4114 self._locked = False
4116 def disconnect(self, inuse_connections: bool = True):
4117 """
4118 Disconnects either all connections in the pool or just the free connections.
4119 """
4120 self._checkpid()
4121 try:
4122 if self._in_maintenance:
4123 self._lock.acquire()
4124 self._locked = True
4126 if inuse_connections:
4127 connections = self._connections
4128 else:
4129 connections = self._get_free_connections()
4131 for connection in connections:
4132 connection.disconnect()
4133 finally:
4134 if self._locked:
4135 try:
4136 self._lock.release()
4137 except Exception:
4138 pass
4139 self._locked = False
4141 def _get_free_connections(self):
4142 with self._lock:
4143 return {conn for conn in self.pool.queue if conn}
4145 def _get_in_use_connections(self):
4146 with self._lock:
4147 # free connections
4148 connections_in_queue = {conn for conn in self.pool.queue if conn}
4149 # in self._connections we keep all created connections
4150 # so the ones that are not in the queue are the in use ones
4151 return {
4152 conn for conn in self._connections if conn not in connections_in_queue
4153 }
4155 def set_in_maintenance(self, in_maintenance: bool):
4156 """
4157 Sets a flag that this Blocking ConnectionPool is in maintenance mode.
4159 This is used to prevent new connections from being created while we are in maintenance mode.
4160 The pool will be in maintenance mode only when we are processing a MOVING notification.
4161 """
4162 self._in_maintenance = in_maintenance