def _switch_to_reader_connection()

in aws_advanced_python_wrapper/read_write_splitting_plugin.py [0:0]

• 25 lines of code
• 12 McCabe index (conditional complexity)

    def _switch_to_reader_connection(self, hosts: Tuple[HostInfo, ...]):
        current_host = self._plugin_service.current_host_info
        current_conn = self._plugin_service.current_connection
        driver_dialect = self._plugin_service.driver_dialect
        if (current_host is not None and current_host.role == HostRole.READER and
                self._is_connection_usable(current_conn, driver_dialect)):
            return

        hostnames = [host_info.host for host_info in hosts]
        if self._reader_host_info is not None and self._reader_host_info.host not in hostnames:
            # The old reader cannot be used anymore because it is no longer in the list of allowed hosts.
            self._close_connection_if_idle(self._reader_connection)

        self._in_read_write_split = True
        if not self._is_connection_usable(self._reader_connection, driver_dialect):
            self._initialize_reader_connection(hosts)
        elif self._reader_connection is not None and self._reader_host_info is not None:
            try:
                self._switch_current_connection_to(self._reader_connection, self._reader_host_info)
                logger.debug("ReadWriteSplittingPlugin.SwitchedFromWriterToReader", self._reader_host_info.url)
            except Exception:
                logger.debug("ReadWriteSplittingPlugin.ErrorSwitchingToCachedReader", self._reader_host_info.url)

                self._reader_connection.close()
                self._reader_connection = None
                self._reader_host_info = None
                self._initialize_reader_connection(hosts)

        if self._is_writer_conn_from_internal_pool:
            self._close_connection_if_idle(self._writer_connection)