def _update_capacity_blocks_info_from_ec2()

in src/slurm_plugin/capacity_block_manager.py [0:0]

• 19 lines of code
• 4 McCabe index (conditional complexity)

    def _update_capacity_blocks_info_from_ec2(self, capacity_blocks: Dict[str, CapacityBlock]):
        """
        Update capacity blocks in given capacity_blocks by adding capacity reservation info.

        This method is called every time the CapacityBlockManager is re-initialized,
        so when it starts/is restarted or when fleet configuration changes.
        """
        capacity_block_ids = list(capacity_blocks.keys())
        logger.info("Retrieving Capacity Blocks information from EC2 for %s", ",".join(capacity_block_ids))
        try:
            capacity_block_reservations_info: List[CapacityReservationInfo] = (
                self.ec2_client.describe_capacity_reservations(capacity_block_ids)
            )

            for capacity_block_reservation_info in capacity_block_reservations_info:
                capacity_block_id = capacity_block_reservation_info.capacity_reservation_id()
                try:
                    capacity_blocks[capacity_block_id].update_capacity_block_reservation_info(
                        capacity_block_reservation_info
                    )
                except KeyError:
                    # should never happen
                    logger.error(f"Unable to find Capacity Block {capacity_block_id} in the internal map.")
        except AWSClientError as e:
            msg = f"Unable to retrieve Capacity Blocks information from EC2. {e}"
            logger.error(msg)
            raise CapacityBlockManagerError(msg)