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 copy
3import inspect
4import math
5import socket
6import sys
7import time
8import warnings
9import weakref
10from abc import ABC, abstractmethod
11from itertools import chain
12from types import MappingProxyType
13from typing import (
14 Any,
15 Callable,
16 Iterable,
17 List,
18 Mapping,
19 Optional,
20 Protocol,
21 Set,
22 Tuple,
23 Type,
24 TypedDict,
25 TypeVar,
26 Union,
27)
28from urllib.parse import ParseResult, parse_qs, unquote, urlparse
30from ..observability.attributes import (
31 DB_CLIENT_CONNECTION_POOL_NAME,
32 DB_CLIENT_CONNECTION_STATE,
33 AttributeBuilder,
34 ConnectionState,
35 get_pool_name,
36)
37from ..utils import SSL_AVAILABLE, deprecated_function
39if SSL_AVAILABLE:
40 import ssl
41 from ssl import SSLContext, TLSVersion, VerifyFlags
42else:
43 ssl = None
44 TLSVersion = None
45 SSLContext = None
46 VerifyFlags = None
48from ..auth.token import TokenInterface
49from ..driver_info import DriverInfo, resolve_driver_info
50from ..event import AsyncAfterConnectionReleasedEvent, EventDispatcher
51from ..utils import deprecated_args, format_error_message
53# the functionality is available in 3.11.x but has a major issue before
54# 3.11.3. See https://github.com/redis/redis-py/issues/2633
55if sys.version_info >= (3, 11, 3):
56 from asyncio import timeout as async_timeout
57else:
58 from async_timeout import timeout as async_timeout
60from redis.asyncio.observability.recorder import (
61 record_connection_closed,
62 record_connection_count,
63 record_connection_create_time,
64 record_connection_wait_time,
65 record_error_count,
66)
67from redis.asyncio.retry import Retry
68from redis.backoff import NoBackoff
69from redis.credentials import CredentialProvider, UsernamePasswordCredentialProvider
70from redis.exceptions import (
71 AuthenticationError,
72 AuthenticationWrongNumberOfArgsError,
73 ConnectionError,
74 DataError,
75 MaxConnectionsError,
76 RedisError,
77 ResponseError,
78 TimeoutError,
79)
80from redis.observability.metrics import CloseReason
81from redis.typing import EncodableT
82from redis.utils import (
83 DEFAULT_RESP_VERSION,
84 HIREDIS_AVAILABLE,
85 SENTINEL,
86 str_if_bytes,
87)
89from .._defaults import (
90 DEFAULT_SOCKET_CONNECT_TIMEOUT,
91 DEFAULT_SOCKET_READ_SIZE,
92 DEFAULT_SOCKET_TIMEOUT,
93 get_default_socket_keepalive_options,
94)
95from .._parsers import (
96 BaseParser,
97 Encoder,
98 _AsyncHiredisParser,
99 _AsyncRESP2Parser,
100 _AsyncRESP3Parser,
101)
103SYM_STAR = b"*"
104SYM_DOLLAR = b"$"
105SYM_CRLF = b"\r\n"
106SYM_LF = b"\n"
107SYM_EMPTY = b""
110DefaultParser: Type[Union[_AsyncRESP2Parser, _AsyncRESP3Parser, _AsyncHiredisParser]]
111if HIREDIS_AVAILABLE:
112 DefaultParser = _AsyncHiredisParser
113else:
114 DefaultParser = _AsyncRESP3Parser
117class ConnectCallbackProtocol(Protocol):
118 def __call__(self, connection: "AbstractConnection"): ...
121class AsyncConnectCallbackProtocol(Protocol):
122 async def __call__(self, connection: "AbstractConnection"): ...
125ConnectCallbackT = Union[ConnectCallbackProtocol, AsyncConnectCallbackProtocol]
128class AbstractConnection:
129 """Manages communication to and from a Redis server"""
131 __slots__ = (
132 "db",
133 "username",
134 "client_name",
135 "lib_name",
136 "lib_version",
137 "credential_provider",
138 "password",
139 "socket_timeout",
140 "socket_connect_timeout",
141 "redis_connect_func",
142 "retry_on_timeout",
143 "retry_on_error",
144 "health_check_interval",
145 "next_health_check",
146 "last_active_at",
147 "encoder",
148 "ssl_context",
149 "protocol",
150 "_reader",
151 "_writer",
152 "_parser",
153 "_connect_callbacks",
154 "_buffer_cutoff",
155 "_lock",
156 "_socket_read_size",
157 "__dict__",
158 )
160 @deprecated_args(
161 args_to_warn=["lib_name", "lib_version"],
162 reason="Use 'driver_info' parameter instead. "
163 "lib_name and lib_version will be removed in a future version.",
164 )
165 def __init__(
166 self,
167 *,
168 db: str | int = 0,
169 password: str | None = None,
170 socket_timeout: float | None = DEFAULT_SOCKET_TIMEOUT,
171 socket_connect_timeout: float | None = DEFAULT_SOCKET_CONNECT_TIMEOUT,
172 retry_on_timeout: bool = False,
173 retry_on_error: list | object = SENTINEL,
174 encoding: str = "utf-8",
175 encoding_errors: str = "strict",
176 decode_responses: bool = False,
177 parser_class: Type[BaseParser] = DefaultParser,
178 socket_read_size: int = DEFAULT_SOCKET_READ_SIZE,
179 health_check_interval: float = 0,
180 client_name: str | None = None,
181 lib_name: str | object | None = SENTINEL,
182 lib_version: str | object | None = SENTINEL,
183 driver_info: DriverInfo | object | None = SENTINEL,
184 username: str | None = None,
185 retry: Retry | None = None,
186 redis_connect_func: ConnectCallbackT | None = None,
187 encoder_class: Type[Encoder] = Encoder,
188 credential_provider: CredentialProvider | None = None,
189 protocol: int | None = None,
190 legacy_responses: bool = True,
191 event_dispatcher: EventDispatcher | None = None,
192 ):
193 """
194 Initialize a new async Connection.
196 Parameters
197 ----------
198 driver_info : DriverInfo, optional
199 Driver metadata for CLIENT SETINFO. If provided, lib_name and lib_version
200 are ignored. If not provided, a DriverInfo will be created from lib_name
201 and lib_version. Explicit None disables CLIENT SETINFO.
202 lib_name : str, optional
203 **Deprecated.** Use driver_info instead. Library name for CLIENT SETINFO.
204 lib_version : str, optional
205 **Deprecated.** Use driver_info instead. Library version for CLIENT SETINFO.
206 """
207 if (username or password) and credential_provider is not None:
208 raise DataError(
209 "'username' and 'password' cannot be passed along with 'credential_"
210 "provider'. Please provide only one of the following arguments: \n"
211 "1. 'password' and (optional) 'username'\n"
212 "2. 'credential_provider'"
213 )
214 if event_dispatcher is None:
215 self._event_dispatcher = EventDispatcher()
216 else:
217 self._event_dispatcher = event_dispatcher
218 self.db = db
219 self.client_name = client_name
221 # Handle driver_info: if provided, use it; otherwise create from lib_name/lib_version.
222 self.driver_info = resolve_driver_info(driver_info, lib_name, lib_version)
224 self.credential_provider = credential_provider
225 self.password = password
226 self.username = username
227 self.socket_timeout = socket_timeout
228 if socket_connect_timeout is None:
229 socket_connect_timeout = socket_timeout
230 self.socket_connect_timeout = socket_connect_timeout
231 self.retry_on_timeout = retry_on_timeout
232 if retry_on_error is SENTINEL:
233 retry_on_error = []
234 if retry_on_timeout:
235 retry_on_error.append(TimeoutError)
236 retry_on_error.append(socket.timeout)
237 retry_on_error.append(asyncio.TimeoutError)
238 self.retry_on_error = retry_on_error
239 if retry or retry_on_error:
240 if not retry:
241 self.retry = Retry(NoBackoff(), 1)
242 else:
243 # deep-copy the Retry object as it is mutable
244 self.retry = copy.deepcopy(retry)
245 # Update the retry's supported errors with the specified errors
246 self.retry.update_supported_errors(retry_on_error)
247 else:
248 self.retry = Retry(NoBackoff(), 0)
249 self.health_check_interval = health_check_interval
250 self.next_health_check: float = -1
251 self.encoder = encoder_class(encoding, encoding_errors, decode_responses)
252 self.redis_connect_func = redis_connect_func
253 self._reader: Optional[asyncio.StreamReader] = None
254 self._writer: Optional[asyncio.StreamWriter] = None
255 self._socket_read_size = socket_read_size
256 self._connect_callbacks: List[weakref.WeakMethod[ConnectCallbackT]] = []
257 self._buffer_cutoff = 6000
258 self._re_auth_token: Optional[TokenInterface] = None
259 self._should_reconnect = False
261 try:
262 p = int(protocol)
263 except TypeError:
264 p = DEFAULT_RESP_VERSION
265 except ValueError:
266 raise ConnectionError("protocol must be an integer")
267 else:
268 if p < 2 or p > 3:
269 raise ConnectionError("protocol must be either 2 or 3")
270 self.protocol = p
271 self.legacy_responses = legacy_responses
272 if parser_class != _AsyncHiredisParser:
273 # The Python parsers are protocol-specific; hiredis supports both.
274 if self.protocol == 3 and parser_class == _AsyncRESP2Parser:
275 parser_class = _AsyncRESP3Parser
276 elif self.protocol == 2 and parser_class == _AsyncRESP3Parser:
277 parser_class = _AsyncRESP2Parser
278 self.set_parser(parser_class)
280 def __del__(self, _warnings: Any = warnings):
281 # For some reason, the individual streams don't get properly garbage
282 # collected and therefore produce no resource warnings. We add one
283 # here, in the same style as those from the stdlib.
284 if getattr(self, "_writer", None):
285 _warnings.warn(
286 f"unclosed Connection {self!r}", ResourceWarning, source=self
287 )
289 try:
290 asyncio.get_running_loop()
291 self._close()
292 except RuntimeError:
293 # No actions been taken if pool already closed.
294 pass
296 def _close(self):
297 """
298 Internal method to silently close the connection without waiting
299 """
300 if self._writer:
301 self._writer.close()
302 self._writer = self._reader = None
304 def __repr__(self):
305 repr_args = ",".join((f"{k}={v}" for k, v in self.repr_pieces()))
306 return f"<{self.__class__.__module__}.{self.__class__.__name__}({repr_args})>"
308 @abstractmethod
309 def repr_pieces(self):
310 pass
312 @property
313 def is_connected(self):
314 return self._reader is not None and self._writer is not None
316 def register_connect_callback(self, callback):
317 """
318 Register a callback to be called when the connection is established either
319 initially or reconnected. This allows listeners to issue commands that
320 are ephemeral to the connection, for example pub/sub subscription or
321 key tracking. The callback must be a _method_ and will be kept as
322 a weak reference.
323 """
324 wm = weakref.WeakMethod(callback)
325 if wm not in self._connect_callbacks:
326 self._connect_callbacks.append(wm)
328 def deregister_connect_callback(self, callback):
329 """
330 De-register a previously registered callback. It will no-longer receive
331 notifications on connection events. Calling this is not required when the
332 listener goes away, since the callbacks are kept as weak methods.
333 """
334 try:
335 self._connect_callbacks.remove(weakref.WeakMethod(callback))
336 except ValueError:
337 pass
339 def set_parser(self, parser_class: Type[BaseParser]) -> None:
340 """
341 Creates a new instance of parser_class with socket size:
342 _socket_read_size and assigns it to the parser for the connection
343 :param parser_class: The required parser class
344 """
345 self._parser = parser_class(socket_read_size=self._socket_read_size)
347 async def connect(self):
348 """Connects to the Redis server if not already connected"""
349 # try once the socket connect with the handshake, retry the whole
350 # connect/handshake flow based on retry policy
351 await self.retry.call_with_retry(
352 lambda: self.connect_check_health(
353 check_health=True, retry_socket_connect=False
354 ),
355 lambda error, failure_count: self.disconnect(
356 error=error, failure_count=failure_count
357 ),
358 with_failure_count=True,
359 )
361 async def connect_check_health(
362 self, check_health: bool = True, retry_socket_connect: bool = True
363 ):
364 if self.is_connected:
365 return
366 # Track actual retry attempts for error reporting
367 actual_retry_attempts = 0
369 def failure_callback(error, failure_count):
370 nonlocal actual_retry_attempts
371 actual_retry_attempts = failure_count
372 return self.disconnect(error=error, failure_count=failure_count)
374 try:
375 if retry_socket_connect:
376 await self.retry.call_with_retry(
377 lambda: self._connect(),
378 failure_callback,
379 with_failure_count=True,
380 )
381 else:
382 await self._connect()
383 except asyncio.CancelledError:
384 raise # in 3.7 and earlier, this is an Exception, not BaseException
385 except (socket.timeout, asyncio.TimeoutError):
386 e = TimeoutError("Timeout connecting to server")
387 await record_error_count(
388 server_address=getattr(self, "host", None),
389 server_port=getattr(self, "port", None),
390 network_peer_address=getattr(self, "host", None),
391 network_peer_port=getattr(self, "port", None),
392 error_type=e,
393 retry_attempts=actual_retry_attempts,
394 is_internal=False,
395 )
396 raise e
397 except OSError as e:
398 e = ConnectionError(self._error_message(e))
399 await record_error_count(
400 server_address=getattr(self, "host", None),
401 server_port=getattr(self, "port", None),
402 network_peer_address=getattr(self, "host", None),
403 network_peer_port=getattr(self, "port", None),
404 error_type=e,
405 retry_attempts=actual_retry_attempts,
406 is_internal=False,
407 )
408 raise e
409 except Exception as exc:
410 raise ConnectionError(exc) from exc
412 try:
413 if not self.redis_connect_func:
414 # Use the default on_connect function
415 await self.on_connect_check_health(check_health=check_health)
416 else:
417 # Use the passed function redis_connect_func
418 (
419 await self.redis_connect_func(self)
420 if asyncio.iscoroutinefunction(self.redis_connect_func)
421 else self.redis_connect_func(self)
422 )
423 except RedisError:
424 # clean up after any error in on_connect
425 await self.disconnect()
426 raise
428 # run any user callbacks. right now the only internal callback
429 # is for pubsub channel/pattern resubscription
430 # first, remove any dead weakrefs
431 self._connect_callbacks = [ref for ref in self._connect_callbacks if ref()]
432 for ref in self._connect_callbacks:
433 callback = ref()
434 task = callback(self)
435 if task and inspect.isawaitable(task):
436 await task
438 def mark_for_reconnect(self):
439 self._should_reconnect = True
441 def should_reconnect(self):
442 return self._should_reconnect
444 def reset_should_reconnect(self):
445 self._should_reconnect = False
447 @abstractmethod
448 async def _connect(self):
449 pass
451 @abstractmethod
452 def _host_error(self) -> str:
453 pass
455 def _error_message(self, exception: BaseException) -> str:
456 return format_error_message(self._host_error(), exception)
458 def get_protocol(self):
459 return self.protocol
461 async def on_connect(self) -> None:
462 """Initialize the connection, authenticate and select a database"""
463 await self.on_connect_check_health(check_health=True)
465 async def on_connect_check_health(self, check_health: bool = True) -> None:
466 self._parser.on_connect(self)
467 parser = self._parser
469 auth_args = None
470 # if credential provider or username and/or password are set, authenticate
471 if self.credential_provider or (self.username or self.password):
472 cred_provider = (
473 self.credential_provider
474 or UsernamePasswordCredentialProvider(self.username, self.password)
475 )
476 auth_args = await cred_provider.get_credentials_async()
478 # if resp version is specified and we have auth args,
479 # we need to send them via HELLO
480 if auth_args and self.protocol not in [2, "2"]:
481 if isinstance(self._parser, _AsyncRESP2Parser):
482 self.set_parser(_AsyncRESP3Parser)
483 # update cluster exception classes
484 self._parser.EXCEPTION_CLASSES = parser.EXCEPTION_CLASSES
485 self._parser.on_connect(self)
486 if len(auth_args) == 1:
487 auth_args = ["default", auth_args[0]]
488 # avoid checking health here -- PING will fail if we try
489 # to check the health prior to the AUTH
490 await self.send_command(
491 "HELLO", self.protocol, "AUTH", *auth_args, check_health=False
492 )
493 response = await self.read_response()
494 if response.get(b"proto") != int(self.protocol) and response.get(
495 "proto"
496 ) != int(self.protocol):
497 raise ConnectionError("Invalid RESP version")
498 # avoid checking health here -- PING will fail if we try
499 # to check the health prior to the AUTH
500 elif auth_args:
501 await self.send_command("AUTH", *auth_args, check_health=False)
503 try:
504 auth_response = await self.read_response()
505 except AuthenticationWrongNumberOfArgsError:
506 # a username and password were specified but the Redis
507 # server seems to be < 6.0.0 which expects a single password
508 # arg. retry auth with just the password.
509 # https://github.com/andymccurdy/redis-py/issues/1274
510 await self.send_command("AUTH", auth_args[-1], check_health=False)
511 auth_response = await self.read_response()
513 if str_if_bytes(auth_response) != "OK":
514 raise AuthenticationError("Invalid Username or Password")
516 # if resp version is specified, switch to it
517 elif self.protocol not in [2, "2"]:
518 if isinstance(self._parser, _AsyncRESP2Parser):
519 self.set_parser(_AsyncRESP3Parser)
520 # update cluster exception classes
521 self._parser.EXCEPTION_CLASSES = parser.EXCEPTION_CLASSES
522 self._parser.on_connect(self)
523 await self.send_command("HELLO", self.protocol, check_health=check_health)
524 response = await self.read_response()
525 # if response.get(b"proto") != self.protocol and response.get(
526 # "proto"
527 # ) != self.protocol:
528 # raise ConnectionError("Invalid RESP version")
530 # if a client_name is given, set it
531 if self.client_name:
532 await self.send_command(
533 "CLIENT",
534 "SETNAME",
535 self.client_name,
536 check_health=check_health,
537 )
538 if str_if_bytes(await self.read_response()) != "OK":
539 raise ConnectionError("Error setting client name")
541 # Set the library name and version from driver_info, pipeline for lower startup latency
542 lib_name_sent = False
543 lib_version_sent = False
545 if self.driver_info and self.driver_info.formatted_name:
546 await self.send_command(
547 "CLIENT",
548 "SETINFO",
549 "LIB-NAME",
550 self.driver_info.formatted_name,
551 check_health=check_health,
552 )
553 lib_name_sent = True
555 if self.driver_info and self.driver_info.lib_version:
556 await self.send_command(
557 "CLIENT",
558 "SETINFO",
559 "LIB-VER",
560 self.driver_info.lib_version,
561 check_health=check_health,
562 )
563 lib_version_sent = True
565 # if a database is specified, switch to it. Also pipeline this
566 if self.db:
567 await self.send_command("SELECT", self.db, check_health=check_health)
569 # read responses from pipeline
570 for _ in range(sum([lib_name_sent, lib_version_sent])):
571 try:
572 await self.read_response()
573 except ResponseError:
574 pass
576 if self.db:
577 if str_if_bytes(await self.read_response()) != "OK":
578 raise ConnectionError("Invalid Database")
580 async def disconnect(
581 self,
582 nowait: bool = False,
583 error: Optional[Exception] = None,
584 failure_count: Optional[int] = None,
585 health_check_failed: bool = False,
586 ) -> None:
587 """Disconnects from the Redis server"""
588 # On Python 3.13+, asyncio.timeout() raises RuntimeError when called
589 # outside a running Task (e.g. during GC finalization or event-loop
590 # callbacks). In that context we fall back to a synchronous close.
591 # See https://github.com/redis/redis-py/issues/3856
592 if asyncio.current_task() is None:
593 self._parser.on_disconnect()
594 self.reset_should_reconnect()
595 self._close()
596 return
598 try:
599 async with async_timeout(self.socket_connect_timeout):
600 self._parser.on_disconnect()
601 # Reset the reconnect flag
602 self.reset_should_reconnect()
603 if not self.is_connected:
604 return
605 try:
606 self._writer.close() # type: ignore[union-attr]
607 # wait for close to finish, except when handling errors and
608 # forcefully disconnecting.
609 if not nowait:
610 await self._writer.wait_closed() # type: ignore[union-attr]
611 except OSError:
612 pass
613 finally:
614 self._reader = None
615 self._writer = None
616 except asyncio.TimeoutError:
617 raise TimeoutError(
618 f"Timed out closing connection after {self.socket_connect_timeout}"
619 ) from None
621 if error:
622 if health_check_failed:
623 close_reason = CloseReason.HEALTHCHECK_FAILED
624 else:
625 close_reason = CloseReason.ERROR
627 if failure_count is not None and failure_count > self.retry.get_retries():
628 await record_error_count(
629 server_address=getattr(self, "host", None),
630 server_port=getattr(self, "port", None),
631 network_peer_address=getattr(self, "host", None),
632 network_peer_port=getattr(self, "port", None),
633 error_type=error,
634 retry_attempts=failure_count,
635 )
637 await record_connection_closed(
638 close_reason=close_reason,
639 error_type=error,
640 )
641 else:
642 await record_connection_closed(
643 close_reason=CloseReason.APPLICATION_CLOSE,
644 )
646 async def _send_ping(self):
647 """Send PING, expect PONG in return"""
648 await self.send_command("PING", check_health=False)
649 if str_if_bytes(await self.read_response()) != "PONG":
650 raise ConnectionError("Bad response from PING health check")
652 async def _ping_failed(self, error, failure_count):
653 """Function to call when PING fails"""
654 await self.disconnect(
655 error=error, failure_count=failure_count, health_check_failed=True
656 )
658 async def check_health(self):
659 """Check the health of the connection with a PING/PONG"""
660 if (
661 self.health_check_interval
662 and asyncio.get_running_loop().time() > self.next_health_check
663 ):
664 await self.retry.call_with_retry(
665 self._send_ping, self._ping_failed, with_failure_count=True
666 )
668 async def _send_packed_command(self, command: Iterable[bytes]) -> None:
669 self._writer.writelines(command)
670 await self._writer.drain()
672 async def send_packed_command(
673 self, command: Union[bytes, str, Iterable[bytes]], check_health: bool = True
674 ) -> None:
675 if not self.is_connected:
676 await self.connect_check_health(check_health=False)
677 if check_health:
678 await self.check_health()
680 try:
681 if isinstance(command, str):
682 command = command.encode()
683 if isinstance(command, bytes):
684 command = [command]
685 if self.socket_timeout:
686 await asyncio.wait_for(
687 self._send_packed_command(command), self.socket_timeout
688 )
689 else:
690 self._writer.writelines(command)
691 await self._writer.drain()
692 except asyncio.TimeoutError:
693 await self.disconnect(nowait=True)
694 raise TimeoutError("Timeout writing to socket") from None
695 except OSError as e:
696 await self.disconnect(nowait=True)
697 if len(e.args) == 1:
698 err_no, errmsg = "UNKNOWN", e.args[0]
699 else:
700 err_no = e.args[0]
701 errmsg = e.args[1]
702 raise ConnectionError(
703 f"Error {err_no} while writing to socket. {errmsg}."
704 ) from e
705 except BaseException:
706 # BaseExceptions can be raised when a socket send operation is not
707 # finished, e.g. due to a timeout. Ideally, a caller could then re-try
708 # to send un-sent data. However, the send_packed_command() API
709 # does not support it so there is no point in keeping the connection open.
710 await self.disconnect(nowait=True)
711 raise
713 async def send_command(self, *args: Any, **kwargs: Any) -> None:
714 """Pack and send a command to the Redis server"""
715 await self.send_packed_command(
716 self.pack_command(*args), check_health=kwargs.get("check_health", True)
717 )
719 @deprecated_function(
720 version="8.0.0", reason="Use can_read() instead", name="can_read_destructive"
721 )
722 async def can_read_destructive(self) -> bool:
723 """Check the socket to see if there's data loaded in the buffer."""
724 try:
725 return await self._parser.can_read()
726 except OSError as e:
727 await self.disconnect(nowait=True)
728 host_error = self._host_error()
729 raise ConnectionError(f"Error while reading from {host_error}: {e.args}")
731 async def can_read(self) -> bool:
732 """Check the socket to see if there's data loaded in the buffer."""
733 # TODO: Rename this API; it detects pending data or dirty/closed
734 # connection state, not only whether application data can be read.
735 try:
736 return await self._parser.can_read()
737 except OSError as e:
738 await self.disconnect(nowait=True)
739 host_error = self._host_error()
740 raise ConnectionError(f"Error while reading from {host_error}: {e.args}")
742 async def read_response(
743 self,
744 disable_decoding: bool = False,
745 timeout: Optional[float] = None,
746 *,
747 disconnect_on_error: bool = True,
748 push_request: Optional[bool] = False,
749 ):
750 """Read the response from a previously sent command.
752 ``timeout`` semantics:
753 - ``None`` (default): fall back to ``self.socket_timeout``.
754 - ``math.inf``: block indefinitely with no timeout. Used by PubSub
755 blocking reads (``listen()`` / ``get_message(timeout=None)`` /
756 ``parse_response(block=True)``) where the configured
757 ``socket_timeout`` must not abort the read.
758 - ``float``: apply that timeout in seconds for this single read.
760 TODO(next-major): replace the ``math.inf`` opt-in with a SENTINEL
761 default for ``timeout``. After that change, ``timeout=None`` will
762 mean "no timeout, block until a response arrives" (matching the
763 long-standing PubSub docstring contract) and the SENTINEL default
764 will be the value that falls back to ``self.socket_timeout``.
765 That swap is a breaking change, so it must wait for a major
766 release. Until then, callers that need an indefinitely blocking
767 read pass ``math.inf`` explicitly.
768 """
769 # TODO(next-major): drop the math.inf branch. Use SENTINEL as the
770 # default for ``timeout`` and treat ``timeout is None`` as the
771 # "no timeout" signal (matching the PubSub docstring contract).
772 # Match only positive infinity here. ``-math.inf`` is not a valid
773 # "block forever" signal and historically behaved as an already-
774 # expired timeout; preserve that.
775 if timeout == math.inf:
776 read_timeout = None
777 else:
778 read_timeout = timeout if timeout is not None else self.socket_timeout
779 host_error = self._host_error()
780 try:
781 if read_timeout is not None and self.protocol in ["3", 3]:
782 async with async_timeout(read_timeout):
783 response = await self._parser.read_response(
784 disable_decoding=disable_decoding, push_request=push_request
785 )
786 elif read_timeout is not None:
787 async with async_timeout(read_timeout):
788 response = await self._parser.read_response(
789 disable_decoding=disable_decoding
790 )
791 elif self.protocol in ["3", 3]:
792 response = await self._parser.read_response(
793 disable_decoding=disable_decoding, push_request=push_request
794 )
795 else:
796 response = await self._parser.read_response(
797 disable_decoding=disable_decoding
798 )
799 except asyncio.TimeoutError:
800 if timeout is not None:
801 # user requested timeout, return None. Operation can be retried
802 return None
803 # it was a self.socket_timeout error.
804 if disconnect_on_error:
805 await self.disconnect(nowait=True)
806 raise TimeoutError(f"Timeout reading from {host_error}")
807 except OSError as e:
808 if disconnect_on_error:
809 await self.disconnect(nowait=True)
810 raise ConnectionError(f"Error while reading from {host_error} : {e.args}")
811 except BaseException:
812 # Also by default close in case of BaseException. A lot of code
813 # relies on this behaviour when doing Command/Response pairs.
814 # See #1128.
815 if disconnect_on_error:
816 await self.disconnect(nowait=True)
817 raise
819 if self.health_check_interval:
820 next_time = asyncio.get_running_loop().time() + self.health_check_interval
821 self.next_health_check = next_time
823 if isinstance(response, ResponseError):
824 raise response from None
825 return response
827 def pack_command(self, *args: EncodableT) -> List[bytes]:
828 """Pack a series of arguments into the Redis protocol"""
829 output = []
830 # the client might have included 1 or more literal arguments in
831 # the command name, e.g., 'CONFIG GET'. The Redis server expects these
832 # arguments to be sent separately, so split the first argument
833 # manually. These arguments should be bytestrings so that they are
834 # not encoded.
835 assert not isinstance(args[0], float)
836 if isinstance(args[0], str):
837 args = tuple(args[0].encode().split()) + args[1:]
838 elif b" " in args[0]:
839 args = tuple(args[0].split()) + args[1:]
841 buff = SYM_EMPTY.join((SYM_STAR, str(len(args)).encode(), SYM_CRLF))
843 buffer_cutoff = self._buffer_cutoff
844 for arg in map(self.encoder.encode, args):
845 # to avoid large string mallocs, chunk the command into the
846 # output list if we're sending large values or memoryviews
847 arg_length = len(arg)
848 if (
849 len(buff) > buffer_cutoff
850 or arg_length > buffer_cutoff
851 or isinstance(arg, memoryview)
852 ):
853 buff = SYM_EMPTY.join(
854 (buff, SYM_DOLLAR, str(arg_length).encode(), SYM_CRLF)
855 )
856 output.append(buff)
857 output.append(arg)
858 buff = SYM_CRLF
859 else:
860 buff = SYM_EMPTY.join(
861 (
862 buff,
863 SYM_DOLLAR,
864 str(arg_length).encode(),
865 SYM_CRLF,
866 arg,
867 SYM_CRLF,
868 )
869 )
870 output.append(buff)
871 return output
873 def pack_commands(self, commands: Iterable[Iterable[EncodableT]]) -> List[bytes]:
874 """Pack multiple commands into the Redis protocol"""
875 output: List[bytes] = []
876 pieces: List[bytes] = []
877 buffer_length = 0
878 buffer_cutoff = self._buffer_cutoff
880 for cmd in commands:
881 for chunk in self.pack_command(*cmd):
882 chunklen = len(chunk)
883 if (
884 buffer_length > buffer_cutoff
885 or chunklen > buffer_cutoff
886 or isinstance(chunk, memoryview)
887 ):
888 if pieces:
889 output.append(SYM_EMPTY.join(pieces))
890 buffer_length = 0
891 pieces = []
893 if chunklen > buffer_cutoff or isinstance(chunk, memoryview):
894 output.append(chunk)
895 else:
896 pieces.append(chunk)
897 buffer_length += chunklen
899 if pieces:
900 output.append(SYM_EMPTY.join(pieces))
901 return output
903 def _socket_is_empty(self):
904 """Check if the socket is empty"""
905 return len(self._reader._buffer) == 0
907 async def process_invalidation_messages(self):
908 while not self._socket_is_empty():
909 await self.read_response(push_request=True)
911 def set_re_auth_token(self, token: TokenInterface):
912 self._re_auth_token = token
914 async def re_auth(self):
915 if self._re_auth_token is not None:
916 await self.send_command(
917 "AUTH",
918 self._re_auth_token.try_get("oid"),
919 self._re_auth_token.get_value(),
920 )
921 await self.read_response()
922 self._re_auth_token = None
925class Connection(AbstractConnection):
926 "Manages TCP communication to and from a Redis server"
928 def __init__(
929 self,
930 *,
931 host: str = "localhost",
932 port: str | int = 6379,
933 socket_keepalive: bool = True,
934 socket_keepalive_options: Mapping[int, int | bytes] | object | None = SENTINEL,
935 socket_type: int = 0,
936 **kwargs,
937 ):
938 """
939 Initialize a TCP connection.
941 Parameters
942 ----------
943 socket_keepalive : bool
944 If `True`, TCP keepalive is enabled for TCP socket connections.
945 socket_keepalive_options : Mapping[int, int | bytes] | object | None
946 Mapping of TCP keepalive socket option constants to values, for
947 example `{socket.TCP_KEEPIDLE: 30}`. If left unspecified, redis-py
948 uses TCP keepalive defaults when `socket_keepalive` is enabled:
949 idle 30 seconds, interval 5 seconds, and 3 probes. Platform-specific
950 options that are not available are skipped. Pass `None` or `{}` to
951 avoid setting additional TCP keepalive options.
952 """
953 self.host = host
954 self.port = int(port)
955 self.socket_keepalive = socket_keepalive
956 if socket_keepalive_options is SENTINEL:
957 socket_keepalive_options = get_default_socket_keepalive_options()
958 self.socket_keepalive_options = socket_keepalive_options or {}
959 self.socket_type = socket_type
960 super().__init__(**kwargs)
962 def repr_pieces(self):
963 pieces = [("host", self.host), ("port", self.port), ("db", self.db)]
964 if self.client_name:
965 pieces.append(("client_name", self.client_name))
966 return pieces
968 def _connection_arguments(self) -> Mapping:
969 return {"host": self.host, "port": self.port}
971 async def _connect(self):
972 """Create a TCP socket connection"""
973 async with async_timeout(self.socket_connect_timeout):
974 reader, writer = await asyncio.open_connection(
975 **self._connection_arguments()
976 )
977 self._reader = reader
978 self._writer = writer
979 sock = writer.transport.get_extra_info("socket")
980 if sock:
981 sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
982 try:
983 # TCP_KEEPALIVE
984 if self.socket_keepalive:
985 sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
986 for k, v in self.socket_keepalive_options.items():
987 sock.setsockopt(socket.SOL_TCP, k, v)
989 except (OSError, TypeError):
990 # `socket_keepalive_options` might contain invalid options
991 # causing an error. Do not leave the connection open.
992 writer.close()
993 raise
995 def _host_error(self) -> str:
996 return f"{self.host}:{self.port}"
999class SSLConnection(Connection):
1000 """Manages SSL connections to and from the Redis server(s).
1001 This class extends the Connection class, adding SSL functionality, and making
1002 use of ssl.SSLContext (https://docs.python.org/3/library/ssl.html#ssl.SSLContext)
1003 """
1005 def __init__(
1006 self,
1007 ssl_keyfile: Optional[str] = None,
1008 ssl_certfile: Optional[str] = None,
1009 ssl_cert_reqs: Union[str, ssl.VerifyMode] = "required",
1010 ssl_include_verify_flags: Optional[List["ssl.VerifyFlags"]] = None,
1011 ssl_exclude_verify_flags: Optional[List["ssl.VerifyFlags"]] = None,
1012 ssl_ca_certs: Optional[str] = None,
1013 ssl_ca_data: Optional[str] = None,
1014 ssl_ca_path: Optional[str] = None,
1015 ssl_check_hostname: bool = True,
1016 ssl_min_version: Optional[TLSVersion] = None,
1017 ssl_ciphers: Optional[str] = None,
1018 ssl_password: Optional[str] = None,
1019 **kwargs,
1020 ):
1021 if not SSL_AVAILABLE:
1022 raise RedisError("Python wasn't built with SSL support")
1024 self.ssl_context: RedisSSLContext = RedisSSLContext(
1025 keyfile=ssl_keyfile,
1026 certfile=ssl_certfile,
1027 cert_reqs=ssl_cert_reqs,
1028 include_verify_flags=ssl_include_verify_flags,
1029 exclude_verify_flags=ssl_exclude_verify_flags,
1030 ca_certs=ssl_ca_certs,
1031 ca_data=ssl_ca_data,
1032 ca_path=ssl_ca_path,
1033 check_hostname=ssl_check_hostname,
1034 min_version=ssl_min_version,
1035 ciphers=ssl_ciphers,
1036 password=ssl_password,
1037 )
1038 super().__init__(**kwargs)
1040 def _connection_arguments(self) -> Mapping:
1041 kwargs = super()._connection_arguments()
1042 kwargs["ssl"] = self.ssl_context.get()
1043 return kwargs
1045 @property
1046 def keyfile(self):
1047 return self.ssl_context.keyfile
1049 @property
1050 def certfile(self):
1051 return self.ssl_context.certfile
1053 @property
1054 def cert_reqs(self):
1055 return self.ssl_context.cert_reqs
1057 @property
1058 def include_verify_flags(self):
1059 return self.ssl_context.include_verify_flags
1061 @property
1062 def exclude_verify_flags(self):
1063 return self.ssl_context.exclude_verify_flags
1065 @property
1066 def ca_certs(self):
1067 return self.ssl_context.ca_certs
1069 @property
1070 def ca_data(self):
1071 return self.ssl_context.ca_data
1073 @property
1074 def check_hostname(self):
1075 return self.ssl_context.check_hostname
1077 @property
1078 def min_version(self):
1079 return self.ssl_context.min_version
1082class RedisSSLContext:
1083 __slots__ = (
1084 "keyfile",
1085 "certfile",
1086 "cert_reqs",
1087 "include_verify_flags",
1088 "exclude_verify_flags",
1089 "ca_certs",
1090 "ca_data",
1091 "ca_path",
1092 "context",
1093 "check_hostname",
1094 "min_version",
1095 "ciphers",
1096 "password",
1097 )
1099 def __init__(
1100 self,
1101 keyfile: Optional[str] = None,
1102 certfile: Optional[str] = None,
1103 cert_reqs: Optional[Union[str, ssl.VerifyMode]] = None,
1104 include_verify_flags: Optional[List["ssl.VerifyFlags"]] = None,
1105 exclude_verify_flags: Optional[List["ssl.VerifyFlags"]] = None,
1106 ca_certs: Optional[str] = None,
1107 ca_data: Optional[str] = None,
1108 ca_path: Optional[str] = None,
1109 check_hostname: bool = False,
1110 min_version: Optional[TLSVersion] = None,
1111 ciphers: Optional[str] = None,
1112 password: Optional[str] = None,
1113 ):
1114 if not SSL_AVAILABLE:
1115 raise RedisError("Python wasn't built with SSL support")
1117 self.keyfile = keyfile
1118 self.certfile = certfile
1119 if cert_reqs is None:
1120 cert_reqs = ssl.CERT_NONE
1121 elif isinstance(cert_reqs, str):
1122 CERT_REQS = { # noqa: N806
1123 "none": ssl.CERT_NONE,
1124 "optional": ssl.CERT_OPTIONAL,
1125 "required": ssl.CERT_REQUIRED,
1126 }
1127 if cert_reqs not in CERT_REQS:
1128 raise RedisError(
1129 f"Invalid SSL Certificate Requirements Flag: {cert_reqs}"
1130 )
1131 cert_reqs = CERT_REQS[cert_reqs]
1132 self.cert_reqs = cert_reqs
1133 self.include_verify_flags = include_verify_flags
1134 self.exclude_verify_flags = exclude_verify_flags
1135 self.ca_certs = ca_certs
1136 self.ca_data = ca_data
1137 self.ca_path = ca_path
1138 self.check_hostname = (
1139 check_hostname if self.cert_reqs != ssl.CERT_NONE else False
1140 )
1141 self.min_version = min_version
1142 self.ciphers = ciphers
1143 self.password = password
1144 self.context: Optional[SSLContext] = None
1146 def get(self) -> SSLContext:
1147 if not self.context:
1148 context = ssl.create_default_context()
1149 context.check_hostname = self.check_hostname
1150 context.verify_mode = self.cert_reqs
1151 if self.include_verify_flags:
1152 for flag in self.include_verify_flags:
1153 context.verify_flags |= flag
1154 if self.exclude_verify_flags:
1155 for flag in self.exclude_verify_flags:
1156 context.verify_flags &= ~flag
1157 if self.certfile or self.keyfile:
1158 context.load_cert_chain(
1159 certfile=self.certfile,
1160 keyfile=self.keyfile,
1161 password=self.password,
1162 )
1163 if self.ca_certs or self.ca_data or self.ca_path:
1164 context.load_verify_locations(
1165 cafile=self.ca_certs, capath=self.ca_path, cadata=self.ca_data
1166 )
1167 if self.min_version is not None:
1168 context.minimum_version = self.min_version
1169 if self.ciphers is not None:
1170 context.set_ciphers(self.ciphers)
1171 self.context = context
1172 return self.context
1175class UnixDomainSocketConnection(AbstractConnection):
1176 "Manages UDS communication to and from a Redis server"
1178 def __init__(self, *, path: str = "", **kwargs):
1179 self.path = path
1180 super().__init__(**kwargs)
1182 def repr_pieces(self) -> Iterable[Tuple[str, Union[str, int]]]:
1183 pieces = [("path", self.path), ("db", self.db)]
1184 if self.client_name:
1185 pieces.append(("client_name", self.client_name))
1186 return pieces
1188 async def _connect(self):
1189 async with async_timeout(self.socket_connect_timeout):
1190 reader, writer = await asyncio.open_unix_connection(path=self.path)
1191 self._reader = reader
1192 self._writer = writer
1193 await self.on_connect()
1195 def _host_error(self) -> str:
1196 return self.path
1199FALSE_STRINGS = ("0", "F", "FALSE", "N", "NO")
1202def to_bool(value) -> Optional[bool]:
1203 if value is None or value == "":
1204 return None
1205 if isinstance(value, str) and value.upper() in FALSE_STRINGS:
1206 return False
1207 return bool(value)
1210def parse_ssl_verify_flags(value):
1211 # flags are passed in as a string representation of a list,
1212 # e.g. VERIFY_X509_STRICT, VERIFY_X509_PARTIAL_CHAIN
1213 verify_flags_str = value.replace("[", "").replace("]", "")
1215 verify_flags = []
1216 for flag in verify_flags_str.split(","):
1217 flag = flag.strip()
1218 if not hasattr(VerifyFlags, flag):
1219 raise ValueError(f"Invalid ssl verify flag: {flag}")
1220 verify_flags.append(getattr(VerifyFlags, flag))
1221 return verify_flags
1224URL_QUERY_ARGUMENT_PARSERS: Mapping[str, Callable[..., object]] = MappingProxyType(
1225 {
1226 "db": int,
1227 "socket_timeout": float,
1228 "socket_connect_timeout": float,
1229 "socket_read_size": int,
1230 "socket_keepalive": to_bool,
1231 "retry_on_timeout": to_bool,
1232 "max_connections": int,
1233 "health_check_interval": int,
1234 "ssl_check_hostname": to_bool,
1235 "ssl_include_verify_flags": parse_ssl_verify_flags,
1236 "ssl_exclude_verify_flags": parse_ssl_verify_flags,
1237 "ssl_min_version": int,
1238 "timeout": float,
1239 "protocol": int,
1240 "legacy_responses": to_bool,
1241 }
1242)
1245class ConnectKwargs(TypedDict, total=False):
1246 username: str
1247 password: str
1248 connection_class: Type[AbstractConnection]
1249 host: str
1250 port: int
1251 db: int
1252 path: str
1255def parse_url(url: str) -> ConnectKwargs:
1256 parsed: ParseResult = urlparse(url)
1257 kwargs: ConnectKwargs = {}
1259 for name, value_list in parse_qs(parsed.query).items():
1260 if value_list and len(value_list) > 0:
1261 value = unquote(value_list[0])
1262 parser = URL_QUERY_ARGUMENT_PARSERS.get(name)
1263 if parser:
1264 try:
1265 kwargs[name] = parser(value)
1266 except (TypeError, ValueError):
1267 raise ValueError(f"Invalid value for '{name}' in connection URL.")
1268 else:
1269 kwargs[name] = value
1271 if parsed.username:
1272 kwargs["username"] = unquote(parsed.username)
1273 if parsed.password:
1274 kwargs["password"] = unquote(parsed.password)
1276 # We only support redis://, rediss:// and unix:// schemes.
1277 if parsed.scheme == "unix":
1278 if parsed.path:
1279 kwargs["path"] = unquote(parsed.path)
1280 kwargs["connection_class"] = UnixDomainSocketConnection
1282 elif parsed.scheme in ("redis", "rediss"):
1283 if parsed.hostname:
1284 kwargs["host"] = unquote(parsed.hostname)
1285 if parsed.port:
1286 kwargs["port"] = int(parsed.port)
1288 # If there's a path argument, use it as the db argument if a
1289 # querystring value wasn't specified
1290 if parsed.path and "db" not in kwargs:
1291 try:
1292 kwargs["db"] = int(unquote(parsed.path).replace("/", ""))
1293 except (AttributeError, ValueError):
1294 pass
1296 if parsed.scheme == "rediss":
1297 kwargs["connection_class"] = SSLConnection
1299 else:
1300 valid_schemes = "redis://, rediss://, unix://"
1301 raise ValueError(
1302 f"Redis URL must specify one of the following schemes ({valid_schemes})"
1303 )
1305 return kwargs
1308_CP = TypeVar("_CP", bound="ConnectionPool")
1311class ConnectionPoolInterface(ABC):
1312 @abstractmethod
1313 def get_protocol(self):
1314 pass
1316 @abstractmethod
1317 def reset(self) -> None:
1318 pass
1320 @abstractmethod
1321 @deprecated_args(
1322 args_to_warn=["*"],
1323 reason="Use get_connection() without args instead",
1324 version="5.3.0",
1325 )
1326 async def get_connection(
1327 self, command_name: Optional[str] = None, *keys: Any, **options: Any
1328 ) -> "AbstractConnection":
1329 pass
1331 @abstractmethod
1332 def get_encoder(self) -> "Encoder":
1333 pass
1335 @abstractmethod
1336 async def release(self, connection: "AbstractConnection") -> None:
1337 pass
1339 @abstractmethod
1340 async def disconnect(self, inuse_connections: bool = True) -> None:
1341 pass
1343 @abstractmethod
1344 async def aclose(self) -> None:
1345 pass
1347 @abstractmethod
1348 def set_retry(self, retry: "Retry") -> None:
1349 pass
1351 @abstractmethod
1352 async def re_auth_callback(self, token: TokenInterface) -> None:
1353 pass
1355 @abstractmethod
1356 def get_connection_count(self) -> List[Tuple[int, dict]]:
1357 """
1358 Returns a connection count (both idle and in use).
1359 """
1360 pass
1363class ConnectionPool(ConnectionPoolInterface):
1364 """
1365 Create a connection pool. ``If max_connections`` is set, then this
1366 object raises :py:class:`~redis.ConnectionError` when the pool's
1367 limit is reached.
1369 By default, TCP connections are created unless ``connection_class``
1370 is specified. Use :py:class:`~redis.UnixDomainSocketConnection` for
1371 unix sockets.
1372 :py:class:`~redis.SSLConnection` can be used for SSL enabled connections.
1374 Any additional keyword arguments are passed to the constructor of
1375 ``connection_class``.
1376 """
1378 @classmethod
1379 def from_url(cls: Type[_CP], url: str, **kwargs) -> _CP:
1380 """
1381 Return a connection pool configured from the given URL.
1383 For example::
1385 redis://[[username]:[password]]@localhost:6379/0
1386 rediss://[[username]:[password]]@localhost:6379/0
1387 unix://[username@]/path/to/socket.sock?db=0[&password=password]
1389 Three URL schemes are supported:
1391 - `redis://` creates a TCP socket connection. See more at:
1392 <https://www.iana.org/assignments/uri-schemes/prov/redis>
1393 - `rediss://` creates a SSL wrapped TCP socket connection. See more at:
1394 <https://www.iana.org/assignments/uri-schemes/prov/rediss>
1395 - ``unix://``: creates a Unix Domain Socket connection.
1397 The username, password, hostname, path and all querystring values
1398 are passed through urllib.parse.unquote in order to replace any
1399 percent-encoded values with their corresponding characters.
1401 There are several ways to specify a database number. The first value
1402 found will be used:
1404 1. A ``db`` querystring option, e.g. redis://localhost?db=0
1406 2. If using the redis:// or rediss:// schemes, the path argument
1407 of the url, e.g. redis://localhost/0
1409 3. A ``db`` keyword argument to this function.
1411 If none of these options are specified, the default db=0 is used.
1413 All querystring options are cast to their appropriate Python types.
1414 Boolean arguments can be specified with string values "True"/"False"
1415 or "Yes"/"No". Values that cannot be properly cast cause a
1416 ``ValueError`` to be raised. Once parsed, the querystring arguments
1417 and keyword arguments are passed to the ``ConnectionPool``'s
1418 class initializer. In the case of conflicting arguments, querystring
1419 arguments always win.
1420 """
1421 url_options = parse_url(url)
1422 kwargs.update(url_options)
1423 return cls(**kwargs)
1425 def __init__(
1426 self,
1427 connection_class: Type[AbstractConnection] = Connection,
1428 max_connections: Optional[int] = None,
1429 **connection_kwargs,
1430 ):
1431 max_connections = max_connections or 100
1432 if not isinstance(max_connections, int) or max_connections < 0:
1433 raise ValueError('"max_connections" must be a positive integer')
1435 self.connection_class = connection_class
1436 self.connection_kwargs = connection_kwargs
1437 self.max_connections = max_connections
1439 self._available_connections: List[AbstractConnection] = []
1440 self._in_use_connections: Set[AbstractConnection] = set()
1441 self.encoder_class = self.connection_kwargs.get("encoder_class", Encoder)
1442 self._lock = asyncio.Lock()
1443 self._event_dispatcher = self.connection_kwargs.get("event_dispatcher", None)
1444 if self._event_dispatcher is None:
1445 self._event_dispatcher = EventDispatcher()
1447 # Keys that should be redacted in __repr__ to avoid exposing sensitive information
1448 SENSITIVE_REPR_KEYS = frozenset(
1449 {
1450 "password",
1451 "username",
1452 "ssl_password",
1453 "credential_provider",
1454 }
1455 )
1457 def __repr__(self):
1458 conn_kwargs = ",".join(
1459 [
1460 f"{k}={'<REDACTED>' if k in self.SENSITIVE_REPR_KEYS else v}"
1461 for k, v in self.connection_kwargs.items()
1462 ]
1463 )
1464 return (
1465 f"<{self.__class__.__module__}.{self.__class__.__name__}"
1466 f"(<{self.connection_class.__module__}.{self.connection_class.__name__}"
1467 f"({conn_kwargs})>)>"
1468 )
1470 def get_protocol(self):
1471 """
1472 Returns:
1473 The RESP protocol version, or ``None`` if the protocol is not specified,
1474 in which case the server default will be used.
1475 """
1476 return self.connection_kwargs.get("protocol", None)
1478 def reset(self):
1479 # Record metrics for connections being removed before clearing
1480 # (only if attributes exist - they won't during __init__)
1481 if hasattr(self, "_available_connections") and hasattr(
1482 self, "_in_use_connections"
1483 ):
1484 idle_count = len(self._available_connections)
1485 in_use_count = len(self._in_use_connections)
1486 if idle_count > 0 or in_use_count > 0:
1487 pool_name = get_pool_name(self)
1488 # Note: Using sync version since reset() is sync
1489 from redis.observability.recorder import (
1490 record_connection_count as sync_record_connection_count,
1491 )
1493 if idle_count > 0:
1494 sync_record_connection_count(
1495 pool_name=pool_name,
1496 connection_state=ConnectionState.IDLE,
1497 counter=-idle_count,
1498 )
1499 if in_use_count > 0:
1500 sync_record_connection_count(
1501 pool_name=pool_name,
1502 connection_state=ConnectionState.USED,
1503 counter=-in_use_count,
1504 )
1506 self._available_connections = []
1507 self._in_use_connections = weakref.WeakSet()
1509 def __del__(self) -> None:
1510 """Clean up connection pool and record metrics when garbage collected."""
1511 try:
1512 if not hasattr(self, "_available_connections") or not hasattr(
1513 self, "_in_use_connections"
1514 ):
1515 return
1516 idle_count = len(self._available_connections)
1517 in_use_count = len(self._in_use_connections)
1518 if idle_count > 0 or in_use_count > 0:
1519 pool_name = get_pool_name(self)
1520 # Note: Using sync version since __del__ is sync
1521 from redis.observability.recorder import (
1522 record_connection_count as sync_record_connection_count,
1523 )
1525 if idle_count > 0:
1526 sync_record_connection_count(
1527 pool_name=pool_name,
1528 connection_state=ConnectionState.IDLE,
1529 counter=-idle_count,
1530 )
1531 if in_use_count > 0:
1532 sync_record_connection_count(
1533 pool_name=pool_name,
1534 connection_state=ConnectionState.USED,
1535 counter=-in_use_count,
1536 )
1537 except Exception:
1538 pass
1540 def can_get_connection(self) -> bool:
1541 """Return True if a connection can be retrieved from the pool."""
1542 return (
1543 self._available_connections
1544 or len(self._in_use_connections) < self.max_connections
1545 )
1547 @deprecated_args(
1548 args_to_warn=["*"],
1549 reason="Use get_connection() without args instead",
1550 version="5.3.0",
1551 )
1552 async def get_connection(self, command_name=None, *keys, **options):
1553 """Get a connected connection from the pool"""
1554 # Track connection count before to detect if a new connection is created
1555 async with self._lock:
1556 connections_before = len(self._available_connections) + len(
1557 self._in_use_connections
1558 )
1559 start_time_created = time.monotonic()
1560 connection = self.get_available_connection()
1561 connections_after = len(self._available_connections) + len(
1562 self._in_use_connections
1563 )
1564 is_created = connections_after > connections_before
1566 # Record state transition for observability
1567 # This ensures counters stay balanced if ensure_connection() fails and release() is called
1568 pool_name = get_pool_name(self)
1569 if is_created:
1570 # New connection created and acquired: just USED +1
1571 await record_connection_count(
1572 pool_name=pool_name,
1573 connection_state=ConnectionState.USED,
1574 counter=1,
1575 )
1576 else:
1577 # Existing connection acquired from pool: IDLE -> USED
1578 await record_connection_count(
1579 pool_name=pool_name,
1580 connection_state=ConnectionState.IDLE,
1581 counter=-1,
1582 )
1583 await record_connection_count(
1584 pool_name=pool_name,
1585 connection_state=ConnectionState.USED,
1586 counter=1,
1587 )
1589 # We now perform the connection check outside of the lock.
1590 try:
1591 await self.ensure_connection(connection)
1593 if is_created:
1594 await record_connection_create_time(
1595 connection_pool=self,
1596 duration_seconds=time.monotonic() - start_time_created,
1597 )
1599 return connection
1600 except BaseException:
1601 await self.release(connection)
1602 raise
1604 def get_available_connection(self):
1605 """Get a connection from the pool, without making sure it is connected"""
1606 try:
1607 connection = self._available_connections.pop()
1608 except IndexError:
1609 if len(self._in_use_connections) >= self.max_connections:
1610 raise MaxConnectionsError("Too many connections") from None
1611 connection = self.make_connection()
1612 self._in_use_connections.add(connection)
1613 return connection
1615 def get_encoder(self):
1616 """Return an encoder based on encoding settings"""
1617 kwargs = self.connection_kwargs
1618 return self.encoder_class(
1619 encoding=kwargs.get("encoding", "utf-8"),
1620 encoding_errors=kwargs.get("encoding_errors", "strict"),
1621 decode_responses=kwargs.get("decode_responses", False),
1622 )
1624 def make_connection(self):
1625 """Create a new connection. Can be overridden by child classes."""
1626 # Note: We don't record IDLE here because async uses a sync make_connection
1627 # but async record_connection_count. The recording is handled in get_connection.
1628 return self.connection_class(**self.connection_kwargs)
1630 async def ensure_connection(self, connection: AbstractConnection):
1631 """Ensure that the connection object is connected and valid"""
1632 await connection.connect()
1633 # connections that the pool provides should be ready to send
1634 # a command. if not, the connection was either returned to the
1635 # pool before all data has been read or the socket has been
1636 # closed. either way, reconnect and verify everything is good.
1637 try:
1638 if await connection.can_read():
1639 raise ConnectionError("Connection has data") from None
1640 except (ConnectionError, TimeoutError, OSError):
1641 await connection.disconnect()
1642 await connection.connect()
1643 if await connection.can_read():
1644 raise ConnectionError("Connection not ready") from None
1646 async def release(self, connection: AbstractConnection):
1647 """Releases the connection back to the pool"""
1648 # Connections should always be returned to the correct pool,
1649 # not doing so is an error that will cause an exception here.
1650 self._in_use_connections.remove(connection)
1652 if connection.should_reconnect():
1653 await connection.disconnect()
1655 self._available_connections.append(connection)
1656 await self._event_dispatcher.dispatch_async(
1657 AsyncAfterConnectionReleasedEvent(connection)
1658 )
1660 # Record state transition: USED -> IDLE
1661 pool_name = get_pool_name(self)
1662 await record_connection_count(
1663 pool_name=pool_name,
1664 connection_state=ConnectionState.USED,
1665 counter=-1,
1666 )
1667 await record_connection_count(
1668 pool_name=pool_name,
1669 connection_state=ConnectionState.IDLE,
1670 counter=1,
1671 )
1673 async def disconnect(self, inuse_connections: bool = True):
1674 """
1675 Disconnects connections in the pool
1677 If ``inuse_connections`` is True, disconnect connections that are
1678 current in use, potentially by other tasks. Otherwise only disconnect
1679 connections that are idle in the pool.
1680 """
1681 if inuse_connections:
1682 connections: Iterable[AbstractConnection] = chain(
1683 self._available_connections, self._in_use_connections
1684 )
1685 else:
1686 connections = self._available_connections
1687 resp = await asyncio.gather(
1688 *(connection.disconnect() for connection in connections),
1689 return_exceptions=True,
1690 )
1692 exc = next((r for r in resp if isinstance(r, BaseException)), None)
1693 if exc:
1694 raise exc
1696 async def update_active_connections_for_reconnect(self):
1697 """
1698 Mark all active connections for reconnect.
1699 """
1700 async with self._lock:
1701 for conn in self._in_use_connections:
1702 conn.mark_for_reconnect()
1704 async def aclose(self) -> None:
1705 """Close the pool, disconnecting all connections"""
1706 await self.disconnect()
1708 async def __aenter__(self: _CP) -> _CP:
1709 return self
1711 async def __aexit__(self, exc_type, exc_value, traceback) -> None:
1712 await self.aclose()
1714 def set_retry(self, retry: "Retry") -> None:
1715 for conn in self._available_connections:
1716 conn.retry = retry
1717 for conn in self._in_use_connections:
1718 conn.retry = retry
1720 async def re_auth_callback(self, token: TokenInterface):
1721 async with self._lock:
1722 for conn in self._available_connections:
1723 await conn.retry.call_with_retry(
1724 lambda: conn.send_command(
1725 "AUTH", token.try_get("oid"), token.get_value()
1726 ),
1727 lambda error: self._mock(error),
1728 )
1729 await conn.retry.call_with_retry(
1730 lambda: conn.read_response(), lambda error: self._mock(error)
1731 )
1732 for conn in self._in_use_connections:
1733 conn.set_re_auth_token(token)
1735 async def _mock(self, error: RedisError):
1736 """
1737 Dummy functions, needs to be passed as error callback to retry object.
1738 :param error:
1739 :return:
1740 """
1741 pass
1743 def get_connection_count(self) -> List[tuple[int, dict]]:
1744 """
1745 Returns a connection count (both idle and in use).
1746 """
1747 attributes = AttributeBuilder.build_base_attributes()
1748 attributes[DB_CLIENT_CONNECTION_POOL_NAME] = get_pool_name(self)
1749 free_connections_attributes = attributes.copy()
1750 in_use_connections_attributes = attributes.copy()
1752 free_connections_attributes[DB_CLIENT_CONNECTION_STATE] = (
1753 ConnectionState.IDLE.value
1754 )
1755 in_use_connections_attributes[DB_CLIENT_CONNECTION_STATE] = (
1756 ConnectionState.USED.value
1757 )
1759 return [
1760 (len(self._available_connections), free_connections_attributes),
1761 (len(self._in_use_connections), in_use_connections_attributes),
1762 ]
1765class BlockingConnectionPool(ConnectionPool):
1766 """
1767 A blocking connection pool::
1769 >>> from redis.asyncio import Redis, BlockingConnectionPool
1770 >>> client = Redis.from_pool(BlockingConnectionPool())
1772 It performs the same function as the default
1773 :py:class:`~redis.asyncio.ConnectionPool` implementation, in that,
1774 it maintains a pool of reusable connections that can be shared by
1775 multiple async redis clients.
1777 The difference is that, in the event that a client tries to get a
1778 connection from the pool when all of connections are in use, rather than
1779 raising a :py:class:`~redis.ConnectionError` (as the default
1780 :py:class:`~redis.asyncio.ConnectionPool` implementation does), it
1781 blocks the current `Task` for a specified number of seconds until
1782 a connection becomes available.
1784 Use ``max_connections`` to increase / decrease the pool size::
1786 >>> pool = BlockingConnectionPool(max_connections=10)
1788 Use ``timeout`` to tell it either how many seconds to wait for a connection
1789 to become available, or to block forever:
1791 >>> # Block forever.
1792 >>> pool = BlockingConnectionPool(timeout=None)
1794 >>> # Raise a ``ConnectionError`` after five seconds if a connection is
1795 >>> # not available.
1796 >>> pool = BlockingConnectionPool(timeout=5)
1797 """
1799 def __init__(
1800 self,
1801 max_connections: int = 50,
1802 timeout: Optional[float] = 20,
1803 connection_class: Type[AbstractConnection] = Connection,
1804 queue_class: Type[asyncio.Queue] = asyncio.LifoQueue, # deprecated
1805 **connection_kwargs,
1806 ):
1807 super().__init__(
1808 connection_class=connection_class,
1809 max_connections=max_connections,
1810 **connection_kwargs,
1811 )
1812 self._condition = asyncio.Condition()
1813 self.timeout = timeout
1815 @deprecated_args(
1816 args_to_warn=["*"],
1817 reason="Use get_connection() without args instead",
1818 version="5.3.0",
1819 )
1820 async def get_connection(self, command_name=None, *keys, **options):
1821 """Gets a connection from the pool, blocking until one is available"""
1822 # Start timing for wait time observability
1823 start_time_acquired = time.monotonic()
1825 try:
1826 async with self._condition:
1827 async with async_timeout(self.timeout):
1828 await self._condition.wait_for(self.can_get_connection)
1829 # Track connection count before to detect if a new connection is created
1830 connections_before = len(self._available_connections) + len(
1831 self._in_use_connections
1832 )
1833 start_time_created = time.monotonic()
1834 connection = super().get_available_connection()
1835 connections_after = len(self._available_connections) + len(
1836 self._in_use_connections
1837 )
1838 is_created = connections_after > connections_before
1839 except asyncio.TimeoutError as err:
1840 raise ConnectionError("No connection available.") from err
1842 # We now perform the connection check outside of the lock.
1843 try:
1844 await self.ensure_connection(connection)
1846 if is_created:
1847 await record_connection_create_time(
1848 connection_pool=self,
1849 duration_seconds=time.monotonic() - start_time_created,
1850 )
1852 await record_connection_wait_time(
1853 pool_name=get_pool_name(self),
1854 duration_seconds=time.monotonic() - start_time_acquired,
1855 )
1857 return connection
1858 except BaseException:
1859 await self.release(connection)
1860 raise
1862 async def release(self, connection: AbstractConnection):
1863 """Releases the connection back to the pool."""
1864 async with self._condition:
1865 await super().release(connection)
1866 self._condition.notify()