def on_connect_check_health()

in redis/connection.py [0:0]

• 80 lines of code
• 37 McCabe index (conditional complexity)

    def on_connect_check_health(self, check_health: bool = True):
        "Initialize the connection, authenticate and select a database"
        self._parser.on_connect(self)
        parser = self._parser

        auth_args = None
        # if credential provider or username and/or password are set, authenticate
        if self.credential_provider or (self.username or self.password):
            cred_provider = (
                self.credential_provider
                or UsernamePasswordCredentialProvider(self.username, self.password)
            )
            auth_args = cred_provider.get_credentials()

        # if resp version is specified and we have auth args,
        # we need to send them via HELLO
        if auth_args and 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)
            if len(auth_args) == 1:
                auth_args = ["default", auth_args[0]]
            # avoid checking health here -- PING will fail if we try
            # to check the health prior to the AUTH
            self.send_command(
                "HELLO", self.protocol, "AUTH", *auth_args, check_health=False
            )
            self.handshake_metadata = self.read_response()
            # if response.get(b"proto") != self.protocol and response.get(
            #     "proto"
            # ) != self.protocol:
            #     raise ConnectionError("Invalid RESP version")
        elif auth_args:
            # avoid checking health here -- PING will fail if we try
            # to check the health prior to the AUTH
            self.send_command("AUTH", *auth_args, check_health=False)

            try:
                auth_response = self.read_response()
            except AuthenticationWrongNumberOfArgsError:
                # a username and password were specified but the Redis
                # server seems to be < 6.0.0 which expects a single password
                # arg. retry auth with just the password.
                # https://github.com/andymccurdy/redis-py/issues/1274
                self.send_command("AUTH", auth_args[-1], check_health=False)
                auth_response = self.read_response()

            if str_if_bytes(auth_response) != "OK":
                raise AuthenticationError("Invalid Username or Password")

        # if resp version is specified, switch 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)
            self.handshake_metadata = self.read_response()
            if (
                self.handshake_metadata.get(b"proto") != self.protocol
                and self.handshake_metadata.get("proto") != self.protocol
            ):
                raise ConnectionError("Invalid RESP version")

        # if a client_name is given, set it
        if self.client_name:
            self.send_command(
                "CLIENT",
                "SETNAME",
                self.client_name,
                check_health=check_health,
            )
            if str_if_bytes(self.read_response()) != "OK":
                raise ConnectionError("Error setting client name")

        try:
            # set the library name and version
            if self.lib_name:
                self.send_command(
                    "CLIENT",
                    "SETINFO",
                    "LIB-NAME",
                    self.lib_name,
                    check_health=check_health,
                )
                self.read_response()
        except ResponseError:
            pass

        try:
            if self.lib_version:
                self.send_command(
                    "CLIENT",
                    "SETINFO",
                    "LIB-VER",
                    self.lib_version,
                    check_health=check_health,
                )
                self.read_response()
        except ResponseError:
            pass

        # if a database is specified, switch to it
        if self.db:
            self.send_command("SELECT", self.db, check_health=check_health)
            if str_if_bytes(self.read_response()) != "OK":
                raise ConnectionError("Invalid Database")