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 os
3import socket
4import sys
5import threading
6import time
7import weakref
8from abc import ABC, abstractmethod
9from itertools import chain
10from queue import Empty, Full, LifoQueue
11from typing import (
12 Any,
13 Callable,
14 Dict,
15 Iterable,
16 List,
17 Literal,
18 Optional,
19 Type,
20 TypeVar,
21 Union,
22)
23from urllib.parse import parse_qs, unquote, urlparse
25from redis.cache import (
26 CacheEntry,
27 CacheEntryStatus,
28 CacheFactory,
29 CacheFactoryInterface,
30 CacheInterface,
31 CacheKey,
32 CacheProxy,
33)
35from ._defaults import (
36 DEFAULT_SOCKET_CONNECT_TIMEOUT,
37 DEFAULT_SOCKET_READ_SIZE,
38 DEFAULT_SOCKET_TIMEOUT,
39 get_default_socket_keepalive_options,
40)
41from ._parsers import BaseParser, Encoder, _HiredisParser, _RESP2Parser, _RESP3Parser
42from .auth.token import TokenInterface
43from .backoff import NoBackoff
44from .credentials import CredentialProvider, UsernamePasswordCredentialProvider
45from .driver_info import DriverInfo, resolve_driver_info
46from .event import AfterConnectionReleasedEvent, EventDispatcher
47from .exceptions import (
48 AuthenticationError,
49 AuthenticationWrongNumberOfArgsError,
50 ChildDeadlockedError,
51 ConnectionError,
52 DataError,
53 MaxConnectionsError,
54 RedisError,
55 ResponseError,
56 TimeoutError,
57)
58from .maint_notifications import (
59 MaintenanceState,
60 MaintNotificationsConfig,
61 MaintNotificationsConnectionHandler,
62 MaintNotificationsPoolHandler,
63 OSSMaintNotificationsHandler,
64)
65from .observability.attributes import (
66 DB_CLIENT_CONNECTION_POOL_NAME,
67 DB_CLIENT_CONNECTION_STATE,
68 AttributeBuilder,
69 ConnectionState,
70 CSCReason,
71 CSCResult,
72 get_pool_name,
73)
74from .observability.metrics import CloseReason
75from .observability.recorder import (
76 init_csc_items,
77 record_connection_closed,
78 record_connection_count,
79 record_connection_create_time,
80 record_connection_wait_time,
81 record_csc_eviction,
82 record_csc_network_saved,
83 record_csc_request,
84 record_error_count,
85 register_csc_items_callback,
86)
87from .retry import Retry
88from .utils import (
89 CRYPTOGRAPHY_AVAILABLE,
90 DEFAULT_RESP_VERSION,
91 HIREDIS_AVAILABLE,
92 SENTINEL,
93 SSL_AVAILABLE,
94 check_protocol_version,
95 compare_versions,
96 deprecated_args,
97 ensure_string,
98 format_error_message,
99 str_if_bytes,
100)
102if SSL_AVAILABLE:
103 import ssl
104 from ssl import VerifyFlags
105else:
106 ssl = None
107 VerifyFlags = None
109if HIREDIS_AVAILABLE:
110 import hiredis
112SYM_STAR = b"*"
113SYM_DOLLAR = b"$"
114SYM_CRLF = b"\r\n"
115SYM_EMPTY = b""
117DefaultParser: Type[Union[_RESP2Parser, _RESP3Parser, _HiredisParser]]
118if HIREDIS_AVAILABLE:
119 DefaultParser = _HiredisParser
120else:
121 DefaultParser = _RESP2Parser
124class HiredisRespSerializer:
125 def pack(self, *args: List):
126 """Pack a series of arguments into the Redis protocol"""
127 output = []
129 if isinstance(args[0], str):
130 args = tuple(args[0].encode().split()) + args[1:]
131 elif b" " in args[0]:
132 args = tuple(args[0].split()) + args[1:]
133 args = tuple(
134 bytes(arg) if isinstance(arg, (bytearray, memoryview)) else arg
135 for arg in args
136 )
137 try:
138 output.append(hiredis.pack_command(args))
139 except TypeError:
140 _, value, traceback = sys.exc_info()
141 raise DataError(value).with_traceback(traceback)
143 return output
146class PythonRespSerializer:
147 def __init__(self, buffer_cutoff, encode) -> None:
148 self._buffer_cutoff = buffer_cutoff
149 self.encode = encode
151 def pack(self, *args):
152 """Pack a series of arguments into the Redis protocol"""
153 output = []
154 # the client might have included 1 or more literal arguments in
155 # the command name, e.g., 'CONFIG GET'. The Redis server expects these
156 # arguments to be sent separately, so split the first argument
157 # manually. These arguments should be bytestrings so that they are
158 # not encoded.
159 if isinstance(args[0], str):
160 args = tuple(args[0].encode().split()) + args[1:]
161 elif b" " in args[0]:
162 args = tuple(args[0].split()) + args[1:]
164 buff = SYM_EMPTY.join((SYM_STAR, str(len(args)).encode(), SYM_CRLF))
166 buffer_cutoff = self._buffer_cutoff
167 for arg in map(self.encode, args):
168 # to avoid large string mallocs, chunk the command into the
169 # output list if we're sending large values or memoryviews
170 arg_length = len(arg)
171 if (
172 len(buff) > buffer_cutoff
173 or arg_length > buffer_cutoff
174 or isinstance(arg, memoryview)
175 ):
176 buff = SYM_EMPTY.join(
177 (buff, SYM_DOLLAR, str(arg_length).encode(), SYM_CRLF)
178 )
179 output.append(buff)
180 output.append(arg)
181 buff = SYM_CRLF
182 else:
183 buff = SYM_EMPTY.join(
184 (
185 buff,
186 SYM_DOLLAR,
187 str(arg_length).encode(),
188 SYM_CRLF,
189 arg,
190 SYM_CRLF,
191 )
192 )
193 output.append(buff)
194 return output
197class ConnectionInterface:
198 @abstractmethod
199 def repr_pieces(self):
200 pass
202 @abstractmethod
203 def register_connect_callback(self, callback):
204 pass
206 @abstractmethod
207 def deregister_connect_callback(self, callback):
208 pass
210 @abstractmethod
211 def set_parser(self, parser_class):
212 pass
214 @abstractmethod
215 def get_protocol(self):
216 pass
218 @abstractmethod
219 def connect(self):
220 pass
222 @abstractmethod
223 def on_connect(self):
224 pass
226 @abstractmethod
227 def disconnect(self, *args, **kwargs):
228 pass
230 @abstractmethod
231 def check_health(self):
232 pass
234 @abstractmethod
235 def send_packed_command(self, command, check_health=True):
236 pass
238 @abstractmethod
239 def send_command(self, *args, **kwargs):
240 pass
242 @abstractmethod
243 def can_read(self, timeout: float = 0) -> bool:
244 # TODO: Rename this API; it detects pending data or dirty/closed
245 # connection state, not only whether application data can be read.
246 pass
248 @abstractmethod
249 def read_response(
250 self,
251 disable_decoding=False,
252 *,
253 timeout: Union[float, object] = SENTINEL,
254 disconnect_on_error=True,
255 push_request=False,
256 ):
257 pass
259 @abstractmethod
260 def pack_command(self, *args):
261 pass
263 @abstractmethod
264 def pack_commands(self, commands):
265 pass
267 @property
268 @abstractmethod
269 def handshake_metadata(self) -> Union[Dict[bytes, bytes], Dict[str, str]]:
270 pass
272 @abstractmethod
273 def set_re_auth_token(self, token: TokenInterface):
274 pass
276 @abstractmethod
277 def re_auth(self):
278 pass
280 @abstractmethod
281 def mark_for_reconnect(self):
282 """
283 Mark the connection to be reconnected on the next command.
284 This is useful when a connection is moved to a different node.
285 """
286 pass
288 @abstractmethod
289 def should_reconnect(self):
290 """
291 Returns True if the connection should be reconnected.
292 """
293 pass
295 @abstractmethod
296 def reset_should_reconnect(self):
297 """
298 Reset the internal flag to False.
299 """
300 pass
302 @abstractmethod
303 def extract_connection_details(self) -> str:
304 pass
306 @property
307 @abstractmethod
308 def is_connected(self) -> bool:
309 """
310 Return ``True`` if the connection to the server is active.
311 """
312 pass
315class MaintNotificationsAbstractConnection:
316 """
317 Abstract class for handling maintenance notifications logic.
318 This class is expected to be used as base class together with ConnectionInterface.
320 This class is intended to be used with multiple inheritance!
322 All logic related to maintenance notifications is encapsulated in this class.
323 """
325 def __init__(
326 self,
327 maint_notifications_config: Optional[MaintNotificationsConfig],
328 maint_notifications_pool_handler: Optional[
329 MaintNotificationsPoolHandler
330 ] = None,
331 maintenance_state: "MaintenanceState" = MaintenanceState.NONE,
332 maintenance_notification_hash: Optional[int] = None,
333 orig_host_address: Optional[str] = None,
334 orig_socket_timeout: Optional[float] = None,
335 orig_socket_connect_timeout: Optional[float] = None,
336 oss_cluster_maint_notifications_handler: Optional[
337 OSSMaintNotificationsHandler
338 ] = None,
339 parser: Optional[BaseParser] = None,
340 event_dispatcher: Optional[EventDispatcher] = None,
341 ):
342 """
343 Initialize the maintenance notifications for the connection.
345 Args:
346 maint_notifications_config (MaintNotificationsConfig): The configuration for maintenance notifications.
347 maint_notifications_pool_handler (Optional[MaintNotificationsPoolHandler]): The pool handler for maintenance notifications.
348 maintenance_state (MaintenanceState): The current maintenance state of the connection.
349 maintenance_notification_hash (Optional[int]): The current maintenance notification hash of the connection.
350 orig_host_address (Optional[str]): The original host address of the connection.
351 orig_socket_timeout (Optional[float]): The original socket timeout of the connection.
352 orig_socket_connect_timeout (Optional[float]): The original socket connect timeout of the connection.
353 oss_cluster_maint_notifications_handler (Optional[OSSMaintNotificationsHandler]): The OSS cluster handler for maintenance notifications.
354 parser (Optional[BaseParser]): The parser to use for maintenance notifications.
355 If not provided, the parser from the connection is used.
356 This is useful when the parser is created after this object.
357 """
358 self.maint_notifications_config = maint_notifications_config
359 self.maintenance_state = maintenance_state
360 self.maintenance_notification_hash = maintenance_notification_hash
362 if event_dispatcher is not None:
363 self.event_dispatcher = event_dispatcher
364 else:
365 self.event_dispatcher = EventDispatcher()
367 self._configure_maintenance_notifications(
368 maint_notifications_pool_handler,
369 orig_host_address,
370 orig_socket_timeout,
371 orig_socket_connect_timeout,
372 oss_cluster_maint_notifications_handler,
373 parser,
374 )
375 self._processed_start_maint_notifications = set()
376 self._skipped_end_maint_notifications = set()
378 @abstractmethod
379 def _get_parser(self) -> BaseParser:
380 pass
382 def _get_push_notifications_parser(self) -> Union[_HiredisParser, _RESP3Parser]:
383 parser = self._get_parser()
384 if not isinstance(parser, (_HiredisParser, _RESP3Parser)):
385 raise RedisError(
386 "Maintenance notifications are only supported with hiredis and RESP3 parsers!"
387 )
388 return parser
390 @abstractmethod
391 def _get_socket(self) -> Optional[socket.socket]:
392 pass
394 @abstractmethod
395 def get_protocol(self) -> Union[int, str]:
396 """
397 Returns:
398 The RESP protocol version, or ``None`` if the protocol is not specified,
399 in which case the server default will be used.
400 """
401 pass
403 @property
404 @abstractmethod
405 def host(self) -> str:
406 pass
408 @host.setter
409 @abstractmethod
410 def host(self, value: str):
411 pass
413 @property
414 @abstractmethod
415 def socket_timeout(self) -> Optional[Union[float, int]]:
416 pass
418 @socket_timeout.setter
419 @abstractmethod
420 def socket_timeout(self, value: Optional[Union[float, int]]):
421 pass
423 @property
424 @abstractmethod
425 def socket_connect_timeout(self) -> Optional[Union[float, int]]:
426 pass
428 @socket_connect_timeout.setter
429 @abstractmethod
430 def socket_connect_timeout(self, value: Optional[Union[float, int]]):
431 pass
433 @abstractmethod
434 def send_command(self, *args, **kwargs):
435 pass
437 @abstractmethod
438 def read_response(
439 self,
440 disable_decoding=False,
441 *,
442 timeout: Union[float, object] = SENTINEL,
443 disconnect_on_error=True,
444 push_request=False,
445 ):
446 pass
448 @abstractmethod
449 def disconnect(self, *args, **kwargs):
450 pass
452 @abstractmethod
453 def mark_for_reconnect(self):
454 pass
456 def _configure_maintenance_notifications(
457 self,
458 maint_notifications_pool_handler: Optional[
459 MaintNotificationsPoolHandler
460 ] = None,
461 orig_host_address=None,
462 orig_socket_timeout=None,
463 orig_socket_connect_timeout=None,
464 oss_cluster_maint_notifications_handler: Optional[
465 OSSMaintNotificationsHandler
466 ] = None,
467 parser: Optional[BaseParser] = None,
468 ):
469 """
470 Enable maintenance notifications by setting up
471 handlers and storing original connection parameters.
473 Should be used ONLY with parsers that support push notifications.
474 """
475 if (
476 not self.maint_notifications_config
477 or not self.maint_notifications_config.enabled
478 ):
479 self._maint_notifications_pool_handler = None
480 self._maint_notifications_connection_handler = None
481 self._oss_cluster_maint_notifications_handler = None
482 return
484 if not parser:
485 raise RedisError(
486 "To configure maintenance notifications, a parser must be provided!"
487 )
489 if not isinstance(parser, _HiredisParser) and not isinstance(
490 parser, _RESP3Parser
491 ):
492 raise RedisError(
493 "Maintenance notifications are only supported with hiredis and RESP3 parsers!"
494 )
496 if maint_notifications_pool_handler:
497 # Extract a reference to a new pool handler that copies all properties
498 # of the original one and has a different connection reference
499 # This is needed because when we attach the handler to the parser
500 # we need to make sure that the handler has a reference to the
501 # connection that the parser is attached to.
502 self._maint_notifications_pool_handler = (
503 maint_notifications_pool_handler.get_handler_for_connection()
504 )
505 self._maint_notifications_pool_handler.set_connection(self)
506 else:
507 self._maint_notifications_pool_handler = None
509 self._maint_notifications_connection_handler = (
510 MaintNotificationsConnectionHandler(self, self.maint_notifications_config)
511 )
513 if oss_cluster_maint_notifications_handler:
514 self._oss_cluster_maint_notifications_handler = (
515 oss_cluster_maint_notifications_handler
516 )
517 # Set up OSS cluster handler to parser
518 parser.set_oss_cluster_maint_push_handler(
519 self._oss_cluster_maint_notifications_handler.handle_notification
520 )
521 else:
522 self._oss_cluster_maint_notifications_handler = None
524 # Set up pool handler to parser if available
525 if self._maint_notifications_pool_handler:
526 parser.set_node_moving_push_handler(
527 self._maint_notifications_pool_handler.handle_notification
528 )
530 # Set up connection handler
531 parser.set_maintenance_push_handler(
532 self._maint_notifications_connection_handler.handle_notification
533 )
535 # Store original connection parameters
536 self.orig_host_address = orig_host_address if orig_host_address else self.host
537 self.orig_socket_timeout = (
538 orig_socket_timeout if orig_socket_timeout else self.socket_timeout
539 )
540 self.orig_socket_connect_timeout = (
541 orig_socket_connect_timeout
542 if orig_socket_connect_timeout
543 else self.socket_connect_timeout
544 )
546 def set_maint_notifications_pool_handler_for_connection(
547 self, maint_notifications_pool_handler: MaintNotificationsPoolHandler
548 ):
549 # Deep copy the pool handler to avoid sharing the same pool handler
550 # between multiple connections, because otherwise each connection will override
551 # the connection reference and the pool handler will only hold a reference
552 # to the last connection that was set.
553 maint_notifications_pool_handler_copy = (
554 maint_notifications_pool_handler.get_handler_for_connection()
555 )
557 maint_notifications_pool_handler_copy.set_connection(self)
558 parser = self._get_push_notifications_parser()
559 parser.set_node_moving_push_handler(
560 maint_notifications_pool_handler_copy.handle_notification
561 )
563 self._maint_notifications_pool_handler = maint_notifications_pool_handler_copy
565 # Update maintenance notification connection handler if it doesn't exist
566 if not self._maint_notifications_connection_handler:
567 self._maint_notifications_connection_handler = (
568 MaintNotificationsConnectionHandler(
569 self, maint_notifications_pool_handler.config
570 )
571 )
572 parser.set_maintenance_push_handler(
573 self._maint_notifications_connection_handler.handle_notification
574 )
575 else:
576 self._maint_notifications_connection_handler.config = (
577 maint_notifications_pool_handler.config
578 )
580 def set_maint_notifications_cluster_handler_for_connection(
581 self, oss_cluster_maint_notifications_handler: OSSMaintNotificationsHandler
582 ):
583 parser = self._get_push_notifications_parser()
584 parser.set_oss_cluster_maint_push_handler(
585 oss_cluster_maint_notifications_handler.handle_notification
586 )
587 # OSS cluster mode and pool-handler mode are mutually exclusive. Clear
588 # any node-moving/pool handler a default (RESP3 "auto") pool wired in
589 # __init__ so this existing connection is not configured with both.
590 parser.set_node_moving_push_handler(None)
591 self._maint_notifications_pool_handler = None
593 self._oss_cluster_maint_notifications_handler = (
594 oss_cluster_maint_notifications_handler
595 )
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, oss_cluster_maint_notifications_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 oss_cluster_maint_notifications_handler.config
610 )
612 def activate_maint_notifications_handling_if_enabled(self, check_health=True):
613 # Send maintenance notifications handshake if RESP3 is active
614 # and maintenance notifications are enabled
615 # and we have a host to determine the endpoint type from
616 # When the maint_notifications_config enabled mode is "auto",
617 # we just log a warning if the handshake fails
618 # When the mode is enabled=True, we raise an exception in case of failure
619 host = getattr(self, "host", None)
620 if (
621 check_protocol_version(self.get_protocol(), 3)
622 and self.maint_notifications_config
623 and self.maint_notifications_config.enabled
624 and self._maint_notifications_connection_handler
625 and host is not None
626 ):
627 self._enable_maintenance_notifications(
628 maint_notifications_config=self.maint_notifications_config,
629 check_health=check_health,
630 )
632 def _enable_maintenance_notifications(
633 self, maint_notifications_config: MaintNotificationsConfig, check_health=True
634 ):
635 try:
636 host = getattr(self, "host", None)
637 if host is None:
638 raise ValueError(
639 "Cannot enable maintenance notifications for connection"
640 " object that doesn't have a host attribute."
641 )
642 else:
643 endpoint_type = maint_notifications_config.get_endpoint_type(host, self)
644 self.send_command(
645 "CLIENT",
646 "MAINT_NOTIFICATIONS",
647 "ON",
648 "moving-endpoint-type",
649 endpoint_type.value,
650 check_health=check_health,
651 )
652 response = self.read_response()
653 if not response or str_if_bytes(response) != "OK":
654 raise ResponseError(
655 "The server doesn't support maintenance notifications"
656 )
657 except Exception as e:
658 if (
659 isinstance(e, ResponseError)
660 and maint_notifications_config.enabled == "auto"
661 ):
662 # Log warning but don't fail the connection
663 import logging
665 logger = logging.getLogger(__name__)
666 logger.debug(f"Failed to enable maintenance notifications: {e}")
667 else:
668 raise
670 def get_resolved_ip(self) -> Optional[str]:
671 """
672 Extract the resolved IP address from an
673 established connection or resolve it from the host.
675 First tries to get the actual IP from the socket (most accurate),
676 then falls back to DNS resolution if needed.
678 Returns:
679 str: The resolved IP address, or None if it cannot be determined
680 """
682 # Method 1: Try to get the actual IP from the established socket connection
683 # This is most accurate as it shows the exact IP being used
684 try:
685 conn_socket = self._get_socket()
686 if conn_socket is not None:
687 peer_addr = conn_socket.getpeername()
688 if peer_addr and len(peer_addr) >= 1:
689 # For TCP sockets, peer_addr is typically (host, port) tuple
690 # Return just the host part
691 return peer_addr[0]
692 except (AttributeError, OSError):
693 # Socket might not be connected or getpeername() might fail
694 pass
696 # Method 2: Fallback to DNS resolution of the host
697 # This is less accurate but works when socket is not available
698 try:
699 host = getattr(self, "host", "localhost")
700 port = getattr(self, "port", 6379)
701 if host:
702 # Use getaddrinfo to resolve the hostname to IP
703 # This mimics what the connection would do during _connect()
704 addr_info = socket.getaddrinfo(
705 host, port, socket.AF_UNSPEC, socket.SOCK_STREAM
706 )
707 if addr_info:
708 # Return the IP from the first result
709 # addr_info[0] is (family, socktype, proto, canonname, sockaddr)
710 # sockaddr[0] is the IP address
711 return str(addr_info[0][4][0])
712 except (AttributeError, OSError, socket.gaierror):
713 # DNS resolution might fail
714 pass
716 return None
718 @property
719 def maintenance_state(self) -> MaintenanceState:
720 return self._maintenance_state
722 @maintenance_state.setter
723 def maintenance_state(self, state: "MaintenanceState"):
724 self._maintenance_state = state
726 def add_maint_start_notification(self, id: int):
727 self._processed_start_maint_notifications.add(id)
729 def get_processed_start_notifications(self) -> set:
730 return self._processed_start_maint_notifications
732 def add_skipped_end_notification(self, id: int):
733 self._skipped_end_maint_notifications.add(id)
735 def get_skipped_end_notifications(self) -> set:
736 return self._skipped_end_maint_notifications
738 def reset_received_notifications(self):
739 self._processed_start_maint_notifications.clear()
740 self._skipped_end_maint_notifications.clear()
742 def getpeername(self):
743 """
744 Returns the peer name of the connection.
745 """
746 conn_socket = self._get_socket()
747 if conn_socket:
748 return conn_socket.getpeername()[0]
749 return None
751 def update_current_socket_timeout(self, relaxed_timeout: Optional[float] = None):
752 conn_socket = self._get_socket()
753 if conn_socket:
754 timeout = relaxed_timeout if relaxed_timeout != -1 else self.socket_timeout
755 # if the current timeout is 0 it means we are in the middle of a can_read call
756 # in this case we don't want to change the timeout because the operation
757 # is non-blocking and should return immediately
758 # Changing the state from non-blocking to blocking in the middle of a read operation
759 # will lead to a deadlock
760 if conn_socket.gettimeout() != 0:
761 conn_socket.settimeout(timeout)
762 self.update_parser_timeout(timeout)
764 def update_parser_timeout(self, timeout: Optional[float] = None):
765 parser = self._get_parser()
766 if parser and parser._buffer:
767 if isinstance(parser, _RESP3Parser) and timeout:
768 parser._buffer.socket_timeout = timeout
769 elif isinstance(parser, _HiredisParser):
770 parser._socket_timeout = timeout
772 def set_tmp_settings(
773 self,
774 tmp_host_address: Optional[Union[str, object]] = SENTINEL,
775 tmp_relaxed_timeout: Optional[float] = -1,
776 ):
777 """
778 SENTINEL keeps the host unchanged. -1 keeps the relaxed timeout unchanged.
779 """
780 if tmp_host_address and tmp_host_address != SENTINEL:
781 self.host = str(tmp_host_address)
782 if tmp_relaxed_timeout != -1:
783 self.socket_timeout = tmp_relaxed_timeout
784 self.socket_connect_timeout = tmp_relaxed_timeout
786 def reset_tmp_settings(
787 self,
788 reset_host_address: bool = False,
789 reset_relaxed_timeout: bool = False,
790 ):
791 if reset_host_address:
792 self.host = self.orig_host_address
793 if reset_relaxed_timeout:
794 self.socket_timeout = self.orig_socket_timeout
795 self.socket_connect_timeout = self.orig_socket_connect_timeout
798class AbstractConnection(MaintNotificationsAbstractConnection, ConnectionInterface):
799 "Manages communication to and from a Redis server"
801 @deprecated_args(
802 args_to_warn=["lib_name", "lib_version"],
803 reason="Use 'driver_info' parameter instead. "
804 "lib_name and lib_version will be removed in a future version.",
805 )
806 def __init__(
807 self,
808 db: int = 0,
809 password: Optional[str] = None,
810 socket_timeout: Optional[float] = DEFAULT_SOCKET_TIMEOUT,
811 socket_connect_timeout: Optional[float] = DEFAULT_SOCKET_CONNECT_TIMEOUT,
812 retry_on_timeout: bool = False,
813 retry_on_error: Union[Iterable[Type[Exception]], object] = SENTINEL,
814 encoding: str = "utf-8",
815 encoding_errors: str = "strict",
816 decode_responses: bool = False,
817 parser_class=DefaultParser,
818 socket_read_size: int = DEFAULT_SOCKET_READ_SIZE,
819 health_check_interval: int = 0,
820 client_name: Optional[str] = None,
821 lib_name: Union[Optional[str], object] = SENTINEL,
822 lib_version: Union[Optional[str], object] = SENTINEL,
823 driver_info: Union[Optional[DriverInfo], object] = SENTINEL,
824 username: Optional[str] = None,
825 retry: Union[Any, None] = None,
826 redis_connect_func: Optional[Callable[[], None]] = None,
827 credential_provider: Optional[CredentialProvider] = None,
828 protocol: Optional[int] = None,
829 legacy_responses: bool = True,
830 command_packer: Optional[Callable[[], None]] = None,
831 event_dispatcher: Optional[EventDispatcher] = None,
832 maint_notifications_config: Optional[MaintNotificationsConfig] = None,
833 maint_notifications_pool_handler: Optional[
834 MaintNotificationsPoolHandler
835 ] = None,
836 maintenance_state: "MaintenanceState" = MaintenanceState.NONE,
837 maintenance_notification_hash: Optional[int] = None,
838 orig_host_address: Optional[str] = None,
839 orig_socket_timeout: Optional[float] = None,
840 orig_socket_connect_timeout: Optional[float] = None,
841 oss_cluster_maint_notifications_handler: Optional[
842 OSSMaintNotificationsHandler
843 ] = None,
844 ):
845 """
846 Initialize a new Connection.
848 To specify a retry policy for specific errors, first set
849 `retry_on_error` to a list of the error/s to retry on, then set
850 `retry` to a valid `Retry` object.
851 To retry on TimeoutError, `retry_on_timeout` can also be set to `True`.
853 Parameters
854 ----------
855 driver_info : DriverInfo, optional
856 Driver metadata for CLIENT SETINFO. If provided, lib_name and lib_version
857 are ignored. If not provided, a DriverInfo will be created from lib_name
858 and lib_version. Explicit None disables CLIENT SETINFO.
859 lib_name : str, optional
860 **Deprecated.** Use driver_info instead. Library name for CLIENT SETINFO.
861 lib_version : str, optional
862 **Deprecated.** Use driver_info instead. Library version for CLIENT SETINFO.
863 """
864 if (username or password) and credential_provider is not None:
865 raise DataError(
866 "'username' and 'password' cannot be passed along with 'credential_"
867 "provider'. Please provide only one of the following arguments: \n"
868 "1. 'password' and (optional) 'username'\n"
869 "2. 'credential_provider'"
870 )
871 if event_dispatcher is None:
872 self._event_dispatcher = EventDispatcher()
873 else:
874 self._event_dispatcher = event_dispatcher
875 self.pid = os.getpid()
876 self.db = db
877 self.client_name = client_name
879 # Handle driver_info: if provided, use it; otherwise create from lib_name/lib_version.
880 self.driver_info = resolve_driver_info(driver_info, lib_name, lib_version)
882 self.credential_provider = credential_provider
883 self.password = password
884 self.username = username
885 self._socket_timeout = socket_timeout
886 if socket_connect_timeout is None:
887 socket_connect_timeout = socket_timeout
888 self._socket_connect_timeout = socket_connect_timeout
889 self.retry_on_timeout = retry_on_timeout
890 if retry_on_error is SENTINEL:
891 retry_on_errors_list = []
892 else:
893 retry_on_errors_list = list(retry_on_error)
894 if retry_on_timeout:
895 # Add TimeoutError to the errors list to retry on
896 retry_on_errors_list.append(TimeoutError)
897 self.retry_on_error = retry_on_errors_list
898 if retry or self.retry_on_error:
899 if retry is None:
900 self.retry = Retry(NoBackoff(), 1)
901 else:
902 # deep-copy the Retry object as it is mutable
903 self.retry = copy.deepcopy(retry)
904 if self.retry_on_error:
905 # Update the retry's supported errors with the specified errors
906 self.retry.update_supported_errors(self.retry_on_error)
907 else:
908 self.retry = Retry(NoBackoff(), 0)
909 self.health_check_interval = health_check_interval
910 self.next_health_check = 0
911 self.redis_connect_func = redis_connect_func
912 self.encoder = Encoder(encoding, encoding_errors, decode_responses)
913 self.handshake_metadata = None
914 self._sock = None
915 self._socket_read_size = socket_read_size
916 self._connect_callbacks = []
917 self._buffer_cutoff = 6000
918 self._re_auth_token: Optional[TokenInterface] = None
919 try:
920 p = int(protocol)
921 except TypeError:
922 p = DEFAULT_RESP_VERSION
923 except ValueError:
924 raise ConnectionError("protocol must be an integer")
925 else:
926 if p < 2 or p > 3:
927 raise ConnectionError("protocol must be either 2 or 3")
928 self.protocol = p
929 self.legacy_responses = legacy_responses
930 if self.protocol == 3 and parser_class == _RESP2Parser:
931 # If the protocol is 3 but the parser is RESP2, change it to RESP3
932 # This is needed because the parser might be set before the protocol
933 # or might be provided as a kwarg to the constructor
934 # We need to react on discrepancy only for RESP2 and RESP3
935 # as hiredis supports both
936 parser_class = _RESP3Parser
937 self.set_parser(parser_class)
939 self._command_packer = self._construct_command_packer(command_packer)
940 self._should_reconnect = False
942 # Set up maintenance notifications
943 MaintNotificationsAbstractConnection.__init__(
944 self,
945 maint_notifications_config,
946 maint_notifications_pool_handler,
947 maintenance_state,
948 maintenance_notification_hash,
949 orig_host_address,
950 orig_socket_timeout,
951 orig_socket_connect_timeout,
952 oss_cluster_maint_notifications_handler,
953 self._parser,
954 event_dispatcher=self._event_dispatcher,
955 )
957 def __repr__(self):
958 repr_args = ",".join([f"{k}={v}" for k, v in self.repr_pieces()])
959 return f"<{self.__class__.__module__}.{self.__class__.__name__}({repr_args})>"
961 @abstractmethod
962 def repr_pieces(self):
963 pass
965 def __del__(self):
966 try:
967 self.disconnect()
968 except Exception:
969 pass
971 @property
972 def is_connected(self) -> bool:
973 return self._sock is not None
975 def _construct_command_packer(self, packer):
976 if packer is not None:
977 return packer
978 elif HIREDIS_AVAILABLE:
979 return HiredisRespSerializer()
980 else:
981 return PythonRespSerializer(self._buffer_cutoff, self.encoder.encode)
983 def register_connect_callback(self, callback):
984 """
985 Register a callback to be called when the connection is established either
986 initially or reconnected. This allows listeners to issue commands that
987 are ephemeral to the connection, for example pub/sub subscription or
988 key tracking. The callback must be a _method_ and will be kept as
989 a weak reference.
990 """
991 wm = weakref.WeakMethod(callback)
992 if wm not in self._connect_callbacks:
993 self._connect_callbacks.append(wm)
995 def deregister_connect_callback(self, callback):
996 """
997 De-register a previously registered callback. It will no-longer receive
998 notifications on connection events. Calling this is not required when the
999 listener goes away, since the callbacks are kept as weak methods.
1000 """
1001 try:
1002 self._connect_callbacks.remove(weakref.WeakMethod(callback))
1003 except ValueError:
1004 pass
1006 def set_parser(self, parser_class):
1007 """
1008 Creates a new instance of parser_class with socket size:
1009 _socket_read_size and assigns it to the parser for the connection
1010 :param parser_class: The required parser class
1011 """
1012 self._parser = parser_class(socket_read_size=self._socket_read_size)
1014 def _get_parser(self) -> Union[_HiredisParser, _RESP3Parser, _RESP2Parser]:
1015 return self._parser
1017 def connect(self):
1018 "Connects to the Redis server if not already connected"
1019 # try once the socket connect with the handshake, retry the whole
1020 # connect/handshake flow based on retry policy
1021 self.retry.call_with_retry(
1022 lambda: self.connect_check_health(
1023 check_health=True, retry_socket_connect=False
1024 ),
1025 lambda error: self.disconnect(error),
1026 )
1028 def connect_check_health(
1029 self, check_health: bool = True, retry_socket_connect: bool = True
1030 ):
1031 if self._sock:
1032 return
1033 # Track actual retry attempts for error reporting
1034 actual_retry_attempts = [0]
1036 def failure_callback(error, failure_count):
1037 actual_retry_attempts[0] = failure_count
1038 self.disconnect(error=error, failure_count=failure_count)
1040 try:
1041 if retry_socket_connect:
1042 sock = self.retry.call_with_retry(
1043 self._connect,
1044 failure_callback,
1045 with_failure_count=True,
1046 )
1047 else:
1048 sock = self._connect()
1049 except socket.timeout:
1050 e = TimeoutError("Timeout connecting to server")
1051 record_error_count(
1052 server_address=self.host,
1053 server_port=self.port,
1054 network_peer_address=self.host,
1055 network_peer_port=self.port,
1056 error_type=e,
1057 retry_attempts=actual_retry_attempts[0],
1058 )
1059 raise e
1060 except OSError as e:
1061 e = ConnectionError(self._error_message(e))
1062 record_error_count(
1063 server_address=getattr(self, "host", None),
1064 server_port=getattr(self, "port", None),
1065 network_peer_address=getattr(self, "host", None),
1066 network_peer_port=getattr(self, "port", None),
1067 error_type=e,
1068 retry_attempts=actual_retry_attempts[0],
1069 )
1070 raise e
1072 self._sock = sock
1073 try:
1074 if self.redis_connect_func is None:
1075 # Use the default on_connect function
1076 self.on_connect_check_health(check_health=check_health)
1077 else:
1078 # Use the passed function redis_connect_func
1079 self.redis_connect_func(self)
1080 except RedisError:
1081 # clean up after any error in on_connect
1082 self.disconnect()
1083 raise
1085 # run any user callbacks. right now the only internal callback
1086 # is for pubsub channel/pattern resubscription
1087 # first, remove any dead weakrefs
1088 self._connect_callbacks = [ref for ref in self._connect_callbacks if ref()]
1089 for ref in self._connect_callbacks:
1090 callback = ref()
1091 if callback:
1092 callback(self)
1094 @abstractmethod
1095 def _connect(self):
1096 pass
1098 @abstractmethod
1099 def _host_error(self):
1100 pass
1102 def _error_message(self, exception):
1103 return format_error_message(self._host_error(), exception)
1105 def on_connect(self):
1106 self.on_connect_check_health(check_health=True)
1108 def on_connect_check_health(self, check_health: bool = True):
1109 "Initialize the connection, authenticate and select a database"
1110 self._parser.on_connect(self)
1111 parser = self._parser
1113 auth_args = None
1114 # if credential provider or username and/or password are set, authenticate
1115 if self.credential_provider or (self.username or self.password):
1116 cred_provider = (
1117 self.credential_provider
1118 or UsernamePasswordCredentialProvider(self.username, self.password)
1119 )
1120 auth_args = cred_provider.get_credentials()
1122 # if resp version is specified and we have auth args,
1123 # we need to send them via HELLO
1124 if auth_args and check_protocol_version(self.protocol, 3):
1125 if isinstance(self._parser, _RESP2Parser):
1126 self.set_parser(_RESP3Parser)
1127 # update cluster exception classes
1128 self._parser.EXCEPTION_CLASSES = parser.EXCEPTION_CLASSES
1129 self._parser.on_connect(self)
1130 if len(auth_args) == 1:
1131 auth_args = ["default", auth_args[0]]
1132 # avoid checking health here -- PING will fail if we try
1133 # to check the health prior to the AUTH
1134 self.send_command(
1135 "HELLO", self.protocol, "AUTH", *auth_args, check_health=False
1136 )
1137 self.handshake_metadata = self.read_response()
1138 # if response.get(b"proto") != self.protocol and response.get(
1139 # "proto"
1140 # ) != self.protocol:
1141 # raise ConnectionError("Invalid RESP version")
1142 elif auth_args:
1143 # avoid checking health here -- PING will fail if we try
1144 # to check the health prior to the AUTH
1145 self.send_command("AUTH", *auth_args, check_health=False)
1147 try:
1148 auth_response = self.read_response()
1149 except AuthenticationWrongNumberOfArgsError:
1150 # a username and password were specified but the Redis
1151 # server seems to be < 6.0.0 which expects a single password
1152 # arg. retry auth with just the password.
1153 # https://github.com/andymccurdy/redis-py/issues/1274
1154 self.send_command("AUTH", auth_args[-1], check_health=False)
1155 auth_response = self.read_response()
1157 if str_if_bytes(auth_response) != "OK":
1158 raise AuthenticationError("Invalid Username or Password")
1160 # if resp version is specified, switch to it
1161 elif check_protocol_version(self.protocol, 3):
1162 if isinstance(self._parser, _RESP2Parser):
1163 self.set_parser(_RESP3Parser)
1164 # update cluster exception classes
1165 self._parser.EXCEPTION_CLASSES = parser.EXCEPTION_CLASSES
1166 self._parser.on_connect(self)
1167 self.send_command("HELLO", self.protocol, check_health=check_health)
1168 self.handshake_metadata = self.read_response()
1169 if (
1170 self.handshake_metadata.get(b"proto") != self.protocol
1171 and self.handshake_metadata.get("proto") != self.protocol
1172 ):
1173 raise ConnectionError("Invalid RESP version")
1175 # Activate maintenance notifications for this connection
1176 # if enabled in the configuration
1177 # This is a no-op if maintenance notifications are not enabled
1178 self.activate_maint_notifications_handling_if_enabled(check_health=check_health)
1180 # if a client_name is given, set it
1181 if self.client_name:
1182 self.send_command(
1183 "CLIENT",
1184 "SETNAME",
1185 self.client_name,
1186 check_health=check_health,
1187 )
1188 if str_if_bytes(self.read_response()) != "OK":
1189 raise ConnectionError("Error setting client name")
1191 # Set the library name and version from driver_info
1192 try:
1193 if self.driver_info and self.driver_info.formatted_name:
1194 self.send_command(
1195 "CLIENT",
1196 "SETINFO",
1197 "LIB-NAME",
1198 self.driver_info.formatted_name,
1199 check_health=check_health,
1200 )
1201 self.read_response()
1202 except ResponseError:
1203 pass
1205 try:
1206 if self.driver_info and self.driver_info.lib_version:
1207 self.send_command(
1208 "CLIENT",
1209 "SETINFO",
1210 "LIB-VER",
1211 self.driver_info.lib_version,
1212 check_health=check_health,
1213 )
1214 self.read_response()
1215 except ResponseError:
1216 pass
1218 # if a database is specified, switch to it
1219 if self.db:
1220 self.send_command("SELECT", self.db, check_health=check_health)
1221 if str_if_bytes(self.read_response()) != "OK":
1222 raise ConnectionError("Invalid Database")
1224 def disconnect(self, *args, **kwargs):
1225 "Disconnects from the Redis server"
1226 self._parser.on_disconnect()
1228 conn_sock = self._sock
1229 self._sock = None
1230 # reset the reconnect flag
1231 self.reset_should_reconnect()
1233 if conn_sock is None:
1234 return
1236 if os.getpid() == self.pid:
1237 try:
1238 conn_sock.shutdown(socket.SHUT_RDWR)
1239 except (OSError, TypeError):
1240 pass
1242 try:
1243 conn_sock.close()
1244 except OSError:
1245 pass
1247 error = kwargs.get("error")
1248 failure_count = kwargs.get("failure_count")
1249 health_check_failed = kwargs.get("health_check_failed")
1251 if error:
1252 if health_check_failed:
1253 close_reason = CloseReason.HEALTHCHECK_FAILED
1254 else:
1255 close_reason = CloseReason.ERROR
1257 if failure_count is not None and failure_count > self.retry.get_retries():
1258 record_error_count(
1259 server_address=self.host,
1260 server_port=self.port,
1261 network_peer_address=self.host,
1262 network_peer_port=self.port,
1263 error_type=error,
1264 retry_attempts=failure_count,
1265 )
1267 record_connection_closed(
1268 close_reason=close_reason,
1269 error_type=error,
1270 )
1271 else:
1272 record_connection_closed(
1273 close_reason=CloseReason.APPLICATION_CLOSE,
1274 )
1276 if self.maintenance_state == MaintenanceState.MAINTENANCE:
1277 # this block will be executed only if the connection was in maintenance state
1278 # and the connection was closed.
1279 # The state change won't be applied on connections that are in Moving state
1280 # because their state and configurations will be handled when the moving ttl expires.
1281 self.reset_tmp_settings(reset_relaxed_timeout=True)
1282 self.maintenance_state = MaintenanceState.NONE
1283 # reset the sets that keep track of received start maint
1284 # notifications and skipped end maint notifications
1285 self.reset_received_notifications()
1287 def mark_for_reconnect(self):
1288 self._should_reconnect = True
1290 def should_reconnect(self):
1291 return self._should_reconnect
1293 def reset_should_reconnect(self):
1294 self._should_reconnect = False
1296 def _send_ping(self):
1297 """Send PING, expect PONG in return"""
1298 self.send_command("PING", check_health=False)
1299 if str_if_bytes(self.read_response()) != "PONG":
1300 raise ConnectionError("Bad response from PING health check")
1302 def _ping_failed(self, error, failure_count):
1303 """Function to call when PING fails"""
1304 self.disconnect(
1305 error=error, failure_count=failure_count, health_check_failed=True
1306 )
1308 def check_health(self):
1309 """Check the health of the connection with a PING/PONG"""
1310 if self.health_check_interval and time.monotonic() > self.next_health_check:
1311 self.retry.call_with_retry(
1312 self._send_ping,
1313 self._ping_failed,
1314 with_failure_count=True,
1315 )
1317 def send_packed_command(self, command, check_health=True):
1318 """Send an already packed command to the Redis server"""
1319 if not self._sock:
1320 self.connect_check_health(check_health=False)
1321 # guard against health check recursion
1322 if check_health:
1323 self.check_health()
1324 try:
1325 if isinstance(command, str):
1326 command = [command]
1327 for item in command:
1328 self._sock.sendall(item)
1329 except socket.timeout:
1330 self.disconnect()
1331 raise TimeoutError("Timeout writing to socket")
1332 except OSError as e:
1333 self.disconnect()
1334 if len(e.args) == 1:
1335 errno, errmsg = "UNKNOWN", e.args[0]
1336 else:
1337 errno = e.args[0]
1338 errmsg = e.args[1]
1339 raise ConnectionError(f"Error {errno} while writing to socket. {errmsg}.")
1340 except BaseException:
1341 # BaseExceptions can be raised when a socket send operation is not
1342 # finished, e.g. due to a timeout. Ideally, a caller could then re-try
1343 # to send un-sent data. However, the send_packed_command() API
1344 # does not support it so there is no point in keeping the connection open.
1345 self.disconnect()
1346 raise
1348 def send_command(self, *args, **kwargs):
1349 """Pack and send a command to the Redis server"""
1350 self.send_packed_command(
1351 self._command_packer.pack(*args),
1352 check_health=kwargs.get("check_health", True),
1353 )
1355 def can_read(self, timeout: float = 0) -> bool:
1356 """Poll the socket to see if there's data that can be read."""
1357 # TODO: Rename this API; it detects pending data or dirty/closed
1358 # connection state, not only whether application data can be read.
1359 sock = self._sock
1360 if not sock:
1361 self.connect()
1363 host_error = self._host_error()
1365 try:
1366 return self._parser.can_read(timeout)
1368 except OSError as e:
1369 self.disconnect()
1370 raise ConnectionError(f"Error while reading from {host_error}: {e.args}")
1372 def read_response(
1373 self,
1374 disable_decoding=False,
1375 *,
1376 timeout: Union[float, object] = SENTINEL,
1377 disconnect_on_error=True,
1378 push_request=False,
1379 ):
1380 """Read the response from a previously sent command"""
1382 host_error = self._host_error()
1384 try:
1385 if self.protocol in ["3", 3]:
1386 response = self._parser.read_response(
1387 disable_decoding=disable_decoding,
1388 push_request=push_request,
1389 timeout=timeout,
1390 )
1391 else:
1392 response = self._parser.read_response(
1393 disable_decoding=disable_decoding, timeout=timeout
1394 )
1395 except socket.timeout:
1396 if disconnect_on_error:
1397 self.disconnect()
1398 raise TimeoutError(f"Timeout reading from {host_error}")
1399 except OSError as e:
1400 if disconnect_on_error:
1401 self.disconnect()
1402 raise ConnectionError(f"Error while reading from {host_error} : {e.args}")
1403 except BaseException:
1404 # Also by default close in case of BaseException. A lot of code
1405 # relies on this behaviour when doing Command/Response pairs.
1406 # See #1128.
1407 if disconnect_on_error:
1408 self.disconnect()
1409 raise
1411 if self.health_check_interval:
1412 self.next_health_check = time.monotonic() + self.health_check_interval
1414 if isinstance(response, ResponseError):
1415 try:
1416 raise response
1417 finally:
1418 del response # avoid creating ref cycles
1419 return response
1421 def pack_command(self, *args):
1422 """Pack a series of arguments into the Redis protocol"""
1423 return self._command_packer.pack(*args)
1425 def pack_commands(self, commands):
1426 """Pack multiple commands into the Redis protocol"""
1427 output = []
1428 pieces = []
1429 buffer_length = 0
1430 buffer_cutoff = self._buffer_cutoff
1432 for cmd in commands:
1433 for chunk in self._command_packer.pack(*cmd):
1434 chunklen = len(chunk)
1435 if (
1436 buffer_length > buffer_cutoff
1437 or chunklen > buffer_cutoff
1438 or isinstance(chunk, memoryview)
1439 ):
1440 if pieces:
1441 output.append(SYM_EMPTY.join(pieces))
1442 buffer_length = 0
1443 pieces = []
1445 if chunklen > buffer_cutoff or isinstance(chunk, memoryview):
1446 output.append(chunk)
1447 else:
1448 pieces.append(chunk)
1449 buffer_length += chunklen
1451 if pieces:
1452 output.append(SYM_EMPTY.join(pieces))
1453 return output
1455 def get_protocol(self) -> Union[int, str]:
1456 return self.protocol
1458 @property
1459 def handshake_metadata(self) -> Union[Dict[bytes, bytes], Dict[str, str]]:
1460 return self._handshake_metadata
1462 @handshake_metadata.setter
1463 def handshake_metadata(self, value: Union[Dict[bytes, bytes], Dict[str, str]]):
1464 self._handshake_metadata = value
1466 def set_re_auth_token(self, token: TokenInterface):
1467 self._re_auth_token = token
1469 def re_auth(self):
1470 if self._re_auth_token is not None:
1471 self.send_command(
1472 "AUTH",
1473 self._re_auth_token.try_get("oid"),
1474 self._re_auth_token.get_value(),
1475 )
1476 self.read_response()
1477 self._re_auth_token = None
1479 def _get_socket(self) -> Optional[socket.socket]:
1480 return self._sock
1482 @property
1483 def socket_timeout(self) -> Optional[Union[float, int]]:
1484 return self._socket_timeout
1486 @socket_timeout.setter
1487 def socket_timeout(self, value: Optional[Union[float, int]]):
1488 self._socket_timeout = value
1490 @property
1491 def socket_connect_timeout(self) -> Optional[Union[float, int]]:
1492 return self._socket_connect_timeout
1494 @socket_connect_timeout.setter
1495 def socket_connect_timeout(self, value: Optional[Union[float, int]]):
1496 self._socket_connect_timeout = value
1498 def extract_connection_details(self) -> str:
1499 socket_address = None
1500 if self._sock is None:
1501 return "not connected"
1502 try:
1503 socket_address = self._sock.getsockname() if self._sock else None
1504 socket_address = socket_address[1] if socket_address else None
1505 except (AttributeError, OSError):
1506 pass
1508 return f"connected to ip {self.get_resolved_ip()}, local socket port: {socket_address}"
1511class Connection(AbstractConnection):
1512 "Manages TCP communication to and from a Redis server"
1514 def __init__(
1515 self,
1516 host="localhost",
1517 port=6379,
1518 socket_keepalive=True,
1519 socket_keepalive_options=SENTINEL,
1520 socket_type=0,
1521 **kwargs,
1522 ):
1523 """
1524 Initialize a TCP connection.
1526 Parameters
1527 ----------
1528 socket_keepalive : bool
1529 If `True`, TCP keepalive is enabled for TCP socket connections.
1530 socket_keepalive_options : Mapping[int, int | bytes] | object | None
1531 Mapping of TCP keepalive socket option constants to values, for
1532 example `{socket.TCP_KEEPIDLE: 30}`. If left unspecified, redis-py
1533 uses TCP keepalive defaults when `socket_keepalive` is enabled:
1534 idle 30 seconds, interval 5 seconds, and 3 probes. Platform-specific
1535 options that are not available are skipped. Pass `None` or `{}` to
1536 avoid setting additional TCP keepalive options.
1537 """
1538 self._host = host
1539 self.port = int(port)
1540 self.socket_keepalive = socket_keepalive
1541 if socket_keepalive_options is SENTINEL:
1542 socket_keepalive_options = get_default_socket_keepalive_options()
1543 self.socket_keepalive_options = socket_keepalive_options or {}
1544 self.socket_type = socket_type
1545 super().__init__(**kwargs)
1547 def repr_pieces(self):
1548 pieces = [("host", self.host), ("port", self.port), ("db", self.db)]
1549 if self.client_name:
1550 pieces.append(("client_name", self.client_name))
1551 return pieces
1553 def _connect(self):
1554 "Create a TCP socket connection"
1555 # we want to mimic what socket.create_connection does to support
1556 # ipv4/ipv6, but we want to set options prior to calling
1557 # socket.connect()
1558 err = None
1560 for res in socket.getaddrinfo(
1561 self.host, self.port, self.socket_type, socket.SOCK_STREAM
1562 ):
1563 family, socktype, proto, canonname, socket_address = res
1564 sock = None
1565 try:
1566 sock = socket.socket(family, socktype, proto)
1567 # TCP_NODELAY
1568 sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
1570 # TCP_KEEPALIVE
1571 if self.socket_keepalive:
1572 sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
1573 for k, v in self.socket_keepalive_options.items():
1574 sock.setsockopt(socket.IPPROTO_TCP, k, v)
1576 # set the socket_connect_timeout before we connect
1577 sock.settimeout(self.socket_connect_timeout)
1579 # connect
1580 sock.connect(socket_address)
1582 # set the socket_timeout now that we're connected
1583 sock.settimeout(self.socket_timeout)
1584 return sock
1586 except OSError as _:
1587 err = _
1588 if sock is not None:
1589 try:
1590 sock.shutdown(socket.SHUT_RDWR) # ensure a clean close
1591 except OSError:
1592 pass
1593 sock.close()
1595 if err is not None:
1596 raise err
1597 raise OSError("socket.getaddrinfo returned an empty list")
1599 def _host_error(self):
1600 return f"{self.host}:{self.port}"
1602 @property
1603 def host(self) -> str:
1604 return self._host
1606 @host.setter
1607 def host(self, value: str):
1608 self._host = value
1611class CacheProxyConnection(MaintNotificationsAbstractConnection, ConnectionInterface):
1612 DUMMY_CACHE_VALUE = b"foo"
1613 MIN_ALLOWED_VERSION = "7.4.0"
1614 DEFAULT_SERVER_NAME = "redis"
1616 def __init__(
1617 self,
1618 conn: ConnectionInterface,
1619 cache: CacheInterface,
1620 pool_lock: threading.RLock,
1621 ):
1622 self.pid = os.getpid()
1623 self._conn = conn
1624 self.retry = self._conn.retry
1625 self.host = self._conn.host
1626 self.port = self._conn.port
1627 self.db = self._conn.db
1628 self._event_dispatcher = self._conn._event_dispatcher
1629 self.credential_provider = conn.credential_provider
1630 self._pool_lock = pool_lock
1631 self._cache = cache
1632 self._cache_lock = threading.RLock()
1633 self._current_command_cache_key = None
1634 self._current_options = None
1635 self.register_connect_callback(self._enable_tracking_callback)
1637 if isinstance(self._conn, MaintNotificationsAbstractConnection):
1638 MaintNotificationsAbstractConnection.__init__(
1639 self,
1640 self._conn.maint_notifications_config,
1641 self._conn._maint_notifications_pool_handler,
1642 self._conn.maintenance_state,
1643 self._conn.maintenance_notification_hash,
1644 self._conn.host,
1645 self._conn.socket_timeout,
1646 self._conn.socket_connect_timeout,
1647 self._conn._oss_cluster_maint_notifications_handler,
1648 self._conn._get_parser(),
1649 event_dispatcher=self._conn.event_dispatcher,
1650 )
1652 def repr_pieces(self):
1653 return self._conn.repr_pieces()
1655 @property
1656 def is_connected(self) -> bool:
1657 return self._conn.is_connected
1659 def register_connect_callback(self, callback):
1660 self._conn.register_connect_callback(callback)
1662 def deregister_connect_callback(self, callback):
1663 self._conn.deregister_connect_callback(callback)
1665 def set_parser(self, parser_class):
1666 self._conn.set_parser(parser_class)
1668 def set_maint_notifications_pool_handler_for_connection(
1669 self, maint_notifications_pool_handler
1670 ):
1671 if isinstance(self._conn, MaintNotificationsAbstractConnection):
1672 self._conn.set_maint_notifications_pool_handler_for_connection(
1673 maint_notifications_pool_handler
1674 )
1676 def set_maint_notifications_cluster_handler_for_connection(
1677 self, oss_cluster_maint_notifications_handler
1678 ):
1679 if isinstance(self._conn, MaintNotificationsAbstractConnection):
1680 self._conn.set_maint_notifications_cluster_handler_for_connection(
1681 oss_cluster_maint_notifications_handler
1682 )
1684 def get_protocol(self):
1685 return self._conn.get_protocol()
1687 def connect(self):
1688 self._conn.connect()
1690 server_name = self._conn.handshake_metadata.get(b"server", None)
1691 if server_name is None:
1692 server_name = self._conn.handshake_metadata.get("server", None)
1693 server_ver = self._conn.handshake_metadata.get(b"version", None)
1694 if server_ver is None:
1695 server_ver = self._conn.handshake_metadata.get("version", None)
1696 if server_ver is None or server_name is None:
1697 raise ConnectionError("Cannot retrieve information about server version")
1699 server_ver = ensure_string(server_ver)
1700 server_name = ensure_string(server_name)
1702 if (
1703 server_name != self.DEFAULT_SERVER_NAME
1704 or compare_versions(server_ver, self.MIN_ALLOWED_VERSION) == 1
1705 ):
1706 raise ConnectionError(
1707 "To maximize compatibility with all Redis products, client-side caching is supported by Redis 7.4 or later" # noqa: E501
1708 )
1710 def on_connect(self):
1711 self._conn.on_connect()
1713 def disconnect(self, *args, **kwargs):
1714 with self._cache_lock:
1715 self._cache.flush()
1716 self._conn.disconnect(*args, **kwargs)
1718 def check_health(self):
1719 self._conn.check_health()
1721 def send_packed_command(self, command, check_health=True):
1722 # TODO: Investigate if it's possible to unpack command
1723 # or extract keys from packed command
1724 self._conn.send_packed_command(command)
1726 def send_command(self, *args, **kwargs):
1727 self._process_pending_invalidations()
1729 with self._cache_lock:
1730 # Command is write command or not allowed
1731 # to be cached.
1732 if not self._cache.is_cachable(
1733 CacheKey(command=args[0], redis_keys=(), redis_args=())
1734 ):
1735 self._current_command_cache_key = None
1736 self._conn.send_command(*args, **kwargs)
1737 return
1739 if kwargs.get("keys") is None:
1740 raise ValueError("Cannot create cache key.")
1742 # Creates cache key.
1743 self._current_command_cache_key = CacheKey(
1744 command=args[0], redis_keys=tuple(kwargs.get("keys")), redis_args=args
1745 )
1747 with self._cache_lock:
1748 # We have to trigger invalidation processing in case if
1749 # it was cached by another connection to avoid
1750 # queueing invalidations in stale connections.
1751 if self._cache.get(self._current_command_cache_key):
1752 entry = self._cache.get(self._current_command_cache_key)
1754 with self._pool_lock:
1755 while entry.connection_ref.can_read():
1756 try:
1757 entry.connection_ref.read_response(
1758 push_request=True,
1759 timeout=0,
1760 disconnect_on_error=False,
1761 )
1762 except TimeoutError:
1763 break
1765 # Re-check: if the entry was invalidated during the drain,
1766 # fall through to send the command over the network.
1767 if self._cache.get(self._current_command_cache_key):
1768 return
1770 # Set temporary entry value to prevent
1771 # race condition from another connection.
1772 self._cache.set(
1773 CacheEntry(
1774 cache_key=self._current_command_cache_key,
1775 cache_value=self.DUMMY_CACHE_VALUE,
1776 status=CacheEntryStatus.IN_PROGRESS,
1777 connection_ref=self._conn,
1778 )
1779 )
1781 # Send command over socket only if it's allowed
1782 # read-only command that not yet cached.
1783 self._conn.send_command(*args, **kwargs)
1785 def can_read(self, timeout: float = 0) -> bool:
1786 # TODO: Rename this API; it detects pending data or dirty/closed
1787 # connection state, not only whether application data can be read.
1788 return self._conn.can_read(timeout)
1790 def read_response(
1791 self,
1792 disable_decoding=False,
1793 *,
1794 timeout: Union[float, object] = SENTINEL,
1795 disconnect_on_error=True,
1796 push_request=False,
1797 ):
1798 with self._cache_lock:
1799 # Check if command response exists in a cache and it's not in progress.
1800 if self._current_command_cache_key is not None:
1801 if (
1802 self._cache.get(self._current_command_cache_key) is not None
1803 and self._cache.get(self._current_command_cache_key).status
1804 != CacheEntryStatus.IN_PROGRESS
1805 ):
1806 res = copy.deepcopy(
1807 self._cache.get(self._current_command_cache_key).cache_value
1808 )
1809 self._current_command_cache_key = None
1810 record_csc_request(
1811 result=CSCResult.HIT,
1812 )
1813 record_csc_network_saved(
1814 bytes_saved=len(res) if hasattr(res, "__len__") else 0,
1815 )
1816 return res
1817 record_csc_request(
1818 result=CSCResult.MISS,
1819 )
1821 response = self._conn.read_response(
1822 disable_decoding=disable_decoding,
1823 timeout=timeout,
1824 disconnect_on_error=disconnect_on_error,
1825 push_request=push_request,
1826 )
1828 with self._cache_lock:
1829 # Prevent not-allowed command from caching.
1830 if self._current_command_cache_key is None:
1831 return response
1832 # If response is None prevent from caching.
1833 if response is None:
1834 self._cache.delete_by_cache_keys([self._current_command_cache_key])
1835 return response
1837 cache_entry = self._cache.get(self._current_command_cache_key)
1839 # Cache only responses that still valid
1840 # and wasn't invalidated by another connection in meantime.
1841 if cache_entry is not None:
1842 cache_entry.status = CacheEntryStatus.VALID
1843 cache_entry.cache_value = response
1844 self._cache.set(cache_entry)
1846 self._current_command_cache_key = None
1848 return response
1850 def pack_command(self, *args):
1851 return self._conn.pack_command(*args)
1853 def pack_commands(self, commands):
1854 return self._conn.pack_commands(commands)
1856 @property
1857 def handshake_metadata(self) -> Union[Dict[bytes, bytes], Dict[str, str]]:
1858 return self._conn.handshake_metadata
1860 def set_re_auth_token(self, token: TokenInterface):
1861 self._conn.set_re_auth_token(token)
1863 def re_auth(self):
1864 self._conn.re_auth()
1866 def mark_for_reconnect(self):
1867 self._conn.mark_for_reconnect()
1869 def should_reconnect(self):
1870 return self._conn.should_reconnect()
1872 def reset_should_reconnect(self):
1873 self._conn.reset_should_reconnect()
1875 @property
1876 def host(self) -> str:
1877 return self._conn.host
1879 @host.setter
1880 def host(self, value: str):
1881 self._conn.host = value
1883 @property
1884 def socket_timeout(self) -> Optional[Union[float, int]]:
1885 return self._conn.socket_timeout
1887 @socket_timeout.setter
1888 def socket_timeout(self, value: Optional[Union[float, int]]):
1889 self._conn.socket_timeout = value
1891 @property
1892 def socket_connect_timeout(self) -> Optional[Union[float, int]]:
1893 return self._conn.socket_connect_timeout
1895 @socket_connect_timeout.setter
1896 def socket_connect_timeout(self, value: Optional[Union[float, int]]):
1897 self._conn.socket_connect_timeout = value
1899 @property
1900 def _maint_notifications_connection_handler(
1901 self,
1902 ) -> Optional[MaintNotificationsConnectionHandler]:
1903 if isinstance(self._conn, MaintNotificationsAbstractConnection):
1904 return self._conn._maint_notifications_connection_handler
1906 @_maint_notifications_connection_handler.setter
1907 def _maint_notifications_connection_handler(
1908 self, value: Optional[MaintNotificationsConnectionHandler]
1909 ):
1910 self._conn._maint_notifications_connection_handler = value
1912 def _get_socket(self) -> Optional[socket.socket]:
1913 if isinstance(self._conn, MaintNotificationsAbstractConnection):
1914 return self._conn._get_socket()
1915 else:
1916 raise NotImplementedError(
1917 "Maintenance notifications are not supported by this connection type"
1918 )
1920 def _get_maint_notifications_connection_instance(
1921 self, connection
1922 ) -> MaintNotificationsAbstractConnection:
1923 """
1924 Validate that connection instance supports maintenance notifications.
1925 With this helper method we ensure that we are working
1926 with the correct connection type.
1927 After twe validate that connection instance supports maintenance notifications
1928 we can safely return the connection instance
1929 as MaintNotificationsAbstractConnection.
1930 """
1931 if not isinstance(connection, MaintNotificationsAbstractConnection):
1932 raise NotImplementedError(
1933 "Maintenance notifications are not supported by this connection type"
1934 )
1935 else:
1936 return connection
1938 @property
1939 def maintenance_state(self) -> MaintenanceState:
1940 con = self._get_maint_notifications_connection_instance(self._conn)
1941 return con.maintenance_state
1943 @maintenance_state.setter
1944 def maintenance_state(self, state: MaintenanceState):
1945 con = self._get_maint_notifications_connection_instance(self._conn)
1946 con.maintenance_state = state
1948 def getpeername(self):
1949 con = self._get_maint_notifications_connection_instance(self._conn)
1950 return con.getpeername()
1952 def get_resolved_ip(self):
1953 con = self._get_maint_notifications_connection_instance(self._conn)
1954 return con.get_resolved_ip()
1956 def update_current_socket_timeout(self, relaxed_timeout: Optional[float] = None):
1957 con = self._get_maint_notifications_connection_instance(self._conn)
1958 con.update_current_socket_timeout(relaxed_timeout)
1960 def set_tmp_settings(
1961 self,
1962 tmp_host_address: Optional[str] = None,
1963 tmp_relaxed_timeout: Optional[float] = -1,
1964 ):
1965 con = self._get_maint_notifications_connection_instance(self._conn)
1966 con.set_tmp_settings(tmp_host_address, tmp_relaxed_timeout)
1968 def reset_tmp_settings(
1969 self,
1970 reset_host_address: bool = False,
1971 reset_relaxed_timeout: bool = False,
1972 ):
1973 con = self._get_maint_notifications_connection_instance(self._conn)
1974 con.reset_tmp_settings(reset_host_address, reset_relaxed_timeout)
1976 def _connect(self):
1977 self._conn._connect()
1979 def _host_error(self):
1980 self._conn._host_error()
1982 def _enable_tracking_callback(self, conn: ConnectionInterface) -> None:
1983 conn.send_command("CLIENT", "TRACKING", "ON")
1984 conn.read_response()
1985 conn._parser.set_invalidation_push_handler(self._on_invalidation_callback)
1987 def _process_pending_invalidations(self):
1988 while self.can_read():
1989 try:
1990 self._conn.read_response(
1991 push_request=True, timeout=0, disconnect_on_error=False
1992 )
1993 except TimeoutError:
1994 break
1996 def _on_invalidation_callback(self, data: List[Union[str, Optional[List[bytes]]]]):
1997 with self._cache_lock:
1998 # Flush cache when DB flushed on server-side
1999 if data[1] is None:
2000 self._cache.flush()
2001 else:
2002 keys_deleted = self._cache.delete_by_redis_keys(data[1])
2004 if len(keys_deleted) > 0:
2005 record_csc_eviction(
2006 count=len(keys_deleted),
2007 reason=CSCReason.INVALIDATION,
2008 )
2010 def extract_connection_details(self) -> str:
2011 return self._conn.extract_connection_details()
2014class SSLConnection(Connection):
2015 """Manages SSL connections to and from the Redis server(s).
2016 This class extends the Connection class, adding SSL functionality, and making
2017 use of ssl.SSLContext (https://docs.python.org/3/library/ssl.html#ssl.SSLContext)
2018 """ # noqa
2020 def __init__(
2021 self,
2022 ssl_keyfile=None,
2023 ssl_certfile=None,
2024 ssl_cert_reqs="required",
2025 ssl_include_verify_flags: Optional[List["VerifyFlags"]] = None,
2026 ssl_exclude_verify_flags: Optional[List["VerifyFlags"]] = None,
2027 ssl_ca_certs=None,
2028 ssl_ca_data=None,
2029 ssl_check_hostname=True,
2030 ssl_ca_path=None,
2031 ssl_password=None,
2032 ssl_validate_ocsp=False,
2033 ssl_validate_ocsp_stapled=False,
2034 ssl_ocsp_context=None,
2035 ssl_ocsp_expected_cert=None,
2036 ssl_min_version=None,
2037 ssl_ciphers=None,
2038 **kwargs,
2039 ):
2040 """Constructor
2042 Args:
2043 ssl_keyfile: Path to an ssl private key. Defaults to None.
2044 ssl_certfile: Path to an ssl certificate. Defaults to None.
2045 ssl_cert_reqs: The string value for the SSLContext.verify_mode (none, optional, required),
2046 or an ssl.VerifyMode. Defaults to "required".
2047 ssl_include_verify_flags: A list of flags to be included in the SSLContext.verify_flags. Defaults to None.
2048 ssl_exclude_verify_flags: A list of flags to be excluded from the SSLContext.verify_flags. Defaults to None.
2049 ssl_ca_certs: The path to a file of concatenated CA certificates in PEM format. Defaults to None.
2050 ssl_ca_data: Either an ASCII string of one or more PEM-encoded certificates or a bytes-like object of DER-encoded certificates.
2051 ssl_check_hostname: If set, match the hostname during the SSL handshake. Defaults to True.
2052 ssl_ca_path: The path to a directory containing several CA certificates in PEM format. Defaults to None.
2053 ssl_password: Password for unlocking an encrypted private key. Defaults to None.
2055 ssl_validate_ocsp: If set, perform a full ocsp validation (i.e not a stapled verification)
2056 ssl_validate_ocsp_stapled: If set, perform a validation on a stapled ocsp response
2057 ssl_ocsp_context: A fully initialized OpenSSL.SSL.Context object to be used in verifying the ssl_ocsp_expected_cert
2058 ssl_ocsp_expected_cert: A PEM armoured string containing the expected certificate to be returned from the ocsp verification service.
2059 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.
2060 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.
2062 Raises:
2063 RedisError
2064 """ # noqa
2065 if not SSL_AVAILABLE:
2066 raise RedisError("Python wasn't built with SSL support")
2068 self.keyfile = ssl_keyfile
2069 self.certfile = ssl_certfile
2070 if ssl_cert_reqs is None:
2071 ssl_cert_reqs = ssl.CERT_NONE
2072 elif isinstance(ssl_cert_reqs, str):
2073 CERT_REQS = { # noqa: N806
2074 "none": ssl.CERT_NONE,
2075 "optional": ssl.CERT_OPTIONAL,
2076 "required": ssl.CERT_REQUIRED,
2077 }
2078 if ssl_cert_reqs not in CERT_REQS:
2079 raise RedisError(
2080 f"Invalid SSL Certificate Requirements Flag: {ssl_cert_reqs}"
2081 )
2082 ssl_cert_reqs = CERT_REQS[ssl_cert_reqs]
2083 self.cert_reqs = ssl_cert_reqs
2084 self.ssl_include_verify_flags = ssl_include_verify_flags
2085 self.ssl_exclude_verify_flags = ssl_exclude_verify_flags
2086 self.ca_certs = ssl_ca_certs
2087 self.ca_data = ssl_ca_data
2088 self.ca_path = ssl_ca_path
2089 self.check_hostname = (
2090 ssl_check_hostname if self.cert_reqs != ssl.CERT_NONE else False
2091 )
2092 self.certificate_password = ssl_password
2093 self.ssl_validate_ocsp = ssl_validate_ocsp
2094 self.ssl_validate_ocsp_stapled = ssl_validate_ocsp_stapled
2095 self.ssl_ocsp_context = ssl_ocsp_context
2096 self.ssl_ocsp_expected_cert = ssl_ocsp_expected_cert
2097 self.ssl_min_version = ssl_min_version
2098 self.ssl_ciphers = ssl_ciphers
2099 super().__init__(**kwargs)
2101 def _connect(self):
2102 """
2103 Wrap the socket with SSL support, handling potential errors.
2104 """
2105 sock = super()._connect()
2106 try:
2107 return self._wrap_socket_with_ssl(sock)
2108 except (OSError, RedisError):
2109 sock.close()
2110 raise
2112 def _wrap_socket_with_ssl(self, sock):
2113 """
2114 Wraps the socket with SSL support.
2116 Args:
2117 sock: The plain socket to wrap with SSL.
2119 Returns:
2120 An SSL wrapped socket.
2121 """
2122 context = ssl.create_default_context()
2123 context.check_hostname = self.check_hostname
2124 context.verify_mode = self.cert_reqs
2125 if self.ssl_include_verify_flags:
2126 for flag in self.ssl_include_verify_flags:
2127 context.verify_flags |= flag
2128 if self.ssl_exclude_verify_flags:
2129 for flag in self.ssl_exclude_verify_flags:
2130 context.verify_flags &= ~flag
2131 if self.certfile or self.keyfile:
2132 context.load_cert_chain(
2133 certfile=self.certfile,
2134 keyfile=self.keyfile,
2135 password=self.certificate_password,
2136 )
2137 if (
2138 self.ca_certs is not None
2139 or self.ca_path is not None
2140 or self.ca_data is not None
2141 ):
2142 context.load_verify_locations(
2143 cafile=self.ca_certs, capath=self.ca_path, cadata=self.ca_data
2144 )
2145 if self.ssl_min_version is not None:
2146 context.minimum_version = self.ssl_min_version
2147 if self.ssl_ciphers:
2148 context.set_ciphers(self.ssl_ciphers)
2149 if self.ssl_validate_ocsp is True and CRYPTOGRAPHY_AVAILABLE is False:
2150 raise RedisError("cryptography is not installed.")
2152 if self.ssl_validate_ocsp_stapled and self.ssl_validate_ocsp:
2153 raise RedisError(
2154 "Either an OCSP staple or pure OCSP connection must be validated "
2155 "- not both."
2156 )
2158 sslsock = context.wrap_socket(sock, server_hostname=self.host)
2160 # validation for the stapled case
2161 if self.ssl_validate_ocsp_stapled:
2162 import OpenSSL
2164 from .ocsp import ocsp_staple_verifier
2166 # if a context is provided use it - otherwise, a basic context
2167 if self.ssl_ocsp_context is None:
2168 staple_ctx = OpenSSL.SSL.Context(OpenSSL.SSL.SSLv23_METHOD)
2169 staple_ctx.use_certificate_file(self.certfile)
2170 staple_ctx.use_privatekey_file(self.keyfile)
2171 else:
2172 staple_ctx = self.ssl_ocsp_context
2174 staple_ctx.set_ocsp_client_callback(
2175 ocsp_staple_verifier, self.ssl_ocsp_expected_cert
2176 )
2178 # need another socket
2179 con = OpenSSL.SSL.Connection(staple_ctx, socket.socket())
2180 con.request_ocsp()
2181 con.connect((self.host, self.port))
2182 con.do_handshake()
2183 con.shutdown()
2184 return sslsock
2186 # pure ocsp validation
2187 if self.ssl_validate_ocsp is True and CRYPTOGRAPHY_AVAILABLE:
2188 from .ocsp import OCSPVerifier
2190 o = OCSPVerifier(sslsock, self.host, self.port, self.ca_certs)
2191 if o.is_valid():
2192 return sslsock
2193 else:
2194 raise ConnectionError("ocsp validation error")
2195 return sslsock
2198class UnixDomainSocketConnection(AbstractConnection):
2199 "Manages UDS communication to and from a Redis server"
2201 def __init__(self, path="", socket_timeout=DEFAULT_SOCKET_TIMEOUT, **kwargs):
2202 super().__init__(**kwargs)
2203 self.path = path
2204 self.socket_timeout = socket_timeout
2206 def repr_pieces(self):
2207 pieces = [("path", self.path), ("db", self.db)]
2208 if self.client_name:
2209 pieces.append(("client_name", self.client_name))
2210 return pieces
2212 def _connect(self):
2213 "Create a Unix domain socket connection"
2214 sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
2215 sock.settimeout(self.socket_connect_timeout)
2216 try:
2217 sock.connect(self.path)
2218 except OSError:
2219 # Prevent ResourceWarnings for unclosed sockets.
2220 try:
2221 sock.shutdown(socket.SHUT_RDWR) # ensure a clean close
2222 except OSError:
2223 pass
2224 sock.close()
2225 raise
2226 sock.settimeout(self.socket_timeout)
2227 return sock
2229 def _host_error(self):
2230 return self.path
2233FALSE_STRINGS = ("0", "F", "FALSE", "N", "NO")
2236def to_bool(value):
2237 if value is None or value == "":
2238 return None
2239 if isinstance(value, str) and value.upper() in FALSE_STRINGS:
2240 return False
2241 return bool(value)
2244def parse_ssl_verify_flags(value):
2245 # flags are passed in as a string representation of a list,
2246 # e.g. VERIFY_X509_STRICT, VERIFY_X509_PARTIAL_CHAIN
2247 verify_flags_str = value.replace("[", "").replace("]", "")
2249 verify_flags = []
2250 for flag in verify_flags_str.split(","):
2251 flag = flag.strip()
2252 if not hasattr(VerifyFlags, flag):
2253 raise ValueError(f"Invalid ssl verify flag: {flag}")
2254 verify_flags.append(getattr(VerifyFlags, flag))
2255 return verify_flags
2258URL_QUERY_ARGUMENT_PARSERS = {
2259 "db": int,
2260 "socket_timeout": float,
2261 "socket_connect_timeout": float,
2262 "socket_read_size": int,
2263 "socket_keepalive": to_bool,
2264 "retry_on_timeout": to_bool,
2265 "retry_on_error": list,
2266 "max_connections": int,
2267 "health_check_interval": int,
2268 "ssl_check_hostname": to_bool,
2269 "ssl_include_verify_flags": parse_ssl_verify_flags,
2270 "ssl_exclude_verify_flags": parse_ssl_verify_flags,
2271 "ssl_min_version": int,
2272 "timeout": float,
2273 "protocol": int,
2274 "legacy_responses": to_bool,
2275}
2278def parse_url(url):
2279 if not (
2280 url.startswith("redis://")
2281 or url.startswith("rediss://")
2282 or url.startswith("unix://")
2283 ):
2284 raise ValueError(
2285 "Redis URL must specify one of the following "
2286 "schemes (redis://, rediss://, unix://)"
2287 )
2289 url = urlparse(url)
2290 kwargs = {}
2292 for name, value in parse_qs(url.query).items():
2293 if value and len(value) > 0:
2294 value = unquote(value[0])
2295 parser = URL_QUERY_ARGUMENT_PARSERS.get(name)
2296 if parser:
2297 try:
2298 kwargs[name] = parser(value)
2299 except (TypeError, ValueError):
2300 raise ValueError(f"Invalid value for '{name}' in connection URL.")
2301 else:
2302 kwargs[name] = value
2304 if url.username:
2305 kwargs["username"] = unquote(url.username)
2306 if url.password:
2307 kwargs["password"] = unquote(url.password)
2309 # We only support redis://, rediss:// and unix:// schemes.
2310 if url.scheme == "unix":
2311 if url.path:
2312 kwargs["path"] = unquote(url.path)
2313 kwargs["connection_class"] = UnixDomainSocketConnection
2315 else: # implied: url.scheme in ("redis", "rediss"):
2316 if url.hostname:
2317 kwargs["host"] = unquote(url.hostname)
2318 if url.port:
2319 kwargs["port"] = int(url.port)
2321 # If there's a path argument, use it as the db argument if a
2322 # querystring value wasn't specified
2323 if url.path and "db" not in kwargs:
2324 try:
2325 kwargs["db"] = int(unquote(url.path).replace("/", ""))
2326 except (AttributeError, ValueError):
2327 pass
2329 if url.scheme == "rediss":
2330 kwargs["connection_class"] = SSLConnection
2332 return kwargs
2335_CP = TypeVar("_CP", bound="ConnectionPool")
2338class ConnectionPoolInterface(ABC):
2339 @abstractmethod
2340 def get_protocol(self):
2341 pass
2343 @abstractmethod
2344 def reset(self):
2345 pass
2347 @abstractmethod
2348 @deprecated_args(
2349 args_to_warn=["*"],
2350 reason="Use get_connection() without args instead",
2351 version="5.3.0",
2352 )
2353 def get_connection(
2354 self, command_name: Optional[str], *keys, **options
2355 ) -> ConnectionInterface:
2356 pass
2358 @abstractmethod
2359 def get_encoder(self):
2360 pass
2362 @abstractmethod
2363 def release(self, connection: ConnectionInterface):
2364 pass
2366 @abstractmethod
2367 def disconnect(self, inuse_connections: bool = True):
2368 pass
2370 @abstractmethod
2371 def close(self):
2372 pass
2374 @abstractmethod
2375 def set_retry(self, retry: Retry):
2376 pass
2378 @abstractmethod
2379 def re_auth_callback(self, token: TokenInterface):
2380 pass
2382 @abstractmethod
2383 def get_connection_count(self) -> list[tuple[int, dict]]:
2384 """
2385 Returns a connection count (both idle and in use).
2386 """
2387 pass
2390class MaintNotificationsAbstractConnectionPool:
2391 """
2392 Abstract class for handling maintenance notifications logic.
2393 This class is mixed into the ConnectionPool classes.
2395 This class is not intended to be used directly!
2397 All logic related to maintenance notifications and
2398 connection pool handling is encapsulated in this class.
2399 """
2401 def __init__(
2402 self,
2403 maint_notifications_config: Optional[MaintNotificationsConfig] = None,
2404 oss_cluster_maint_notifications_handler: Optional[
2405 OSSMaintNotificationsHandler
2406 ] = None,
2407 **kwargs,
2408 ):
2409 # Initialize maintenance notifications
2410 is_protocol_supported = check_protocol_version(kwargs.get("protocol"), 3)
2412 if maint_notifications_config is None and is_protocol_supported:
2413 maint_notifications_config = MaintNotificationsConfig()
2415 if maint_notifications_config and maint_notifications_config.enabled:
2416 if not is_protocol_supported:
2417 raise RedisError(
2418 "Maintenance notifications handlers on connection are only supported with RESP version 3"
2419 )
2421 self._event_dispatcher = kwargs.get("event_dispatcher", None)
2422 if self._event_dispatcher is None:
2423 self._event_dispatcher = EventDispatcher()
2425 self._maint_notifications_pool_handler = MaintNotificationsPoolHandler(
2426 self, maint_notifications_config
2427 )
2428 if oss_cluster_maint_notifications_handler:
2429 self._oss_cluster_maint_notifications_handler = (
2430 oss_cluster_maint_notifications_handler
2431 )
2432 self._update_connection_kwargs_for_maint_notifications(
2433 oss_cluster_maint_notifications_handler=self._oss_cluster_maint_notifications_handler
2434 )
2435 self._maint_notifications_pool_handler = None
2436 else:
2437 self._oss_cluster_maint_notifications_handler = None
2438 self._maint_notifications_pool_handler = MaintNotificationsPoolHandler(
2439 self, maint_notifications_config
2440 )
2442 self._update_connection_kwargs_for_maint_notifications(
2443 maint_notifications_pool_handler=self._maint_notifications_pool_handler
2444 )
2445 else:
2446 self._maint_notifications_pool_handler = None
2447 self._oss_cluster_maint_notifications_handler = None
2449 @property
2450 @abstractmethod
2451 def connection_kwargs(self) -> Dict[str, Any]:
2452 pass
2454 @connection_kwargs.setter
2455 @abstractmethod
2456 def connection_kwargs(self, value: Dict[str, Any]):
2457 pass
2459 @abstractmethod
2460 def _get_pool_lock(self) -> threading.RLock:
2461 pass
2463 @abstractmethod
2464 def _get_free_connections(self) -> Iterable["MaintNotificationsAbstractConnection"]:
2465 pass
2467 @abstractmethod
2468 def _get_in_use_connections(
2469 self,
2470 ) -> Iterable["MaintNotificationsAbstractConnection"]:
2471 pass
2473 def maint_notifications_enabled(self):
2474 """
2475 Returns:
2476 True if the maintenance notifications are enabled, False otherwise.
2477 The maintenance notifications config is stored in the pool handler.
2478 If the pool handler is not set, the maintenance notifications are not enabled.
2479 """
2480 if self._oss_cluster_maint_notifications_handler:
2481 maint_notifications_config = (
2482 self._oss_cluster_maint_notifications_handler.config
2483 )
2484 else:
2485 maint_notifications_config = (
2486 self._maint_notifications_pool_handler.config
2487 if self._maint_notifications_pool_handler
2488 else None
2489 )
2491 return maint_notifications_config and maint_notifications_config.enabled
2493 def update_maint_notifications_config(
2494 self,
2495 maint_notifications_config: MaintNotificationsConfig,
2496 oss_cluster_maint_notifications_handler: Optional[
2497 OSSMaintNotificationsHandler
2498 ] = None,
2499 ):
2500 """
2501 Updates the maintenance notifications configuration.
2502 This method should be called only if the pool was created
2503 without enabling the maintenance notifications and
2504 in a later point in time maintenance notifications
2505 are requested to be enabled.
2506 """
2507 if (
2508 self.maint_notifications_enabled()
2509 and not maint_notifications_config.enabled
2510 ):
2511 raise ValueError(
2512 "Cannot disable maintenance notifications after enabling them"
2513 )
2514 if oss_cluster_maint_notifications_handler:
2515 self._oss_cluster_maint_notifications_handler = (
2516 oss_cluster_maint_notifications_handler
2517 )
2518 # OSS cluster mode and pool-handler mode are mutually exclusive
2519 # (see __init__). A pool created with the default RESP3 "auto"
2520 # config wires a pool handler before this method runs; clear it so
2521 # new and existing connections are not configured with both handlers.
2522 self._maint_notifications_pool_handler = None
2523 else:
2524 # first update pool settings
2525 if self._oss_cluster_maint_notifications_handler:
2526 # Pool already in OSS cluster mode; update the OSS handler config
2527 # instead of creating a mutually-exclusive pool handler (which
2528 # would be silently ignored because the OSS handler wins priority
2529 # in both update helpers below).
2530 self._oss_cluster_maint_notifications_handler.config = (
2531 maint_notifications_config
2532 )
2533 elif not self._maint_notifications_pool_handler:
2534 self._maint_notifications_pool_handler = MaintNotificationsPoolHandler(
2535 self, maint_notifications_config
2536 )
2537 else:
2538 self._maint_notifications_pool_handler.config = (
2539 maint_notifications_config
2540 )
2542 # then update connection kwargs and existing connections
2543 self._update_connection_kwargs_for_maint_notifications(
2544 maint_notifications_pool_handler=self._maint_notifications_pool_handler,
2545 oss_cluster_maint_notifications_handler=self._oss_cluster_maint_notifications_handler,
2546 )
2547 self._update_maint_notifications_configs_for_connections(
2548 maint_notifications_pool_handler=self._maint_notifications_pool_handler,
2549 oss_cluster_maint_notifications_handler=self._oss_cluster_maint_notifications_handler,
2550 )
2552 def _update_connection_kwargs_for_maint_notifications(
2553 self,
2554 maint_notifications_pool_handler: Optional[
2555 MaintNotificationsPoolHandler
2556 ] = None,
2557 oss_cluster_maint_notifications_handler: Optional[
2558 OSSMaintNotificationsHandler
2559 ] = None,
2560 ):
2561 """
2562 Update the connection kwargs for all future connections.
2563 """
2564 if not self.maint_notifications_enabled():
2565 return
2566 if maint_notifications_pool_handler:
2567 self.connection_kwargs.update(
2568 {
2569 "maint_notifications_pool_handler": maint_notifications_pool_handler,
2570 "maint_notifications_config": maint_notifications_pool_handler.config,
2571 }
2572 )
2573 if oss_cluster_maint_notifications_handler:
2574 self.connection_kwargs.update(
2575 {
2576 "oss_cluster_maint_notifications_handler": oss_cluster_maint_notifications_handler,
2577 "maint_notifications_config": oss_cluster_maint_notifications_handler.config,
2578 }
2579 )
2580 # OSS cluster mode and pool-handler mode are mutually exclusive.
2581 # Drop any pool handler a default (RESP3 "auto") pool creation may
2582 # have wired so future connections are not configured with both.
2583 self.connection_kwargs.pop("maint_notifications_pool_handler", None)
2585 # Store original connection parameters for maintenance notifications.
2586 if self.connection_kwargs.get("orig_host_address", None) is None:
2587 # If orig_host_address is None it means we haven't
2588 # configured the original values yet
2589 self.connection_kwargs.update(
2590 {
2591 "orig_host_address": self.connection_kwargs.get("host"),
2592 "orig_socket_timeout": self.connection_kwargs.get(
2593 "socket_timeout", DEFAULT_SOCKET_TIMEOUT
2594 ),
2595 "orig_socket_connect_timeout": self.connection_kwargs.get(
2596 "socket_connect_timeout", DEFAULT_SOCKET_CONNECT_TIMEOUT
2597 ),
2598 }
2599 )
2601 def _update_maint_notifications_configs_for_connections(
2602 self,
2603 maint_notifications_pool_handler: Optional[
2604 MaintNotificationsPoolHandler
2605 ] = None,
2606 oss_cluster_maint_notifications_handler: Optional[
2607 OSSMaintNotificationsHandler
2608 ] = None,
2609 ):
2610 """Update the maintenance notifications config for all connections in the pool."""
2611 with self._get_pool_lock():
2612 for conn in self._get_free_connections():
2613 if oss_cluster_maint_notifications_handler:
2614 # set cluster handler for conn
2615 conn.set_maint_notifications_cluster_handler_for_connection(
2616 oss_cluster_maint_notifications_handler
2617 )
2618 conn.maint_notifications_config = (
2619 oss_cluster_maint_notifications_handler.config
2620 )
2621 elif maint_notifications_pool_handler:
2622 conn.set_maint_notifications_pool_handler_for_connection(
2623 maint_notifications_pool_handler
2624 )
2625 conn.maint_notifications_config = (
2626 maint_notifications_pool_handler.config
2627 )
2628 else:
2629 raise ValueError(
2630 "Either maint_notifications_pool_handler or oss_cluster_maint_notifications_handler must be set"
2631 )
2632 conn.disconnect()
2633 for conn in self._get_in_use_connections():
2634 if oss_cluster_maint_notifications_handler:
2635 # Use set_maint_notifications_cluster_handler_for_connection
2636 # (not _configure_maintenance_notifications) so the parser is
2637 # obtained from the connection itself. _configure_* requires a
2638 # parser argument and would raise here; it would also reset the
2639 # connection's orig_* settings, which is wrong for an in-use
2640 # (active) connection. This mirrors the idle-connection branch
2641 # above and the pool-handler branches.
2642 conn.set_maint_notifications_cluster_handler_for_connection(
2643 oss_cluster_maint_notifications_handler
2644 )
2645 conn.maint_notifications_config = (
2646 oss_cluster_maint_notifications_handler.config
2647 )
2648 elif maint_notifications_pool_handler:
2649 conn.set_maint_notifications_pool_handler_for_connection(
2650 maint_notifications_pool_handler
2651 )
2652 conn.maint_notifications_config = (
2653 maint_notifications_pool_handler.config
2654 )
2655 else:
2656 raise ValueError(
2657 "Either maint_notifications_pool_handler or oss_cluster_maint_notifications_handler must be set"
2658 )
2659 conn.mark_for_reconnect()
2661 def _should_update_connection(
2662 self,
2663 conn: "MaintNotificationsAbstractConnection",
2664 matching_pattern: Literal[
2665 "connected_address", "configured_address", "notification_hash"
2666 ] = "connected_address",
2667 matching_address: Optional[str] = None,
2668 matching_notification_hash: Optional[int] = None,
2669 ) -> bool:
2670 """
2671 Check if the connection should be updated based on the matching criteria.
2672 """
2673 if matching_pattern == "connected_address":
2674 if matching_address and conn.getpeername() != matching_address:
2675 return False
2676 elif matching_pattern == "configured_address":
2677 if matching_address and conn.host != matching_address:
2678 return False
2679 elif matching_pattern == "notification_hash":
2680 if (
2681 matching_notification_hash is not None
2682 and conn.maintenance_notification_hash != matching_notification_hash
2683 ):
2684 return False
2685 return True
2687 def update_connection_settings(
2688 self,
2689 conn: "MaintNotificationsAbstractConnection",
2690 state: Optional["MaintenanceState"] = None,
2691 maintenance_notification_hash: Optional[int] = None,
2692 host_address: Optional[str] = None,
2693 relaxed_timeout: Optional[float] = None,
2694 update_notification_hash: bool = False,
2695 reset_host_address: bool = False,
2696 reset_relaxed_timeout: bool = False,
2697 ):
2698 """
2699 Update the settings for a single connection.
2700 """
2701 if state:
2702 conn.maintenance_state = state
2704 if update_notification_hash:
2705 # update the notification hash only if requested
2706 conn.maintenance_notification_hash = maintenance_notification_hash
2708 if host_address is not None:
2709 conn.set_tmp_settings(tmp_host_address=host_address)
2711 if relaxed_timeout is not None:
2712 conn.set_tmp_settings(tmp_relaxed_timeout=relaxed_timeout)
2714 if reset_relaxed_timeout or reset_host_address:
2715 conn.reset_tmp_settings(
2716 reset_host_address=reset_host_address,
2717 reset_relaxed_timeout=reset_relaxed_timeout,
2718 )
2720 conn.update_current_socket_timeout(relaxed_timeout)
2722 def update_connections_settings(
2723 self,
2724 state: Optional["MaintenanceState"] = None,
2725 maintenance_notification_hash: Optional[int] = None,
2726 host_address: Optional[str] = None,
2727 relaxed_timeout: Optional[float] = None,
2728 matching_address: Optional[str] = None,
2729 matching_notification_hash: Optional[int] = None,
2730 matching_pattern: Literal[
2731 "connected_address", "configured_address", "notification_hash"
2732 ] = "connected_address",
2733 update_notification_hash: bool = False,
2734 reset_host_address: bool = False,
2735 reset_relaxed_timeout: bool = False,
2736 include_free_connections: bool = True,
2737 ):
2738 """
2739 Update the settings for all matching connections in the pool.
2741 This method does not create new connections.
2742 This method does not affect the connection kwargs.
2744 :param state: The maintenance state to set for the connection.
2745 :param maintenance_notification_hash: The hash of the maintenance notification
2746 to set for the connection.
2747 :param host_address: The host address to set for the connection.
2748 :param relaxed_timeout: The relaxed timeout to set for the connection.
2749 :param matching_address: The address to match for the connection.
2750 :param matching_notification_hash: The notification hash to match for the connection.
2751 :param matching_pattern: The pattern to match for the connection.
2752 :param update_notification_hash: Whether to update the notification hash for the connection.
2753 :param reset_host_address: Whether to reset the host address to the original address.
2754 :param reset_relaxed_timeout: Whether to reset the relaxed timeout to the original timeout.
2755 :param include_free_connections: Whether to include free/available connections.
2756 """
2757 with self._get_pool_lock():
2758 for conn in self._get_in_use_connections():
2759 if self._should_update_connection(
2760 conn,
2761 matching_pattern,
2762 matching_address,
2763 matching_notification_hash,
2764 ):
2765 self.update_connection_settings(
2766 conn,
2767 state=state,
2768 maintenance_notification_hash=maintenance_notification_hash,
2769 host_address=host_address,
2770 relaxed_timeout=relaxed_timeout,
2771 update_notification_hash=update_notification_hash,
2772 reset_host_address=reset_host_address,
2773 reset_relaxed_timeout=reset_relaxed_timeout,
2774 )
2776 if include_free_connections:
2777 for conn in self._get_free_connections():
2778 if self._should_update_connection(
2779 conn,
2780 matching_pattern,
2781 matching_address,
2782 matching_notification_hash,
2783 ):
2784 self.update_connection_settings(
2785 conn,
2786 state=state,
2787 maintenance_notification_hash=maintenance_notification_hash,
2788 host_address=host_address,
2789 relaxed_timeout=relaxed_timeout,
2790 update_notification_hash=update_notification_hash,
2791 reset_host_address=reset_host_address,
2792 reset_relaxed_timeout=reset_relaxed_timeout,
2793 )
2795 def update_connection_kwargs(
2796 self,
2797 **kwargs,
2798 ):
2799 """
2800 Update the connection kwargs for all future connections.
2802 This method updates the connection kwargs for all future connections created by the pool.
2803 Existing connections are not affected.
2804 """
2805 self.connection_kwargs.update(kwargs)
2807 def update_active_connections_for_reconnect(
2808 self,
2809 moving_address_src: Optional[str] = None,
2810 ):
2811 """
2812 Mark all active connections for reconnect.
2813 This is used when a cluster node is migrated to a different address.
2815 :param moving_address_src: The address of the node that is being moved.
2816 """
2817 with self._get_pool_lock():
2818 for conn in self._get_in_use_connections():
2819 if self._should_update_connection(
2820 conn, "connected_address", moving_address_src
2821 ):
2822 conn.mark_for_reconnect()
2824 def disconnect_free_connections(
2825 self,
2826 moving_address_src: Optional[str] = None,
2827 ):
2828 """
2829 Disconnect all free/available connections.
2830 This is used when a cluster node is migrated to a different address.
2832 :param moving_address_src: The address of the node that is being moved.
2833 """
2834 with self._get_pool_lock():
2835 for conn in self._get_free_connections():
2836 if self._should_update_connection(
2837 conn, "connected_address", moving_address_src
2838 ):
2839 conn.disconnect()
2842class ConnectionPool(MaintNotificationsAbstractConnectionPool, ConnectionPoolInterface):
2843 """
2844 Create a connection pool. ``If max_connections`` is set, then this
2845 object raises :py:class:`~redis.exceptions.ConnectionError` when the pool's
2846 limit is reached.
2848 By default, TCP connections are created unless ``connection_class``
2849 is specified. Use class:`.UnixDomainSocketConnection` for
2850 unix sockets.
2851 :py:class:`~redis.SSLConnection` can be used for SSL enabled connections.
2853 If ``maint_notifications_config`` is provided, the connection pool will support
2854 maintenance notifications.
2855 Maintenance notifications are supported only with RESP3.
2856 If the ``maint_notifications_config`` is not provided but the ``protocol`` is 3,
2857 the maintenance notifications will be enabled by default.
2859 Any additional keyword arguments are passed to the constructor of
2860 ``connection_class``.
2861 """
2863 @classmethod
2864 def from_url(cls: Type[_CP], url: str, **kwargs) -> _CP:
2865 """
2866 Return a connection pool configured from the given URL.
2868 For example::
2870 redis://[[username]:[password]]@localhost:6379/0
2871 rediss://[[username]:[password]]@localhost:6379/0
2872 unix://[username@]/path/to/socket.sock?db=0[&password=password]
2874 Three URL schemes are supported:
2876 - `redis://` creates a TCP socket connection. See more at:
2877 <https://www.iana.org/assignments/uri-schemes/prov/redis>
2878 - `rediss://` creates a SSL wrapped TCP socket connection. See more at:
2879 <https://www.iana.org/assignments/uri-schemes/prov/rediss>
2880 - ``unix://``: creates a Unix Domain Socket connection.
2882 The username, password, hostname, path and all querystring values
2883 are passed through urllib.parse.unquote in order to replace any
2884 percent-encoded values with their corresponding characters.
2886 There are several ways to specify a database number. The first value
2887 found will be used:
2889 1. A ``db`` querystring option, e.g. redis://localhost?db=0
2890 2. If using the redis:// or rediss:// schemes, the path argument
2891 of the url, e.g. redis://localhost/0
2892 3. A ``db`` keyword argument to this function.
2894 If none of these options are specified, the default db=0 is used.
2896 All querystring options are cast to their appropriate Python types.
2897 Boolean arguments can be specified with string values "True"/"False"
2898 or "Yes"/"No". Values that cannot be properly cast cause a
2899 ``ValueError`` to be raised. Once parsed, the querystring arguments
2900 and keyword arguments are passed to the ``ConnectionPool``'s
2901 class initializer. In the case of conflicting arguments, querystring
2902 arguments always win.
2903 """
2904 url_options = parse_url(url)
2906 if "connection_class" in kwargs:
2907 url_options["connection_class"] = kwargs["connection_class"]
2909 kwargs.update(url_options)
2910 return cls(**kwargs)
2912 def __init__(
2913 self,
2914 connection_class=Connection,
2915 max_connections: Optional[int] = None,
2916 cache_factory: Optional[CacheFactoryInterface] = None,
2917 maint_notifications_config: Optional[MaintNotificationsConfig] = None,
2918 **connection_kwargs,
2919 ):
2920 max_connections = max_connections or 100
2921 if not isinstance(max_connections, int) or max_connections < 0:
2922 raise ValueError('"max_connections" must be a positive integer')
2924 self.connection_class = connection_class
2925 self._connection_kwargs = connection_kwargs
2926 self.max_connections = max_connections
2927 self.cache = None
2928 self._cache_factory = cache_factory
2930 try:
2931 supports_maint_notifications = issubclass(
2932 connection_class, MaintNotificationsAbstractConnection
2933 )
2934 is_unix_domain_socket_connection = issubclass(
2935 connection_class, UnixDomainSocketConnection
2936 )
2937 except TypeError:
2938 supports_maint_notifications = False
2939 is_unix_domain_socket_connection = False
2941 if is_unix_domain_socket_connection or not supports_maint_notifications:
2942 if (
2943 maint_notifications_config
2944 and maint_notifications_config.enabled is True
2945 ):
2946 raise RedisError(
2947 "Maintenance notifications are not supported with "
2948 f"{connection_class}"
2949 )
2950 maint_notifications_config = MaintNotificationsConfig(enabled=False)
2952 self._event_dispatcher = self._connection_kwargs.get("event_dispatcher", None)
2953 if self._event_dispatcher is None:
2954 self._event_dispatcher = EventDispatcher()
2956 if connection_kwargs.get("cache_config") or connection_kwargs.get("cache"):
2957 if not check_protocol_version(self._connection_kwargs.get("protocol"), 3):
2958 raise RedisError("Client caching is only supported with RESP version 3")
2960 cache = self._connection_kwargs.get("cache")
2962 if cache is not None:
2963 if not isinstance(cache, CacheInterface):
2964 raise ValueError("Cache must implement CacheInterface")
2966 self.cache = cache
2967 else:
2968 if self._cache_factory is not None:
2969 self.cache = CacheProxy(self._cache_factory.get_cache())
2970 else:
2971 self.cache = CacheFactory(
2972 self._connection_kwargs.get("cache_config")
2973 ).get_cache()
2975 init_csc_items()
2976 register_csc_items_callback(
2977 callback=lambda: self.cache.size,
2978 pool_name=get_pool_name(self),
2979 )
2981 connection_kwargs.pop("cache", None)
2982 connection_kwargs.pop("cache_config", None)
2984 # a lock to protect the critical section in _checkpid().
2985 # this lock is acquired when the process id changes, such as
2986 # after a fork. during this time, multiple threads in the child
2987 # process could attempt to acquire this lock. the first thread
2988 # to acquire the lock will reset the data structures and lock
2989 # object of this pool. subsequent threads acquiring this lock
2990 # will notice the first thread already did the work and simply
2991 # release the lock.
2993 self._fork_lock = threading.RLock()
2994 self._lock = threading.RLock()
2996 # Generate unique pool ID for observability (matches go-redis behavior)
2997 import secrets
2999 self._pool_id = secrets.token_hex(4)
3001 MaintNotificationsAbstractConnectionPool.__init__(
3002 self,
3003 maint_notifications_config=maint_notifications_config,
3004 **connection_kwargs,
3005 )
3007 self.reset()
3009 # Keys that should be redacted in __repr__ to avoid exposing sensitive information
3010 SENSITIVE_REPR_KEYS = frozenset(
3011 {
3012 "password",
3013 "username",
3014 "ssl_password",
3015 "credential_provider",
3016 }
3017 )
3019 def __repr__(self) -> str:
3020 conn_kwargs = ",".join(
3021 [
3022 f"{k}={'<REDACTED>' if k in self.SENSITIVE_REPR_KEYS else v}"
3023 for k, v in self.connection_kwargs.items()
3024 ]
3025 )
3026 return (
3027 f"<{self.__class__.__module__}.{self.__class__.__name__}"
3028 f"(<{self.connection_class.__module__}.{self.connection_class.__name__}"
3029 f"({conn_kwargs})>)>"
3030 )
3032 @property
3033 def connection_kwargs(self) -> Dict[str, Any]:
3034 return self._connection_kwargs
3036 @connection_kwargs.setter
3037 def connection_kwargs(self, value: Dict[str, Any]):
3038 self._connection_kwargs = value
3040 def get_protocol(self):
3041 """
3042 Returns:
3043 The RESP protocol version, or ``None`` if the protocol is not specified,
3044 in which case the server default will be used.
3045 """
3046 return self.connection_kwargs.get("protocol", None)
3048 def reset(self) -> None:
3049 # Record metrics for connections being removed before clearing
3050 # (only if attributes exist - they won't during __init__)
3051 if hasattr(self, "_available_connections") and hasattr(
3052 self, "_in_use_connections"
3053 ):
3054 with self._lock:
3055 idle_count = len(self._available_connections)
3056 in_use_count = len(self._in_use_connections)
3057 if idle_count > 0 or in_use_count > 0:
3058 pool_name = get_pool_name(self)
3059 if idle_count > 0:
3060 record_connection_count(
3061 pool_name=pool_name,
3062 connection_state=ConnectionState.IDLE,
3063 counter=-idle_count,
3064 )
3065 if in_use_count > 0:
3066 record_connection_count(
3067 pool_name=pool_name,
3068 connection_state=ConnectionState.USED,
3069 counter=-in_use_count,
3070 )
3072 self._created_connections = 0
3073 self._available_connections = []
3074 self._in_use_connections = set()
3076 # this must be the last operation in this method. while reset() is
3077 # called when holding _fork_lock, other threads in this process
3078 # can call _checkpid() which compares self.pid and os.getpid() without
3079 # holding any lock (for performance reasons). keeping this assignment
3080 # as the last operation ensures that those other threads will also
3081 # notice a pid difference and block waiting for the first thread to
3082 # release _fork_lock. when each of these threads eventually acquire
3083 # _fork_lock, they will notice that another thread already called
3084 # reset() and they will immediately release _fork_lock and continue on.
3085 self.pid = os.getpid()
3087 def __del__(self) -> None:
3088 """Clean up connection pool and record metrics when garbage collected."""
3089 try:
3090 if not hasattr(self, "_available_connections") or not hasattr(
3091 self, "_in_use_connections"
3092 ):
3093 return
3094 # Record metrics for all connections being removed
3095 idle_count = len(self._available_connections)
3096 in_use_count = len(self._in_use_connections)
3097 if idle_count > 0 or in_use_count > 0:
3098 pool_name = get_pool_name(self)
3099 if idle_count > 0:
3100 record_connection_count(
3101 pool_name=pool_name,
3102 connection_state=ConnectionState.IDLE,
3103 counter=-idle_count,
3104 )
3105 if in_use_count > 0:
3106 record_connection_count(
3107 pool_name=pool_name,
3108 connection_state=ConnectionState.USED,
3109 counter=-in_use_count,
3110 )
3111 except Exception:
3112 pass
3114 def _checkpid(self) -> None:
3115 # _checkpid() attempts to keep ConnectionPool fork-safe on modern
3116 # systems. this is called by all ConnectionPool methods that
3117 # manipulate the pool's state such as get_connection() and release().
3118 #
3119 # _checkpid() determines whether the process has forked by comparing
3120 # the current process id to the process id saved on the ConnectionPool
3121 # instance. if these values are the same, _checkpid() simply returns.
3122 #
3123 # when the process ids differ, _checkpid() assumes that the process
3124 # has forked and that we're now running in the child process. the child
3125 # process cannot use the parent's file descriptors (e.g., sockets).
3126 # therefore, when _checkpid() sees the process id change, it calls
3127 # reset() in order to reinitialize the child's ConnectionPool. this
3128 # will cause the child to make all new connection objects.
3129 #
3130 # _checkpid() is protected by self._fork_lock to ensure that multiple
3131 # threads in the child process do not call reset() multiple times.
3132 #
3133 # there is an extremely small chance this could fail in the following
3134 # scenario:
3135 # 1. process A calls _checkpid() for the first time and acquires
3136 # self._fork_lock.
3137 # 2. while holding self._fork_lock, process A forks (the fork()
3138 # could happen in a different thread owned by process A)
3139 # 3. process B (the forked child process) inherits the
3140 # ConnectionPool's state from the parent. that state includes
3141 # a locked _fork_lock. process B will not be notified when
3142 # process A releases the _fork_lock and will thus never be
3143 # able to acquire the _fork_lock.
3144 #
3145 # to mitigate this possible deadlock, _checkpid() will only wait 5
3146 # seconds to acquire _fork_lock. if _fork_lock cannot be acquired in
3147 # that time it is assumed that the child is deadlocked and a
3148 # redis.ChildDeadlockedError error is raised.
3149 if self.pid != os.getpid():
3150 acquired = self._fork_lock.acquire(timeout=5)
3151 if not acquired:
3152 raise ChildDeadlockedError
3153 # reset() the instance for the new process if another thread
3154 # hasn't already done so
3155 try:
3156 if self.pid != os.getpid():
3157 self.reset()
3158 finally:
3159 self._fork_lock.release()
3161 @deprecated_args(
3162 args_to_warn=["*"],
3163 reason="Use get_connection() without args instead",
3164 version="5.3.0",
3165 )
3166 def get_connection(self, command_name=None, *keys, **options) -> "Connection":
3167 "Get a connection from the pool"
3169 # Start timing for observability
3170 self._checkpid()
3171 is_created = False
3173 with self._lock:
3174 try:
3175 connection = self._available_connections.pop()
3176 except IndexError:
3177 # Start timing for observability
3178 start_time_created = time.monotonic()
3180 connection = self.make_connection()
3181 is_created = True
3182 self._in_use_connections.add(connection)
3184 # Record state transition: IDLE -> USED
3185 # (make_connection already recorded IDLE +1 for new connections)
3186 # This ensures counters stay balanced if connect() fails and release() is called
3187 pool_name = get_pool_name(self)
3188 record_connection_count(
3189 pool_name=pool_name,
3190 connection_state=ConnectionState.IDLE,
3191 counter=-1,
3192 )
3193 record_connection_count(
3194 pool_name=pool_name,
3195 connection_state=ConnectionState.USED,
3196 counter=1,
3197 )
3199 try:
3200 # ensure this connection is connected to Redis
3201 connection.connect()
3202 # connections that the pool provides should be ready to send
3203 # a command. if not, the connection was either returned to the
3204 # pool before all data has been read or the socket has been
3205 # closed. either way, reconnect and verify everything is good.
3206 try:
3207 if (
3208 connection.can_read()
3209 and self.cache is None
3210 and not self.maint_notifications_enabled()
3211 ):
3212 raise ConnectionError("Connection has data")
3213 except (ConnectionError, TimeoutError, OSError):
3214 connection.disconnect()
3215 connection.connect()
3216 if (
3217 connection.can_read()
3218 and self.cache is None
3219 and not self.maint_notifications_enabled()
3220 ):
3221 raise ConnectionError("Connection not ready")
3222 except BaseException:
3223 # release the connection back to the pool so that we don't
3224 # leak it
3225 self.release(connection)
3226 raise
3228 if is_created:
3229 record_connection_create_time(
3230 connection_pool=self,
3231 duration_seconds=time.monotonic() - start_time_created,
3232 )
3234 return connection
3236 def get_encoder(self) -> Encoder:
3237 "Return an encoder based on encoding settings"
3238 kwargs = self.connection_kwargs
3239 return Encoder(
3240 encoding=kwargs.get("encoding", "utf-8"),
3241 encoding_errors=kwargs.get("encoding_errors", "strict"),
3242 decode_responses=kwargs.get("decode_responses", False),
3243 )
3245 def make_connection(self) -> "ConnectionInterface":
3246 "Create a new connection"
3247 if self._created_connections >= self.max_connections:
3248 raise MaxConnectionsError("Too many connections")
3249 self._created_connections += 1
3251 kwargs = dict(self.connection_kwargs)
3253 # Create the connection first, then record metrics only on success
3254 if self.cache is not None:
3255 connection = CacheProxyConnection(
3256 self.connection_class(**kwargs), self.cache, self._lock
3257 )
3258 else:
3259 connection = self.connection_class(**kwargs)
3261 # Record new connection created (starts as IDLE) - only after successful construction
3262 record_connection_count(
3263 pool_name=get_pool_name(self),
3264 connection_state=ConnectionState.IDLE,
3265 counter=1,
3266 )
3268 return connection
3270 def release(self, connection: "Connection") -> None:
3271 "Releases the connection back to the pool"
3272 self._checkpid()
3273 with self._lock:
3274 try:
3275 self._in_use_connections.remove(connection)
3276 except KeyError:
3277 # Gracefully fail when a connection is returned to this pool
3278 # that the pool doesn't actually own
3279 return
3281 if self.owns_connection(connection):
3282 if connection.should_reconnect():
3283 connection.disconnect()
3284 self._available_connections.append(connection)
3285 self._event_dispatcher.dispatch(
3286 AfterConnectionReleasedEvent(connection)
3287 )
3289 # Record state transition: USED -> IDLE
3290 pool_name = get_pool_name(self)
3291 record_connection_count(
3292 pool_name=pool_name,
3293 connection_state=ConnectionState.USED,
3294 counter=-1,
3295 )
3296 record_connection_count(
3297 pool_name=pool_name,
3298 connection_state=ConnectionState.IDLE,
3299 counter=1,
3300 )
3301 else:
3302 # Pool doesn't own this connection, do not add it back
3303 # to the pool.
3304 # Still need to decrement USED since it was counted in get_connection()
3305 connection.disconnect()
3306 # Subclasses such as SentinelConnectionPool can override
3307 # owns_connection() with a comparison different from local PID
3308 # ownership. When such a subclass rejects a connection, also require
3309 # connection.pid == self.pid before reclaiming its slot.
3310 if connection.pid == self.pid:
3311 self._created_connections -= 1
3312 record_connection_count(
3313 pool_name="unknown_pool",
3314 connection_state=ConnectionState.USED,
3315 counter=-1,
3316 )
3317 return
3319 def owns_connection(self, connection: "Connection") -> int:
3320 return connection.pid == self.pid
3322 def disconnect(self, inuse_connections: bool = True) -> None:
3323 """
3324 Disconnects connections in the pool
3326 If ``inuse_connections`` is True, disconnect connections that are
3327 currently in use, potentially by other threads. Otherwise only disconnect
3328 connections that are idle in the pool.
3329 """
3330 self._checkpid()
3331 with self._lock:
3332 if inuse_connections:
3333 connections = chain(
3334 self._available_connections, self._in_use_connections
3335 )
3336 else:
3337 connections = self._available_connections
3339 for connection in connections:
3340 connection.disconnect()
3342 def close(self) -> None:
3343 """Close the pool, disconnecting all connections"""
3344 self.disconnect()
3346 def __enter__(self: _CP) -> _CP:
3347 return self
3349 def __exit__(self, exc_type, exc_value, traceback) -> None:
3350 self.close()
3352 def set_retry(self, retry: Retry) -> None:
3353 self.connection_kwargs.update({"retry": retry})
3354 for conn in self._available_connections:
3355 conn.retry = retry
3356 for conn in self._in_use_connections:
3357 conn.retry = retry
3359 def re_auth_callback(self, token: TokenInterface):
3360 with self._lock:
3361 for conn in self._available_connections:
3362 conn.retry.call_with_retry(
3363 lambda: conn.send_command(
3364 "AUTH", token.try_get("oid"), token.get_value()
3365 ),
3366 lambda error: self._mock(error),
3367 )
3368 conn.retry.call_with_retry(
3369 lambda: conn.read_response(), lambda error: self._mock(error)
3370 )
3371 for conn in self._in_use_connections:
3372 conn.set_re_auth_token(token)
3374 def _get_pool_lock(self):
3375 return self._lock
3377 def _get_free_connections(self):
3378 with self._lock:
3379 return list(self._available_connections)
3381 def _get_in_use_connections(self):
3382 with self._lock:
3383 return set(self._in_use_connections)
3385 def _mock(self, error: RedisError):
3386 """
3387 Dummy functions, needs to be passed as error callback to retry object.
3388 :param error:
3389 :return:
3390 """
3391 pass
3393 def get_connection_count(self) -> List[tuple[int, dict]]:
3394 from redis.observability.attributes import get_pool_name
3396 attributes = AttributeBuilder.build_base_attributes()
3397 attributes[DB_CLIENT_CONNECTION_POOL_NAME] = get_pool_name(self)
3398 free_connections_attributes = attributes.copy()
3399 in_use_connections_attributes = attributes.copy()
3401 free_connections_attributes[DB_CLIENT_CONNECTION_STATE] = (
3402 ConnectionState.IDLE.value
3403 )
3404 in_use_connections_attributes[DB_CLIENT_CONNECTION_STATE] = (
3405 ConnectionState.USED.value
3406 )
3408 return [
3409 (len(self._get_free_connections()), free_connections_attributes),
3410 (len(self._get_in_use_connections()), in_use_connections_attributes),
3411 ]
3414class BlockingConnectionPool(ConnectionPool):
3415 """
3416 Thread-safe blocking connection pool::
3418 >>> from redis.client import Redis
3419 >>> client = Redis(connection_pool=BlockingConnectionPool())
3421 It performs the same function as the default
3422 :py:class:`~redis.ConnectionPool` implementation, in that,
3423 it maintains a pool of reusable connections that can be shared by
3424 multiple redis clients (safely across threads if required).
3426 The difference is that, in the event that a client tries to get a
3427 connection from the pool when all of connections are in use, rather than
3428 raising a :py:class:`~redis.ConnectionError` (as the default
3429 :py:class:`~redis.ConnectionPool` implementation does), it
3430 makes the client wait ("blocks") for a specified number of seconds until
3431 a connection becomes available.
3433 Use ``max_connections`` to increase / decrease the pool size::
3435 >>> pool = BlockingConnectionPool(max_connections=10)
3437 Use ``timeout`` to tell it either how many seconds to wait for a connection
3438 to become available, or to block forever:
3440 >>> # Block forever.
3441 >>> pool = BlockingConnectionPool(timeout=None)
3443 >>> # Raise a ``ConnectionError`` after five seconds if a connection is
3444 >>> # not available.
3445 >>> pool = BlockingConnectionPool(timeout=5)
3446 """
3448 def __init__(
3449 self,
3450 max_connections=50,
3451 timeout=20,
3452 connection_class=Connection,
3453 queue_class=LifoQueue,
3454 **connection_kwargs,
3455 ):
3456 self.queue_class = queue_class
3457 self.timeout = timeout
3458 self._in_maintenance = False
3459 self._locked = False
3460 super().__init__(
3461 connection_class=connection_class,
3462 max_connections=max_connections,
3463 **connection_kwargs,
3464 )
3466 def reset(self):
3467 # Create and fill up a thread safe queue with ``None`` values.
3468 try:
3469 if self._in_maintenance:
3470 self._lock.acquire()
3471 self._locked = True
3473 # Record metrics for connections being removed before clearing
3474 # Note: Access pool.queue directly to avoid deadlock since we may
3475 # already hold self._lock (which is non-reentrant)
3476 if (
3477 hasattr(self, "_connections")
3478 and self._connections
3479 and hasattr(self, "pool")
3480 ):
3481 with self._lock:
3482 connections_in_queue = {conn for conn in self.pool.queue if conn}
3483 idle_count = len(connections_in_queue)
3484 in_use_count = len(self._connections) - idle_count
3485 if idle_count > 0 or in_use_count > 0:
3486 pool_name = get_pool_name(self)
3487 if idle_count > 0:
3488 record_connection_count(
3489 pool_name=pool_name,
3490 connection_state=ConnectionState.IDLE,
3491 counter=-idle_count,
3492 )
3493 if in_use_count > 0:
3494 record_connection_count(
3495 pool_name=pool_name,
3496 connection_state=ConnectionState.USED,
3497 counter=-in_use_count,
3498 )
3500 self.pool = self.queue_class(self.max_connections)
3501 while True:
3502 try:
3503 self.pool.put_nowait(None)
3504 except Full:
3505 break
3507 # Keep a list of actual connection instances so that we can
3508 # disconnect them later.
3509 self._connections = []
3510 finally:
3511 if self._locked:
3512 try:
3513 self._lock.release()
3514 except Exception:
3515 pass
3516 self._locked = False
3518 # this must be the last operation in this method. while reset() is
3519 # called when holding _fork_lock, other threads in this process
3520 # can call _checkpid() which compares self.pid and os.getpid() without
3521 # holding any lock (for performance reasons). keeping this assignment
3522 # as the last operation ensures that those other threads will also
3523 # notice a pid difference and block waiting for the first thread to
3524 # release _fork_lock. when each of these threads eventually acquire
3525 # _fork_lock, they will notice that another thread already called
3526 # reset() and they will immediately release _fork_lock and continue on.
3527 self.pid = os.getpid()
3529 def __del__(self) -> None:
3530 """Clean up connection pool and record metrics when garbage collected."""
3531 try:
3532 # Note: Access pool.queue directly to avoid potential deadlock
3533 # if GC runs while the lock is held by the same thread
3534 if (
3535 hasattr(self, "_connections")
3536 and self._connections
3537 and hasattr(self, "pool")
3538 ):
3539 connections_in_queue = {conn for conn in self.pool.queue if conn}
3540 idle_count = len(connections_in_queue)
3541 in_use_count = len(self._connections) - idle_count
3542 if idle_count > 0 or in_use_count > 0:
3543 pool_name = get_pool_name(self)
3544 if idle_count > 0:
3545 record_connection_count(
3546 pool_name=pool_name,
3547 connection_state=ConnectionState.IDLE,
3548 counter=-idle_count,
3549 )
3550 if in_use_count > 0:
3551 record_connection_count(
3552 pool_name=pool_name,
3553 connection_state=ConnectionState.USED,
3554 counter=-in_use_count,
3555 )
3556 except Exception:
3557 pass
3559 def make_connection(self):
3560 "Make a fresh connection."
3561 try:
3562 if self._in_maintenance:
3563 self._lock.acquire()
3564 self._locked = True
3566 if self.cache is not None:
3567 connection = CacheProxyConnection(
3568 self.connection_class(**self.connection_kwargs),
3569 self.cache,
3570 self._lock,
3571 )
3572 else:
3573 connection = self.connection_class(**self.connection_kwargs)
3574 self._connections.append(connection)
3576 # Record new connection created (starts as IDLE)
3577 record_connection_count(
3578 pool_name=get_pool_name(self),
3579 connection_state=ConnectionState.IDLE,
3580 counter=1,
3581 )
3583 return connection
3584 finally:
3585 if self._locked:
3586 try:
3587 self._lock.release()
3588 except Exception:
3589 pass
3590 self._locked = False
3592 @deprecated_args(
3593 args_to_warn=["*"],
3594 reason="Use get_connection() without args instead",
3595 version="5.3.0",
3596 )
3597 def get_connection(self, command_name=None, *keys, **options):
3598 """
3599 Get a connection, blocking for ``self.timeout`` until a connection
3600 is available from the pool.
3602 If the connection returned is ``None`` then creates a new connection.
3603 Because we use a last-in first-out queue, the existing connections
3604 (having been returned to the pool after the initial ``None`` values
3605 were added) will be returned before ``None`` values. This means we only
3606 create new connections when we need to, i.e.: the actual number of
3607 connections will only increase in response to demand.
3608 """
3609 start_time_acquired = time.monotonic()
3610 # Make sure we haven't changed process.
3611 self._checkpid()
3612 is_created = False
3614 # Try and get a connection from the pool. If one isn't available within
3615 # self.timeout then raise a ``ConnectionError``.
3616 connection = None
3617 try:
3618 if self._in_maintenance:
3619 self._lock.acquire()
3620 self._locked = True
3621 try:
3622 connection = self.pool.get(block=True, timeout=self.timeout)
3623 except Empty:
3624 # Note that this is not caught by the redis client and will be
3625 # raised unless handled by application code. If you want never to
3626 raise ConnectionError("No connection available.")
3628 # If the ``connection`` is actually ``None`` then that's a cue to make
3629 # a new connection to add to the pool.
3630 if connection is None:
3631 # Start timing for observability
3632 start_time_created = time.monotonic()
3633 connection = self.make_connection()
3634 is_created = True
3635 finally:
3636 if self._locked:
3637 try:
3638 self._lock.release()
3639 except Exception:
3640 pass
3641 self._locked = False
3643 # Record state transition: IDLE -> USED
3644 # (make_connection already recorded IDLE +1 for new connections)
3645 # This ensures counters stay balanced if connect() fails and release() is called
3646 pool_name = get_pool_name(self)
3647 record_connection_count(
3648 pool_name=pool_name,
3649 connection_state=ConnectionState.IDLE,
3650 counter=-1,
3651 )
3652 record_connection_count(
3653 pool_name=pool_name,
3654 connection_state=ConnectionState.USED,
3655 counter=1,
3656 )
3658 try:
3659 # ensure this connection is connected to Redis
3660 connection.connect()
3661 # connections that the pool provides should be ready to send
3662 # a command. if not, the connection was either returned to the
3663 # pool before all data has been read or the socket has been
3664 # closed. either way, reconnect and verify everything is good.
3665 try:
3666 if (
3667 connection.can_read()
3668 and self.cache is None
3669 and not self.maint_notifications_enabled()
3670 ):
3671 raise ConnectionError("Connection has data")
3672 except (ConnectionError, TimeoutError, OSError):
3673 connection.disconnect()
3674 connection.connect()
3675 if (
3676 connection.can_read()
3677 and self.cache is None
3678 and not self.maint_notifications_enabled()
3679 ):
3680 raise ConnectionError("Connection not ready")
3681 except BaseException:
3682 # release the connection back to the pool so that we don't leak it
3683 self.release(connection)
3684 raise
3686 if is_created:
3687 record_connection_create_time(
3688 connection_pool=self,
3689 duration_seconds=time.monotonic() - start_time_created,
3690 )
3692 record_connection_wait_time(
3693 pool_name=pool_name,
3694 duration_seconds=time.monotonic() - start_time_acquired,
3695 )
3697 return connection
3699 def release(self, connection):
3700 "Releases the connection back to the pool."
3701 # Make sure we haven't changed process.
3702 self._checkpid()
3704 try:
3705 if self._in_maintenance:
3706 self._lock.acquire()
3707 self._locked = True
3708 if not self.owns_connection(connection):
3709 # pool doesn't own this connection. do not add it back
3710 # to the pool. instead add a None value which is a placeholder
3711 # that will cause the pool to recreate the connection if
3712 # its needed.
3713 connection.disconnect()
3714 self.pool.put_nowait(None)
3715 # Still need to decrement USED since it was counted in get_connection()
3716 record_connection_count(
3717 pool_name="unknown_pool",
3718 connection_state=ConnectionState.USED,
3719 counter=-1,
3720 )
3721 return
3722 if connection.should_reconnect():
3723 connection.disconnect()
3724 # Put the connection back into the pool.
3725 pool_name = get_pool_name(self)
3726 try:
3727 self.pool.put_nowait(connection)
3729 # Record state transition: USED -> IDLE
3730 record_connection_count(
3731 pool_name=pool_name,
3732 connection_state=ConnectionState.USED,
3733 counter=-1,
3734 )
3735 record_connection_count(
3736 pool_name=pool_name,
3737 connection_state=ConnectionState.IDLE,
3738 counter=1,
3739 )
3740 except Full:
3741 pass
3742 finally:
3743 if self._locked:
3744 try:
3745 self._lock.release()
3746 except Exception:
3747 pass
3748 self._locked = False
3750 def disconnect(self, inuse_connections: bool = True):
3751 """
3752 Disconnects either all connections in the pool or just the free connections.
3753 """
3754 self._checkpid()
3755 try:
3756 if self._in_maintenance:
3757 self._lock.acquire()
3758 self._locked = True
3760 if inuse_connections:
3761 connections = self._connections
3762 else:
3763 connections = self._get_free_connections()
3765 for connection in connections:
3766 connection.disconnect()
3767 finally:
3768 if self._locked:
3769 try:
3770 self._lock.release()
3771 except Exception:
3772 pass
3773 self._locked = False
3775 def _get_free_connections(self):
3776 with self._lock:
3777 return {conn for conn in self.pool.queue if conn}
3779 def _get_in_use_connections(self):
3780 with self._lock:
3781 # free connections
3782 connections_in_queue = {conn for conn in self.pool.queue if conn}
3783 # in self._connections we keep all created connections
3784 # so the ones that are not in the queue are the in use ones
3785 return {
3786 conn for conn in self._connections if conn not in connections_in_queue
3787 }
3789 def set_in_maintenance(self, in_maintenance: bool):
3790 """
3791 Sets a flag that this Blocking ConnectionPool is in maintenance mode.
3793 This is used to prevent new connections from being created while we are in maintenance mode.
3794 The pool will be in maintenance mode only when we are processing a MOVING notification.
3795 """
3796 self._in_maintenance = in_maintenance