def open_connection()

in mysql-connector-python/lib/mysql/connector/network.py [0:0]
53 lines of code
13 McCabe index (conditional complexity)

    def open_connection(self) -> None:
        """Open the TCP/IP connection to the MySQL server."""
        # pylint: disable=no-member
        # Get address information
        addrinfo: Tuple[
            socket.AddressFamily,
            socket.SocketKind,
            int,
            str,
            Union[tuple[str, int], tuple[str, int, int, int], tuple[int, bytes]],
        ] = (None, None, None, None, None)
        try:
            addrinfos = socket.getaddrinfo(
                self.server_host,
                self.server_port,
                0,
                socket.SOCK_STREAM,
                socket.SOL_TCP,
            )
            # If multiple results we favor IPv4, unless IPv6 was forced.
            for info in addrinfos:
                if self.force_ipv6 and info[0] == socket.AF_INET6:
                    addrinfo = info
                    break
                if info[0] == socket.AF_INET:
                    addrinfo = info
                    break
            if self.force_ipv6 and addrinfo[0] is None:
                raise InterfaceError(f"No IPv6 address found for {self.server_host}")
            if addrinfo[0] is None:
                addrinfo = addrinfos[0]
        except IOError as err:
            raise InterfaceError(
                errno=2003,
                values=(self.server_host, self.server_port, _strioerror(err)),
            ) from err

        (self._family, socktype, proto, _, sockaddr) = addrinfo

        # Instantiate the socket and connect
        try:
            self.sock = socket.socket(self._family, socktype, proto)
            self.sock.settimeout(self._connection_timeout)
            self.sock.connect(sockaddr)
        except (socket.timeout, TimeoutError) as err:
            raise ConnectionTimeoutError(
                errno=2003,
                values=(
                    self.server_host,
                    self.server_port,
                    _strioerror(err),
                ),
            ) from err
        except IOError as err:
            raise InterfaceError(
                errno=2003,
                values=(self.server_host, self.server_port, _strioerror(err)),
            ) from err
        except Exception as err:
            raise InterfaceError(str(err)) from err