def _check_host_status()

in aws_advanced_python_wrapper/host_monitoring_plugin.py [0:0]

• 30 lines of code
• 9 McCabe index (conditional complexity)

    def _check_host_status(self, host_check_timeout_ms: int) -> HostStatus:
        context = self._telemetry_factory.open_telemetry_context(
            "connection status check", TelemetryTraceLevel.FORCE_TOP_LEVEL)
        context.set_attribute("url", self._host_info.url)

        start_ns = perf_counter_ns()
        try:
            driver_dialect = self._plugin_service.driver_dialect
            if self._monitoring_conn is None or driver_dialect.is_closed(self._monitoring_conn):
                props_copy: Properties = copy(self._props)
                for key, value in self._props.items():
                    if key.startswith(Monitor._MONITORING_PROPERTY_PREFIX):
                        props_copy[key[len(Monitor._MONITORING_PROPERTY_PREFIX):len(key)]] = value
                        props_copy.pop(key, None)

                # Set a default connect timeout if the user hasn't configured one
                if props_copy.get(WrapperProperties.CONNECT_TIMEOUT_SEC.name, None) is None:
                    props_copy[WrapperProperties.CONNECT_TIMEOUT_SEC.name] = Monitor._DEFAULT_CONNECT_TIMEOUT_SEC

                logger.debug("Monitor.OpeningMonitorConnection", self._host_info.url)
                start_ns = perf_counter_ns()
                self._monitoring_conn = self._plugin_service.force_connect(self._host_info, props_copy, None)
                logger.debug("Monitor.OpenedMonitorConnection", self._host_info.url)
                return Monitor.HostStatus(True, perf_counter_ns() - start_ns)

            start_ns = perf_counter_ns()
            is_available = self._is_host_available(self._monitoring_conn, host_check_timeout_ms / 1000)
            if not is_available:
                self._host_invalid_counter.inc()
            return Monitor.HostStatus(is_available, perf_counter_ns() - start_ns)
        except Exception:
            self._host_invalid_counter.inc()
            return Monitor.HostStatus(False, perf_counter_ns() - start_ns)
        finally:
            context.close_context()