def __init_

def init()

in aws_advanced_python_wrapper/failover_plugin.py [0:0]
38 lines of code
1 McCabe index (conditional complexity)

    def __init__(self, plugin_service: PluginService, props: Properties):
        self._plugin_service = plugin_service
        self._properties = props
        self._host_list_provider_service: HostListProviderService
        self._reader_failover_handler: ReaderFailoverHandler
        self._writer_failover_handler: WriterFailoverHandler

        self._enable_failover_setting = WrapperProperties.ENABLE_FAILOVER.get_bool(self._properties)
        self._failover_timeout_sec = WrapperProperties.FAILOVER_TIMEOUT_SEC.get_float(self._properties)
        self._failover_cluster_topology_refresh_rate_sec = WrapperProperties.FAILOVER_CLUSTER_TOPOLOGY_REFRESH_RATE_SEC.get_float(
            self._properties)
        self._failover_writer_reconnect_interval_sec = WrapperProperties.FAILOVER_WRITER_RECONNECT_INTERVAL_SEC.get_float(
            self._properties)
        self._failover_reader_connect_timeout_sec = WrapperProperties.FAILOVER_READER_CONNECT_TIMEOUT_SEC.get_float(
            self._properties)
        self._telemetry_failover_additional_top_trace_setting = (
            WrapperProperties.TELEMETRY_FAILOVER_ADDITIONAL_TOP_TRACE.get_bool(self._properties))
        self._failover_mode: FailoverMode
        self._is_in_transaction: bool = False
        self._is_closed: bool = False
        self._closed_explicitly: bool = False
        self._last_exception: Optional[Exception] = None
        self._rds_utils = RdsUtils()
        self._rds_url_type: RdsUrlType = self._rds_utils.identify_rds_type(self._properties.get("host"))
        self._stale_dns_helper: StaleDnsHelper = StaleDnsHelper(plugin_service)
        self._saved_read_only_status: bool = False
        self._saved_auto_commit_status: bool = False

        telemetry_factory = self._plugin_service.get_telemetry_factory()
        self._failover_writer_triggered_counter = telemetry_factory.create_counter("writer_failover.triggered.count")
        self._failover_writer_success_counter = telemetry_factory.create_counter(
            "writer_failover.completed.success.count")
        self._failover_writer_failed_counter = telemetry_factory.create_counter(
            "writer_failover.completed.failed.count")
        self._failover_reader_triggered_counter = telemetry_factory.create_counter("reader_failover.triggered.count")
        self._failover_reader_success_counter = telemetry_factory.create_counter(
            "reader_failover.completed.success.count")
        self._failover_reader_failed_counter = telemetry_factory.create_counter(
            "reader_failover.completed.failed.count")

        FailoverPlugin._SUBSCRIBED_METHODS.update(self._plugin_service.network_bound_methods)
def __init__()

def init()
