def _create_host()

in aws_advanced_python_wrapper/host_list_provider.py [0:0]

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

    def _create_host(self, record: Tuple) -> HostInfo:
        """
        Convert a topology query record into a :py:class:`HostInfo`
        object containing the information for a database instance in the cluster.

        :param record: a query record containing information about a database instance in the cluster.
        :return: a :py:class:`HostInfo` object representing a database instance in the cluster.
        """

        # According to TopologyAwareDatabaseDialect.topology_query the result set
        # should contain 4 columns: instance ID, 1/0 (writer/reader), CPU utilization, host lag in ms.
        # There might be a 5th column specifying the last update time.
        if not self._cluster_instance_template:
            raise AwsWrapperError(Messages.get("RdsHostListProvider.UninitializedClusterInstanceTemplate"))
        if not self._initial_host_info:
            raise AwsWrapperError(Messages.get("RdsHostListProvider.UninitializedInitialHostInfo"))

        host_id: str = record[0]
        is_writer: bool = record[1]
        last_update: datetime
        if len(record) > 4 and isinstance(record[4], datetime):
            last_update = record[4]
        else:
            last_update = datetime.now()

        host_id = host_id if host_id else "?"
        endpoint = self._cluster_instance_template.host.replace("?", host_id)
        port = self._cluster_instance_template.port \
            if self._cluster_instance_template.is_port_specified() \
            else self._initial_host_info.port
        host_info = HostInfo(
            host=endpoint,
            port=port,
            availability=HostAvailability.AVAILABLE,
            host_availability_strategy=create_host_availability_strategy(self._props),
            role=HostRole.WRITER if is_writer else HostRole.READER,
            last_update_time=last_update,
            host_id=host_id)
        host_info.add_alias(host_id)
        return host_info