def _process_query_results()

in aws_advanced_python_wrapper/host_list_provider.py [0:0]

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

    def _process_query_results(self, cursor: Cursor) -> Tuple[HostInfo, ...]:
        """
        Form a list of hosts from the results of the topology query.
        :param cursor: The Cursor object containing a reference to the results of the topology query.
        :return: a tuple of hosts representing the database topology.
        An empty tuple will be returned if the query results did not include a writer instance.
        """
        host_map = {}
        for record in cursor:
            host = self._create_host(record)
            host_map[host.host] = host

        hosts = []
        writers = []
        for host in host_map.values():
            if host.role == HostRole.WRITER:
                writers.append(host)
            else:
                hosts.append(host)

        if len(writers) == 0:
            logger.error("RdsHostListProvider.InvalidTopology")
            hosts.clear()
        elif len(writers) == 1:
            hosts.append(writers[0])
        else:
            # Take the latest updated writer host as the current writer. All others will be ignored.
            existing_writers: List[HostInfo] = [x for x in writers if x is not None]
            existing_writers.sort(reverse=True, key=lambda h: h.last_update_time is not None and h.last_update_time)
            hosts.append(existing_writers[0])

        return tuple(hosts)