redis-redis-py-3863

redis
tasks.py
tests
util
whitelist.py
/repo/benchmarks/:
__init__.py
base.py
basic_operations.py
cluster_async.py
cluster_async_pipeline.py
command_packer_benchmark.py
socket_read_size.py

/repo/dockers/:
sentinel.conf

/repo/docs/:
Makefile
_static
_templates
advanced_features.rst
backoff.rst
clustering.rst
commands.rst
conf.py
connections.rst
examples
examples.rst
exceptions.rst
genindex.rst
images
index.rst
lock.rst
lua_scripting.rst
multi_database.rst
opentelemetry.rst
redismodules.rst
requirements.txt
resp3_features.rst
retry.rst

/repo/doctests/:
README.md
cmds_cnxmgmt.py
cmds_generic.py
cmds_hash.py
cmds_list.py
cmds_servermgmt.py
cmds_set.py
cmds_sorted_set.py
cmds_string.py
data
dt_bitfield.py
dt_bitmap.py

727:        if retry_on_timeout:
728:            # Add TimeoutError to the errors list to retry on
729:            retry_on_errors_list.append(TimeoutError)
730:        self.retry_on_error = retry_on_errors_list
731:        if
…[862 chars omitted]…
   def set_retry(self, retry: Retry):
2703:    def set_retry(self, retry: Retry) -> None:
2704:        self.connection_kwargs.update({"retry": retry})
2706:            conn.retry = retry
2708:            conn.retry = retry
2713:                conn.retry.call_with_retry(
2719:                conn.retry.call_with_retry(
2738:        Dummy functions, needs to be passed as error callback to retry object.

        try:
            if retry_socket_connect:
                sock = self.retry.call_with_retry(
                    lambda: self._connect(), lambda error: self.disconnect(error)
                )
            else:
                sock = self._connect()
        except socket.timeout:
            raise TimeoutError("Timeout connecting to server")
        except OSError as e:
            raise ConnectionError(self._error_message(e))

        self._sock = sock
        try:
            i
…[3668 chars omitted]…
h to it
        elif self.protocol not in [2, "2"]:
            if isinstance(self._parser, _RESP2Parser):
                self.set_parser(_RESP3Parser)
                # update cluster exception classes
                self._parser.EXCEPTION_CLASSES = parser.EXCEPTION_CLASSES
                self._parser.on_connect(self)
            self.send_command("HELLO", self.protocol, check_health=check_health)

    TypeVar,
)

from redis.exceptions import ConnectionError, TimeoutError

T = TypeVar("T")
E = TypeVar("E", bound=Exception, covariant=True)

if TYPE_CHECKING:
    from redis.backoff import AbstractBackoff


class AbstractRetry(Generic[E], abc.ABC):
    """Retry a specific number of times after a failure"""

    _supported_errors: Tuple[Type[E], ...]

    def __init__(
        self,
        backoff: "AbstractBackoff",
        retries: int,
        supported_errors: Tuple[Type[E], ...],
    ):
        """
        Initialize a `Retry` object with a `Backoff` object
        that retries a maximum of `retries` times.
        `retries` can be negative to retry forever.
        You can specify the types of supported er
…[2309 chars omitted]…
t self._supported_errors as error:
                if is_retryable and not is_retryable(error):
                    raise
                failures += 1
                fail(error)
                if self._retries >= 0 and failures > self._retries:
                    raise error
                backoff = self._backoff.compute(failures)
                if backoff > 0:
                    sleep(backoff)

        # if a client_name is given, set it
        if self.client_name:
            self.send_command(
                "CLIENT",
                "SETNAME",
                self.client_name,
                check_health=check_health,
            )
 
…[614 chars omitted]…
                "CLIENT",
                    "SETINFO",
                    "LIB-VER",
                    self.lib_version,
                    check_health=check_health,
                )
                self.read_response()
        except ResponseError:
            pass

/repo/redis/sentinel.py:43:        self.connect_check_health(
/repo/redis/sentinel.py:45:            retry_socket_connect=False,

entraid_utils.py
helpers.py
mocks.py
ssl_utils.py
test_asyncio
test_auth
test_background.py
test_backoff.py
test_bloom.py
test_cache.py
test_cluster.py
test_cluster_transaction.py
test_command_parser.py
test_command_policies.py
test_commands.py
test_connect.py
test_connection.py
test_connection_pool.py
test_credentials.py
test_data_structure.py
test_encoding.py
test_event.py
test_function.py
test_hash.py
test_he
…[15 chars omitted]…
ttp
test_json.py
test_lock.py
test_maint_notifications.py
test_maint_notifications_handling.py
test_max_connections_error.py
test_monitor.py
test_multidb
test_multiprocessing.py
test_parsers
test_pipeline.py
test_pubsub.py
test_retry.py
test_scenario
test_scripting.py
test_search.py
test_sentinel.py
test_sentinel_managed_connection.py
test_ssl.py
test_timeseries.py
test_utils.py
test_vsets.py
testdata

429:                self._parser.on_connect(self)
529:            await self.connect_check_health(check_health=False)
972:        await self.on_connect()

                    lambda: self._connect(), lambda error: self.disconnect()
                )
            else:
                await self._connect()
        except asyncio.CancelledError:
            raise  # in 3.7 and earlier, this is an Exception, not BaseException
        except (socket.timeout, asyncio.TimeoutError):
            raise TimeoutError("Timeout connecting to server")
        except OSError as e:
            raise ConnectionError(self._error_message(
…[717 chars omitted]…
   # run any user callbacks. right now the only internal callback
        # is for pubsub channel/pattern resubscription
        # first, remove any dead weakrefs
        self._connect_callbacks = [ref for ref in self._connect_callbacks if ref()]
        for ref in self._connect_callbacks:
            callback = ref()
            task = callback(self)
            if task and inspect.isawaitable(task):

/repo/re
…[214 chars omitted]…
self):
/repo/redis/connection.py:1853:    def _connect(self):
/repo/redis/connection.py:2004:    def disconnect(self, inuse_connections: bool = True):
/repo/redis/connection.py:2346:    def disconnect_free_connections(
/repo/redis/connection.py:2679:    def disconnect(self, inuse_connections: bool = True) -> None:
/repo/redis/connection.py:2959:    def disconnect(self, inuse_connections: bool = True):

        # reset the reconnect flag
        self.reset_should_reconnect()
        if conn_sock is None:
            return

        if os.getpid() == self.pid:
            try:
                conn_sock.shutdown(socket.SHUT_RDWR)
            except (OSError, TypeError):
                pass

        try:
            conn_sock.close()
        except OSError:
            pass

    def mark_for_reconnect(self):
        self._should_reconnect = True

    def s
…[1383 chars omitted]…
ait:
                        await self._writer.wait_closed()  # type: ignore[union-attr]
                except OSError:
                    pass
                finally:
                    self._reader = None
                    self._writer = None
        except asyncio.TimeoutError:
            raise TimeoutError(
                f"Timed out closing connection after {self.socket_connect_timeout}"

                    lambda error: self.disconnect(error),
                )
            else:
                self._connect_and_handshake(check_health)
        except socket.timeout:
            raise TimeoutError("Timeout connecting to server")
        except OSError as e:
            raise ConnectionError(self._error_message(e))
        except RedisError:
            # clean up after any error in the handshake
            self.disconnect()
            ra
…[924 chars omitted]…
onnect_check_health(check_health=check_health)
            else:
                # Use the passed function redis_connect_func
                self.redis_connect_func(self)
        except BaseException:
            # clean up after any error in the handshake
            self.disconnect()
            raise
        return sock

    @abstractmethod
    def _connect(self):
        pass

    @abstractmethod

                    lambda: self._connect_and_handshake(check_health),
                    lambda error: self.disconnect(),
                )
            else:
                await self._connect_and_handshake(check_health)
        except asyncio.CancelledError:
            raise  # in 3.7 and earlier, this is an Exception, not BaseException
        except (socket.timeout, asyncio.TimeoutError):
            raise TimeoutError("Timeout connecting to server")
        ex
…[1676 chars omitted]…
                else self.redis_connect_func(self)
                )
        except BaseException:
            # clean up after any error in the handshake
            await self.disconnect()
            raise

    @abstractmethod
    async def _connect(self):
        pass

    @abstractmethod
    def _host_error(self) -> str:
        pass

    def _error_message(self, exception: BaseException) -> str:

  File "/repo/redis/connection.py", line 855, in connect_check_health
    self.retry.call_with_retry(
  File "/repo/redis/retry.py", line 116, in call_with_retry
    return do()
           ^^^^
  File "/repo/redis/connection.py", line 856, in <lambda>
    lambda: self._connect_and_handshake(check_health),
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/redis/connection.py", line 897, in _connect_and_handshake
    self.disconnect()
  File "/repo/redis/connection.py", line 1044, in disconnect
    conn_sock.shutdown(socket.SHUT_RDWR)
    ^^^^^^^^^^^^^^^^^^
AttributeError: 'object' object has no attribute 'shutdown'

  File "/repo/redis/connection.py", line 855, in connect_check_health
    self.retry.call_with_retry(
  File "/repo/redis/retry.py", line 116, in call_with_retry
    return do()
           ^^^^
  File "/repo/redis/connection.py", line 856, in <lambda>
    lambda: self._connect_and_handshake(check_health),
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/redis/connection.py", line 897, in _connect_and_handshake
    self.disconnect()
  File "/repo/redis/connection.py", line 1044, in disconnect
    conn_sock.shutdown(socket.SHUT_RDWR)
    ^^^^^^^^^^^^^^^^^^
AttributeError: 'object' object has no attribute 'shutdown'

52:class OutOfMemoryError(ResponseError):
64:class ExecAbortError(ResponseError):
68:class ReadOnlyError(ResponseError):
72:class NoPermissionError(ResponseError):
76:class ModuleError(ResponseError):
80:class LockError(RedisError, ValueError):
83:    # NOTE: For backwards compatibility, this class derives from ValueError.
91:class LockNotOwnedError(LockError):
97:class ChildDeadlockedError(Exception):
103:class AuthenticationWrongNumberOfAr
…[128 chars omitted]…
skError(ResponseError):
171:class TryAgainError(ResponseError):
182:class ClusterCrossSlotError(ResponseError):
192:class MovedError(AskError):
202:class MasterDownError(ClusterDownError):
211:class SlotNotCoveredError(RedisClusterException):
223:class MaxConnectionsError(ConnectionError):
232:class CrossSlotTransactionError(RedisClusterException):
250:class ExternalAuthProviderError(ConnectionError):

        self.retry = Retry(NoBackoff(), 2)
    def _connect(self):
        attempts["n"] += 1
        import socket as _s; return _s.socket()
    def _host_error(self):
        return ("localhost", 6379)
    def _error_message(self, e):
        return str(e)

# 1. retryable handshake failure -> retried, then ConnectionError
attempts["n"] = 0
c = FakeConn()
def failing_handshake(check_health=True):
    raise ConnectionError("ECONNREFUSED during handshake")
c.on_connect_check_health = failing_handshake
try:
    c.connect_check_health()
    print("FAIL
…[447 chars omitted]…
, attempts["n"])
assert attempts["n"] == 1, attempts

# 3. socket connect fails -> still retried
attempts["n"] = 0
c = FakeConn()
def fail_connect():
    raise ConnectionError("[Errno 111] Connection refused")
c._connect = fail_connect
try:
    c.connect_check_health()
except ConnectionError as e:
    print("connect-only:", e, "attempts:", attempts["n"])
assert attempts["n"] == 3, attempts
print("OK")

           ^^^^^^^^^^^^^^^
  File "/tmp/repro_a.py", line 31, in main
    assert attempts["n"] == 3
           ^^^^^^^^^^^^^^^^^^
AssertionError

           ^^^^^^^^^^^^^^^
  File "/tmp/repro_a.py", line 32, in main
    assert attempts["n"] == 3
           ^^^^^^^^^^^^^^^^^^
AssertionError

class SSLConnection(Connection):
    """Manages SSL connections to and from the Redis server(s).
    This class extends the Connection class, adding SSL functionality, and making
    use of ssl.SSLContext (https://docs.python.org/3/library/ssl.html#ssl.SSLContext)
    """

    def __init__(
        self,
        ssl_keyfile: Optional[str] = None,
        ssl_certfile: Optional[str] = None,
        ssl_cert_reqs: Union[str, ssl.VerifyMode] = "required",
        ssl_include_verify_flags: Optional[List["ssl.Verify
…[964 chars omitted]…
  def _connection_arguments(self) -> Mapping:
        kwargs = super()._connection_arguments()
        kwargs["ssl"] = self.ssl_context.get()
        return kwargs

    @property
    def keyfile(self):
        return self.ssl_context.keyfile

    @property
    def certfile(self):
        return self.ssl_context.certfile

    @property
    def cert_reqs(self):
        return self.ssl_context.cert_reqs

        if self.client_name:
            pieces.append(("client_name", self.client_name))
        return pieces

    def _connection_arguments(self) -> Mapping:
        return {"host": self.host, "port": self.port}

    async def _connect(self):
        """Create a TCP socket connection"""
        async with async_timeout(self.socket_connect_timeout):
            reader, writer = await asyncio.open_connection(
                **self._connection_arguments()
            )
        self.
…[90 chars omitted]…
nfo("socket")
        if sock:
            sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
            try:
                # TCP_KEEPALIVE
                if self.socket_keepalive:
                    sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
                    for k, v in self.socket_keepalive_options.items():
                        sock.setsockopt(socket.SOL_TCP, k, v)

           ^^^^^^^^^^^^^^^
  File "/tmp/repro_a.py", line 50, in run
    await main(srv)
  File "/tmp/repro_a.py", line 33, in main
    assert attempts["n"] == 3
           ^^^^^^^^^^^^^^^^^^
AssertionError

           ^^^^^^^^^^^^^^^
  File "/tmp/repro_a.py", line 50, in run
    await main()
  File "/tmp/repro_a.py", line 33, in main
    assert attempts["n"] == 3
           ^^^^^^^^^^^^^^^^^^
AssertionError

On branch master
Changes not staged for commit:
  (use "git add <file>..." to update what will be committed)
  (use "git restore <file>..." to discard changes in working directory)
	modified:   redis/asyncio/connection.py
	modified:   redis/connection.py

no changes added to commit (use "git add" and/or "git commit -a")
Dropped refs/stash@{0} (617f50ebfdc1fe2430e86b2d450649a2d5369dc9)

             else:
-                await self._connect()
+                await self._connect_and_handshake(check_health)
         except asyncio.CancelledError:
             raise  # in 3.7 and earlier, this is an Exception, not BaseException
         except (socket.timeout, asyncio.TimeoutError):
             raise TimeoutError("Timeout connecting t
…[4501 chars omitted]…
self.on_connect_check_health(check_health=check_health)
+            else:
+                # Use the passed function redis_connect_func
+                self.redis_connect_func(self)
+        except BaseException:
+            # clean up after any error in the handshake
+            self.disconnect()
+            raise
+        return sock
+
     @abstractmethod
     def _connect(self):
         pass

class UnixDomainSocketConnection(AbstractConnection):
    "Manages UDS communication to and from a Redis server"

    def __init__(self, *, path: str = "", **kwargs):
        self.path = path
        super().__init__(**kwargs)

    def repr_pieces(self) -> Iterable[Tuple[str, Union[str, int]]]:
        pieces = [("path", self.path), ("db", self.db)]
        if self.c
…[143 chars omitted]…
h async_timeout(self.socket_connect_timeout):
            reader, writer = await asyncio.open_unix_connection(path=self.path)
        self._reader = reader
        self._writer = writer
        await self.on_connect()
redis/cluster.py:671:        # inside the RedisCluster.on_connect() function
redis/cluster.py:782:        connection.on_connect()
redis/connection.py:1419:        self._conn.on_connect()

        connection.on_connect()

        if self.read_from_replicas or self.load_balancing_strategy:
            # Sending READONLY command to server to configure connection as
            # readonly. Since each cluster node may change its server type due
            # to a failover, we should establish a READONLY connection
            # regardless of the server type. If this is a primary connection,
            # READONLY would not aff
…[620 chars omitted]…
ERSION) == 1
        ):
            raise ConnectionError(
                "To maximize compatibility with all Redis products, client-side caching is supported by Redis 7.4 or later"  # noqa: E501
            )

    def on_connect(self):
        self._conn.on_connect()

    def disconnect(self, *args):
        with self._cache_lock:
            self._cache.flush()
        self._conn.disconnect(*args)

        super().__init__(**kwargs)

    def repr_pieces(self) -> Iterable[Tuple[str, Union[str, int]]]:
        pieces = [("path", self.path), ("db", self.db)]
        if self.client_name:
            pieces.append(("client_name", self.client_name))
        return pieces

    async def _connect(self):
        async with async_timeout(self.socket_connect_timeout):
            reader, writer = await asyncio.open_unix_connection(path=self.path)
        self._reader = reader
        self._writer = writer
        await self.on_connect()

    def _host_error(self) -> str:
        return self.path


FALSE_STRINGS = ("0", "F", "FALSE", "N", "NO")


def to_bool(value) -> Optional[bool]:
    if value is None or value == "":
        return None
    if isinstance(value, str) and value.upper() in FALSE_STRINGS:
        return False
    return bool(value)