def connect_to_reader()

in aws_advanced_python_wrapper/writer_failover_handler.py [0:0]

• 15 lines of code
• 8 McCabe index (conditional complexity)

    def connect_to_reader(self) -> None:
        while not self._timeout_event.is_set():
            if self._current_topology is None:
                raise AwsWrapperError(Messages.get("WriterFailoverHandler.CurrentTopologyNone"))
            try:
                conn_result: ReaderFailoverResult = self._reader_failover_handler.get_reader_connection(self._current_topology)

                # check if valid reader connection
                if conn_result.is_connected and conn_result.connection is not None and conn_result.new_host is not None:
                    self._current_reader_connection = conn_result.connection
                    self._current_reader_host = conn_result.new_host
                    logger.debug("WriterFailoverHandler.TaskBConnectedToReader", self._current_reader_host.url)
                    break

            except Exception:
                pass
            logger.debug("WriterFailoverHandler.TaskBFailedToConnectToAnyReader")
            sleep(1)