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 asyncio
2import contextlib
3import copy
4import inspect
5import logging
6import math
7import socket
8import sys
9import time
10import warnings
11import weakref
12from abc import ABC, abstractmethod
13from itertools import chain
14from types import MappingProxyType
15from typing import (
16 Any,
17 AsyncIterator,
18 Callable,
19 Iterable,
20 List,
21 Literal,
22 Mapping,
23 Optional,
24 Protocol,
25 Set,
26 Tuple,
27 Type,
28 TypedDict,
29 TypeVar,
30 Union,
31)
32from urllib.parse import ParseResult, parse_qs, unquote, urlparse
34from ..observability.attributes import (
35 DB_CLIENT_CONNECTION_POOL_NAME,
36 DB_CLIENT_CONNECTION_STATE,
37 AttributeBuilder,
38 ConnectionState,
39 get_pool_name,
40)
41from ..utils import SSL_AVAILABLE, deprecated_function
43if SSL_AVAILABLE:
44 import ssl
45 from ssl import SSLContext, TLSVersion, VerifyFlags
46else:
47 ssl = None
48 TLSVersion = None
49 SSLContext = None
50 VerifyFlags = None
52from ..auth.token import TokenInterface
53from ..driver_info import DriverInfo, resolve_driver_info
54from ..event import AsyncAfterConnectionReleasedEvent, EventDispatcher
55from ..utils import deprecated_args, format_error_message
57# the functionality is available in 3.11.x but has a major issue before
58# 3.11.3. See https://github.com/redis/redis-py/issues/2633
59if sys.version_info >= (3, 11, 3):
60 from asyncio import timeout as async_timeout
61else:
62 from async_timeout import timeout as async_timeout
64from redis import exceptions as redis_exceptions
65from redis.asyncio.maint_notifications import (
66 AsyncMaintNotificationsConnectionHandler,
67 AsyncMaintNotificationsPoolHandler,
68 AsyncOSSMaintNotificationsHandler,
69)
70from redis.asyncio.observability.recorder import (
71 record_connection_closed,
72 record_connection_count,
73 record_connection_create_time,
74 record_connection_timeout,
75 record_connection_wait_time,
76 record_error_count,
77)
78from redis.asyncio.retry import Retry, _to_async_retry
79from redis.backoff import NoBackoff
80from redis.connection import HiredisRespSerializer
81from redis.credentials import CredentialProvider, UsernamePasswordCredentialProvider
82from redis.event import MaintenanceCompletedEvent
83from redis.exceptions import (
84 AuthenticationError,
85 AuthenticationWrongNumberOfArgsError,
86 ConnectionError,
87 DataError,
88 MaxConnectionsError,
89 RedisError,
90 ResponseError,
91 TimeoutError,
92)
93from redis.himport import HImportRegistry
94from redis.maint_notifications import (
95 PENDING_PUSH_NOTIFICATIONS_READ_TIMEOUT,
96 MaintenanceState,
97 MaintNotificationsConfig,
98 NodeMovingNotification,
99 _build_moving_cleanup_connection_kwargs,
100 _build_moving_connection_kwargs,
101 _dispatch_maintenance_event,
102)
103from redis.observability.metrics import CloseReason
104from redis.typing import EncodableT
105from redis.utils import (
106 DEFAULT_RESP_VERSION,
107 HIREDIS_AVAILABLE,
108 SENTINEL,
109 check_protocol_version,
110 str_if_bytes,
111)
113from .._defaults import (
114 DEFAULT_SOCKET_CONNECT_TIMEOUT,
115 DEFAULT_SOCKET_READ_SIZE,
116 DEFAULT_SOCKET_TIMEOUT,
117 get_default_socket_keepalive_options,
118)
119from .._parsers import (
120 UNRECOVERABLE_PARSE_ERRORS,
121 AsyncPushNotificationsParser,
122 BaseParser,
123 Encoder,
124 _AsyncHiredisParser,
125 _AsyncRESP2Parser,
126 _AsyncRESP3Parser,
127)
129SYM_STAR = b"*"
130SYM_DOLLAR = b"$"
131SYM_CRLF = b"\r\n"
132SYM_LF = b"\n"
133SYM_EMPTY = b""
135DefaultParser: Type[Union[_AsyncRESP2Parser, _AsyncRESP3Parser, _AsyncHiredisParser]]
136if HIREDIS_AVAILABLE:
137 DefaultParser = _AsyncHiredisParser
138else:
139 DefaultParser = _AsyncRESP3Parser
141logger = logging.getLogger(__name__)
144def add_debug_log_for_connection_failure(
145 connection: "AbstractConnection",
146 error: BaseException,
147 operation: str,
148) -> None:
149 """
150 Render the connection's live state on a failure that is about to close it.
152 Must be called *before* ``disconnect()``. ``extract_connection_details()``
153 reads the local port and the in-flight read deadline off the transport, so
154 once it is gone it can only report ``not connected`` - which hides exactly
155 the state needed to explain the failure. In particular it is what tells
156 apart a read that ran under the original timeout from one that ran under a
157 relaxed maintenance timeout.
158 """
159 if logger.isEnabledFor(logging.DEBUG):
160 logger.debug(
161 f"{type(error).__name__} while {operation}, "
162 f"with connection: {connection}, "
163 f"details: {connection.extract_connection_details()}, "
164 f"error: {error}",
165 )
168class ConnectCallbackProtocol(Protocol):
169 def __call__(self, connection: "AbstractConnection"): ...
172class AsyncConnectCallbackProtocol(Protocol):
173 async def __call__(self, connection: "AbstractConnection"): ...
176ConnectCallbackT = Union[ConnectCallbackProtocol, AsyncConnectCallbackProtocol]
179class AsyncMaintNotificationsAbstractConnection:
180 """
181 Internal mixin for async maintenance notification state and parser handlers.
183 The sync implementation uses the same mixin-style structure. The async
184 version keeps the notification state and parser handler installation close
185 to the connection without sending the server-side handshake; that is wired
186 in a later step.
187 """
189 __slots__ = ()
191 def __init__(
192 self,
193 maint_notifications_config: MaintNotificationsConfig | None,
194 maint_notifications_pool_handler: (
195 AsyncMaintNotificationsPoolHandler | None
196 ) = None,
197 maintenance_state: MaintenanceState = MaintenanceState.NONE,
198 maintenance_notification_hash: int | None = None,
199 orig_host_address: str | None = None,
200 orig_socket_timeout: float | None = None,
201 orig_socket_connect_timeout: float | None = None,
202 oss_cluster_maint_notifications_handler: (
203 AsyncOSSMaintNotificationsHandler | None
204 ) = None,
205 parser: BaseParser | None = None,
206 ) -> None:
207 self.maint_notifications_config = maint_notifications_config
208 self.maintenance_state = maintenance_state
209 self.maintenance_notification_hash = maintenance_notification_hash
210 self._processed_start_maint_notifications: set[int] = set()
211 self._skipped_end_maint_notifications: set[int] = set()
212 self._configure_maintenance_notifications(
213 maint_notifications_pool_handler,
214 orig_host_address,
215 orig_socket_timeout,
216 orig_socket_connect_timeout,
217 oss_cluster_maint_notifications_handler,
218 parser,
219 )
221 @abstractmethod
222 def _get_parser(self) -> BaseParser:
223 pass
225 def _get_push_notifications_parser(self) -> AsyncPushNotificationsParser:
226 parser = self._get_parser()
227 if not isinstance(parser, (_AsyncHiredisParser, _AsyncRESP3Parser)):
228 raise RedisError(
229 "Maintenance notifications are only supported with hiredis and RESP3 parsers!"
230 )
231 return parser
233 def _complete_maintenance_on_disconnect(self) -> None:
234 """
235 Restore the settings a maintenance relaxed on this connection when it is
236 closed, and report the relaxation as completed - the same way the
237 completion notification would have been reported. The MOVING state is
238 left alone: it is owned by the pool-level TTL cleanup.
239 """
240 if self.maintenance_state != MaintenanceState.MAINTENANCE:
241 return
242 self.reset_tmp_settings(reset_relaxed_timeout=True)
243 self.maintenance_state = MaintenanceState.NONE
244 # reset the sets that keep track of received start maint
245 # notifications and skipped end maint notifications
246 self.reset_received_notifications()
247 if self.maint_notifications_config is not None:
248 pool_handler = self._maint_notifications_pool_handler
249 _dispatch_maintenance_event(
250 self.maint_notifications_config,
251 MaintenanceCompletedEvent(
252 connection_pool=(
253 pool_handler.pool if pool_handler is not None else None
254 ),
255 connection=self,
256 state=MaintenanceState.MAINTENANCE,
257 notification=None,
258 config=self.maint_notifications_config,
259 ),
260 )
262 async def handle_pending_push_notifications(self) -> None:
263 """
264 Process the push notifications buffered on this idle connection if it
265 is still alive, and report a connection that has to be reconnected.
267 A pooled connection that has readable data while idle either has push
268 notifications waiting, or has been closed by the server - or both, when
269 the notifications announced the maintenance that closed it. The socket
270 state cannot be told apart from the pending data without reading it, so
271 the notifications are collected first and handed to their handlers only
272 once the drain has shown the socket to be alive: the server closes the
273 connections of a moved endpoint when the MOVING time-to-live expires,
274 so an alive socket means the maintenance is still in progress and the
275 announced address current. A notification buffered on a socket the
276 server has since closed describes a maintenance that is over, after
277 which the configured address is what to connect to again: it is
278 discarded, and the ``ConnectionError`` raised here makes the pool
279 reconnect before any command is sent.
281 Anything read that is not a maintenance notification or a client-side
282 cache invalidation - a reply, a null or an error reply left unread by an
283 earlier command - means the connection is dirty and is reported the same
284 way, as is a drain that does not complete within
285 ``PENDING_PUSH_NOTIFICATIONS_READ_TIMEOUT``.
286 """
287 parser = self._get_push_notifications_parser()
288 deferred: list = []
290 def defer(handler):
291 async def deferred_handler(notification):
292 deferred.append((handler, notification))
294 return deferred_handler
296 handlers = (
297 (parser.node_moving_push_handler_func, parser.set_node_moving_push_handler),
298 (parser.maintenance_push_handler_func, parser.set_maintenance_push_handler),
299 (
300 parser.oss_cluster_maint_push_handler_func,
301 parser.set_oss_cluster_maint_push_handler,
302 ),
303 # Client-side cache invalidations are pushes too; the ones on a
304 # connection that has to be reconnected are covered by the cache
305 # flush its disconnect performs
306 (
307 parser.invalidation_push_handler_func,
308 parser.set_invalidation_push_handler,
309 ),
310 )
311 for handler, set_handler in handlers:
312 if handler is not None:
313 set_handler(defer(handler))
315 alive = False
316 deadline = time.monotonic() + PENDING_PUSH_NOTIFICATIONS_READ_TIMEOUT
317 try:
318 while await self.can_read():
319 remaining = deadline - time.monotonic()
320 if remaining <= 0:
321 raise ConnectionError(
322 "Timed out reading pending push notifications"
323 )
324 read = len(deferred)
325 # The pool disconnects the connection it handed out on the error
326 try:
327 await asyncio.wait_for(
328 self.read_response(
329 push_request=True, disconnect_on_error=False
330 ),
331 remaining,
332 )
333 except asyncio.TimeoutError:
334 raise ConnectionError(
335 "Timed out reading pending push notifications"
336 ) from None
337 # Only a push reaches a deferred handler; anything else is a
338 # reply left unread by an earlier command
339 if len(deferred) == read:
340 raise ConnectionError("Connection has data")
341 alive = True
342 except ResponseError as e:
343 raise ConnectionError("Connection has data") from e
344 finally:
345 for handler, set_handler in handlers:
346 set_handler(handler)
347 if not alive and deferred and logger.isEnabledFor(logging.DEBUG):
348 logger.debug(
349 f"Discarding {len(deferred)} push notification(s) buffered on "
350 f"a connection that is closed or dirty, reconnecting: {self}"
351 )
353 for handler, notification in deferred:
354 try:
355 await handler(notification)
356 except Exception as e:
357 logger.error(f"Error handling {notification}: {e}")
359 @abstractmethod
360 def get_protocol(self):
361 pass
363 @abstractmethod
364 async def send_command(self, *args: Any, **kwargs: Any) -> None:
365 pass
367 @abstractmethod
368 async def read_response(
369 self,
370 disable_decoding: bool = False,
371 timeout: float | None = None,
372 *,
373 disconnect_on_error: bool = True,
374 push_request: bool | None = False,
375 ) -> Any:
376 pass
378 @abstractmethod
379 def getpeername(self) -> str | None:
380 pass
382 @abstractmethod
383 def extract_connection_details(self) -> str:
384 pass
386 def _configure_maintenance_notifications(
387 self,
388 maint_notifications_pool_handler: (
389 AsyncMaintNotificationsPoolHandler | None
390 ) = None,
391 orig_host_address: str | None = None,
392 orig_socket_timeout: float | None = None,
393 orig_socket_connect_timeout: float | None = None,
394 oss_cluster_maint_notifications_handler: (
395 AsyncOSSMaintNotificationsHandler | None
396 ) = None,
397 parser: BaseParser | None = None,
398 ) -> None:
399 if (
400 not self.maint_notifications_config
401 or not self.maint_notifications_config.enabled
402 ):
403 self._maint_notifications_pool_handler = None
404 self._maint_notifications_connection_handler = None
405 self._oss_cluster_maint_notifications_handler = None
406 return
408 if not parser:
409 raise RedisError(
410 "To configure maintenance notifications, a parser must be provided!"
411 )
413 if not isinstance(parser, _AsyncHiredisParser) and not isinstance(
414 parser, _AsyncRESP3Parser
415 ):
416 raise RedisError(
417 "Maintenance notifications are only supported with hiredis and RESP3 parsers!"
418 )
420 if maint_notifications_pool_handler:
421 # Extract a reference to a new pool handler that copies all properties
422 # of the original one and has a different connection reference
423 # This is needed because when we attach the handler to the parser
424 # we need to make sure that the handler has a reference to the
425 # connection that the parser is attached to.
426 self._maint_notifications_pool_handler = (
427 maint_notifications_pool_handler.get_handler_for_connection()
428 )
429 self._maint_notifications_pool_handler.set_connection(self)
430 else:
431 self._maint_notifications_pool_handler = None
433 self._maint_notifications_connection_handler = (
434 AsyncMaintNotificationsConnectionHandler(
435 self, self.maint_notifications_config
436 )
437 )
439 if oss_cluster_maint_notifications_handler:
440 self._oss_cluster_maint_notifications_handler = (
441 oss_cluster_maint_notifications_handler
442 )
443 parser.set_oss_cluster_maint_push_handler(
444 oss_cluster_maint_notifications_handler.handle_notification
445 )
446 else:
447 self._oss_cluster_maint_notifications_handler = None
449 # Set up pool handler to parser if available
450 if self._maint_notifications_pool_handler:
451 parser.set_node_moving_push_handler(
452 self._maint_notifications_pool_handler.handle_notification
453 )
455 # Set up connection handler
456 parser.set_maintenance_push_handler(
457 self._maint_notifications_connection_handler.handle_notification
458 )
460 self.orig_host_address = orig_host_address if orig_host_address else self.host
461 self.orig_socket_timeout = (
462 orig_socket_timeout if orig_socket_timeout else self.socket_timeout
463 )
464 self.orig_socket_connect_timeout = (
465 orig_socket_connect_timeout
466 if orig_socket_connect_timeout
467 else self.socket_connect_timeout
468 )
470 def set_maint_notifications_pool_handler_for_connection(
471 self, maint_notifications_pool_handler: AsyncMaintNotificationsPoolHandler
472 ) -> None:
473 # Deep copy the pool handler to avoid sharing the same pool handler
474 # between multiple connections, because otherwise each connection will override
475 # the connection reference and the pool handler will only hold a reference
476 # to the last connection that was set.
477 maint_notifications_pool_handler_copy = (
478 maint_notifications_pool_handler.get_handler_for_connection()
479 )
480 maint_notifications_pool_handler_copy.set_connection(self)
481 parser = self._get_push_notifications_parser()
482 parser.set_node_moving_push_handler(
483 maint_notifications_pool_handler_copy.handle_notification
484 )
485 self._maint_notifications_pool_handler = maint_notifications_pool_handler_copy
487 # Update maintenance notification connection handler if it doesn't exist
488 if not self._maint_notifications_connection_handler:
489 self._maint_notifications_connection_handler = (
490 AsyncMaintNotificationsConnectionHandler(
491 self, maint_notifications_pool_handler.config
492 )
493 )
494 parser.set_maintenance_push_handler(
495 self._maint_notifications_connection_handler.handle_notification
496 )
497 else:
498 self._maint_notifications_connection_handler.config = (
499 maint_notifications_pool_handler.config
500 )
502 def set_maint_notifications_cluster_handler_for_connection(
503 self,
504 oss_cluster_maint_notifications_handler: AsyncOSSMaintNotificationsHandler,
505 ) -> None:
506 parser = self._get_push_notifications_parser()
507 parser.set_oss_cluster_maint_push_handler(
508 oss_cluster_maint_notifications_handler.handle_notification
509 )
510 # OSS cluster mode and pool-handler mode are mutually exclusive. Clear
511 # any node-moving/pool handler a default (RESP3 "auto") pool wired in
512 # __init__ so this existing connection is not configured with both.
513 parser.set_node_moving_push_handler(None)
514 self._maint_notifications_pool_handler = None
516 self._oss_cluster_maint_notifications_handler = (
517 oss_cluster_maint_notifications_handler
518 )
520 # Update maintenance notification connection handler if it doesn't exist
521 if not self._maint_notifications_connection_handler:
522 self._maint_notifications_connection_handler = (
523 AsyncMaintNotificationsConnectionHandler(
524 self, oss_cluster_maint_notifications_handler.config
525 )
526 )
527 parser.set_maintenance_push_handler(
528 self._maint_notifications_connection_handler.handle_notification
529 )
530 else:
531 self._maint_notifications_connection_handler.config = (
532 oss_cluster_maint_notifications_handler.config
533 )
535 async def activate_maint_notifications_handling_if_enabled(
536 self, check_health: bool = True
537 ) -> None:
538 # Send maintenance notifications handshake if RESP3 is active
539 # and maintenance notifications are enabled
540 # and we have a host to determine the endpoint type from
541 # When the maint_notifications_config enabled mode is "auto",
542 # we just log a warning if the handshake fails
543 # When the mode is enabled=True, we raise an exception in case of failure
544 host = getattr(self, "host", None)
545 if (
546 check_protocol_version(self.get_protocol(), 3)
547 and self.maint_notifications_config
548 and self.maint_notifications_config.enabled
549 and self._maint_notifications_connection_handler
550 and host is not None
551 ):
552 await self._enable_maintenance_notifications(
553 maint_notifications_config=self.maint_notifications_config,
554 check_health=check_health,
555 )
557 async def _enable_maintenance_notifications(
558 self,
559 maint_notifications_config: MaintNotificationsConfig,
560 check_health: bool = True,
561 ) -> None:
562 try:
563 host = getattr(self, "host", None)
564 if host is None:
565 raise ValueError(
566 "Cannot enable maintenance notifications for connection"
567 " object that doesn't have a host attribute."
568 )
570 endpoint_type = maint_notifications_config.get_endpoint_type(host, self)
571 await self.send_command(
572 "CLIENT",
573 "MAINT_NOTIFICATIONS",
574 "ON",
575 "moving-endpoint-type",
576 endpoint_type.value,
577 check_health=check_health,
578 )
579 response = await self.read_response()
580 if not response or str_if_bytes(response) != "OK":
581 raise ResponseError(
582 "The server doesn't support maintenance notifications"
583 )
584 except Exception as e:
585 if (
586 isinstance(e, ResponseError)
587 and maint_notifications_config.enabled == "auto"
588 ):
589 # Log warning but don't fail the connection
590 logger.debug(f"Failed to enable maintenance notifications: {e}")
591 else:
592 raise
594 def get_resolved_ip(self) -> str | None:
595 """
596 Extract the resolved IP address from an established connection or host.
598 First tries to get the actual peer IP from the async stream writer, then
599 falls back to DNS resolution if needed.
601 Returns:
602 The resolved IP address, or None if it cannot be determined.
603 """
605 # Method 1: Try to get the actual IP from the established stream.
606 # This is most accurate as it shows the exact IP being used.
607 try:
608 peer_addr = self.getpeername()
609 if peer_addr:
610 return peer_addr
611 except (AttributeError, OSError):
612 # Stream might not be connected or peer address lookup might fail.
613 pass
615 # Method 2: Fall back to the configured host (which may be an IP or an
616 # FQDN). Unlike the sync client we intentionally do NOT call
617 # socket.getaddrinfo() here: this method runs on the event loop, so a
618 # blocking DNS resolution would stall it. On the endpoint-type handshake
619 # path (get_endpoint_type) getpeername() above always succeeds because the
620 # writer was just connected, so this fallback is only reached by the
621 # debug-log call sites during reconnects — where returning the host is
622 # fine. A blocking getaddrinfo on an FQDN host there can freeze the loop
623 # for seconds and trip unrelated connect timeouts.
624 return getattr(self, "host", None)
626 @property
627 def maintenance_state(self) -> MaintenanceState:
628 return self._maintenance_state
630 @maintenance_state.setter
631 def maintenance_state(self, state: MaintenanceState) -> None:
632 self._maintenance_state = state
634 def add_maint_start_notification(self, id: int) -> None:
635 self._processed_start_maint_notifications.add(id)
637 def get_processed_start_notifications(self) -> set[int]:
638 return self._processed_start_maint_notifications
640 def add_skipped_end_notification(self, id: int) -> None:
641 self._skipped_end_maint_notifications.add(id)
643 def get_skipped_end_notifications(self) -> set[int]:
644 return self._skipped_end_maint_notifications
646 def reset_received_notifications(self) -> None:
647 self._processed_start_maint_notifications.clear()
648 self._skipped_end_maint_notifications.clear()
650 def update_current_socket_timeout(
651 self, relaxed_timeout: float | None = None
652 ) -> None:
653 timeout = relaxed_timeout if relaxed_timeout != -1 else self.socket_timeout
654 self._reschedule_active_read_timeout(timeout)
656 def _reschedule_active_read_timeout(self, timeout: float | None) -> None:
657 timeout_context = getattr(self, "_active_read_timeout", None)
658 if timeout_context is None:
659 # No read_response call is currently inside its socket timeout
660 # context, so there is no in-flight deadline to relax or restore.
661 return
663 if timeout is None:
664 # A None socket timeout means the active read should become blocking.
665 # Python 3.11's timeout context supports clearing the deadline.
666 if hasattr(timeout_context, "reschedule"):
667 timeout_context.reschedule(None)
668 # Older async-timeout contexts cannot clear a deadline, so reject the
669 # current timeout instead of leaving a stale relaxed deadline active.
670 elif hasattr(timeout_context, "reject"):
671 timeout_context.reject()
672 return
674 # Active read timeouts are stored as loop-time deadlines, not durations.
675 deadline = asyncio.get_running_loop().time() + timeout
676 if hasattr(timeout_context, "reschedule"):
677 # Python 3.11 asyncio.timeout exposes reschedule().
678 timeout_context.reschedule(deadline)
679 elif hasattr(timeout_context, "update"):
680 # async-timeout exposes update() for the same deadline adjustment.
681 timeout_context.update(deadline)
683 def set_tmp_settings(
684 self,
685 tmp_host_address: str | object | None = SENTINEL,
686 tmp_relaxed_timeout: float | None = -1,
687 ) -> None:
688 """
689 SENTINEL keeps the host unchanged. -1 keeps the relaxed timeout unchanged.
690 """
691 if tmp_host_address and tmp_host_address != SENTINEL:
692 self.host = str(tmp_host_address)
693 if tmp_relaxed_timeout != -1:
694 self.socket_timeout = tmp_relaxed_timeout
695 self.socket_connect_timeout = tmp_relaxed_timeout
697 def reset_tmp_settings(
698 self,
699 reset_host_address: bool = False,
700 reset_relaxed_timeout: bool = False,
701 ) -> None:
702 if reset_host_address:
703 self.host = self.orig_host_address
704 if reset_relaxed_timeout:
705 self.socket_timeout = self.orig_socket_timeout
706 self.socket_connect_timeout = self.orig_socket_connect_timeout
709class AbstractConnection(AsyncMaintNotificationsAbstractConnection):
710 """Manages communication to and from a Redis server"""
712 __slots__ = (
713 "db",
714 "username",
715 "client_name",
716 "lib_name",
717 "lib_version",
718 "credential_provider",
719 "password",
720 "socket_timeout",
721 "socket_connect_timeout",
722 "redis_connect_func",
723 "retry_on_timeout",
724 "retry_on_error",
725 "health_check_interval",
726 "next_health_check",
727 "last_active_at",
728 "encoder",
729 "ssl_context",
730 "protocol",
731 "_reader",
732 "_writer",
733 "_parser",
734 "_active_read_timeout",
735 "_connect_callbacks",
736 "_buffer_cutoff",
737 "_lock",
738 "_socket_read_size",
739 "__dict__",
740 )
742 @deprecated_args(
743 args_to_warn=["lib_name", "lib_version"],
744 reason="Use 'driver_info' parameter instead. "
745 "lib_name and lib_version will be removed in a future version.",
746 )
747 def __init__(
748 self,
749 *,
750 db: str | int = 0,
751 password: str | None = None,
752 socket_timeout: float | None = DEFAULT_SOCKET_TIMEOUT,
753 socket_connect_timeout: float | None = DEFAULT_SOCKET_CONNECT_TIMEOUT,
754 retry_on_timeout: bool = False,
755 retry_on_error: Iterable[Type[Exception]] | object = SENTINEL,
756 encoding: str = "utf-8",
757 encoding_errors: str = "strict",
758 decode_responses: bool = False,
759 parser_class: Type[BaseParser] = DefaultParser,
760 socket_read_size: int = DEFAULT_SOCKET_READ_SIZE,
761 health_check_interval: float = 0,
762 client_name: str | None = None,
763 lib_name: str | object | None = SENTINEL,
764 lib_version: str | object | None = SENTINEL,
765 driver_info: DriverInfo | object | None = SENTINEL,
766 username: str | None = None,
767 retry: Retry | None = None,
768 redis_connect_func: ConnectCallbackT | None = None,
769 encoder_class: Type[Encoder] = Encoder,
770 credential_provider: CredentialProvider | None = None,
771 protocol: int | None = None,
772 legacy_responses: bool = True,
773 event_dispatcher: EventDispatcher | None = None,
774 maint_notifications_config: MaintNotificationsConfig | None = None,
775 maint_notifications_pool_handler: (
776 AsyncMaintNotificationsPoolHandler | None
777 ) = None,
778 maintenance_state: MaintenanceState = MaintenanceState.NONE,
779 maintenance_notification_hash: int | None = None,
780 orig_host_address: str | None = None,
781 orig_socket_timeout: float | None = None,
782 orig_socket_connect_timeout: float | None = None,
783 oss_cluster_maint_notifications_handler: (
784 AsyncOSSMaintNotificationsHandler | None
785 ) = None,
786 himport_registry: HImportRegistry | None = None,
787 ):
788 """
789 Initialize a new async Connection.
791 Parameters
792 ----------
793 driver_info : DriverInfo, optional
794 Driver metadata for CLIENT SETINFO. If provided, lib_name and lib_version
795 are ignored. If not provided, a DriverInfo will be created from lib_name
796 and lib_version. Explicit None disables CLIENT SETINFO.
797 lib_name : str, optional
798 **Deprecated.** Use driver_info instead. Library name for CLIENT SETINFO.
799 lib_version : str, optional
800 **Deprecated.** Use driver_info instead. Library version for CLIENT SETINFO.
801 """
802 if (username or password) and credential_provider is not None:
803 raise DataError(
804 "'username' and 'password' cannot be passed along with 'credential_"
805 "provider'. Please provide only one of the following arguments: \n"
806 "1. 'password' and (optional) 'username'\n"
807 "2. 'credential_provider'"
808 )
809 if event_dispatcher is None:
810 self._event_dispatcher = EventDispatcher()
811 else:
812 self._event_dispatcher = event_dispatcher
813 self.db = db
814 self.client_name = client_name
816 # Handle driver_info: if provided, use it; otherwise create from lib_name/lib_version.
817 self.driver_info = resolve_driver_info(driver_info, lib_name, lib_version)
819 self.credential_provider = credential_provider
820 self.password = password
821 self.username = username
822 self.socket_timeout = socket_timeout
823 if socket_connect_timeout is None:
824 socket_connect_timeout = socket_timeout
825 self.socket_connect_timeout = socket_connect_timeout
826 self.retry_on_timeout = retry_on_timeout
827 if retry_on_error is SENTINEL:
828 retry_on_error = []
829 else:
830 # Copy so we never mutate the caller-supplied list (parity with sync).
831 retry_on_error = list(retry_on_error)
832 if retry_on_timeout:
833 retry_on_error.append(TimeoutError)
834 retry_on_error.append(socket.timeout)
835 retry_on_error.append(asyncio.TimeoutError)
836 self.retry_on_error = retry_on_error
837 if retry or retry_on_error:
838 if not retry:
839 self.retry = Retry(NoBackoff(), 1)
840 else:
841 # deep-copy the Retry object as it is mutable
842 self.retry = copy.deepcopy(_to_async_retry(retry))
843 # Update the retry's supported errors with the specified errors
844 self.retry.update_supported_errors(retry_on_error)
845 else:
846 self.retry = Retry(NoBackoff(), 0)
847 self.health_check_interval = health_check_interval
848 self.next_health_check: float = -1
849 self.encoder = encoder_class(encoding, encoding_errors, decode_responses)
850 self.redis_connect_func = redis_connect_func
851 self._reader: Optional[asyncio.StreamReader] = None
852 self._writer: Optional[asyncio.StreamWriter] = None
853 self._socket_read_size = socket_read_size
854 self._active_read_timeout = None
855 self._connect_callbacks: List[weakref.WeakMethod[ConnectCallbackT]] = []
856 self._buffer_cutoff = 6000
857 self._re_auth_token: Optional[TokenInterface] = None
858 self._command_packer = (
859 HiredisRespSerializer(self._pack_command_python, self.encoder.encode)
860 if HIREDIS_AVAILABLE
861 else None
862 )
863 self._should_reconnect = False
865 try:
866 p = int(protocol)
867 except TypeError:
868 p = DEFAULT_RESP_VERSION
869 except ValueError:
870 raise ConnectionError("protocol must be an integer")
871 else:
872 if p < 2 or p > 3:
873 raise ConnectionError("protocol must be either 2 or 3")
874 self.protocol = p
875 self.legacy_responses = legacy_responses
876 if parser_class != _AsyncHiredisParser:
877 # The Python parsers are protocol-specific; hiredis supports both.
878 if self.protocol == 3 and parser_class == _AsyncRESP2Parser:
879 parser_class = _AsyncRESP3Parser
880 elif self.protocol == 2 and parser_class == _AsyncRESP3Parser:
881 parser_class = _AsyncRESP2Parser
882 self.set_parser(parser_class)
884 # HIMPORT client-side state. `himport_registry` is the shared client-level
885 # registry (empty if unconfigured) and persists across reconnects.
886 self.himport_registry = himport_registry
887 self._reset_himport_state()
889 AsyncMaintNotificationsAbstractConnection.__init__(
890 self,
891 maint_notifications_config,
892 maint_notifications_pool_handler,
893 maintenance_state,
894 maintenance_notification_hash,
895 orig_host_address,
896 orig_socket_timeout,
897 orig_socket_connect_timeout,
898 oss_cluster_maint_notifications_handler,
899 self._parser,
900 )
902 def __del__(self, _warnings: Any = warnings):
903 # For some reason, the individual streams don't get properly garbage
904 # collected and therefore produce no resource warnings. We add one
905 # here, in the same style as those from the stdlib.
906 if getattr(self, "_writer", None):
907 _warnings.warn(
908 f"unclosed Connection {self!r}", ResourceWarning, source=self
909 )
911 try:
912 asyncio.get_running_loop()
913 self._close()
914 except RuntimeError:
915 # No actions been taken if pool already closed.
916 pass
918 def _close(self):
919 """
920 Internal method to silently close the connection without waiting
921 """
922 if self._writer:
923 self._writer.close()
924 self._writer = self._reader = None
926 def __repr__(self):
927 repr_args = ",".join((f"{k}={v}" for k, v in self.repr_pieces()))
928 return f"<{self.__class__.__module__}.{self.__class__.__name__}({repr_args})>"
930 @abstractmethod
931 def repr_pieces(self):
932 pass
934 @property
935 def is_connected(self):
936 return self._reader is not None and self._writer is not None
938 def register_connect_callback(self, callback):
939 """
940 Register a callback to be called when the connection is established either
941 initially or reconnected. This allows listeners to issue commands that
942 are ephemeral to the connection, for example pub/sub subscription or
943 key tracking. The callback must be a _method_ and will be kept as
944 a weak reference.
945 """
946 wm = weakref.WeakMethod(callback)
947 if wm not in self._connect_callbacks:
948 self._connect_callbacks.append(wm)
950 def deregister_connect_callback(self, callback):
951 """
952 De-register a previously registered callback. It will no-longer receive
953 notifications on connection events. Calling this is not required when the
954 listener goes away, since the callbacks are kept as weak methods.
955 """
956 try:
957 self._connect_callbacks.remove(weakref.WeakMethod(callback))
958 except ValueError:
959 pass
961 def set_parser(self, parser_class: Type[BaseParser]) -> None:
962 """
963 Creates a new instance of parser_class with socket size:
964 _socket_read_size and assigns it to the parser for the connection
965 :param parser_class: The required parser class
966 """
967 self._parser = parser_class(socket_read_size=self._socket_read_size)
969 def _get_parser(self) -> BaseParser:
970 return self._parser
972 def getpeername(self) -> str | None:
973 """
974 Returns the peer name of the connection.
975 """
976 writer = self._writer
977 if writer is None:
978 return None
979 peername = writer.get_extra_info("peername")
980 if isinstance(peername, tuple) and peername:
981 return str(peername[0])
982 return None
984 def extract_connection_details(self) -> str:
985 """
986 Render the connection's identity, maintenance state and effective timeouts.
988 This is what the debug logs use to explain a failed or timed out command:
990 - ``host`` vs ``orig host`` says whether the connection still points at the
991 node being moved away from, or has already been repointed at the new one.
992 - ``state`` says whether maintenance handling touched this connection at
993 all, so an unaffected node's connections are distinguishable.
994 - ``socket_timeout`` vs ``active read timeout`` says which timeout the read
995 actually ran under. ``active read timeout`` is the remaining deadline of
996 the in-flight ``read_response`` (``None`` when no read is in flight), so
997 the two diverge for a command that was already reading when the relaxed
998 timeout was applied.
999 """
1000 writer = self._writer
1001 if writer is None:
1002 return "not connected"
1004 socket_address = None
1005 try:
1006 socket_name = writer.get_extra_info("sockname")
1007 # AF_UNIX sockets report a path string rather than a (host, port) tuple
1008 if isinstance(socket_name, tuple) and len(socket_name) > 1:
1009 socket_address = socket_name[1]
1010 except (AttributeError, OSError):
1011 pass
1013 # Unlike the sync client there is no timeout armed on the socket; the
1014 # deadline lives in the timeout context wrapping the in-flight read.
1015 active_read_timeout = None
1016 timeout_context = self._active_read_timeout
1017 if timeout_context is not None:
1018 try:
1019 when = timeout_context.when()
1020 if when is not None:
1021 active_read_timeout = round(
1022 when - asyncio.get_running_loop().time(), 3
1023 )
1024 except (AttributeError, RuntimeError):
1025 pass
1027 state = getattr(self.maintenance_state, "value", self.maintenance_state)
1028 return (
1029 f"connected to ip {self.get_resolved_ip()}, "
1030 f"local socket port: {socket_address}, "
1031 f"host: {self._host_error()} "
1032 f"(orig: {getattr(self, 'orig_host_address', None)}), "
1033 f"state: {state}, "
1034 f"socket_timeout: {self.socket_timeout} "
1035 f"(orig: {getattr(self, 'orig_socket_timeout', None)}), "
1036 f"active read timeout: {active_read_timeout}, "
1037 f"should_reconnect: {self.should_reconnect()}, "
1038 f"notification_hash: {self.maintenance_notification_hash}"
1039 )
1041 async def connect(self):
1042 """Connects to the Redis server if not already connected"""
1043 # try once the socket connect with the handshake, retry the whole
1044 # connect/handshake flow based on retry policy
1045 await self.retry.call_with_retry(
1046 lambda: self.connect_check_health(
1047 check_health=True, retry_socket_connect=False
1048 ),
1049 lambda error, failure_count: self.disconnect(
1050 error=error, failure_count=failure_count
1051 ),
1052 with_failure_count=True,
1053 )
1055 async def connect_check_health(
1056 self, check_health: bool = True, retry_socket_connect: bool = True
1057 ):
1058 if self.is_connected:
1059 return
1060 # Track actual retry attempts for error reporting
1061 actual_retry_attempts = 0
1063 def failure_callback(error, failure_count):
1064 nonlocal actual_retry_attempts
1065 actual_retry_attempts = failure_count
1066 return self.disconnect(error=error, failure_count=failure_count)
1068 try:
1069 if retry_socket_connect:
1070 await self.retry.call_with_retry(
1071 lambda: self._connect(),
1072 failure_callback,
1073 with_failure_count=True,
1074 )
1075 else:
1076 await self._connect()
1077 except asyncio.CancelledError:
1078 raise # in 3.7 and earlier, this is an Exception, not BaseException
1079 except (socket.timeout, asyncio.TimeoutError):
1080 e = TimeoutError("Timeout connecting to server")
1081 await record_error_count(
1082 server_address=getattr(self, "host", None),
1083 server_port=getattr(self, "port", None),
1084 network_peer_address=getattr(self, "host", None),
1085 network_peer_port=getattr(self, "port", None),
1086 error_type=e,
1087 retry_attempts=actual_retry_attempts,
1088 is_internal=False,
1089 )
1090 raise e
1091 except OSError as e:
1092 e = ConnectionError(self._error_message(e))
1093 await record_error_count(
1094 server_address=getattr(self, "host", None),
1095 server_port=getattr(self, "port", None),
1096 network_peer_address=getattr(self, "host", None),
1097 network_peer_port=getattr(self, "port", None),
1098 error_type=e,
1099 retry_attempts=actual_retry_attempts,
1100 is_internal=False,
1101 )
1102 raise e
1103 except Exception as exc:
1104 raise ConnectionError(exc) from exc
1106 try:
1107 if not self.redis_connect_func:
1108 # Use the default on_connect function
1109 await self.on_connect_check_health(check_health=check_health)
1110 else:
1111 # Use the passed function redis_connect_func
1112 (
1113 await self.redis_connect_func(self)
1114 if asyncio.iscoroutinefunction(self.redis_connect_func)
1115 else self.redis_connect_func(self)
1116 )
1117 except RedisError:
1118 # clean up after any error in on_connect
1119 await self.disconnect()
1120 raise
1122 # run any user callbacks. right now the only internal callback
1123 # is for pubsub channel/pattern resubscription
1124 # first, remove any dead weakrefs
1125 self._connect_callbacks = [ref for ref in self._connect_callbacks if ref()]
1126 for ref in self._connect_callbacks:
1127 callback = ref()
1128 task = callback(self)
1129 if task and inspect.isawaitable(task):
1130 await task
1132 def mark_for_reconnect(self):
1133 self._should_reconnect = True
1135 def should_reconnect(self):
1136 return self._should_reconnect
1138 def reset_should_reconnect(self):
1139 self._should_reconnect = False
1141 @abstractmethod
1142 async def _connect(self):
1143 pass
1145 @abstractmethod
1146 def _host_error(self) -> str:
1147 pass
1149 def _error_message(self, exception: BaseException) -> str:
1150 return format_error_message(self._host_error(), exception)
1152 def get_protocol(self):
1153 return self.protocol
1155 def _reset_himport_state(self) -> None:
1156 # A fresh server session has no prepared HIMPORT fieldsets, so the next
1157 # himport_set must re-prepare on this connection. ``_himport_prepared`` maps
1158 # fieldset name -> the version prepared on the server; ``_himport_reconciled
1159 # _revision`` is the registry revision this connection last reconciled discards
1160 # against. Both are reset on connect/disconnect since the session is gone.
1161 self._himport_prepared: dict[str, int] = {}
1162 self._himport_reconciled_revision: int = 0
1164 async def on_connect(self) -> None:
1165 """Initialize the connection, authenticate and select a database"""
1166 await self.on_connect_check_health(check_health=True)
1168 async def on_connect_check_health(self, check_health: bool = True) -> None:
1169 # A fresh socket is a new server session: no prepared HIMPORT fieldsets.
1170 self._reset_himport_state()
1171 self._parser.on_connect(self)
1172 parser = self._parser
1174 auth_args = None
1175 # if credential provider or username and/or password are set, authenticate
1176 if self.credential_provider or (self.username or self.password):
1177 cred_provider = (
1178 self.credential_provider
1179 or UsernamePasswordCredentialProvider(self.username, self.password)
1180 )
1181 auth_args = await cred_provider.get_credentials_async()
1183 # if resp version is specified and we have auth args,
1184 # we need to send them via HELLO
1185 if auth_args and check_protocol_version(self.protocol, 3):
1186 if isinstance(self._parser, _AsyncRESP2Parser):
1187 self.set_parser(_AsyncRESP3Parser)
1188 # update cluster exception classes
1189 self._parser.EXCEPTION_CLASSES = parser.EXCEPTION_CLASSES
1190 self._parser.on_connect(self)
1191 if len(auth_args) == 1:
1192 auth_args = ["default", auth_args[0]]
1193 # avoid checking health here -- PING will fail if we try
1194 # to check the health prior to the AUTH
1195 await self.send_command(
1196 "HELLO", self.protocol, "AUTH", *auth_args, check_health=False
1197 )
1198 response = await self.read_response()
1199 if response.get(b"proto") != int(self.protocol) and response.get(
1200 "proto"
1201 ) != int(self.protocol):
1202 raise ConnectionError("Invalid RESP version")
1203 # avoid checking health here -- PING will fail if we try
1204 # to check the health prior to the AUTH
1205 elif auth_args:
1206 await self.send_command("AUTH", *auth_args, check_health=False)
1208 try:
1209 auth_response = await self.read_response()
1210 except AuthenticationWrongNumberOfArgsError:
1211 # a username and password were specified but the Redis
1212 # server seems to be < 6.0.0 which expects a single password
1213 # arg. retry auth with just the password.
1214 # https://github.com/andymccurdy/redis-py/issues/1274
1215 await self.send_command("AUTH", auth_args[-1], check_health=False)
1216 auth_response = await self.read_response()
1218 if str_if_bytes(auth_response) != "OK":
1219 raise AuthenticationError("Invalid Username or Password")
1221 # if resp version is specified, switch to it
1222 elif check_protocol_version(self.protocol, 3):
1223 if isinstance(self._parser, _AsyncRESP2Parser):
1224 self.set_parser(_AsyncRESP3Parser)
1225 # update cluster exception classes
1226 self._parser.EXCEPTION_CLASSES = parser.EXCEPTION_CLASSES
1227 self._parser.on_connect(self)
1228 await self.send_command("HELLO", self.protocol, check_health=check_health)
1229 response = await self.read_response()
1230 # if response.get(b"proto") != self.protocol and response.get(
1231 # "proto"
1232 # ) != self.protocol:
1233 # raise ConnectionError("Invalid RESP version")
1235 # Activate maintenance notifications for this connection
1236 # if enabled in the configuration
1237 # This is a no-op if maintenance notifications are not enabled
1238 await self.activate_maint_notifications_handling_if_enabled(
1239 check_health=check_health
1240 )
1242 # if a client_name is given, set it
1243 if self.client_name:
1244 await self.send_command(
1245 "CLIENT",
1246 "SETNAME",
1247 self.client_name,
1248 check_health=check_health,
1249 )
1250 if str_if_bytes(await self.read_response()) != "OK":
1251 raise ConnectionError("Error setting client name")
1253 # Set the library name and version from driver_info, pipeline for lower startup latency
1254 lib_name_sent = False
1255 lib_version_sent = False
1257 if self.driver_info and self.driver_info.formatted_name:
1258 await self.send_command(
1259 "CLIENT",
1260 "SETINFO",
1261 "LIB-NAME",
1262 self.driver_info.formatted_name,
1263 check_health=check_health,
1264 )
1265 lib_name_sent = True
1267 if self.driver_info and self.driver_info.lib_version:
1268 await self.send_command(
1269 "CLIENT",
1270 "SETINFO",
1271 "LIB-VER",
1272 self.driver_info.lib_version,
1273 check_health=check_health,
1274 )
1275 lib_version_sent = True
1277 # if a database is specified, switch to it. Also pipeline this
1278 if self.db:
1279 await self.send_command("SELECT", self.db, check_health=check_health)
1281 # read responses from pipeline
1282 for _ in range(sum([lib_name_sent, lib_version_sent])):
1283 try:
1284 await self.read_response()
1285 except ResponseError:
1286 pass
1288 if self.db:
1289 if str_if_bytes(await self.read_response()) != "OK":
1290 raise ConnectionError("Invalid Database")
1292 async def disconnect(
1293 self,
1294 nowait: bool = False,
1295 error: Optional[Exception] = None,
1296 failure_count: Optional[int] = None,
1297 health_check_failed: bool = False,
1298 ) -> None:
1299 """Disconnects from the Redis server"""
1300 # The server session is gone, so any HIMPORT fieldsets prepared on this
1301 # socket no longer exist; reset the tracking.
1302 self._reset_himport_state()
1303 # Likewise the relaxation a maintenance applied to it: restored and
1304 # reported here, ahead of a close that may time out or be skipped
1305 self._complete_maintenance_on_disconnect()
1306 # On Python 3.13+, asyncio.timeout() raises RuntimeError when called
1307 # outside a running Task (e.g. during GC finalization or event-loop
1308 # callbacks). In that context we fall back to a synchronous close.
1309 # See https://github.com/redis/redis-py/issues/3856
1310 if asyncio.current_task() is None:
1311 self._parser.on_disconnect()
1312 self.reset_should_reconnect()
1313 self._close()
1314 return
1316 try:
1317 async with async_timeout(self.socket_connect_timeout):
1318 self._parser.on_disconnect()
1319 # Reset the reconnect flag
1320 self.reset_should_reconnect()
1321 if not self.is_connected:
1322 return
1323 try:
1324 self._writer.close() # type: ignore[union-attr]
1325 # wait for close to finish, except when handling errors and
1326 # forcefully disconnecting.
1327 if not nowait:
1328 await self._writer.wait_closed() # type: ignore[union-attr]
1329 except OSError:
1330 pass
1331 finally:
1332 self._reader = None
1333 self._writer = None
1334 except asyncio.TimeoutError:
1335 raise TimeoutError(
1336 f"Timed out closing connection after {self.socket_connect_timeout}"
1337 ) from None
1339 if error:
1340 if health_check_failed:
1341 close_reason = CloseReason.HEALTHCHECK_FAILED
1342 else:
1343 close_reason = CloseReason.ERROR
1345 if failure_count is not None and failure_count > self.retry.get_retries():
1346 await record_error_count(
1347 server_address=getattr(self, "host", None),
1348 server_port=getattr(self, "port", None),
1349 network_peer_address=getattr(self, "host", None),
1350 network_peer_port=getattr(self, "port", None),
1351 error_type=error,
1352 retry_attempts=failure_count,
1353 )
1355 await record_connection_closed(
1356 close_reason=close_reason,
1357 error_type=error,
1358 )
1359 else:
1360 await record_connection_closed(
1361 close_reason=CloseReason.APPLICATION_CLOSE,
1362 )
1364 async def _send_ping(self):
1365 """Send PING, expect PONG in return"""
1366 await self.send_command("PING", check_health=False)
1367 if str_if_bytes(await self.read_response()) != "PONG":
1368 raise ConnectionError("Bad response from PING health check")
1370 async def _ping_failed(self, error, failure_count):
1371 """Function to call when PING fails"""
1372 await self.disconnect(
1373 error=error, failure_count=failure_count, health_check_failed=True
1374 )
1376 async def check_health(self):
1377 """Check the health of the connection with a PING/PONG"""
1378 if (
1379 self.health_check_interval
1380 and asyncio.get_running_loop().time() > self.next_health_check
1381 ):
1382 await self.retry.call_with_retry(
1383 self._send_ping, self._ping_failed, with_failure_count=True
1384 )
1386 async def _send_packed_command(self, command: Iterable[bytes]) -> None:
1387 writer = self._writer
1388 if writer is None or writer.transport.is_closing():
1389 raise ConnectionError("Connection closed by the server before write")
1390 try:
1391 writer.writelines(command)
1392 await writer.drain()
1393 except (TypeError, AttributeError) as e:
1394 # CPython gh-136234 adds the missing connection-lost check in 3.13.10+
1395 # and 3.14.1+. Python 3.12, 3.13.0-3.13.9, and 3.14.0 can instead
1396 # leak TypeError or AttributeError from the transport (#4287).
1397 if writer.transport.is_closing():
1398 raise ConnectionError(
1399 "Connection closed by the server while writing"
1400 ) from e
1401 raise
1403 async def send_packed_command(
1404 self, command: Union[bytes, str, Iterable[bytes]], check_health: bool = True
1405 ) -> None:
1406 if not self.is_connected:
1407 await self.connect_check_health(check_health=False)
1408 if check_health:
1409 await self.check_health()
1411 try:
1412 if isinstance(command, str):
1413 command = command.encode()
1414 if isinstance(command, bytes):
1415 command = [command]
1416 if self.socket_timeout:
1417 await asyncio.wait_for(
1418 self._send_packed_command(command), self.socket_timeout
1419 )
1420 else:
1421 await self._send_packed_command(command)
1422 except asyncio.TimeoutError as e:
1423 add_debug_log_for_connection_failure(self, e, "writing command")
1424 await self.disconnect(nowait=True)
1425 raise TimeoutError("Timeout writing to socket") from None
1426 except OSError as e:
1427 add_debug_log_for_connection_failure(self, e, "writing command")
1428 await self.disconnect(nowait=True)
1429 if len(e.args) == 1:
1430 err_no, errmsg = "UNKNOWN", e.args[0]
1431 else:
1432 err_no = e.args[0]
1433 errmsg = e.args[1]
1434 raise ConnectionError(
1435 f"Error {err_no} while writing to socket. {errmsg}."
1436 ) from e
1437 except BaseException as e:
1438 # BaseExceptions can be raised when a socket send operation is not
1439 # finished, e.g. due to a timeout. Ideally, a caller could then re-try
1440 # to send un-sent data. However, the send_packed_command() API
1441 # does not support it so there is no point in keeping the connection open.
1442 add_debug_log_for_connection_failure(self, e, "writing command")
1443 await self.disconnect(nowait=True)
1444 raise
1446 async def send_command(self, *args: Any, **kwargs: Any) -> None:
1447 """Pack and send a command to the Redis server"""
1448 await self.send_packed_command(
1449 self.pack_command(*args), check_health=kwargs.get("check_health", True)
1450 )
1452 @deprecated_function(
1453 version="8.0.0", reason="Use can_read() instead", name="can_read_destructive"
1454 )
1455 async def can_read_destructive(self) -> bool:
1456 """Check the socket to see if there's data loaded in the buffer."""
1457 try:
1458 return await self._parser.can_read()
1459 except OSError as e:
1460 await self.disconnect(nowait=True)
1461 host_error = self._host_error()
1462 raise ConnectionError(f"Error while reading from {host_error}: {e.args}")
1464 async def can_read(self) -> bool:
1465 """Check the socket to see if there's data loaded in the buffer."""
1466 # TODO: Rename this API; it detects pending data or dirty/closed
1467 # connection state, not only whether application data can be read.
1468 try:
1469 return await self._parser.can_read()
1470 except OSError as e:
1471 await self.disconnect(nowait=True)
1472 host_error = self._host_error()
1473 raise ConnectionError(f"Error while reading from {host_error}: {e.args}")
1475 async def read_response(
1476 self,
1477 disable_decoding: bool = False,
1478 timeout: float | None = None,
1479 *,
1480 disconnect_on_error: bool = True,
1481 push_request: bool | None = False,
1482 ):
1483 """Read the response from a previously sent command.
1485 ``timeout`` semantics:
1486 - ``None`` (default): fall back to ``self.socket_timeout``.
1487 - ``math.inf``: block indefinitely with no timeout. Used by PubSub
1488 blocking reads (``listen()`` / ``get_message(timeout=None)`` /
1489 ``parse_response(block=True)``) where the configured
1490 ``socket_timeout`` must not abort the read.
1491 - ``float``: apply that timeout in seconds for this single read.
1493 TODO(next-major): replace the ``math.inf`` opt-in with a SENTINEL
1494 default for ``timeout``. After that change, ``timeout=None`` will
1495 mean "no timeout, block until a response arrives" (matching the
1496 long-standing PubSub docstring contract) and the SENTINEL default
1497 will be the value that falls back to ``self.socket_timeout``.
1498 That swap is a breaking change, so it must wait for a major
1499 release. Until then, callers that need an indefinitely blocking
1500 read pass ``math.inf`` explicitly.
1501 """
1502 # TODO(next-major): drop the math.inf branch. Use SENTINEL as the
1503 # default for ``timeout`` and treat ``timeout is None`` as the
1504 # "no timeout" signal (matching the PubSub docstring contract).
1505 # Match only positive infinity here. ``-math.inf`` is not a valid
1506 # "block forever" signal and historically behaved as an already-
1507 # expired timeout; preserve that.
1508 if timeout == math.inf:
1509 read_timeout = None
1510 else:
1511 read_timeout = timeout if timeout is not None else self.socket_timeout
1512 host_error = self._host_error()
1513 try:
1514 if read_timeout is not None:
1515 timeout_context = async_timeout(read_timeout)
1516 if timeout is None:
1517 async with timeout_context as active_timeout:
1518 self._active_read_timeout = active_timeout
1519 try:
1520 response = await self._read_response_from_parser(
1521 disable_decoding=disable_decoding,
1522 push_request=push_request,
1523 )
1524 finally:
1525 self._active_read_timeout = None
1526 else:
1527 async with timeout_context:
1528 response = await self._read_response_from_parser(
1529 disable_decoding=disable_decoding,
1530 push_request=push_request,
1531 )
1532 else:
1533 response = await self._read_response_from_parser(
1534 disable_decoding=disable_decoding,
1535 push_request=push_request,
1536 )
1537 except asyncio.TimeoutError as e:
1538 if timeout is not None:
1539 # user requested timeout, return None. Operation can be retried
1540 return None
1541 # it was a self.socket_timeout error.
1542 if disconnect_on_error:
1543 add_debug_log_for_connection_failure(self, e, "reading response")
1544 await self.disconnect(nowait=True)
1545 raise TimeoutError(f"Timeout reading from {host_error}")
1546 except OSError as e:
1547 if disconnect_on_error:
1548 add_debug_log_for_connection_failure(self, e, "reading response")
1549 await self.disconnect(nowait=True)
1550 raise ConnectionError(f"Error while reading from {host_error} : {e.args}")
1551 except UNRECOVERABLE_PARSE_ERRORS as e:
1552 # See the sync Connection.read_response and #4291. The async parser
1553 # re-parses from self._pos = 0 rather than rewinding a socket
1554 # buffer, but the consequence is the same: the bytes that already
1555 # failed to parse are still there, so the connection is not
1556 # reusable no matter what disconnect_on_error says.
1557 add_debug_log_for_connection_failure(self, e, "reading response")
1558 await self.disconnect(nowait=True)
1559 raise
1560 except BaseException as e:
1561 # Also by default close in case of BaseException. A lot of code
1562 # relies on this behaviour when doing Command/Response pairs.
1563 # See #1128.
1564 if disconnect_on_error:
1565 add_debug_log_for_connection_failure(self, e, "reading response")
1566 await self.disconnect(nowait=True)
1567 raise
1569 if self.health_check_interval:
1570 next_time = asyncio.get_running_loop().time() + self.health_check_interval
1571 self.next_health_check = next_time
1573 if isinstance(response, ResponseError):
1574 raise response from None
1575 return response
1577 async def _read_response_from_parser(
1578 self, disable_decoding: bool = False, push_request: bool | None = False
1579 ):
1580 if check_protocol_version(self.protocol, 3):
1581 return await self._parser.read_response(
1582 disable_decoding=disable_decoding, push_request=push_request
1583 )
1584 return await self._parser.read_response(disable_decoding=disable_decoding)
1586 def pack_command(self, *args: EncodableT) -> List[bytes]:
1587 if self._command_packer is not None:
1588 return self._command_packer.pack(*args)
1589 return self._pack_command_python(*args)
1591 def _pack_command_python(self, *args: EncodableT) -> List[bytes]:
1592 """Pack a series of arguments into the Redis protocol"""
1593 output = []
1594 # the client might have included 1 or more literal arguments in
1595 # the command name, e.g., 'CONFIG GET'. The Redis server expects these
1596 # arguments to be sent separately, so split the first argument
1597 # manually. These arguments should be bytestrings so that they are
1598 # not encoded.
1599 assert not isinstance(args[0], float)
1600 if isinstance(args[0], str):
1601 args = tuple(args[0].encode().split()) + args[1:]
1602 elif b" " in args[0]:
1603 args = tuple(args[0].split()) + args[1:]
1605 buff = SYM_EMPTY.join((SYM_STAR, str(len(args)).encode(), SYM_CRLF))
1607 buffer_cutoff = self._buffer_cutoff
1608 for arg in map(self.encoder.encode, args):
1609 # to avoid large string mallocs, chunk the command into the
1610 # output list if we're sending large values or memoryviews
1611 arg_length = len(arg)
1612 if (
1613 len(buff) > buffer_cutoff
1614 or arg_length > buffer_cutoff
1615 or isinstance(arg, memoryview)
1616 ):
1617 buff = SYM_EMPTY.join(
1618 (buff, SYM_DOLLAR, str(arg_length).encode(), SYM_CRLF)
1619 )
1620 output.append(buff)
1621 output.append(arg)
1622 buff = SYM_CRLF
1623 else:
1624 buff = SYM_EMPTY.join(
1625 (
1626 buff,
1627 SYM_DOLLAR,
1628 str(arg_length).encode(),
1629 SYM_CRLF,
1630 arg,
1631 SYM_CRLF,
1632 )
1633 )
1634 output.append(buff)
1635 return output
1637 def pack_commands(self, commands: Iterable[Iterable[EncodableT]]) -> List[bytes]:
1638 """Pack multiple commands into the Redis protocol"""
1639 output: List[bytes] = []
1640 pieces: List[bytes] = []
1641 buffer_length = 0
1642 buffer_cutoff = self._buffer_cutoff
1644 for cmd in commands:
1645 for chunk in self.pack_command(*cmd):
1646 chunklen = len(chunk)
1647 if (
1648 buffer_length > buffer_cutoff
1649 or chunklen > buffer_cutoff
1650 or isinstance(chunk, memoryview)
1651 ):
1652 if pieces:
1653 output.append(SYM_EMPTY.join(pieces))
1654 buffer_length = 0
1655 pieces = []
1657 if chunklen > buffer_cutoff or isinstance(chunk, memoryview):
1658 output.append(chunk)
1659 else:
1660 pieces.append(chunk)
1661 buffer_length += chunklen
1663 if pieces:
1664 output.append(SYM_EMPTY.join(pieces))
1665 return output
1667 def _socket_is_empty(self):
1668 """Check if the socket is empty"""
1669 return len(self._reader._buffer) == 0
1671 async def process_invalidation_messages(self):
1672 while not self._socket_is_empty():
1673 await self.read_response(push_request=True)
1675 def set_re_auth_token(self, token: TokenInterface):
1676 self._re_auth_token = token
1678 async def re_auth(self):
1679 if self._re_auth_token is not None:
1680 await self.send_command(
1681 "AUTH",
1682 self._re_auth_token.try_get("oid"),
1683 self._re_auth_token.get_value(),
1684 )
1685 await self.read_response()
1686 self._re_auth_token = None
1689class Connection(AbstractConnection):
1690 "Manages TCP communication to and from a Redis server"
1692 def __init__(
1693 self,
1694 *,
1695 host: str = "localhost",
1696 port: str | int = 6379,
1697 socket_keepalive: bool = True,
1698 socket_keepalive_options: Mapping[int, int | bytes] | object | None = SENTINEL,
1699 socket_type: int = 0,
1700 **kwargs,
1701 ):
1702 """
1703 Initialize a TCP connection.
1705 Parameters
1706 ----------
1707 socket_keepalive : bool
1708 If `True`, TCP keepalive is enabled for TCP socket connections.
1709 socket_keepalive_options : Mapping[int, int | bytes] | object | None
1710 Mapping of TCP keepalive socket option constants to values, for
1711 example `{socket.TCP_KEEPIDLE: 30}`. If left unspecified, redis-py
1712 uses TCP keepalive defaults when `socket_keepalive` is enabled:
1713 idle 30 seconds, interval 5 seconds, and 3 probes. Platform-specific
1714 options that are not available are skipped. Pass `None` or `{}` to
1715 avoid setting additional TCP keepalive options.
1716 """
1717 self.host = host
1718 # bool subclasses int; port=True would become privileged port 1.
1719 # Numeric strings stay valid. Callers still pass "6379".
1720 if isinstance(port, bool):
1721 raise TypeError("port must be an integer, not bool")
1722 if isinstance(port, str):
1723 try:
1724 port = int(port)
1725 except ValueError:
1726 raise TypeError("port must be an integer, not str") from None
1727 elif not isinstance(port, int):
1728 raise TypeError(f"port must be an integer, not {type(port).__name__}")
1729 if not 0 <= port <= 65535:
1730 raise ValueError(f"port must be in 0..65535, got {port}")
1731 self.port = port
1732 self.socket_keepalive = socket_keepalive
1733 if socket_keepalive_options is SENTINEL:
1734 socket_keepalive_options = get_default_socket_keepalive_options()
1735 self.socket_keepalive_options = socket_keepalive_options or {}
1736 self.socket_type = socket_type
1737 super().__init__(**kwargs)
1739 def repr_pieces(self):
1740 pieces = [("host", self.host), ("port", self.port), ("db", self.db)]
1741 if self.client_name:
1742 pieces.append(("client_name", self.client_name))
1743 return pieces
1745 def _connection_arguments(self) -> Mapping:
1746 return {"host": self.host, "port": self.port}
1748 async def _connect(self):
1749 """Create a TCP socket connection"""
1750 async with async_timeout(self.socket_connect_timeout):
1751 reader, writer = await asyncio.open_connection(
1752 **self._connection_arguments()
1753 )
1754 self._reader = reader
1755 self._writer = writer
1756 sock = writer.transport.get_extra_info("socket")
1757 if sock:
1758 sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
1759 try:
1760 # TCP_KEEPALIVE
1761 if self.socket_keepalive:
1762 sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
1763 for k, v in self.socket_keepalive_options.items():
1764 sock.setsockopt(socket.SOL_TCP, k, v)
1766 except (OSError, TypeError):
1767 # `socket_keepalive_options` might contain invalid options
1768 # causing an error. Do not leave the connection open.
1769 writer.close()
1770 raise
1772 def _host_error(self) -> str:
1773 return f"{self.host}:{self.port}"
1776class SSLConnection(Connection):
1777 """Manages SSL connections to and from the Redis server(s).
1778 This class extends the Connection class, adding SSL functionality, and making
1779 use of ssl.SSLContext (https://docs.python.org/3/library/ssl.html#ssl.SSLContext)
1780 """
1782 def __init__(
1783 self,
1784 ssl_keyfile: Optional[str] = None,
1785 ssl_certfile: Optional[str] = None,
1786 ssl_cert_reqs: Union[str, ssl.VerifyMode] = "required",
1787 ssl_include_verify_flags: Optional[List["ssl.VerifyFlags"]] = None,
1788 ssl_exclude_verify_flags: Optional[List["ssl.VerifyFlags"]] = None,
1789 ssl_ca_certs: Optional[str] = None,
1790 ssl_ca_data: Optional[str] = None,
1791 ssl_ca_path: Optional[str] = None,
1792 ssl_check_hostname: bool = True,
1793 ssl_min_version: Optional[TLSVersion] = None,
1794 ssl_ciphers: Optional[str] = None,
1795 ssl_password: Optional[str] = None,
1796 **kwargs,
1797 ):
1798 if not SSL_AVAILABLE:
1799 raise RedisError("Python wasn't built with SSL support")
1801 self.ssl_context: RedisSSLContext = RedisSSLContext(
1802 keyfile=ssl_keyfile,
1803 certfile=ssl_certfile,
1804 cert_reqs=ssl_cert_reqs,
1805 include_verify_flags=ssl_include_verify_flags,
1806 exclude_verify_flags=ssl_exclude_verify_flags,
1807 ca_certs=ssl_ca_certs,
1808 ca_data=ssl_ca_data,
1809 ca_path=ssl_ca_path,
1810 check_hostname=ssl_check_hostname,
1811 min_version=ssl_min_version,
1812 ciphers=ssl_ciphers,
1813 password=ssl_password,
1814 )
1815 super().__init__(**kwargs)
1817 def _connection_arguments(self) -> Mapping:
1818 kwargs = super()._connection_arguments()
1819 kwargs["ssl"] = self.ssl_context.get()
1820 return kwargs
1822 @property
1823 def keyfile(self):
1824 return self.ssl_context.keyfile
1826 @property
1827 def certfile(self):
1828 return self.ssl_context.certfile
1830 @property
1831 def cert_reqs(self):
1832 return self.ssl_context.cert_reqs
1834 @property
1835 def include_verify_flags(self):
1836 return self.ssl_context.include_verify_flags
1838 @property
1839 def exclude_verify_flags(self):
1840 return self.ssl_context.exclude_verify_flags
1842 @property
1843 def ca_certs(self):
1844 return self.ssl_context.ca_certs
1846 @property
1847 def ca_data(self):
1848 return self.ssl_context.ca_data
1850 @property
1851 def check_hostname(self):
1852 return self.ssl_context.check_hostname
1854 @property
1855 def min_version(self):
1856 return self.ssl_context.min_version
1859class RedisSSLContext:
1860 __slots__ = (
1861 "keyfile",
1862 "certfile",
1863 "cert_reqs",
1864 "include_verify_flags",
1865 "exclude_verify_flags",
1866 "ca_certs",
1867 "ca_data",
1868 "ca_path",
1869 "context",
1870 "check_hostname",
1871 "min_version",
1872 "ciphers",
1873 "password",
1874 )
1876 def __init__(
1877 self,
1878 keyfile: Optional[str] = None,
1879 certfile: Optional[str] = None,
1880 cert_reqs: Optional[Union[str, ssl.VerifyMode]] = None,
1881 include_verify_flags: Optional[List["ssl.VerifyFlags"]] = None,
1882 exclude_verify_flags: Optional[List["ssl.VerifyFlags"]] = None,
1883 ca_certs: Optional[str] = None,
1884 ca_data: Optional[str] = None,
1885 ca_path: Optional[str] = None,
1886 check_hostname: bool = False,
1887 min_version: Optional[TLSVersion] = None,
1888 ciphers: Optional[str] = None,
1889 password: Optional[str] = None,
1890 ):
1891 if not SSL_AVAILABLE:
1892 raise RedisError("Python wasn't built with SSL support")
1894 self.keyfile = keyfile
1895 self.certfile = certfile
1896 if cert_reqs is None:
1897 cert_reqs = ssl.CERT_NONE
1898 elif isinstance(cert_reqs, str):
1899 CERT_REQS = { # noqa: N806
1900 "none": ssl.CERT_NONE,
1901 "optional": ssl.CERT_OPTIONAL,
1902 "required": ssl.CERT_REQUIRED,
1903 }
1904 if cert_reqs not in CERT_REQS:
1905 raise RedisError(
1906 f"Invalid SSL Certificate Requirements Flag: {cert_reqs}"
1907 )
1908 cert_reqs = CERT_REQS[cert_reqs]
1909 self.cert_reqs = cert_reqs
1910 self.include_verify_flags = include_verify_flags
1911 self.exclude_verify_flags = exclude_verify_flags
1912 self.ca_certs = ca_certs
1913 self.ca_data = ca_data
1914 self.ca_path = ca_path
1915 self.check_hostname = (
1916 check_hostname if self.cert_reqs != ssl.CERT_NONE else False
1917 )
1918 self.min_version = min_version
1919 self.ciphers = ciphers
1920 self.password = password
1921 self.context: Optional[SSLContext] = None
1923 def get(self) -> SSLContext:
1924 if not self.context:
1925 context = ssl.create_default_context()
1926 context.check_hostname = self.check_hostname
1927 context.verify_mode = self.cert_reqs
1928 if self.include_verify_flags:
1929 for flag in self.include_verify_flags:
1930 context.verify_flags |= flag
1931 if self.exclude_verify_flags:
1932 for flag in self.exclude_verify_flags:
1933 context.verify_flags &= ~flag
1934 if self.certfile or self.keyfile:
1935 context.load_cert_chain(
1936 certfile=self.certfile,
1937 keyfile=self.keyfile,
1938 password=self.password,
1939 )
1940 if self.ca_certs or self.ca_data or self.ca_path:
1941 context.load_verify_locations(
1942 cafile=self.ca_certs, capath=self.ca_path, cadata=self.ca_data
1943 )
1944 if self.min_version is not None:
1945 context.minimum_version = self.min_version
1946 if self.ciphers is not None:
1947 context.set_ciphers(self.ciphers)
1948 self.context = context
1949 return self.context
1952class UnixDomainSocketConnection(AbstractConnection):
1953 "Manages UDS communication to and from a Redis server"
1955 def __init__(self, *, path: str = "", **kwargs):
1956 self.path = path
1957 super().__init__(**kwargs)
1959 def repr_pieces(self) -> Iterable[Tuple[str, Union[str, int]]]:
1960 pieces = [("path", self.path), ("db", self.db)]
1961 if self.client_name:
1962 pieces.append(("client_name", self.client_name))
1963 return pieces
1965 async def _connect(self):
1966 async with async_timeout(self.socket_connect_timeout):
1967 reader, writer = await asyncio.open_unix_connection(path=self.path)
1968 self._reader = reader
1969 self._writer = writer
1971 def _host_error(self) -> str:
1972 return self.path
1975FALSE_STRINGS = ("0", "F", "FALSE", "N", "NO", "OFF")
1978def to_bool(value) -> Optional[bool]:
1979 if value is None or value == "":
1980 return None
1981 if isinstance(value, str) and value.upper() in FALSE_STRINGS:
1982 return False
1983 return bool(value)
1986def parse_ssl_verify_flags(value):
1987 # flags are passed in as a string representation of a list,
1988 # e.g. VERIFY_X509_STRICT, VERIFY_X509_PARTIAL_CHAIN
1989 verify_flags_str = value.replace("[", "").replace("]", "")
1991 verify_flags = []
1992 for flag in verify_flags_str.split(","):
1993 flag = flag.strip()
1994 if not hasattr(VerifyFlags, flag):
1995 raise ValueError(f"Invalid ssl verify flag: {flag}")
1996 verify_flags.append(getattr(VerifyFlags, flag))
1997 return verify_flags
2000def parse_retry_on_error(value):
2001 # exception class names are passed as a comma-separated list,
2002 # e.g. ConnectionError,TimeoutError
2003 retry_on_error = []
2004 for name in value.replace("[", "").replace("]", "").split(","):
2005 name = name.strip()
2006 if not name:
2007 raise ValueError("Empty retry_on_error entry")
2008 exc = getattr(redis_exceptions, name, None)
2009 if not (isinstance(exc, type) and issubclass(exc, Exception)):
2010 raise ValueError(f"Unknown redis exception {name!r}")
2011 retry_on_error.append(exc)
2012 return retry_on_error
2015URL_QUERY_ARGUMENT_PARSERS: Mapping[str, Callable[..., object]] = MappingProxyType(
2016 {
2017 "db": int,
2018 "socket_timeout": float,
2019 "socket_connect_timeout": float,
2020 "socket_read_size": int,
2021 "socket_keepalive": to_bool,
2022 "retry_on_timeout": to_bool,
2023 "retry_on_error": parse_retry_on_error,
2024 "max_connections": int,
2025 "health_check_interval": int,
2026 "ssl_check_hostname": to_bool,
2027 "ssl_include_verify_flags": parse_ssl_verify_flags,
2028 "ssl_exclude_verify_flags": parse_ssl_verify_flags,
2029 "ssl_min_version": int,
2030 "timeout": float,
2031 "protocol": int,
2032 "legacy_responses": to_bool,
2033 }
2034)
2037class ConnectKwargs(TypedDict, total=False):
2038 username: str
2039 password: str
2040 connection_class: Type[AbstractConnection]
2041 host: str
2042 port: int
2043 db: int
2044 path: str
2047def parse_url(url: str) -> ConnectKwargs:
2048 # Scheme names are case-insensitive (RFC 3986), so normalize before the
2049 # prefix check; the "://" is required so a URL like "redis:foo" (which
2050 # urlparse would still report as the "redis" scheme) is rejected.
2051 if not url.lower().startswith(("redis://", "rediss://", "unix://")):
2052 raise ValueError(
2053 "Redis URL must specify one of the following schemes "
2054 "(redis://, rediss://, unix://)"
2055 )
2057 parsed: ParseResult = urlparse(url)
2058 kwargs: ConnectKwargs = {}
2060 for name, value_list in parse_qs(parsed.query).items():
2061 if value_list and len(value_list) > 0:
2062 # parse_qs() already percent-decodes query values, so use the value
2063 # as-is; unquoting again here would double-decode (e.g. "%2520" ->
2064 # "%20" -> " "). See issue #4208.
2065 value = value_list[0]
2066 parser = URL_QUERY_ARGUMENT_PARSERS.get(name)
2067 if parser:
2068 try:
2069 kwargs[name] = parser(value)
2070 except (TypeError, ValueError):
2071 raise ValueError(f"Invalid value for '{name}' in connection URL.")
2072 else:
2073 kwargs[name] = value
2075 if parsed.username:
2076 kwargs["username"] = unquote(parsed.username)
2077 if parsed.password:
2078 kwargs["password"] = unquote(parsed.password)
2080 # We only support redis://, rediss:// and unix:// schemes.
2081 if parsed.scheme == "unix":
2082 if parsed.path:
2083 kwargs["path"] = unquote(parsed.path)
2084 kwargs["connection_class"] = UnixDomainSocketConnection
2086 else: # implied: parsed.scheme in ("redis", "rediss")
2087 if parsed.hostname:
2088 kwargs["host"] = unquote(parsed.hostname)
2089 if parsed.port is not None:
2090 kwargs["port"] = int(parsed.port)
2092 # If there's a path argument, use it as the db argument if a
2093 # querystring value wasn't specified
2094 if parsed.path and "db" not in kwargs:
2095 try:
2096 kwargs["db"] = int(unquote(parsed.path).replace("/", ""))
2097 except (AttributeError, ValueError):
2098 pass
2100 if parsed.scheme == "rediss":
2101 kwargs["connection_class"] = SSLConnection
2103 return kwargs
2106_CP = TypeVar("_CP", bound="ConnectionPool")
2109class ConnectionPoolInterface(ABC):
2110 @abstractmethod
2111 def get_protocol(self):
2112 pass
2114 @abstractmethod
2115 def reset(self) -> None:
2116 pass
2118 @abstractmethod
2119 @deprecated_args(
2120 args_to_warn=["*"],
2121 reason="Use get_connection() without args instead",
2122 version="5.3.0",
2123 )
2124 async def get_connection(
2125 self, command_name: Optional[str] = None, *keys: Any, **options: Any
2126 ) -> "AbstractConnection":
2127 pass
2129 @abstractmethod
2130 def get_encoder(self) -> "Encoder":
2131 pass
2133 @abstractmethod
2134 async def release(self, connection: "AbstractConnection") -> None:
2135 pass
2137 @abstractmethod
2138 async def disconnect(self, inuse_connections: bool = True) -> None:
2139 pass
2141 @abstractmethod
2142 async def aclose(self) -> None:
2143 pass
2145 @abstractmethod
2146 def set_retry(self, retry: "Retry") -> None:
2147 pass
2149 @abstractmethod
2150 async def re_auth_callback(self, token: TokenInterface) -> None:
2151 pass
2153 @abstractmethod
2154 def get_connection_count(self) -> List[Tuple[int, dict]]:
2155 """
2156 Returns a connection count (both idle and in use).
2157 """
2158 pass
2161class AsyncMaintNotificationsAbstractConnectionPool:
2162 """
2163 Internal mixin for async maintenance notification pool wiring.
2165 The handler owns notification policy.
2166 This mixin owns pool state mutation because `_available_connections`,
2167 `_in_use_connections`, `connection_kwargs`, and the non-reentrant `asyncio.Lock`
2168 all live on the pool.
2169 """
2171 def __init__(
2172 self,
2173 maint_notifications_config: MaintNotificationsConfig | None = None,
2174 oss_cluster_maint_notifications_handler: (
2175 "AsyncOSSMaintNotificationsHandler | None"
2176 ) = None,
2177 **kwargs: Any,
2178 ) -> None:
2179 protocol = kwargs.get("protocol")
2180 is_protocol_supported = check_protocol_version(protocol, 3)
2181 is_connection_supported = self._maintenance_notifications_supported()
2183 if (
2184 maint_notifications_config is None
2185 and is_protocol_supported
2186 and is_connection_supported
2187 ):
2188 maint_notifications_config = MaintNotificationsConfig()
2190 if maint_notifications_config and maint_notifications_config.enabled:
2191 if not is_connection_supported:
2192 if maint_notifications_config.enabled is True:
2193 # Unix sockets do not have a host endpoint for CLIENT
2194 # MAINT_NOTIFICATIONS to describe.
2195 if "path" in self.connection_kwargs:
2196 raise RedisError(
2197 "Maintenance notifications are not supported for "
2198 "Unix domain socket connections"
2199 )
2201 # Custom connection classes must inherit the async maintenance
2202 # mixin so handlers can update connection state safely.
2203 if not self._maintenance_notifications_connection_class_supported():
2204 connection_class = getattr(self, "connection_class", None)
2205 connection_class_name = getattr(
2206 connection_class, "__name__", connection_class
2207 )
2208 raise RedisError(
2209 "Maintenance notifications are not supported for "
2210 f"connection class {connection_class_name}"
2211 )
2213 # TCP-like connections still need a host to identify the
2214 # endpoint that can move during maintenance.
2215 raise RedisError(
2216 "Maintenance notifications are not supported for connections "
2217 "without a host"
2218 )
2219 self._maint_notifications_pool_handler = None
2220 self._oss_cluster_maint_notifications_handler = None
2221 return
2223 if not is_protocol_supported:
2224 raise RedisError(
2225 "Maintenance notifications handlers on connection are only supported with RESP version 3"
2226 )
2228 if oss_cluster_maint_notifications_handler:
2229 self._oss_cluster_maint_notifications_handler = (
2230 oss_cluster_maint_notifications_handler
2231 )
2232 self._update_connection_kwargs_for_maint_notifications(
2233 oss_cluster_maint_notifications_handler=self._oss_cluster_maint_notifications_handler
2234 )
2235 self._maint_notifications_pool_handler = None
2236 else:
2237 self._oss_cluster_maint_notifications_handler = None
2238 self._maint_notifications_pool_handler = (
2239 AsyncMaintNotificationsPoolHandler(self, maint_notifications_config)
2240 )
2241 self._update_connection_kwargs_for_maint_notifications(
2242 maint_notifications_pool_handler=self._maint_notifications_pool_handler
2243 )
2244 else:
2245 self._maint_notifications_pool_handler = None
2246 self._oss_cluster_maint_notifications_handler = None
2248 async def _on_close(self) -> None:
2249 """Hook invoked from the pool's ``aclose()`` before the pool is shut down."""
2250 if self._maint_notifications_pool_handler is not None:
2251 await self._maint_notifications_pool_handler.cancel_scheduled_tasks()
2253 @property
2254 @abstractmethod
2255 def connection_kwargs(self) -> dict[str, Any]:
2256 pass
2258 @connection_kwargs.setter
2259 @abstractmethod
2260 def connection_kwargs(self, value: dict[str, Any]) -> None:
2261 pass
2263 @abstractmethod
2264 def _get_pool_lock(self) -> asyncio.Lock:
2265 pass
2267 @abstractmethod
2268 def _get_free_connections(self) -> Iterable["AbstractConnection"]:
2269 pass
2271 @abstractmethod
2272 def _get_in_use_connections(self) -> Iterable["AbstractConnection"]:
2273 pass
2275 def _maintenance_notifications_supported(self) -> bool:
2276 if "path" in self.connection_kwargs:
2277 return False
2278 if not self._maintenance_notifications_connection_class_supported():
2279 return False
2280 return bool(self.connection_kwargs.get("host"))
2282 def _maintenance_notifications_connection_class_supported(self) -> bool:
2283 connection_class = getattr(self, "connection_class", None)
2284 if connection_class is None:
2285 return False
2286 try:
2287 return issubclass(
2288 connection_class, AsyncMaintNotificationsAbstractConnection
2289 )
2290 except TypeError:
2291 return False
2293 def maint_notifications_enabled(self):
2294 """
2295 Returns:
2296 True if the maintenance notifications are enabled, False otherwise.
2297 The maintenance notifications config is stored in the pool handler.
2298 If the pool handler is not set, the maintenance notifications are not enabled.
2299 """
2300 if self._oss_cluster_maint_notifications_handler:
2301 maint_notifications_config = (
2302 self._oss_cluster_maint_notifications_handler.config
2303 )
2304 else:
2305 maint_notifications_config = (
2306 self._maint_notifications_pool_handler.config
2307 if self._maint_notifications_pool_handler
2308 else None
2309 )
2310 return maint_notifications_config and maint_notifications_config.enabled
2312 async def update_maint_notifications_config(
2313 self,
2314 maint_notifications_config: MaintNotificationsConfig,
2315 oss_cluster_maint_notifications_handler: (
2316 AsyncOSSMaintNotificationsHandler | None
2317 ) = None,
2318 ) -> None:
2319 """
2320 Updates the maintenance notifications configuration.
2321 This method should be called only if the pool was created
2322 without enabling the maintenance notifications and
2323 in a later point in time maintenance notifications
2324 are requested to be enabled.
2325 """
2326 if (
2327 self.maint_notifications_enabled()
2328 and not maint_notifications_config.enabled
2329 ):
2330 raise ValueError(
2331 "Cannot disable maintenance notifications after enabling them"
2332 )
2334 if oss_cluster_maint_notifications_handler:
2335 self._oss_cluster_maint_notifications_handler = (
2336 oss_cluster_maint_notifications_handler
2337 )
2338 # OSS cluster mode and pool-handler mode are mutually exclusive
2339 # (see __init__). A pool created with the default RESP3 "auto"
2340 # config wires a pool handler before this method runs; clear it so
2341 # new and existing connections are not configured with both handlers.
2342 self._maint_notifications_pool_handler = None
2343 else:
2344 if (
2345 maint_notifications_config.enabled
2346 and not self._maintenance_notifications_supported()
2347 ):
2348 if maint_notifications_config.enabled is True:
2349 # Unix sockets do not have a host endpoint for CLIENT
2350 # MAINT_NOTIFICATIONS to describe.
2351 if "path" in self.connection_kwargs:
2352 raise RedisError(
2353 "Maintenance notifications are not supported for "
2354 "Unix domain socket connections"
2355 )
2357 # Custom connection classes must inherit the async maintenance
2358 # mixin so handlers can update connection state safely.
2359 if not self._maintenance_notifications_connection_class_supported():
2360 connection_class = getattr(self, "connection_class", None)
2361 connection_class_name = getattr(
2362 connection_class, "__name__", connection_class
2363 )
2364 raise RedisError(
2365 "Maintenance notifications are not supported for "
2366 f"connection class {connection_class_name}"
2367 )
2369 # TCP-like connections still need a host to identify the
2370 # endpoint that can move during maintenance.
2371 raise RedisError(
2372 "Maintenance notifications are not supported for connections "
2373 "without a host"
2374 )
2375 self._maint_notifications_pool_handler = None
2376 return
2378 if self._oss_cluster_maint_notifications_handler:
2379 # Pool already in OSS cluster mode; update the OSS handler config
2380 # instead of creating a mutually-exclusive pool handler (which
2381 # would be silently ignored because the OSS handler wins priority
2382 # in both update helpers below).
2383 self._oss_cluster_maint_notifications_handler.config = (
2384 maint_notifications_config
2385 )
2386 elif not self._maint_notifications_pool_handler:
2387 self._maint_notifications_pool_handler = (
2388 AsyncMaintNotificationsPoolHandler(self, maint_notifications_config)
2389 )
2390 else:
2391 self._maint_notifications_pool_handler.config = (
2392 maint_notifications_config
2393 )
2395 self._update_connection_kwargs_for_maint_notifications(
2396 maint_notifications_pool_handler=self._maint_notifications_pool_handler,
2397 oss_cluster_maint_notifications_handler=self._oss_cluster_maint_notifications_handler,
2398 )
2399 await self._update_maint_notifications_configs_for_connections(
2400 maint_notifications_pool_handler=self._maint_notifications_pool_handler,
2401 oss_cluster_maint_notifications_handler=self._oss_cluster_maint_notifications_handler,
2402 )
2404 def _update_connection_kwargs_for_maint_notifications(
2405 self,
2406 maint_notifications_pool_handler: (
2407 AsyncMaintNotificationsPoolHandler | None
2408 ) = None,
2409 oss_cluster_maint_notifications_handler: (
2410 AsyncOSSMaintNotificationsHandler | None
2411 ) = None,
2412 ) -> None:
2413 """
2414 Update the connection kwargs for all future connections.
2415 """
2416 if not self.maint_notifications_enabled():
2417 return
2419 if maint_notifications_pool_handler:
2420 self.connection_kwargs.update(
2421 {
2422 "maint_notifications_pool_handler": maint_notifications_pool_handler,
2423 "maint_notifications_config": maint_notifications_pool_handler.config,
2424 }
2425 )
2426 if oss_cluster_maint_notifications_handler:
2427 self.connection_kwargs.update(
2428 {
2429 "oss_cluster_maint_notifications_handler": oss_cluster_maint_notifications_handler,
2430 "maint_notifications_config": oss_cluster_maint_notifications_handler.config,
2431 }
2432 )
2433 # OSS cluster mode and pool-handler mode are mutually exclusive.
2434 # Drop any pool handler a default (RESP3 "auto") pool creation may
2435 # have wired so future connections are not configured with both.
2436 self.connection_kwargs.pop("maint_notifications_pool_handler", None)
2438 # Store original connection parameters for maintenance notifications.
2439 if self.connection_kwargs.get("orig_host_address", None) is None:
2440 # If orig_host_address is None it means we haven't
2441 # configured the original values yet
2442 self.connection_kwargs.update(
2443 {
2444 "orig_host_address": self.connection_kwargs.get("host"),
2445 "orig_socket_timeout": self.connection_kwargs.get(
2446 "socket_timeout", DEFAULT_SOCKET_TIMEOUT
2447 ),
2448 "orig_socket_connect_timeout": self.connection_kwargs.get(
2449 "socket_connect_timeout", DEFAULT_SOCKET_CONNECT_TIMEOUT
2450 ),
2451 }
2452 )
2454 async def _update_maint_notifications_configs_for_connections(
2455 self,
2456 maint_notifications_pool_handler: (
2457 AsyncMaintNotificationsPoolHandler | None
2458 ) = None,
2459 oss_cluster_maint_notifications_handler: (
2460 AsyncOSSMaintNotificationsHandler | None
2461 ) = None,
2462 ) -> None:
2463 """Update the maintenance notifications config for all connections in the pool."""
2464 async with self._get_pool_lock():
2465 for conn in list(self._get_free_connections()):
2466 if oss_cluster_maint_notifications_handler:
2467 conn.set_maint_notifications_cluster_handler_for_connection(
2468 oss_cluster_maint_notifications_handler
2469 )
2470 conn.maint_notifications_config = (
2471 oss_cluster_maint_notifications_handler.config
2472 )
2473 elif maint_notifications_pool_handler:
2474 conn.set_maint_notifications_pool_handler_for_connection(
2475 maint_notifications_pool_handler
2476 )
2477 conn.maint_notifications_config = (
2478 maint_notifications_pool_handler.config
2479 )
2480 else:
2481 raise ValueError(
2482 "Either maint_notifications_pool_handler or "
2483 "oss_cluster_maint_notifications_handler must be set"
2484 )
2485 await conn.disconnect()
2487 for conn in list(self._get_in_use_connections()):
2488 if oss_cluster_maint_notifications_handler:
2489 # Use set_maint_notifications_cluster_handler_for_connection
2490 # (not _configure_maintenance_notifications) so the parser is
2491 # obtained from the connection itself. _configure_* requires a
2492 # parser argument and would raise here; it would also reset the
2493 # connection's orig_* settings, which is wrong for an in-use
2494 # (active) connection. This mirrors the idle-connection branch
2495 # above and the pool-handler branches.
2496 conn.set_maint_notifications_cluster_handler_for_connection(
2497 oss_cluster_maint_notifications_handler
2498 )
2499 conn.maint_notifications_config = (
2500 oss_cluster_maint_notifications_handler.config
2501 )
2502 elif maint_notifications_pool_handler:
2503 conn.set_maint_notifications_pool_handler_for_connection(
2504 maint_notifications_pool_handler
2505 )
2506 conn.maint_notifications_config = (
2507 maint_notifications_pool_handler.config
2508 )
2509 else:
2510 raise ValueError(
2511 "Either maint_notifications_pool_handler or "
2512 "oss_cluster_maint_notifications_handler must be set"
2513 )
2514 if logger.isEnabledFor(logging.DEBUG):
2515 logger.debug(
2516 "Marking active connection for reconnect after config update "
2517 f"config update: {conn}, {conn.extract_connection_details()}"
2518 )
2519 conn.mark_for_reconnect()
2521 def _should_update_connection(
2522 self,
2523 conn: "AbstractConnection",
2524 matching_pattern: str = "connected_address",
2525 matching_address: str | None = None,
2526 matching_notification_hash: int | None = None,
2527 ) -> bool:
2528 """
2529 Check if the connection should be updated based on the matching criteria.
2530 """
2531 if matching_pattern == "connected_address":
2532 if matching_address and conn.getpeername() != matching_address:
2533 return False
2534 elif matching_pattern == "configured_address":
2535 if matching_address and conn.host != matching_address:
2536 return False
2537 elif matching_pattern == "notification_hash":
2538 if (
2539 matching_notification_hash is not None
2540 and conn.maintenance_notification_hash != matching_notification_hash
2541 ):
2542 return False
2543 return True
2545 def update_connection_settings(
2546 self,
2547 conn: "AsyncMaintNotificationsAbstractConnection",
2548 state: MaintenanceState | None = None,
2549 maintenance_notification_hash: int | None = None,
2550 host_address: str | None = None,
2551 relaxed_timeout: float | None = None,
2552 update_notification_hash: bool = False,
2553 reset_host_address: bool = False,
2554 reset_relaxed_timeout: bool = False,
2555 ) -> None:
2556 """
2557 Update the settings for a single connection.
2558 """
2559 if state:
2560 self._complete_maintenance_taken_over_by_handoff(conn, state)
2561 conn.maintenance_state = state
2563 if update_notification_hash:
2564 # update the notification hash only if requested
2565 conn.maintenance_notification_hash = maintenance_notification_hash
2567 if host_address is not None:
2568 conn.set_tmp_settings(tmp_host_address=host_address)
2570 if relaxed_timeout is not None:
2571 conn.set_tmp_settings(tmp_relaxed_timeout=relaxed_timeout)
2573 if reset_relaxed_timeout or reset_host_address:
2574 conn.reset_tmp_settings(
2575 reset_host_address=reset_host_address,
2576 reset_relaxed_timeout=reset_relaxed_timeout,
2577 )
2579 conn.update_current_socket_timeout(relaxed_timeout)
2581 def _complete_maintenance_taken_over_by_handoff(
2582 self, conn: "AsyncMaintNotificationsAbstractConnection", state: MaintenanceState
2583 ) -> None:
2584 """
2585 Report the MAINTENANCE relaxation of a connection a handoff is taking
2586 over as completed. The connection's completion notification is skipped
2587 while it is MOVING, and the handoff's own cleanup restores it without
2588 looking at what it superseded, so this is the only point at which the
2589 pair opened by the start event can be closed.
2590 """
2591 if (
2592 state is not MaintenanceState.MOVING
2593 or conn.maintenance_state is not MaintenanceState.MAINTENANCE
2594 or conn.maint_notifications_config is None
2595 ):
2596 return
2597 _dispatch_maintenance_event(
2598 conn.maint_notifications_config,
2599 MaintenanceCompletedEvent(
2600 connection_pool=self,
2601 connection=conn,
2602 state=MaintenanceState.MAINTENANCE,
2603 notification=None,
2604 config=conn.maint_notifications_config,
2605 ),
2606 )
2608 async def update_connections_settings(
2609 self,
2610 state: MaintenanceState | None = None,
2611 maintenance_notification_hash: int | None = None,
2612 host_address: str | None = None,
2613 relaxed_timeout: float | None = None,
2614 matching_address: str | None = None,
2615 matching_notification_hash: int | None = None,
2616 matching_pattern: Literal[
2617 "connected_address", "configured_address", "notification_hash"
2618 ] = "connected_address",
2619 update_notification_hash: bool = False,
2620 reset_host_address: bool = False,
2621 reset_relaxed_timeout: bool = False,
2622 include_free_connections: bool = True,
2623 ) -> None:
2624 """
2625 Update the settings for all matching connections in the pool.
2627 This method does not create new connections.
2628 This method does not affect the connection kwargs.
2630 :param state: The maintenance state to set for the connection.
2631 :param maintenance_notification_hash: The hash of the maintenance notification
2632 to set for the connection.
2633 :param host_address: The host address to set for the connection.
2634 :param relaxed_timeout: The relaxed timeout to set for the connection.
2635 :param matching_address: The address to match for the connection.
2636 :param matching_notification_hash: The notification hash to match for the connection.
2637 :param matching_pattern: The pattern to match for the connection.
2638 :param update_notification_hash: Whether to update the notification hash for the connection.
2639 :param reset_host_address: Whether to reset the host address to the original address.
2640 :param reset_relaxed_timeout: Whether to reset the relaxed timeout to the original timeout.
2641 :param include_free_connections: Whether to include free/available connections.
2642 """
2643 async with self._get_pool_lock():
2644 self._update_connections_settings_without_locking(
2645 state=state,
2646 maintenance_notification_hash=maintenance_notification_hash,
2647 host_address=host_address,
2648 relaxed_timeout=relaxed_timeout,
2649 matching_address=matching_address,
2650 matching_notification_hash=matching_notification_hash,
2651 matching_pattern=matching_pattern,
2652 update_notification_hash=update_notification_hash,
2653 reset_host_address=reset_host_address,
2654 reset_relaxed_timeout=reset_relaxed_timeout,
2655 include_free_connections=include_free_connections,
2656 )
2658 def _update_connections_settings_without_locking(
2659 self,
2660 state: MaintenanceState | None = None,
2661 maintenance_notification_hash: int | None = None,
2662 host_address: str | None = None,
2663 relaxed_timeout: float | None = None,
2664 matching_address: str | None = None,
2665 matching_notification_hash: int | None = None,
2666 matching_pattern: Literal[
2667 "connected_address", "configured_address", "notification_hash"
2668 ] = "connected_address",
2669 update_notification_hash: bool = False,
2670 reset_host_address: bool = False,
2671 reset_relaxed_timeout: bool = False,
2672 include_free_connections: bool = True,
2673 ) -> None:
2674 """
2675 Update matching connections while the caller already holds the pool lock.
2677 This helper intentionally does not acquire the pool lock so callers can
2678 compose several pool mutations inside one critical section without
2679 deadlocking the non-reentrant `asyncio.Lock`.
2680 """
2681 for conn in self._get_in_use_connections():
2682 if self._should_update_connection(
2683 conn,
2684 matching_pattern,
2685 matching_address,
2686 matching_notification_hash,
2687 ):
2688 self.update_connection_settings(
2689 conn,
2690 state=state,
2691 maintenance_notification_hash=maintenance_notification_hash,
2692 host_address=host_address,
2693 relaxed_timeout=relaxed_timeout,
2694 update_notification_hash=update_notification_hash,
2695 reset_host_address=reset_host_address,
2696 reset_relaxed_timeout=reset_relaxed_timeout,
2697 )
2699 if include_free_connections:
2700 for conn in self._get_free_connections():
2701 if self._should_update_connection(
2702 conn,
2703 matching_pattern,
2704 matching_address,
2705 matching_notification_hash,
2706 ):
2707 self.update_connection_settings(
2708 conn,
2709 state=state,
2710 maintenance_notification_hash=maintenance_notification_hash,
2711 host_address=host_address,
2712 relaxed_timeout=relaxed_timeout,
2713 update_notification_hash=update_notification_hash,
2714 reset_host_address=reset_host_address,
2715 reset_relaxed_timeout=reset_relaxed_timeout,
2716 )
2718 def update_connection_kwargs(self, **kwargs: Any) -> None:
2719 """
2720 Update the connection kwargs for all future connections.
2722 This method updates the connection kwargs for all future connections created by the pool.
2723 Existing connections are not affected.
2724 """
2725 self.connection_kwargs.update(kwargs)
2727 async def apply_moving_notification(
2728 self,
2729 notification: NodeMovingNotification,
2730 config: MaintNotificationsConfig,
2731 moving_address_src: str | None,
2732 run_proactive_reconnect: bool = False,
2733 ) -> None:
2734 """
2735 Apply the pool state transition for a MOVING notification atomically.
2737 Async pools use a non-reentrant `asyncio.Lock`, so the handler cannot
2738 safely compose several separately locked calls. Existing connection
2739 updates, optional proactive reconnect, and future `connection_kwargs`
2740 changes must happen under one pool-owned lock; otherwise a connection
2741 can move between active/free lists and escape handling.
2742 """
2743 if logger.isEnabledFor(logging.DEBUG):
2744 logger.debug(
2745 f"Applying MOVING notification to pool: {notification}, "
2746 f"moving address src: {moving_address_src}, "
2747 f"proactive reconnect: {run_proactive_reconnect}"
2748 )
2749 async with self._get_pool_lock():
2750 # Opt BlockingConnectionPool into serializing its get/release
2751 # with this critical section. Other pools do not define
2752 # set_in_maintenance and this is a no-op for them.
2753 self._set_in_maintenance(True)
2754 try:
2755 self._update_connections_settings_without_locking(
2756 state=MaintenanceState.MOVING,
2757 maintenance_notification_hash=hash(notification),
2758 relaxed_timeout=config.relaxed_timeout,
2759 host_address=notification.new_node_host,
2760 matching_address=moving_address_src,
2761 matching_pattern="connected_address",
2762 update_notification_hash=True,
2763 include_free_connections=True,
2764 )
2766 if run_proactive_reconnect:
2767 await self._run_proactive_reconnect_without_locking(
2768 moving_address_src
2769 )
2771 self.update_connection_kwargs(
2772 **_build_moving_connection_kwargs(notification, config)
2773 )
2774 finally:
2775 self._set_in_maintenance(False)
2777 async def run_proactive_reconnect(
2778 self,
2779 moving_address_src: str | None = None,
2780 ) -> None:
2781 """
2782 Mark active connections and disconnect free connections atomically.
2784 This operation is pool-owned because the active/free lists can change
2785 while tasks acquire or release connections. Keeping the mark/disconnect
2786 pass under one lock avoids a connection moving between lists between
2787 separately locked calls.
2788 """
2789 async with self._get_pool_lock():
2790 await self._run_proactive_reconnect_without_locking(moving_address_src)
2792 async def _run_proactive_reconnect_without_locking(
2793 self,
2794 moving_address_src: str | None = None,
2795 ) -> None:
2796 """
2797 Mark and disconnect matching connections while the caller holds the pool lock.
2799 This helper intentionally does not acquire the pool lock so it can be
2800 reused by larger atomic operations that already hold the non-reentrant
2801 `asyncio.Lock`.
2802 """
2803 debug = logger.isEnabledFor(logging.DEBUG)
2804 for conn in self._get_in_use_connections():
2805 if self._should_update_connection(
2806 conn, "connected_address", moving_address_src
2807 ):
2808 if debug:
2809 logger.debug(
2810 f"Marking active connection for reconnect: {conn}, "
2811 f"{conn.extract_connection_details()}"
2812 )
2813 conn.mark_for_reconnect()
2815 free_connections = [
2816 conn
2817 for conn in self._get_free_connections()
2818 if self._should_update_connection(
2819 conn, "connected_address", moving_address_src
2820 )
2821 ]
2822 if debug:
2823 for conn in free_connections:
2824 logger.debug(
2825 f"Disconnecting free connection: {conn}, "
2826 f"{conn.extract_connection_details()}"
2827 )
2828 await self._disconnect_connections(free_connections)
2830 async def cleanup_moving_notification(
2831 self,
2832 notification_hash: int,
2833 reset_relaxed_timeout: bool,
2834 reset_host_address: bool,
2835 ) -> bool:
2836 """
2837 Revert MOVING pool state atomically after the notification TTL.
2839 Future connection kwargs and existing connection state must be cleaned
2840 up in the same critical section. Splitting the cleanup lets an
2841 acquire/release interleave, which can leave stale MOVING state or undo a
2842 newer overlapping MOVING notification.
2844 Returns True when the pool's connection kwargs were reverted, False when
2845 a newer MOVING notification had superseded this one and they were left
2846 for it to revert.
2847 """
2848 if logger.isEnabledFor(logging.DEBUG):
2849 logger.debug(
2850 "Cleaning up MOVING pool state for notification hash "
2851 f"{notification_hash}, reset_relaxed_timeout="
2852 f"{reset_relaxed_timeout}, reset_host_address={reset_host_address}"
2853 )
2854 async with self._get_pool_lock():
2855 kwargs = _build_moving_cleanup_connection_kwargs(
2856 self.connection_kwargs, notification_hash
2857 )
2858 if kwargs is not None:
2859 self.update_connection_kwargs(**kwargs)
2861 self._update_connections_settings_without_locking(
2862 relaxed_timeout=-1,
2863 state=MaintenanceState.NONE,
2864 maintenance_notification_hash=None,
2865 matching_notification_hash=notification_hash,
2866 matching_pattern="notification_hash",
2867 update_notification_hash=True,
2868 reset_relaxed_timeout=reset_relaxed_timeout,
2869 reset_host_address=reset_host_address,
2870 include_free_connections=True,
2871 )
2872 return kwargs is not None
2874 async def _disconnect_connections(
2875 self, connections: Iterable["AbstractConnection"]
2876 ) -> None:
2877 connections = tuple(connections)
2878 if not connections:
2879 return
2880 results = await asyncio.gather(
2881 *(connection.disconnect() for connection in connections),
2882 return_exceptions=True,
2883 )
2884 exc = next(
2885 (result for result in results if isinstance(result, BaseException)), None
2886 )
2887 if exc:
2888 raise exc
2890 def _set_in_maintenance(self, in_maintenance: bool) -> None:
2891 """Flip the pool's maintenance flag if it exposes one (BlockingConnectionPool)."""
2892 set_in_maintenance = getattr(self, "set_in_maintenance", None)
2893 if callable(set_in_maintenance):
2894 set_in_maintenance(in_maintenance)
2897class ConnectionPool(
2898 AsyncMaintNotificationsAbstractConnectionPool, ConnectionPoolInterface
2899):
2900 """
2901 Create a connection pool. ``If max_connections`` is set, then this
2902 object raises :py:class:`~redis.ConnectionError` when the pool's
2903 limit is reached.
2905 By default, TCP connections are created unless ``connection_class``
2906 is specified. Use :py:class:`~redis.UnixDomainSocketConnection` for
2907 unix sockets.
2908 :py:class:`~redis.SSLConnection` can be used for SSL enabled connections.
2910 Any additional keyword arguments are passed to the constructor of
2911 ``connection_class``.
2912 """
2914 @classmethod
2915 def from_url(cls: Type[_CP], url: str, **kwargs) -> _CP:
2916 """
2917 Return a connection pool configured from the given URL.
2919 For example::
2921 redis://[[username]:[password]]@localhost:6379/0
2922 rediss://[[username]:[password]]@localhost:6379/0
2923 unix://[username@]/path/to/socket.sock?db=0[&password=password]
2925 Three URL schemes are supported:
2927 - `redis://` creates a TCP socket connection. See more at:
2928 <https://www.iana.org/assignments/uri-schemes/prov/redis>
2929 - `rediss://` creates a SSL wrapped TCP socket connection. See more at:
2930 <https://www.iana.org/assignments/uri-schemes/prov/rediss>
2931 - ``unix://``: creates a Unix Domain Socket connection.
2933 The username, password, hostname and path are passed through
2934 urllib.parse.unquote in order to replace any percent-encoded values
2935 with their corresponding characters. Querystring values are decoded
2936 by urllib.parse.parse_qs and are not unquoted again.
2938 There are several ways to specify a database number. The first value
2939 found will be used:
2941 1. A ``db`` querystring option, e.g. redis://localhost?db=0
2943 2. If using the redis:// or rediss:// schemes, the path argument
2944 of the url, e.g. redis://localhost/0
2946 3. A ``db`` keyword argument to this function.
2948 If none of these options are specified, the default db=0 is used.
2950 All querystring options are cast to their appropriate Python types.
2951 Boolean arguments can be specified with string values "True"/"False"
2952 or "Yes"/"No". Values that cannot be properly cast cause a
2953 ``ValueError`` to be raised. Once parsed, the querystring arguments
2954 and keyword arguments are passed to the ``ConnectionPool``'s
2955 class initializer. In the case of conflicting arguments, querystring
2956 arguments always win.
2957 """
2958 url_options = parse_url(url)
2960 if "connection_class" in kwargs:
2961 url_options["connection_class"] = kwargs["connection_class"]
2963 kwargs.update(url_options)
2964 return cls(**kwargs)
2966 def __init__(
2967 self,
2968 connection_class: Type[AbstractConnection] = Connection,
2969 max_connections: Optional[int] = None,
2970 maint_notifications_config: MaintNotificationsConfig | None = None,
2971 **connection_kwargs,
2972 ):
2973 max_connections = max_connections or 100
2974 if not isinstance(max_connections, int) or max_connections < 0:
2975 raise ValueError('"max_connections" must be a positive integer')
2977 retry = connection_kwargs.get("retry")
2978 if retry is not None:
2979 connection_kwargs["retry"] = _to_async_retry(retry)
2981 self.connection_class = connection_class
2982 self._connection_kwargs = connection_kwargs
2983 self.max_connections = max_connections
2985 # Resolve the HIMPORT registry. A pre-built ``himport_registry`` (shared, e.g.
2986 # from the cluster client) takes precedence; otherwise build a fresh empty one.
2987 # A registry always exists so runtime ``himport_prepare`` mutates a single object
2988 # every connection already shares. The object stays in ``connection_kwargs`` so
2989 # it reaches every connection. It is injected unconditionally (like other
2990 # auto-added pool kwargs), so a custom ``connection_class`` must accept
2991 # ``**kwargs`` (or a ``himport_registry`` parameter), as built-ins do.
2992 himport_registry = connection_kwargs.get("himport_registry")
2993 if himport_registry is None:
2994 himport_registry = HImportRegistry()
2995 connection_kwargs["himport_registry"] = himport_registry
2996 self.himport_registry = himport_registry
2998 self._available_connections: List[AbstractConnection] = []
2999 self._in_use_connections: Set[AbstractConnection] = set()
3000 self.encoder_class = self.connection_kwargs.get("encoder_class", Encoder)
3001 self._lock = asyncio.Lock()
3002 self._event_dispatcher = self.connection_kwargs.get("event_dispatcher", None)
3003 if self._event_dispatcher is None:
3004 self._event_dispatcher = EventDispatcher()
3006 AsyncMaintNotificationsAbstractConnectionPool.__init__(
3007 self,
3008 maint_notifications_config=maint_notifications_config,
3009 **connection_kwargs,
3010 )
3012 # Keys that should be redacted in __repr__ to avoid exposing sensitive information
3013 SENSITIVE_REPR_KEYS = frozenset(
3014 {
3015 "password",
3016 "username",
3017 "ssl_password",
3018 "credential_provider",
3019 }
3020 )
3022 # Internal plumbing kwargs omitted from __repr__ (not user-facing config).
3023 OMIT_REPR_KEYS = frozenset({"himport_registry"})
3025 def __repr__(self):
3026 conn_kwargs = ",".join(
3027 [
3028 f"{k}={'<REDACTED>' if k in self.SENSITIVE_REPR_KEYS else v}"
3029 for k, v in self.connection_kwargs.items()
3030 if k not in self.OMIT_REPR_KEYS
3031 ]
3032 )
3033 return (
3034 f"<{self.__class__.__module__}.{self.__class__.__name__}"
3035 f"(<{self.connection_class.__module__}.{self.connection_class.__name__}"
3036 f"({conn_kwargs})>)>"
3037 )
3039 @property
3040 def connection_kwargs(self) -> dict[str, Any]:
3041 return self._connection_kwargs
3043 @connection_kwargs.setter
3044 def connection_kwargs(self, value: dict[str, Any]) -> None:
3045 self._connection_kwargs = value
3047 def _get_pool_lock(self) -> asyncio.Lock:
3048 return self._lock
3050 def _get_free_connections(self) -> Iterable[AbstractConnection]:
3051 return self._available_connections
3053 def _get_in_use_connections(self) -> Iterable[AbstractConnection]:
3054 return self._in_use_connections
3056 def get_protocol(self):
3057 """
3058 Returns:
3059 The RESP protocol version, or ``None`` if the protocol is not specified,
3060 in which case the server default will be used.
3061 """
3062 return self.connection_kwargs.get("protocol", None)
3064 def reset(self):
3065 # Record metrics for connections being removed before clearing
3066 # (only if attributes exist - they won't during __init__)
3067 if hasattr(self, "_available_connections") and hasattr(
3068 self, "_in_use_connections"
3069 ):
3070 idle_count = len(self._available_connections)
3071 in_use_count = len(self._in_use_connections)
3072 if idle_count > 0 or in_use_count > 0:
3073 pool_name = get_pool_name(self)
3074 # Note: Using sync version since reset() is sync
3075 from redis.observability.recorder import (
3076 record_connection_count as sync_record_connection_count,
3077 )
3079 if idle_count > 0:
3080 sync_record_connection_count(
3081 pool_name=pool_name,
3082 connection_state=ConnectionState.IDLE,
3083 counter=-idle_count,
3084 )
3085 if in_use_count > 0:
3086 sync_record_connection_count(
3087 pool_name=pool_name,
3088 connection_state=ConnectionState.USED,
3089 counter=-in_use_count,
3090 )
3092 self._available_connections = []
3093 self._in_use_connections = weakref.WeakSet()
3095 def __del__(self) -> None:
3096 """Clean up connection pool and record metrics when garbage collected."""
3097 try:
3098 if not hasattr(self, "_available_connections") or not hasattr(
3099 self, "_in_use_connections"
3100 ):
3101 return
3102 idle_count = len(self._available_connections)
3103 in_use_count = len(self._in_use_connections)
3104 if idle_count > 0 or in_use_count > 0:
3105 pool_name = get_pool_name(self)
3106 # Note: Using sync version since __del__ is sync
3107 from redis.observability.recorder import (
3108 record_connection_count as sync_record_connection_count,
3109 )
3111 if idle_count > 0:
3112 sync_record_connection_count(
3113 pool_name=pool_name,
3114 connection_state=ConnectionState.IDLE,
3115 counter=-idle_count,
3116 )
3117 if in_use_count > 0:
3118 sync_record_connection_count(
3119 pool_name=pool_name,
3120 connection_state=ConnectionState.USED,
3121 counter=-in_use_count,
3122 )
3123 except Exception:
3124 pass
3126 def can_get_connection(self) -> bool:
3127 """Return True if a connection can be retrieved from the pool."""
3128 return (
3129 self._available_connections
3130 or len(self._in_use_connections) < self.max_connections
3131 )
3133 @deprecated_args(
3134 args_to_warn=["*"],
3135 reason="Use get_connection() without args instead",
3136 version="5.3.0",
3137 )
3138 async def get_connection(self, command_name=None, *keys, **options):
3139 """Get a connected connection from the pool"""
3140 # Track connection count before to detect if a new connection is created
3141 async with self._lock:
3142 connections_before = len(self._available_connections) + len(
3143 self._in_use_connections
3144 )
3145 start_time_created = time.monotonic()
3146 connection = self.get_available_connection()
3147 connections_after = len(self._available_connections) + len(
3148 self._in_use_connections
3149 )
3150 is_created = connections_after > connections_before
3152 # Record state transition for observability
3153 # This ensures counters stay balanced if ensure_connection() fails and release() is called
3154 pool_name = get_pool_name(self)
3155 if is_created:
3156 # New connection created and acquired: just USED +1
3157 await record_connection_count(
3158 pool_name=pool_name,
3159 connection_state=ConnectionState.USED,
3160 counter=1,
3161 )
3162 else:
3163 # Existing connection acquired from pool: IDLE -> USED
3164 await record_connection_count(
3165 pool_name=pool_name,
3166 connection_state=ConnectionState.IDLE,
3167 counter=-1,
3168 )
3169 await record_connection_count(
3170 pool_name=pool_name,
3171 connection_state=ConnectionState.USED,
3172 counter=1,
3173 )
3175 # We now perform the connection check outside of the lock.
3176 try:
3177 await self.ensure_connection(connection)
3179 if is_created:
3180 await record_connection_create_time(
3181 connection_pool=self,
3182 duration_seconds=time.monotonic() - start_time_created,
3183 )
3185 return connection
3186 except BaseException:
3187 await self.release(connection)
3188 raise
3190 def get_available_connection(self):
3191 """Get a connection from the pool, without making sure it is connected"""
3192 try:
3193 connection = self._available_connections.pop()
3194 except IndexError:
3195 if len(self._in_use_connections) >= self.max_connections:
3196 raise MaxConnectionsError("Too many connections") from None
3197 connection = self.make_connection()
3198 self._in_use_connections.add(connection)
3199 return connection
3201 def get_encoder(self):
3202 """Return an encoder based on encoding settings"""
3203 kwargs = self.connection_kwargs
3204 return self.encoder_class(
3205 encoding=kwargs.get("encoding", "utf-8"),
3206 encoding_errors=kwargs.get("encoding_errors", "strict"),
3207 decode_responses=kwargs.get("decode_responses", False),
3208 )
3210 def make_connection(self):
3211 """Create a new connection. Can be overridden by child classes."""
3212 # Note: We don't record IDLE here because async uses a sync make_connection
3213 # but async record_connection_count. The recording is handled in get_connection.
3214 return self.connection_class(**self.connection_kwargs)
3216 async def ensure_connection(self, connection: AbstractConnection):
3217 """Ensure that the connection object is connected and valid"""
3218 await connection.connect()
3219 # connections that the pool provides should be ready to send
3220 # a command. if not, the connection was either returned to the
3221 # pool before all data has been read or the socket has been
3222 # closed. either way, reconnect and verify everything is good.
3223 # With maintenance notifications enabled the pending data may be
3224 # push notifications that have to be applied rather than a reason
3225 # to reconnect; reading them is what tells the two apart, and it
3226 # raises for a socket the server has closed in the meantime.
3227 try:
3228 if await connection.can_read():
3229 if self.maint_notifications_enabled():
3230 await connection.handle_pending_push_notifications()
3231 else:
3232 raise ConnectionError("Connection has data") from None
3233 except (ConnectionError, TimeoutError, OSError):
3234 await connection.disconnect()
3235 await connection.connect()
3236 if await connection.can_read() and not self.maint_notifications_enabled():
3237 raise ConnectionError("Connection not ready") from None
3239 async def release(self, connection: AbstractConnection):
3240 """Releases the connection back to the pool"""
3241 # Connections should always be returned to the correct pool,
3242 # not doing so is an error that will cause an exception here.
3243 async with self._lock:
3244 self._in_use_connections.remove(connection)
3246 if connection.should_reconnect():
3247 if logger.isEnabledFor(logging.DEBUG):
3248 logger.debug(
3249 "Disconnecting released connection marked for reconnect: "
3250 f"{connection}, {connection.extract_connection_details()}"
3251 )
3252 await connection.disconnect()
3254 self._available_connections.append(connection)
3256 await self._event_dispatcher.dispatch_async(
3257 AsyncAfterConnectionReleasedEvent(connection)
3258 )
3260 # Record state transition: USED -> IDLE
3261 pool_name = get_pool_name(self)
3262 await record_connection_count(
3263 pool_name=pool_name,
3264 connection_state=ConnectionState.USED,
3265 counter=-1,
3266 )
3267 await record_connection_count(
3268 pool_name=pool_name,
3269 connection_state=ConnectionState.IDLE,
3270 counter=1,
3271 )
3273 async def disconnect(self, inuse_connections: bool = True):
3274 """
3275 Disconnects connections in the pool
3277 If ``inuse_connections`` is True, disconnect connections that are
3278 current in use, potentially by other tasks. Otherwise only disconnect
3279 connections that are idle in the pool.
3280 """
3281 if inuse_connections:
3282 connections: Iterable[AbstractConnection] = chain(
3283 self._available_connections, self._in_use_connections
3284 )
3285 else:
3286 connections = self._available_connections
3287 resp = await asyncio.gather(
3288 *(connection.disconnect() for connection in connections),
3289 return_exceptions=True,
3290 )
3292 exc = next((r for r in resp if isinstance(r, BaseException)), None)
3293 if exc:
3294 raise exc
3296 async def update_active_connections_for_reconnect(self):
3297 """
3298 Mark all active connections for reconnect.
3299 """
3300 debug = logger.isEnabledFor(logging.DEBUG)
3301 async with self._lock:
3302 for conn in self._in_use_connections:
3303 if debug:
3304 logger.debug(
3305 f"Marking active connection for reconnect: {conn}, "
3306 f"{conn.extract_connection_details()}"
3307 )
3308 conn.mark_for_reconnect()
3310 async def aclose(self) -> None:
3311 """Close the pool, disconnecting all connections"""
3312 await self._on_close()
3313 await self.disconnect()
3315 async def __aenter__(self: _CP) -> _CP:
3316 return self
3318 async def __aexit__(self, exc_type, exc_value, traceback) -> None:
3319 await self.aclose()
3321 def set_retry(self, retry: "Retry") -> None:
3322 retry = _to_async_retry(retry)
3323 self.connection_kwargs["retry"] = retry
3324 for conn in self._available_connections:
3325 conn.retry = retry
3326 for conn in self._in_use_connections:
3327 conn.retry = retry
3329 async def re_auth_callback(self, token: TokenInterface):
3330 async with self._lock:
3331 for conn in self._available_connections:
3332 await conn.retry.call_with_retry(
3333 lambda: conn.send_command(
3334 "AUTH", token.try_get("oid"), token.get_value()
3335 ),
3336 lambda error: self._mock(error),
3337 )
3338 await conn.retry.call_with_retry(
3339 lambda: conn.read_response(), lambda error: self._mock(error)
3340 )
3341 for conn in self._in_use_connections:
3342 conn.set_re_auth_token(token)
3344 async def _mock(self, error: RedisError):
3345 """
3346 Dummy functions, needs to be passed as error callback to retry object.
3347 :param error:
3348 :return:
3349 """
3350 pass
3352 def get_connection_count(self) -> List[tuple[int, dict]]:
3353 """
3354 Returns a connection count (both idle and in use).
3355 """
3356 attributes = AttributeBuilder.build_base_attributes()
3357 attributes[DB_CLIENT_CONNECTION_POOL_NAME] = get_pool_name(self)
3358 free_connections_attributes = attributes.copy()
3359 in_use_connections_attributes = attributes.copy()
3361 free_connections_attributes[DB_CLIENT_CONNECTION_STATE] = (
3362 ConnectionState.IDLE.value
3363 )
3364 in_use_connections_attributes[DB_CLIENT_CONNECTION_STATE] = (
3365 ConnectionState.USED.value
3366 )
3368 return [
3369 (len(self._available_connections), free_connections_attributes),
3370 (len(self._in_use_connections), in_use_connections_attributes),
3371 ]
3374class BlockingConnectionPool(ConnectionPool):
3375 """
3376 A blocking connection pool::
3378 >>> from redis.asyncio import Redis, BlockingConnectionPool
3379 >>> client = Redis.from_pool(BlockingConnectionPool())
3381 It performs the same function as the default
3382 :py:class:`~redis.asyncio.ConnectionPool` implementation, in that,
3383 it maintains a pool of reusable connections that can be shared by
3384 multiple async redis clients.
3386 The difference is that, in the event that a client tries to get a
3387 connection from the pool when all of connections are in use, rather than
3388 raising a :py:class:`~redis.ConnectionError` (as the default
3389 :py:class:`~redis.asyncio.ConnectionPool` implementation does), it
3390 blocks the current `Task` for a specified number of seconds until
3391 a connection becomes available.
3393 Use ``max_connections`` to increase / decrease the pool size::
3395 >>> pool = BlockingConnectionPool(max_connections=10)
3397 Use ``timeout`` to tell it either how many seconds to wait for a connection
3398 to become available, or to block forever:
3400 >>> # Block forever.
3401 >>> pool = BlockingConnectionPool(timeout=None)
3403 >>> # Raise a ``ConnectionError`` after five seconds if a connection is
3404 >>> # not available.
3405 >>> pool = BlockingConnectionPool(timeout=5)
3406 """
3408 def __init__(
3409 self,
3410 max_connections: int = 50,
3411 timeout: Optional[float] = 20,
3412 connection_class: Type[AbstractConnection] = Connection,
3413 queue_class: Type[asyncio.Queue] = asyncio.LifoQueue, # deprecated
3414 **connection_kwargs,
3415 ):
3416 super().__init__(
3417 connection_class=connection_class,
3418 max_connections=max_connections,
3419 **connection_kwargs,
3420 )
3421 self._condition = asyncio.Condition()
3422 self.timeout = timeout
3423 self._in_maintenance = False
3425 def set_in_maintenance(self, in_maintenance: bool) -> None:
3426 """
3427 Toggle the pool's maintenance mode.
3429 While maintenance mode is on, ``get_connection`` and ``release``
3430 serialize their pool mutations through ``self._lock`` so they cannot
3431 interleave with a MOVING notification handler that is currently
3432 rewriting pool state under the same lock. Outside of maintenance the
3433 mutations skip the lock, since their critical sections are pure-Python
3434 and already atomic under asyncio's single-threaded scheduling.
3435 """
3436 self._in_maintenance = in_maintenance
3438 @contextlib.asynccontextmanager
3439 async def _maybe_pool_lock(self) -> AsyncIterator[None]:
3440 if self._in_maintenance:
3441 async with self._lock:
3442 yield
3443 else:
3444 yield
3446 @deprecated_args(
3447 args_to_warn=["*"],
3448 reason="Use get_connection() without args instead",
3449 version="5.3.0",
3450 )
3451 async def get_connection(self, command_name=None, *keys, **options):
3452 """Gets a connection from the pool, blocking until one is available"""
3453 # Start timing for wait time observability
3454 start_time_acquired = time.monotonic()
3456 try:
3457 async with self._condition:
3458 async with async_timeout(self.timeout):
3459 await self._condition.wait_for(self.can_get_connection)
3460 async with self._maybe_pool_lock():
3461 # Track connection count before to detect if a new connection is created
3462 connections_before = len(self._available_connections) + len(
3463 self._in_use_connections
3464 )
3465 start_time_created = time.monotonic()
3466 connection = super().get_available_connection()
3467 connections_after = len(self._available_connections) + len(
3468 self._in_use_connections
3469 )
3470 is_created = connections_after > connections_before
3471 except asyncio.TimeoutError as err:
3472 await record_connection_timeout(pool_name=get_pool_name(self))
3473 raise ConnectionError("No connection available.") from err
3475 # Record state transition for observability.
3476 pool_name = get_pool_name(self)
3477 if is_created:
3478 # New connection created and acquired: just USED +1
3479 await record_connection_count(
3480 pool_name=pool_name,
3481 connection_state=ConnectionState.USED,
3482 counter=1,
3483 )
3484 else:
3485 # Existing connection acquired from pool: IDLE -> USED
3486 await record_connection_count(
3487 pool_name=pool_name,
3488 connection_state=ConnectionState.IDLE,
3489 counter=-1,
3490 )
3491 await record_connection_count(
3492 pool_name=pool_name,
3493 connection_state=ConnectionState.USED,
3494 counter=1,
3495 )
3497 # We now perform the connection check outside of the lock.
3498 try:
3499 await self.ensure_connection(connection)
3501 if is_created:
3502 await record_connection_create_time(
3503 connection_pool=self,
3504 duration_seconds=time.monotonic() - start_time_created,
3505 )
3507 await record_connection_wait_time(
3508 pool_name=get_pool_name(self),
3509 duration_seconds=time.monotonic() - start_time_acquired,
3510 )
3512 return connection
3513 except BaseException:
3514 await self.release(connection)
3515 raise
3517 async def release(self, connection: AbstractConnection):
3518 """Releases the connection back to the pool."""
3519 async with self._condition:
3520 await super().release(connection)
3521 self._condition.notify()