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