def connect()

in aws_advanced_python_wrapper/custom_endpoint_plugin.py [0:0]

• 27 lines of code
• 5 McCabe index (conditional complexity)

    def connect(
            self,
            target_driver_func: Callable,
            driver_dialect: DriverDialect,
            host_info: HostInfo,
            props: Properties,
            is_initial_connection: bool,
            connect_func: Callable) -> Connection:
        if not self._rds_utils.is_rds_custom_cluster_dns(host_info.host):
            return connect_func()

        self._custom_endpoint_host_info = host_info
        logger.debug("CustomEndpointPlugin.ConnectionRequestToCustomEndpoint", host_info.host)

        self._custom_endpoint_id = self._rds_utils.get_cluster_id(host_info.host)
        if not self._custom_endpoint_id:
            raise AwsWrapperError(
                Messages.get_formatted(
                    "CustomEndpointPlugin.ErrorParsingEndpointIdentifier", self._custom_endpoint_host_info.host))

        hostname = self._custom_endpoint_host_info.host
        self._region = self._region_utils.get_region_from_hostname(hostname)
        if not self._region:
            error_message = "RdsUtils.UnsupportedHostname"
            logger.debug(error_message, hostname)
            raise AwsWrapperError(Messages.get_formatted(error_message, hostname))

        monitor = self._create_monitor_if_absent(props)
        if self._should_wait_for_info:
            self._wait_for_info(monitor)

        return connect_func()