Repository: skypilot-org/skypilot (branch: master)
Commit date: 2023-11-07T23:13:33Z

Workflow file: .github/workflows/format.yml (format)
name: format

on:
  # Trigger the workflow on push or pull request,
  # but only for the main branch
  push:
    branches:
      - master
      - 'releases/**'
  pull_request:
    branches:
      - master
      - 'releases/**'
jobs:
  format:
    runs-on: ubuntu-latest
    services:
      pypi_wayback:
        image: ghcr.io/waleko/pypi-wayback
        ports:
          - 8629:8080
    strategy:
      matrix:
        python-version: ["3.8"]
    steps:
    - uses: actions/checkout@v3
    - name: Set up Python ${{ matrix.python-version }}
      uses: actions/setup-python@v4
      with:
        python-version: ${{ matrix.python-version }}
    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install yapf==0.32.0
        pip install toml==0.10.2
        pip install black==22.10.0
        pip install isort==5.12.0
    - name: Running yapf
      run: |
        yapf --diff --recursive ./ --exclude 'sky/skylet/ray_patches/**' \
            --exclude 'sky/skylet/providers/aws/**' \
            --exclude 'sky/skylet/providers/gcp/**' \
            --exclude 'sky/skylet/providers/azure/**' \
            --exclude 'sky/skylet/providers/ibm/**'
    - name: Running black
      run: |
        black --diff --check sky/skylet/providers/aws/ \
            sky/skylet/providers/gcp/ \
            sky/skylet/providers/azure/ \
            sky/skylet/providers/ibm/
    - name: Running isort for black formatted files
      run: |
        isort --diff --check --profile black -l 88 -m 3 \
            sky/skylet/providers/ibm/
    - name: Running isort for yapf formatted files
      run: |
        isort --diff --check ./ --sg 'sky/skylet/ray_patches/**' \
            --sg 'sky/skylet/providers/aws/**' \
            --sg 'sky/skylet/providers/gcp/**' \
            --sg 'sky/skylet/providers/azure/**' \
            --sg 'sky/skylet/providers/ibm/**'

env:
  PIP_INDEX_URL: http://localhost:8629/2023-11-07
  UV_INDEX_URL: http://localhost:8629/2023-11-07


Failed build logs:
step_name: format (3.8)/5_Running yapf.txt
log: ##[group]Run yapf --diff --recursive ./ --exclude 'sky/skylet/ray_patches/**' \
[36;1myapf --diff --recursive ./ --exclude 'sky/skylet/ray_patches/**' \[0m
[36;1m    --exclude 'sky/skylet/providers/aws/**' \[0m
[36;1m    --exclude 'sky/skylet/providers/gcp/**' \[0m
[36;1m    --exclude 'sky/skylet/providers/azure/**' \[0m
[36;1m    --exclude 'sky/skylet/providers/ibm/**'[0m
shell: /usr/bin/bash -e {0}
env:
  pythonLocation: /opt/hostedtoolcache/Python/3.8.18/x64
  PKG_CONFIG_PATH: /opt/hostedtoolcache/Python/3.8.18/x64/lib/pkgconfig
  Python_ROOT_DIR: /opt/hostedtoolcache/Python/3.8.18/x64
  Python2_ROOT_DIR: /opt/hostedtoolcache/Python/3.8.18/x64
  Python3_ROOT_DIR: /opt/hostedtoolcache/Python/3.8.18/x64
  LD_LIBRARY_PATH: /opt/hostedtoolcache/Python/3.8.18/x64/lib
##[endgroup]
--- ./sky/resources.py	(original)
+++ ./sky/resources.py	(reformatted)
@@ -152,8 +152,7 @@
             if round(disk_size) != disk_size:
                 with ux_utils.print_exception_no_traceback():
                     raise ValueError(
-                        f"OS disk size must be an integer. Got: {disk_size}."
-                    )
+                        f"OS disk size must be an integer. Got: {disk_size}.")
             self._disk_size = int(disk_size)
         else:
             self._disk_size = _DEFAULT_DISK_SIZE_GB
@@ -167,7 +166,9 @@
             if None in image_id:
                 self._image_id = {self._region: image_id[None].strip()}
             else:
-                self._image_id = {k.strip(): v.strip() for k, v in image_id.items()}
+                self._image_id = {
+                    k.strip(): v.strip() for k, v in image_id.items()
+                }
         self._is_image_managed = _is_image_managed
 
         self._disk_tier = disk_tier
@@ -177,7 +178,8 @@
                 ports = list(ports)
             if not isinstance(ports, list):
                 ports = [ports]
-            ports = resources_utils.simplify_ports([str(port) for port in ports])
+            ports = resources_utils.simplify_ports(
+                [str(port) for port in ports])
             if not ports:
                 # Set to None if empty. This is mainly for resources from
                 # cli, which will comes in as an empty tuple.
@@ -280,8 +282,7 @@
         hardware_str = (
             f"{instance_type}{use_spot}"
             f"{cpus}{memory}{accelerators}{accelerator_args}{image_id}"
-            f"{disk_tier}{disk_size}{ports}"
-        )
+            f"{disk_tier}{disk_size}{ports}")
         # It may have leading ',' (for example, instance_type not set) or empty
         # spaces.  Remove them.
         while hardware_str and hardware_str[0] in (",", " "):
@@ -363,7 +364,8 @@
         if self._accelerators is not None:
             return self._accelerators
         if self.cloud is not None and self._instance_type is not None:
-            return self.cloud.get_accelerators_from_instance_type(self._instance_type)
+            return self.cloud.get_accelerators_from_instance_type(
+                self._instance_type)
         return None
 
     @property
@@ -423,16 +425,14 @@
                 with ux_utils.print_exception_no_traceback():
                     raise ValueError(
                         f'The "cpus" field should be either a number or '
-                        f'a string "<number>+". Found: {cpus!r}'
-                    ) from None
+                        f'a string "<number>+". Found: {cpus!r}') from None
         else:
             num_cpus = float(cpus)
 
         if num_cpus <= 0:
             with ux_utils.print_exception_no_traceback():
                 raise ValueError(
-                    f'The "cpus" field should be positive. Found: {cpus!r}'
-                )
+                    f'The "cpus" field should be positive. Found: {cpus!r}')
 
     def _set_memory(
         self,
@@ -455,16 +455,14 @@
                 with ux_utils.print_exception_no_traceback():
                     raise ValueError(
                         f'The "memory" field should be either a number or '
-                        f'a string "<number>+". Found: {memory!r}'
-                    ) from None
+                        f'a string "<number>+". Found: {memory!r}') from None
         else:
             memory_gb = float(memory)
 
         if memory_gb <= 0:
             with ux_utils.print_exception_no_traceback():
                 raise ValueError(
-                    f'The "cpus" field should be positive. Found: {memory!r}'
-                )
+                    f'The "cpus" field should be positive. Found: {memory!r}')
 
     def _set_accelerators(
         self,
@@ -483,11 +481,9 @@
                     accelerators = {accelerators: 1}
                 else:
                     splits = accelerators.split(":")
-                    parse_error = (
-                        'The "accelerators" field as a str '
-                        "should be <name> or <name>:<cnt>. "
-                        f"Found: {accelerators!r}"
-                    )
+                    parse_error = ('The "accelerators" field as a str '
+                                   "should be <name> or <name>:<cnt>. "
+                                   f"Found: {accelerators!r}")
                     if len(splits) != 2:
                         with ux_utils.print_exception_no_traceback():
                             raise ValueError(parse_error)
@@ -507,15 +503,16 @@
 
             # Canonicalize the accelerator names.
             accelerators = {
-                accelerator_registry.canonicalize_accelerator_name(acc): acc_count
-                for acc, acc_count in accelerators.items()
+                accelerator_registry.canonicalize_accelerator_name(acc):
+                acc_count for acc, acc_count in accelerators.items()
             }
 
             acc, _ = list(accelerators.items())[0]
             if "tpu" in acc.lower():
                 if self.cloud is None:
                     self._cloud = clouds.GCP()
-                assert self.cloud.is_same_cloud(clouds.GCP()), "Cloud must be GCP."
+                assert self.cloud.is_same_cloud(
+                    clouds.GCP()), "Cloud must be GCP."
                 if accelerator_args is None:
                     accelerator_args = {}
                 use_tpu_vm = accelerator_args.get("tpu_vm", False)
@@ -526,8 +523,7 @@
                         with ux_utils.print_exception_no_traceback():
                             raise ValueError(
                                 "Cannot specify instance type"
-                                f' (got "{self.instance_type}") for TPU VM.'
-                            )
+                                f' (got "{self.instance_type}") for TPU VM.')
                 if "runtime_version" not in accelerator_args:
                     if use_tpu_vm:
                         accelerator_args["runtime_version"] = "tpu-vm-base"
@@ -535,8 +531,7 @@
                         accelerator_args["runtime_version"] = "2.12.0"
                     logger.info(
                         "Missing runtime_version in accelerator_args, using"
-                        f' default ({accelerator_args["runtime_version"]})'
-                    )
+                        f' default ({accelerator_args["runtime_version"]})')
 
         self._accelerators = accelerators
         self._accelerator_args = accelerator_args
@@ -549,9 +544,8 @@
         assert self.is_launchable(), self
         return self.cloud.need_cleanup_after_preemption(self)
 
-    def _validate_and_set_region_zone(
-        self, region: Optional[str], zone: Optional[str]
-    ) -> None:
+    def _validate_and_set_region_zone(self, region: Optional[str],
+                                      zone: Optional[str]) -> None:
         if region is None and zone is None:
             return
 
@@ -582,30 +576,28 @@
                         table = log_utils.create_table(["Cloud", "Hint"])
                         table.add_row(["-----", "----"])
                         for cloud, error in cloud_to_errors.items():
-                            reason_str = "\n".join(textwrap.wrap(str(error), 80))
+                            reason_str = "\n".join(textwrap.wrap(
+                                str(error), 80))
                             table.add_row([str(cloud), reason_str])
                         hint = table.get_string()
                     raise ValueError(
                         f"Invalid (region {region!r}, zone {zone!r}) "
-                        f"{cloud_str}. Details:\n{hint}"
-                    )
+                        f"{cloud_str}. Details:\n{hint}")
             elif len(valid_clouds) > 1:
                 with ux_utils.print_exception_no_traceback():
                     raise ValueError(
                         f"Cannot infer cloud from (region {region!r}, zone "
                         f"{zone!r}). Multiple enabled clouds have region/zone "
                         f"of the same names: {valid_clouds}. "
-                        f"To fix: explicitly specify `cloud`."
-                    )
-            logger.debug(
-                f"Cloud is not specified, using {valid_clouds[0]} "
-                f"inferred from region {region!r} and zone {zone!r}"
-            )
+                        f"To fix: explicitly specify `cloud`.")
+            logger.debug(f"Cloud is not specified, using {valid_clouds[0]} "
+                         f"inferred from region {region!r} and zone {zone!r}")
             self._cloud = valid_clouds[0]
 
         # Validate if region and zone exist in the catalog, and set the region
         # if zone is specified.
-        self._region, self._zone = self._cloud.validate_region_zone(region, zone)
+        self._region, self._zone = self._cloud.validate_region_zone(
+            region, zone)
 
     def get_valid_regions_for_launchable(self) -> List[clouds.Region]:
         """Returns a set of `Region`s that can provision this Resources.
@@ -630,12 +622,9 @@
 
         # Filter the regions by the skypilot_config
         ssh_proxy_command_config = skypilot_config.get_nested(
-            (str(self._cloud).lower(), "ssh_proxy_command"), None
-        )
-        if (
-            isinstance(ssh_proxy_command_config, str)
-            or ssh_proxy_command_config is None
-        ):
+            (str(self._cloud).lower(), "ssh_proxy_command"), None)
+        if (isinstance(ssh_proxy_command_config, str) or
+                ssh_proxy_command_config is None):
             # All regions are valid as the regions are not specified for the
             # ssh_proxy_command config.
             return regions
@@ -659,8 +648,7 @@
             logger.warning(
                 f"{yellow}Request {self} cannot be satisfied by any feasible "
                 "region. To fix, check that ssh_proxy_command's region keys "
-                f"include the regions to use.{reset}"
-            )
+                f"include the regions to use.{reset}")
 
         return filtered_regions
 
@@ -675,8 +663,7 @@
                 with ux_utils.print_exception_no_traceback():
                     raise ValueError(
                         f"Invalid instance type {self._instance_type!r} "
-                        f"for cloud {self.cloud}."
-                    )
+                        f"for cloud {self.cloud}.")
         else:
             # If cloud not specified
             valid_clouds = []
@@ -692,8 +679,7 @@
                 with ux_utils.print_exception_no_traceback():
                     raise ValueError(
                         f"Invalid instance type {self._instance_type!r} "
-                        f"{cloud_str}."
-                    )
+                        f"{cloud_str}.")
             if len(valid_clouds) > 1:
                 with ux_utils.print_exception_no_traceback():
                     raise ValueError(
@@ -702,8 +688,7 @@
                     )
             logger.debug(
                 f"Cloud is not specified, using {valid_clouds[0]} "
-                f"inferred from the instance_type {self.instance_type!r}."
-            )
+                f"inferred from the instance_type {self.instance_type!r}.")
             self._cloud = valid_clouds[0]
 
     def _try_validate_cpus_mem(self) -> None:
@@ -715,7 +700,8 @@
             # The _try_validate_instance_type() method infers and sets
             # self.cloud if self.instance_type is not None.
             assert self.cloud is not None
-            cpus, mem = self.cloud.get_vcpus_mem_from_instance_type(self.instance_type)
+            cpus, mem = self.cloud.get_vcpus_mem_from_instance_type(
+                self.instance_type)
             if self.cpus is not None:
                 if self.cpus.endswith("+"):
                     if cpus < float(self.cpus[:-1]):
@@ -723,15 +709,13 @@
                             raise ValueError(
                                 f"{self.instance_type} does not have enough "
                                 f"vCPUs. {self.instance_type} has {cpus} "
-                                f"vCPUs, but {self.cpus} is requested."
-                            )
+                                f"vCPUs, but {self.cpus} is requested.")
                 elif cpus != float(self.cpus):
                     with ux_utils.print_exception_no_traceback():
                         raise ValueError(
                             f"{self.instance_type} does not have the requested "
                             f"number of vCPUs. {self.instance_type} has {cpus} "
-                            f"vCPUs, but {self.cpus} is requested."
-                        )
+                            f"vCPUs, but {self.cpus} is requested.")
             if self.memory is not None:
                 if self.memory.endswith("+"):
                     if mem < float(self.memory[:-1]):
@@ -739,15 +723,13 @@
                             raise ValueError(
                                 f"{self.instance_type} does not have enough "
                                 f"memory. {self.instance_type} has {mem} GB "
-                                f"memory, but {self.memory} is requested."
-                            )
+                                f"memory, but {self.memory} is requested.")
                 elif mem != float(self.memory):
                     with ux_utils.print_exception_no_traceback():
                         raise ValueError(
                             f"{self.instance_type} does not have the requested "
                             f"memory. {self.instance_type} has {mem} GB "
-                            f"memory, but {self.memory} is requested."
-                        )
+                            f"memory, but {self.memory} is requested.")
 
     def _try_validate_spot(self) -> None:
         if self._spot_recovery is None:
@@ -762,31 +744,26 @@
                 raise ValueError(
                     f"Spot recovery strategy {self._spot_recovery} "
                     "is not supported. The strategy should be among "
-                    f"{list(spot.SPOT_STRATEGIES.keys())}"
-                )
+                    f"{list(spot.SPOT_STRATEGIES.keys())}")
 
     def _try_validate_local(self) -> None:
         if isinstance(self._cloud, clouds.Local):
             if self._use_spot:
                 with ux_utils.print_exception_no_traceback():
-                    raise ValueError(
-                        "Local/On-prem mode does not support spot " "instances."
-                    )
+                    raise ValueError("Local/On-prem mode does not support spot "
+                                     "instances.")
             local_instance = clouds.Local.get_default_instance_type()
-            if (
-                self._instance_type is not None
-                and self._instance_type != local_instance
-            ):
+            if (self._instance_type is not None and
+                    self._instance_type != local_instance):
                 with ux_utils.print_exception_no_traceback():
                     raise ValueError(
                         "Local/On-prem mode does not support instance type:"
-                        f" {self._instance_type}."
-                    )
+                        f" {self._instance_type}.")
             if self._image_id is not None:
                 with ux_utils.print_exception_no_traceback():
                     raise ValueError(
-                        "Local/On-prem mode does not support custom " "images."
-                    )
+                        "Local/On-prem mode does not support custom "
+                        "images.")
 
     def extract_docker_image(self) -> Optional[str]:
         if self.image_id is None:
@@ -794,7 +771,7 @@
         if len(self.image_id) == 1 and self.region in self.image_id:
             image_id = self.image_id[self.region]
             if image_id.startswith("docker:"):
-                return image_id[len("docker:") :]
+                return image_id[len("docker:"):]
         return None
 
     def _try_validate_image_id(self) -> None:
@@ -808,63 +785,53 @@
                     if acc.lower().startswith("tpu"):
                         with ux_utils.print_exception_no_traceback():
                             raise ValueError(
-                                "Docker image is not supported for TPU VM."
-                            )
+                                "Docker image is not supported for TPU VM.")
             if self.cloud is not None:
                 self.cloud.check_features_are_supported(
-                    {clouds.CloudImplementationFeatures.DOCKER_IMAGE}
-                )
+                    {clouds.CloudImplementationFeatures.DOCKER_IMAGE})
             return
 
         if self.cloud is None:
             with ux_utils.print_exception_no_traceback():
-                raise ValueError("Cloud must be specified when image_id is provided.")
+                raise ValueError(
+                    "Cloud must be specified when image_id is provided.")
 
         # Apr, 2023 by Hysun(hysun.he@oracle.com): Added support for OCI
-        if (
-            not self._cloud.is_same_cloud(clouds.AWS())
-            and not self._cloud.is_same_cloud(clouds.GCP())
-            and not self._cloud.is_same_cloud(clouds.IBM())
-            and not self._cloud.is_same_cloud(clouds.OCI())
-        ):
+        if (not self._cloud.is_same_cloud(clouds.AWS()) and
+                not self._cloud.is_same_cloud(clouds.GCP()) and
+                not self._cloud.is_same_cloud(clouds.IBM()) and
+                not self._cloud.is_same_cloud(clouds.OCI())):
             with ux_utils.print_exception_no_traceback():
                 raise ValueError(
                     "image_id is only supported for AWS/GCP/IBM/OCI, please "
-                    "explicitly specify the cloud."
-                )
+                    "explicitly specify the cloud.")
 
         if self._region is not None:
             if self._region not in self._image_id:
                 with ux_utils.print_exception_no_traceback():
                     raise ValueError(
                         f"image_id {self._image_id} should contain the image "
-                        f"for the specified region {self._region}."
-                    )
+                        f"for the specified region {self._region}.")
             # Narrow down the image_id to the specified region.
             self._image_id = {self._region: self._image_id[self._region]}
 
         # Check the image_id's are valid.
         for region, image_id in self._image_id.items():
-            if image_id.startswith("skypilot:") and not self._cloud.is_image_tag_valid(
-                image_id, region
-            ):
+            if image_id.startswith(
+                    "skypilot:") and not self._cloud.is_image_tag_valid(
+                        image_id, region):
                 region_str = f" ({region})" if region else ""
                 with ux_utils.print_exception_no_traceback():
                     raise ValueError(
                         f"Image tag {image_id!r} is not valid, please make sure"
-                        f" the tag exists in {self._cloud}{region_str}."
-                    )
-
-            if (
-                self._cloud.is_same_cloud(clouds.AWS())
-                and not image_id.startswith("skypilot:")
-                and region is None
-            ):
+                        f" the tag exists in {self._cloud}{region_str}.")
+
+            if (self._cloud.is_same_cloud(clouds.AWS()) and
+                    not image_id.startswith("skypilot:") and region is None):
                 with ux_utils.print_exception_no_traceback():
                     raise ValueError(
                         "image_id is only supported for AWS in a specific "
-                        "region, please explicitly specify the region."
-                    )
+                        "region, please explicitly specify the region.")
 
         # Validate the image exists and the size is smaller than the disk size.
         for region, image_id in self._image_id.items():
@@ -873,9 +840,8 @@
             image_size = self.cloud.get_image_size(image_id, region)
             if image_size >= self.disk_size:
                 with ux_utils.print_exception_no_traceback():
-                    size_comparison = (
-                        "larger" if image_size > self.disk_size else "equal"
-                    )
+                    size_comparison = ("larger" if image_size > self.disk_size
+                                       else "equal")
                     raise ValueError(
                         f"Image {image_id!r} is {image_size}GB, which is "
                         f"{size_comparison} than the specified disk_size: {self.disk_size}"
@@ -889,25 +855,25 @@
             with ux_utils.print_exception_no_traceback():
                 raise ValueError(
                     f"Invalid disk_tier {self.disk_tier}. "
-                    'Please use one of "high", "medium", or "low".'
-                )
+                    'Please use one of "high", "medium", or "low".')
         if self.instance_type is None:
             return
         if self.cloud is not None:
-            self.cloud.check_disk_tier_enabled(self.instance_type, self.disk_tier)
+            self.cloud.check_disk_tier_enabled(self.instance_type,
+                                               self.disk_tier)
 
     def _try_validate_ports(self) -> None:
         if self.ports is None:
             return
-        if skypilot_config.get_nested(("aws", "security_group_name"), None) is not None:
+        if skypilot_config.get_nested(("aws", "security_group_name"),
+                                      None) is not None:
             with ux_utils.print_exception_no_traceback():
                 raise ValueError(
-                    "Cannot specify ports when AWS security group name is " "specified."
-                )
+                    "Cannot specify ports when AWS security group name is "
+                    "specified.")
         if self.cloud is not None:
             self.cloud.check_features_are_supported(
-                {clouds.CloudImplementationFeatures.OPEN_PORTS}
-            )
+                {clouds.CloudImplementationFeatures.OPEN_PORTS})
         # We don't need to check the ports format since we already done it
         # in resources_utils.simplify_ports
 
@@ -916,13 +882,11 @@
         hours = seconds / 3600
         # Instance.
         hourly_cost = self.cloud.instance_type_to_hourly_cost(
-            self._instance_type, self.use_spot, self._region, self._zone
-        )
+            self._instance_type, self.use_spot, self._region, self._zone)
         # Accelerators (if any).
         if self.accelerators is not None:
             hourly_cost += self.cloud.accelerators_to_hourly_cost(
-                self.accelerators, self.use_spot, self._region, self._zone
-            )
+                self.accelerators, self.use_spot, self._region, self._zone)
         return hourly_cost * hours
 
     def make_deploy_variables(
@@ -939,8 +903,7 @@
         variables are generated by this method.
         """
         cloud_specific_variables = self.cloud.make_deploy_resources_variables(
-            self, cluster_name_on_cloud, region, zones
-        )
+            self, cluster_name_on_cloud, region, zones)
         docker_image = self.extract_docker_image()
         return dict(
             cloud_specific_variables,
@@ -951,7 +914,8 @@
                 "docker_image": docker_image,
                 # Docker container name. The name of the container. Default to
                 # `sky_container`.
-                "docker_container_name": constants.DEFAULT_DOCKER_CONTAINER_NAME,
+                "docker_container_name":
+                    constants.DEFAULT_DOCKER_CONTAINER_NAME,
                 # Docker login config (if any). This helps pull the image from
                 # private registries.
                 "docker_login_config": self._docker_login_config,
@@ -959,8 +923,7 @@
         )
 
     def get_reservations_available_resources(
-        self, specific_reservations: Set[str]
-    ) -> Dict[str, int]:
+            self, specific_reservations: Set[str]) -> Dict[str, int]:
         """Returns the number of available reservation resources."""
         if self.use_spot:
             # GCP's & AWS's reservations do not support spot instances. We
@@ -968,8 +931,8 @@
             # to each cloud if any cloud supports reservations for spot.
             return {}
         return self.cloud.get_reservations_available_resources(
-            self._instance_type, self._region, self._zone, specific_reservations
-        )
+            self._instance_type, self._region, self._zone,
+            specific_reservations)
 
     def less_demanding_than(
         self,
@@ -1009,17 +972,15 @@
                 if not self.image_id.items() <= other.image_id.items():
                     return False
             else:
-                this_image = self.image_id.get(other.region) or self.image_id.get(None)
-                other_image = other.image_id.get(other.region) or other.image_id.get(
-                    None
-                )
+                this_image = self.image_id.get(
+                    other.region) or self.image_id.get(None)
+                other_image = other.image_id.get(
+                    other.region) or other.image_id.get(None)
                 if this_image != other_image:
                     return False
 
-        if (
-            self._instance_type is not None
-            and self._instance_type != other.instance_type
-        ):
+        if (self._instance_type is not None and
+                self._instance_type != other.instance_type):
             return False
         # self._instance_type <= other.instance_type
 
@@ -1034,10 +995,8 @@
                     return False
         # self.accelerators <= other.accelerators
 
-        if (
-            self.accelerator_args is not None
-            and self.accelerator_args != other.accelerator_args
-        ):
+        if (self.accelerator_args is not None and
+                self.accelerator_args != other.accelerator_args):
             return False
         # self.accelerator_args == other.accelerator_args
 
@@ -1049,7 +1008,8 @@
                 return False
             if self.disk_tier != other.disk_tier:
                 types = ["low", "medium", "high"]
-                return types.index(self.disk_tier) < types.index(other.disk_tier)
+                return types.index(self.disk_tier) < types.index(
+                    other.disk_tier)
 
         if check_ports:
             if self.ports is not None:
@@ -1070,42 +1030,37 @@
         for that field.
         """
         is_matched = True
-        if blocked.cloud is not None and not self.cloud.is_same_cloud(blocked.cloud):
+        if blocked.cloud is not None and not self.cloud.is_same_cloud(
+                blocked.cloud):
             is_matched = False
-        if (
-            blocked.instance_type is not None
-            and self.instance_type != blocked.instance_type
-        ):
+        if (blocked.instance_type is not None and
+                self.instance_type != blocked.instance_type):
             is_matched = False
         if blocked.region is not None and self._region != blocked.region:
             is_matched = False
         if blocked.zone is not None and self._zone != blocked.zone:
             is_matched = False
-        if (
-            blocked.accelerators is not None
-            and self.accelerators != blocked.accelerators
-        ):
+        if (blocked.accelerators is not None and
+                self.accelerators != blocked.accelerators):
             is_matched = False
         return is_matched
 
     def is_empty(self) -> bool:
         """Is this Resources an empty request (all fields None)?"""
-        return all(
-            [
-                self.cloud is None,
-                self._instance_type is None,
-                self.cpus is None,
-                self.memory is None,
-                self.accelerators is None,
-                self.accelerator_args is None,
-                not self._use_spot_specified,
-                self.disk_size == _DEFAULT_DISK_SIZE_GB,
-                self.disk_tier is None,
-                self._image_id is None,
-                self.ports is None,
-                self._docker_login_config is None,
-            ]
-        )
+        return all([
+            self.cloud is None,
+            self._instance_type is None,
+            self.cpus is None,
+            self.memory is None,
+            self.accelerators is None,
+            self.accelerator_args is None,
+            not self._use_spot_specified,
+            self.disk_size == _DEFAULT_DISK_SIZE_GB,
+            self.disk_tier is None,
+            self._image_id is None,
+            self.ports is None,
+            self._docker_login_config is None,
+        ])
 
     def copy(self, **override) -> "Resources":
         """Returns a copy of the given Resources."""
@@ -1116,7 +1071,8 @@
             cpus=override.pop("cpus", self.cpus),
             memory=override.pop("memory", self.memory),
             accelerators=override.pop("accelerators", self.accelerators),
-            accelerator_args=override.pop("accelerator_args", self.accelerator_args),
+            accelerator_args=override.pop("accelerator_args",
+                                          self.accelerator_args),
             use_spot=override.pop("use_spot", use_spot),
             spot_recovery=override.pop("spot_recovery", self.spot_recovery),
             disk_size=override.pop("disk_size", self.disk_size),
@@ -1125,10 +1081,10 @@
             image_id=override.pop("image_id", self.image_id),
             disk_tier=override.pop("disk_tier", self.disk_tier),
             ports=override.pop("ports", self.ports),
-            _docker_login_config=override.pop(
-                "_docker_login_config", self._docker_login_config
-            ),
-            _is_image_managed=override.pop("_is_image_managed", self._is_image_managed),
+            _docker_login_config=override.pop("_docker_login_config",
+                                              self._docker_login_config),
+            _is_image_managed=override.pop("_is_image_managed",
+                                           self._is_image_managed),
         )
         assert len(override) == 0
         return resources
@@ -1144,7 +1100,8 @@
                 return False
         return True
 
-    def get_required_cloud_features(self) -> Set[clouds.CloudImplementationFeatures]:
+    def get_required_cloud_features(
+            self) -> Set[clouds.CloudImplementationFeatures]:
         """Returns the set of cloud features required by this Resources."""
         features = set()
         if self.use_spot:
@@ -1162,19 +1119,18 @@
         if config is None:
             return Resources()
 
-        common_utils.validate_schema(
-            config, schemas.get_resources_schema(), "Invalid resources YAML: "
-        )
+        common_utils.validate_schema(config, schemas.get_resources_schema(),
+                                     "Invalid resources YAML: ")
 
         resources_fields = {}
         resources_fields["cloud"] = clouds.CLOUD_REGISTRY.from_str(
-            config.pop("cloud", None)
-        )
+            config.pop("cloud", None))
         resources_fields["instance_type"] = config.pop("instance_type", None)
         resources_fields["cpus"] = config.pop("cpus", None)
         resources_fields["memory"] = config.pop("memory", None)
         resources_fields["accelerators"] = config.pop("accelerators", None)
-        resources_fields["accelerator_args"] = config.pop("accelerator_args", None)
+        resources_fields["accelerator_args"] = config.pop(
+            "accelerator_args", None)
         resources_fields["use_spot"] = config.pop("use_spot", None)
         resources_fields["spot_recovery"] = config.pop("spot_recovery", None)
         resources_fields["disk_size"] = config.pop("disk_size", None)
@@ -1184,9 +1140,9 @@
         resources_fields["disk_tier"] = config.pop("disk_tier", None)
         resources_fields["ports"] = config.pop("ports", None)
         resources_fields["_docker_login_config"] = config.pop(
-            "_docker_login_config", None
-        )
-        resources_fields["_is_image_managed"] = config.pop("_is_image_managed", None)
+            "_docker_login_config", None)
+        resources_fields["_is_image_managed"] = config.pop(
+            "_is_image_managed", None)
 
         if resources_fields["cpus"] is not None:
             resources_fields["cpus"] = str(resources_fields["cpus"])
@@ -1194,8 +1150,7 @@
             resources_fields["memory"] = str(resources_fields["memory"])
         if resources_fields["accelerator_args"] is not None:
             resources_fields["accelerator_args"] = dict(
-                resources_fields["accelerator_args"]
-            )
+                resources_fields["accelerator_args"])
         if resources_fields["disk_size"] is not None:
             resources_fields["disk_size"] = int(resources_fields["disk_size"])
 
@@ -1273,8 +1228,8 @@
             accelerators = state.pop("_accelerators", None)
             if accelerators is not None:
                 accelerators = {
-                    accelerator_registry.canonicalize_accelerator_name(acc): acc_count
-                    for acc, acc_count in accelerators.items()
+                    accelerator_registry.canonicalize_accelerator_name(acc):
+                    acc_count for acc, acc_count in accelerators.items()
                 }
             state["_accelerators"] = accelerators
 
@@ -1304,7 +1259,6 @@
             original_ports = state.get("_ports", None)
             if original_ports is not None:
                 state["_ports"] = resources_utils.simplify_ports(
-                    [str(port) for port in original_ports]
-                )
+                    [str(port) for port in original_ports])
 
         self.__dict__.update(state)
##[error]Process completed with exit code 1.


Changed files:
sky/resources.py

Fix diff:
diff --git a/sky/resources.py b/sky/resources.py
index ef8e888b..2f7aebce 100644
--- a/sky/resources.py
+++ b/sky/resources.py
@@ -42,7 +42,6 @@ class Resources:
     * for provisioning on a cloud
 
     """
-
     # If any fields changed, increment the version. For backward compatibility,
     # modify the __setstate__ method to handle the old version.
     _VERSION = 13
@@ -61,7 +60,7 @@ class Resources:
         zone: Optional[str] = None,
         image_id: Union[Dict[str, str], str, None] = None,
         disk_size: Optional[int] = None,
-        disk_tier: Optional[Literal["high", "medium", "low"]] = None,
+        disk_tier: Optional[Literal['high', 'medium', 'low']] = None,
         ports: Optional[Union[int, str, List[str], Tuple[str]]] = None,
         # Internal use only.
         _docker_login_config: Optional[docker_utils.DockerLoginConfig] = None,
@@ -145,15 +144,14 @@ class Resources:
         self._use_spot = use_spot if use_spot is not None else False
         self._spot_recovery = None
         if spot_recovery is not None:
-            if spot_recovery.strip().lower() != "none":
+            if spot_recovery.strip().lower() != 'none':
                 self._spot_recovery = spot_recovery.upper()
 
         if disk_size is not None:
             if round(disk_size) != disk_size:
                 with ux_utils.print_exception_no_traceback():
                     raise ValueError(
-                        f"OS disk size must be an integer. Got: {disk_size}."
-                    )
+                        f'OS disk size must be an integer. Got: {disk_size}.')
             self._disk_size = int(disk_size)
         else:
             self._disk_size = _DEFAULT_DISK_SIZE_GB
@@ -167,7 +165,9 @@ class Resources:
             if None in image_id:
                 self._image_id = {self._region: image_id[None].strip()}
             else:
-                self._image_id = {k.strip(): v.strip() for k, v in image_id.items()}
+                self._image_id = {
+                    k.strip(): v.strip() for k, v in image_id.items()
+                }
         self._is_image_managed = _is_image_managed
 
         self._disk_tier = disk_tier
@@ -177,7 +177,8 @@ class Resources:
                 ports = list(ports)
             if not isinstance(ports, list):
                 ports = [ports]
-            ports = resources_utils.simplify_ports([str(port) for port in ports])
+            ports = resources_utils.simplify_ports(
+                [str(port) for port in ports])
             if not ports:
                 # Set to None if empty. This is mainly for resources from
                 # cli, which will comes in as an empty tuple.
@@ -228,84 +229,83 @@ class Resources:
             >>> sky.Resources(disk_size=100)
             <Cloud>(disk_size=100)
         """
-        accelerators = ""
-        accelerator_args = ""
+        accelerators = ''
+        accelerator_args = ''
         if self.accelerators is not None:
-            accelerators = f", {self.accelerators}"
+            accelerators = f', {self.accelerators}'
             if self.accelerator_args is not None:
-                accelerator_args = f", accelerator_args={self.accelerator_args}"
+                accelerator_args = f', accelerator_args={self.accelerator_args}'
 
-        cpus = ""
+        cpus = ''
         if self.cpus is not None:
-            cpus = f", cpus={self.cpus}"
+            cpus = f', cpus={self.cpus}'
 
-        memory = ""
+        memory = ''
         if self.memory is not None:
-            memory = f", mem={self.memory}"
+            memory = f', mem={self.memory}'
 
         if isinstance(self.cloud, clouds.Local):
-            return f"{self.cloud}({self.accelerators})"
+            return f'{self.cloud}({self.accelerators})'
 
-        use_spot = ""
+        use_spot = ''
         if self.use_spot:
-            use_spot = "[Spot]"
+            use_spot = '[Spot]'
 
-        image_id = ""
+        image_id = ''
         if self.image_id is not None:
             if None in self.image_id:
-                image_id = f", image_id={self.image_id[None]}"
+                image_id = f', image_id={self.image_id[None]}'
             else:
-                image_id = f", image_id={self.image_id}"
+                image_id = f', image_id={self.image_id}'
 
-        disk_tier = ""
+        disk_tier = ''
         if self.disk_tier is not None:
-            disk_tier = f", disk_tier={self.disk_tier}"
+            disk_tier = f', disk_tier={self.disk_tier}'
 
-        disk_size = ""
+        disk_size = ''
         if self.disk_size != _DEFAULT_DISK_SIZE_GB:
-            disk_size = f", disk_size={self.disk_size}"
+            disk_size = f', disk_size={self.disk_size}'
 
-        ports = ""
+        ports = ''
         if self.ports is not None:
-            ports = f", ports={self.ports}"
+            ports = f', ports={self.ports}'
 
         if self._instance_type is not None:
-            instance_type = f"{self._instance_type}"
+            instance_type = f'{self._instance_type}'
         else:
-            instance_type = ""
+            instance_type = ''
 
         # Do not show region/zone here as `sky status -a` would show them as
         # separate columns. Also, Resources repr will be printed during
         # failover, and the region may be dynamically determined.
         hardware_str = (
-            f"{instance_type}{use_spot}"
-            f"{cpus}{memory}{accelerators}{accelerator_args}{image_id}"
-            f"{disk_tier}{disk_size}{ports}"
-        )
+            f'{instance_type}{use_spot}'
+            f'{cpus}{memory}{accelerators}{accelerator_args}{image_id}'
+            f'{disk_tier}{disk_size}{ports}')
         # It may have leading ',' (for example, instance_type not set) or empty
         # spaces.  Remove them.
-        while hardware_str and hardware_str[0] in (",", " "):
+        while hardware_str and hardware_str[0] in (',', ' '):
             hardware_str = hardware_str[1:]
 
-        cloud_str = "<Cloud>"
+        cloud_str = '<Cloud>'
         if self.cloud is not None:
-            cloud_str = f"{self.cloud}"
+            cloud_str = f'{self.cloud}'
 
-        return f"{cloud_str}({hardware_str})"
+        return f'{cloud_str}({hardware_str})'
 
     @property
     def repr_with_region_zone(self) -> str:
-        region_str = ""
+        region_str = ''
         if self.region is not None:
-            region_str = f", region={self.region}"
-        zone_str = ""
+            region_str = f', region={self.region}'
+        zone_str = ''
         if self.zone is not None:
-            zone_str = f", zone={self.zone}"
+            zone_str = f', zone={self.zone}'
         repr_str = str(self)
-        if repr_str.endswith(")"):
-            repr_str = repr_str[:-1] + f"{region_str}{zone_str})"
+        if repr_str.endswith(')'):
+            repr_str = repr_str[:-1] + f'{region_str}{zone_str})'
         else:
-            repr_str += f"{region_str}{zone_str}"
+            repr_str += f'{region_str}{zone_str}'
         return repr_str
 
     @property
@@ -363,7 +363,8 @@ class Resources:
         if self._accelerators is not None:
             return self._accelerators
         if self.cloud is not None and self._instance_type is not None:
-            return self.cloud.get_accelerators_from_instance_type(self._instance_type)
+            return self.cloud.get_accelerators_from_instance_type(
+                self._instance_type)
         return None
 
     @property
@@ -412,7 +413,7 @@ class Resources:
 
         self._cpus = str(cpus)
         if isinstance(cpus, str):
-            if cpus.endswith("+"):
+            if cpus.endswith('+'):
                 num_cpus_str = cpus[:-1]
             else:
                 num_cpus_str = cpus
@@ -423,16 +424,14 @@ class Resources:
                 with ux_utils.print_exception_no_traceback():
                     raise ValueError(
                         f'The "cpus" field should be either a number or '
-                        f'a string "<number>+". Found: {cpus!r}'
-                    ) from None
+                        f'a string "<number>+". Found: {cpus!r}') from None
         else:
             num_cpus = float(cpus)
 
         if num_cpus <= 0:
             with ux_utils.print_exception_no_traceback():
                 raise ValueError(
-                    f'The "cpus" field should be positive. Found: {cpus!r}'
-                )
+                    f'The "cpus" field should be positive. Found: {cpus!r}')
 
     def _set_memory(
         self,
@@ -444,7 +443,7 @@ class Resources:
 
         self._memory = str(memory)
         if isinstance(memory, str):
-            if memory.endswith("+"):
+            if memory.endswith('+'):
                 num_memory_gb = memory[:-1]
             else:
                 num_memory_gb = memory
@@ -455,16 +454,14 @@ class Resources:
                 with ux_utils.print_exception_no_traceback():
                     raise ValueError(
                         f'The "memory" field should be either a number or '
-                        f'a string "<number>+". Found: {memory!r}'
-                    ) from None
+                        f'a string "<number>+". Found: {memory!r}') from None
         else:
             memory_gb = float(memory)
 
         if memory_gb <= 0:
             with ux_utils.print_exception_no_traceback():
                 raise ValueError(
-                    f'The "cpus" field should be positive. Found: {memory!r}'
-                )
+                    f'The "cpus" field should be positive. Found: {memory!r}')
 
     def _set_accelerators(
         self,
@@ -479,15 +476,13 @@ class Resources:
         """
         if accelerators is not None:
             if isinstance(accelerators, str):  # Convert to Dict[str, int].
-                if ":" not in accelerators:
+                if ':' not in accelerators:
                     accelerators = {accelerators: 1}
                 else:
-                    splits = accelerators.split(":")
-                    parse_error = (
-                        'The "accelerators" field as a str '
-                        "should be <name> or <name>:<cnt>. "
-                        f"Found: {accelerators!r}"
-                    )
+                    splits = accelerators.split(':')
+                    parse_error = ('The "accelerators" field as a str '
+                                   'should be <name> or <name>:<cnt>. '
+                                   f'Found: {accelerators!r}')
                     if len(splits) != 2:
                         with ux_utils.print_exception_no_traceback():
                             raise ValueError(parse_error)
@@ -507,36 +502,35 @@ class Resources:
 
             # Canonicalize the accelerator names.
             accelerators = {
-                accelerator_registry.canonicalize_accelerator_name(acc): acc_count
-                for acc, acc_count in accelerators.items()
+                accelerator_registry.canonicalize_accelerator_name(acc):
+                acc_count for acc, acc_count in accelerators.items()
             }
 
             acc, _ = list(accelerators.items())[0]
-            if "tpu" in acc.lower():
+            if 'tpu' in acc.lower():
                 if self.cloud is None:
                     self._cloud = clouds.GCP()
-                assert self.cloud.is_same_cloud(clouds.GCP()), "Cloud must be GCP."
+                assert self.cloud.is_same_cloud(
+                    clouds.GCP()), 'Cloud must be GCP.'
                 if accelerator_args is None:
                     accelerator_args = {}
-                use_tpu_vm = accelerator_args.get("tpu_vm", False)
+                use_tpu_vm = accelerator_args.get('tpu_vm', False)
                 if use_tpu_vm:
                     tpu_utils.check_gcp_cli_include_tpu_vm()
                 if self.instance_type is not None and use_tpu_vm:
-                    if self.instance_type != "TPU-VM":
+                    if self.instance_type != 'TPU-VM':
                         with ux_utils.print_exception_no_traceback():
                             raise ValueError(
-                                "Cannot specify instance type"
-                                f' (got "{self.instance_type}") for TPU VM.'
-                            )
-                if "runtime_version" not in accelerator_args:
+                                'Cannot specify instance type'
+                                f' (got "{self.instance_type}") for TPU VM.')
+                if 'runtime_version' not in accelerator_args:
                     if use_tpu_vm:
-                        accelerator_args["runtime_version"] = "tpu-vm-base"
+                        accelerator_args['runtime_version'] = 'tpu-vm-base'
                     else:
-                        accelerator_args["runtime_version"] = "2.12.0"
+                        accelerator_args['runtime_version'] = '2.12.0'
                     logger.info(
-                        "Missing runtime_version in accelerator_args, using"
-                        f' default ({accelerator_args["runtime_version"]})'
-                    )
+                        'Missing runtime_version in accelerator_args, using'
+                        f' default ({accelerator_args["runtime_version"]})')
 
         self._accelerators = accelerators
         self._accelerator_args = accelerator_args
@@ -549,9 +543,8 @@ class Resources:
         assert self.is_launchable(), self
         return self.cloud.need_cleanup_after_preemption(self)
 
-    def _validate_and_set_region_zone(
-        self, region: Optional[str], zone: Optional[str]
-    ) -> None:
+    def _validate_and_set_region_zone(self, region: Optional[str],
+                                      zone: Optional[str]) -> None:
         if region is None and zone is None:
             return
 
@@ -571,41 +564,39 @@ class Resources:
 
             if len(valid_clouds) == 0:
                 if len(enabled_clouds) == 1:
-                    cloud_str = f"for cloud {enabled_clouds[0]}"
+                    cloud_str = f'for cloud {enabled_clouds[0]}'
                 else:
-                    cloud_str = f"for any cloud among {enabled_clouds}"
+                    cloud_str = f'for any cloud among {enabled_clouds}'
                 with ux_utils.print_exception_no_traceback():
                     if len(cloud_to_errors) == 1:
                         # UX: if 1 cloud, don't print a table.
                         hint = list(cloud_to_errors.items())[0][-1]
                     else:
-                        table = log_utils.create_table(["Cloud", "Hint"])
-                        table.add_row(["-----", "----"])
+                        table = log_utils.create_table(['Cloud', 'Hint'])
+                        table.add_row(['-----', '----'])
                         for cloud, error in cloud_to_errors.items():
-                            reason_str = "\n".join(textwrap.wrap(str(error), 80))
+                            reason_str = '\n'.join(textwrap.wrap(
+                                str(error), 80))
                             table.add_row([str(cloud), reason_str])
                         hint = table.get_string()
                     raise ValueError(
-                        f"Invalid (region {region!r}, zone {zone!r}) "
-                        f"{cloud_str}. Details:\n{hint}"
-                    )
+                        f'Invalid (region {region!r}, zone {zone!r}) '
+                        f'{cloud_str}. Details:\n{hint}')
             elif len(valid_clouds) > 1:
                 with ux_utils.print_exception_no_traceback():
                     raise ValueError(
-                        f"Cannot infer cloud from (region {region!r}, zone "
-                        f"{zone!r}). Multiple enabled clouds have region/zone "
-                        f"of the same names: {valid_clouds}. "
-                        f"To fix: explicitly specify `cloud`."
-                    )
-            logger.debug(
-                f"Cloud is not specified, using {valid_clouds[0]} "
-                f"inferred from region {region!r} and zone {zone!r}"
-            )
+                        f'Cannot infer cloud from (region {region!r}, zone '
+                        f'{zone!r}). Multiple enabled clouds have region/zone '
+                        f'of the same names: {valid_clouds}. '
+                        f'To fix: explicitly specify `cloud`.')
+            logger.debug(f'Cloud is not specified, using {valid_clouds[0]} '
+                         f'inferred from region {region!r} and zone {zone!r}')
             self._cloud = valid_clouds[0]
 
         # Validate if region and zone exist in the catalog, and set the region
         # if zone is specified.
-        self._region, self._zone = self._cloud.validate_region_zone(region, zone)
+        self._region, self._zone = self._cloud.validate_region_zone(
+            region, zone)
 
     def get_valid_regions_for_launchable(self) -> List[clouds.Region]:
         """Returns a set of `Region`s that can provision this Resources.
@@ -618,24 +609,18 @@ class Resources:
         """
         assert self.is_launchable(), self
 
-        regions = self._cloud.regions_with_offering(
-            self._instance_type,
-            self.accelerators,
-            self._use_spot,
-            self._region,
-            self._zone,
-        )
+        regions = self._cloud.regions_with_offering(self._instance_type,
+                                                    self.accelerators,
+                                                    self._use_spot,
+                                                    self._region, self._zone)
         if self._image_id is not None and None not in self._image_id:
             regions = [r for r in regions if r.name in self._image_id]
 
         # Filter the regions by the skypilot_config
         ssh_proxy_command_config = skypilot_config.get_nested(
-            (str(self._cloud).lower(), "ssh_proxy_command"), None
-        )
-        if (
-            isinstance(ssh_proxy_command_config, str)
-            or ssh_proxy_command_config is None
-        ):
+            (str(self._cloud).lower(), 'ssh_proxy_command'), None)
+        if (isinstance(ssh_proxy_command_config, str) or
+                ssh_proxy_command_config is None):
             # All regions are valid as the regions are not specified for the
             # ssh_proxy_command config.
             return regions
@@ -657,10 +642,9 @@ class Resources:
             yellow = colorama.Fore.YELLOW
             reset = colorama.Style.RESET_ALL
             logger.warning(
-                f"{yellow}Request {self} cannot be satisfied by any feasible "
-                "region. To fix, check that ssh_proxy_command's region keys "
-                f"include the regions to use.{reset}"
-            )
+                f'{yellow}Request {self} cannot be satisfied by any feasible '
+                'region. To fix, check that ssh_proxy_command\'s region keys '
+                f'include the regions to use.{reset}')
 
         return filtered_regions
 
@@ -674,9 +658,8 @@ class Resources:
             if not valid:
                 with ux_utils.print_exception_no_traceback():
                     raise ValueError(
-                        f"Invalid instance type {self._instance_type!r} "
-                        f"for cloud {self.cloud}."
-                    )
+                        f'Invalid instance type {self._instance_type!r} '
+                        f'for cloud {self.cloud}.')
         else:
             # If cloud not specified
             valid_clouds = []
@@ -686,24 +669,22 @@ class Resources:
                     valid_clouds.append(cloud)
             if len(valid_clouds) == 0:
                 if len(enabled_clouds) == 1:
-                    cloud_str = f"for cloud {enabled_clouds[0]}"
+                    cloud_str = f'for cloud {enabled_clouds[0]}'
                 else:
-                    cloud_str = f"for any cloud among {enabled_clouds}"
+                    cloud_str = f'for any cloud among {enabled_clouds}'
                 with ux_utils.print_exception_no_traceback():
                     raise ValueError(
-                        f"Invalid instance type {self._instance_type!r} "
-                        f"{cloud_str}."
-                    )
+                        f'Invalid instance type {self._instance_type!r} '
+                        f'{cloud_str}.')
             if len(valid_clouds) > 1:
                 with ux_utils.print_exception_no_traceback():
                     raise ValueError(
-                        f"Ambiguous instance type {self._instance_type!r}. "
-                        f"Please specify cloud explicitly among {valid_clouds}."
+                        f'Ambiguous instance type {self._instance_type!r}. '
+                        f'Please specify cloud explicitly among {valid_clouds}.'
                     )
             logger.debug(
-                f"Cloud is not specified, using {valid_clouds[0]} "
-                f"inferred from the instance_type {self.instance_type!r}."
-            )
+                f'Cloud is not specified, using {valid_clouds[0]} '
+                f'inferred from the instance_type {self.instance_type!r}.')
             self._cloud = valid_clouds[0]
 
     def _try_validate_cpus_mem(self) -> None:
@@ -715,39 +696,36 @@ class Resources:
             # The _try_validate_instance_type() method infers and sets
             # self.cloud if self.instance_type is not None.
             assert self.cloud is not None
-            cpus, mem = self.cloud.get_vcpus_mem_from_instance_type(self.instance_type)
+            cpus, mem = self.cloud.get_vcpus_mem_from_instance_type(
+                self.instance_type)
             if self.cpus is not None:
-                if self.cpus.endswith("+"):
+                if self.cpus.endswith('+'):
                     if cpus < float(self.cpus[:-1]):
                         with ux_utils.print_exception_no_traceback():
                             raise ValueError(
-                                f"{self.instance_type} does not have enough "
-                                f"vCPUs. {self.instance_type} has {cpus} "
-                                f"vCPUs, but {self.cpus} is requested."
-                            )
+                                f'{self.instance_type} does not have enough '
+                                f'vCPUs. {self.instance_type} has {cpus} '
+                                f'vCPUs, but {self.cpus} is requested.')
                 elif cpus != float(self.cpus):
                     with ux_utils.print_exception_no_traceback():
                         raise ValueError(
-                            f"{self.instance_type} does not have the requested "
-                            f"number of vCPUs. {self.instance_type} has {cpus} "
-                            f"vCPUs, but {self.cpus} is requested."
-                        )
+                            f'{self.instance_type} does not have the requested '
+                            f'number of vCPUs. {self.instance_type} has {cpus} '
+                            f'vCPUs, but {self.cpus} is requested.')
             if self.memory is not None:
-                if self.memory.endswith("+"):
+                if self.memory.endswith('+'):
                     if mem < float(self.memory[:-1]):
                         with ux_utils.print_exception_no_traceback():
                             raise ValueError(
-                                f"{self.instance_type} does not have enough "
-                                f"memory. {self.instance_type} has {mem} GB "
-                                f"memory, but {self.memory} is requested."
-                            )
+                                f'{self.instance_type} does not have enough '
+                                f'memory. {self.instance_type} has {mem} GB '
+                                f'memory, but {self.memory} is requested.')
                 elif mem != float(self.memory):
                     with ux_utils.print_exception_no_traceback():
                         raise ValueError(
-                            f"{self.instance_type} does not have the requested "
-                            f"memory. {self.instance_type} has {mem} GB "
-                            f"memory, but {self.memory} is requested."
-                        )
+                            f'{self.instance_type} does not have the requested '
+                            f'memory. {self.instance_type} has {mem} GB '
+                            f'memory, but {self.memory} is requested.')
 
     def _try_validate_spot(self) -> None:
         if self._spot_recovery is None:
@@ -755,46 +733,41 @@ class Resources:
         if not self._use_spot:
             with ux_utils.print_exception_no_traceback():
                 raise ValueError(
-                    "Cannot specify spot_recovery without use_spot set to True."
+                    'Cannot specify spot_recovery without use_spot set to True.'
                 )
         if self._spot_recovery not in spot.SPOT_STRATEGIES:
             with ux_utils.print_exception_no_traceback():
                 raise ValueError(
-                    f"Spot recovery strategy {self._spot_recovery} "
-                    "is not supported. The strategy should be among "
-                    f"{list(spot.SPOT_STRATEGIES.keys())}"
-                )
+                    f'Spot recovery strategy {self._spot_recovery} '
+                    'is not supported. The strategy should be among '
+                    f'{list(spot.SPOT_STRATEGIES.keys())}')
 
     def _try_validate_local(self) -> None:
         if isinstance(self._cloud, clouds.Local):
             if self._use_spot:
                 with ux_utils.print_exception_no_traceback():
-                    raise ValueError(
-                        "Local/On-prem mode does not support spot " "instances."
-                    )
+                    raise ValueError('Local/On-prem mode does not support spot '
+                                     'instances.')
             local_instance = clouds.Local.get_default_instance_type()
-            if (
-                self._instance_type is not None
-                and self._instance_type != local_instance
-            ):
+            if (self._instance_type is not None and
+                    self._instance_type != local_instance):
                 with ux_utils.print_exception_no_traceback():
                     raise ValueError(
-                        "Local/On-prem mode does not support instance type:"
-                        f" {self._instance_type}."
-                    )
+                        'Local/On-prem mode does not support instance type:'
+                        f' {self._instance_type}.')
             if self._image_id is not None:
                 with ux_utils.print_exception_no_traceback():
                     raise ValueError(
-                        "Local/On-prem mode does not support custom " "images."
-                    )
+                        'Local/On-prem mode does not support custom '
+                        'images.')
 
     def extract_docker_image(self) -> Optional[str]:
         if self.image_id is None:
             return None
         if len(self.image_id) == 1 and self.region in self.image_id:
             image_id = self.image_id[self.region]
-            if image_id.startswith("docker:"):
-                return image_id[len("docker:") :]
+            if image_id.startswith('docker:'):
+                return image_id[len('docker:'):]
         return None
 
     def _try_validate_image_id(self) -> None:
@@ -805,66 +778,56 @@ class Resources:
             # TODO(tian): validate the docker image exists / of reasonable size
             if self.accelerators is not None:
                 for acc in self.accelerators.keys():
-                    if acc.lower().startswith("tpu"):
+                    if acc.lower().startswith('tpu'):
                         with ux_utils.print_exception_no_traceback():
                             raise ValueError(
-                                "Docker image is not supported for TPU VM."
-                            )
+                                'Docker image is not supported for TPU VM.')
             if self.cloud is not None:
                 self.cloud.check_features_are_supported(
-                    {clouds.CloudImplementationFeatures.DOCKER_IMAGE}
-                )
+                    {clouds.CloudImplementationFeatures.DOCKER_IMAGE})
             return
 
         if self.cloud is None:
             with ux_utils.print_exception_no_traceback():
-                raise ValueError("Cloud must be specified when image_id is provided.")
+                raise ValueError(
+                    'Cloud must be specified when image_id is provided.')
 
         # Apr, 2023 by Hysun(hysun.he@oracle.com): Added support for OCI
-        if (
-            not self._cloud.is_same_cloud(clouds.AWS())
-            and not self._cloud.is_same_cloud(clouds.GCP())
-            and not self._cloud.is_same_cloud(clouds.IBM())
-            and not self._cloud.is_same_cloud(clouds.OCI())
-        ):
+        if not self._cloud.is_same_cloud(
+                clouds.AWS()) and not self._cloud.is_same_cloud(
+                    clouds.GCP()) and not self._cloud.is_same_cloud(
+                        clouds.IBM()) and not self._cloud.is_same_cloud(
+                            clouds.OCI()):
             with ux_utils.print_exception_no_traceback():
                 raise ValueError(
-                    "image_id is only supported for AWS/GCP/IBM/OCI, please "
-                    "explicitly specify the cloud."
-                )
+                    'image_id is only supported for AWS/GCP/IBM/OCI, please '
+                    'explicitly specify the cloud.')
 
         if self._region is not None:
             if self._region not in self._image_id:
                 with ux_utils.print_exception_no_traceback():
                     raise ValueError(
-                        f"image_id {self._image_id} should contain the image "
-                        f"for the specified region {self._region}."
-                    )
+                        f'image_id {self._image_id} should contain the image '
+                        f'for the specified region {self._region}.')
             # Narrow down the image_id to the specified region.
             self._image_id = {self._region: self._image_id[self._region]}
 
         # Check the image_id's are valid.
         for region, image_id in self._image_id.items():
-            if image_id.startswith("skypilot:") and not self._cloud.is_image_tag_valid(
-                image_id, region
-            ):
-                region_str = f" ({region})" if region else ""
+            if (image_id.startswith('skypilot:') and
+                    not self._cloud.is_image_tag_valid(image_id, region)):
+                region_str = f' ({region})' if region else ''
                 with ux_utils.print_exception_no_traceback():
                     raise ValueError(
-                        f"Image tag {image_id!r} is not valid, please make sure"
-                        f" the tag exists in {self._cloud}{region_str}."
-                    )
+                        f'Image tag {image_id!r} is not valid, please make sure'
+                        f' the tag exists in {self._cloud}{region_str}.')
 
-            if (
-                self._cloud.is_same_cloud(clouds.AWS())
-                and not image_id.startswith("skypilot:")
-                and region is None
-            ):
+            if (self._cloud.is_same_cloud(clouds.AWS()) and
+                    not image_id.startswith('skypilot:') and region is None):
                 with ux_utils.print_exception_no_traceback():
                     raise ValueError(
-                        "image_id is only supported for AWS in a specific "
-                        "region, please explicitly specify the region."
-                    )
+                        'image_id is only supported for AWS in a specific '
+                        'region, please explicitly specify the region.')
 
         # Validate the image exists and the size is smaller than the disk size.
         for region, image_id in self._image_id.items():
@@ -873,41 +836,40 @@ class Resources:
             image_size = self.cloud.get_image_size(image_id, region)
             if image_size >= self.disk_size:
                 with ux_utils.print_exception_no_traceback():
-                    size_comparison = (
-                        "larger" if image_size > self.disk_size else "equal"
-                    )
+                    size_compare = 'larger than' if image_size > self.disk_size \
+                        else 'equal to'
                     raise ValueError(
-                        f"Image {image_id!r} is {image_size}GB, which is "
-                        f"{size_comparison} than the specified disk_size: {self.disk_size}"
-                        " GB. Please specify a larger disk_size to use this image."
-                    )
+                        f'Image {image_id!r} is {image_size}GB, which is '
+                        f'{size_compare} the specified disk_size: '
+                        f'{self.disk_size} GB. Please specify a larger '
+                        'disk_size to use this image.')
 
     def _try_validate_disk_tier(self) -> None:
         if self.disk_tier is None:
             return
-        if self.disk_tier not in ["high", "medium", "low"]:
+        if self.disk_tier not in ['high', 'medium', 'low']:
             with ux_utils.print_exception_no_traceback():
                 raise ValueError(
-                    f"Invalid disk_tier {self.disk_tier}. "
-                    'Please use one of "high", "medium", or "low".'
-                )
+                    f'Invalid disk_tier {self.disk_tier}. '
+                    'Please use one of "high", "medium", or "low".')
         if self.instance_type is None:
             return
         if self.cloud is not None:
-            self.cloud.check_disk_tier_enabled(self.instance_type, self.disk_tier)
+            self.cloud.check_disk_tier_enabled(self.instance_type,
+                                               self.disk_tier)
 
     def _try_validate_ports(self) -> None:
         if self.ports is None:
             return
-        if skypilot_config.get_nested(("aws", "security_group_name"), None) is not None:
+        if skypilot_config.get_nested(('aws', 'security_group_name'),
+                                      None) is not None:
             with ux_utils.print_exception_no_traceback():
                 raise ValueError(
-                    "Cannot specify ports when AWS security group name is " "specified."
-                )
+                    'Cannot specify ports when AWS security group name is '
+                    'specified.')
         if self.cloud is not None:
             self.cloud.check_features_are_supported(
-                {clouds.CloudImplementationFeatures.OPEN_PORTS}
-            )
+                {clouds.CloudImplementationFeatures.OPEN_PORTS})
         # We don't need to check the ports format since we already done it
         # in resources_utils.simplify_ports
 
@@ -916,21 +878,16 @@ class Resources:
         hours = seconds / 3600
         # Instance.
         hourly_cost = self.cloud.instance_type_to_hourly_cost(
-            self._instance_type, self.use_spot, self._region, self._zone
-        )
+            self._instance_type, self.use_spot, self._region, self._zone)
         # Accelerators (if any).
         if self.accelerators is not None:
             hourly_cost += self.cloud.accelerators_to_hourly_cost(
-                self.accelerators, self.use_spot, self._region, self._zone
-            )
+                self.accelerators, self.use_spot, self._region, self._zone)
         return hourly_cost * hours
 
     def make_deploy_variables(
-        self,
-        cluster_name_on_cloud: str,
-        region: clouds.Region,
-        zones: Optional[List[clouds.Zone]],
-    ) -> Dict[str, Optional[str]]:
+            self, cluster_name_on_cloud: str, region: clouds.Region,
+            zones: Optional[List[clouds.Zone]]) -> Dict[str, Optional[str]]:
         """Converts planned sky.Resources to resource variables.
 
         These variables are divided into two categories: cloud-specific and
@@ -939,8 +896,7 @@ class Resources:
         variables are generated by this method.
         """
         cloud_specific_variables = self.cloud.make_deploy_resources_variables(
-            self, cluster_name_on_cloud, region, zones
-        )
+            self, cluster_name_on_cloud, region, zones)
         docker_image = self.extract_docker_image()
         return dict(
             cloud_specific_variables,
@@ -948,19 +904,18 @@ class Resources:
                 # Docker config
                 # Docker image. The image name used to pull the image, e.g.
                 # ubuntu:latest.
-                "docker_image": docker_image,
+                'docker_image': docker_image,
                 # Docker container name. The name of the container. Default to
                 # `sky_container`.
-                "docker_container_name": constants.DEFAULT_DOCKER_CONTAINER_NAME,
+                'docker_container_name':
+                    constants.DEFAULT_DOCKER_CONTAINER_NAME,
                 # Docker login config (if any). This helps pull the image from
                 # private registries.
-                "docker_login_config": self._docker_login_config,
-            },
-        )
+                'docker_login_config': self._docker_login_config
+            })
 
     def get_reservations_available_resources(
-        self, specific_reservations: Set[str]
-    ) -> Dict[str, int]:
+            self, specific_reservations: Set[str]) -> Dict[str, int]:
         """Returns the number of available reservation resources."""
         if self.use_spot:
             # GCP's & AWS's reservations do not support spot instances. We
@@ -968,12 +923,12 @@ class Resources:
             # to each cloud if any cloud supports reservations for spot.
             return {}
         return self.cloud.get_reservations_available_resources(
-            self._instance_type, self._region, self._zone, specific_reservations
-        )
+            self._instance_type, self._region, self._zone,
+            specific_reservations)
 
     def less_demanding_than(
         self,
-        other: Union[List["Resources"], "Resources"],
+        other: Union[List['Resources'], 'Resources'],
         requested_num_nodes: int = 1,
         check_ports: bool = False,
     ) -> bool:
@@ -1009,17 +964,15 @@ class Resources:
                 if not self.image_id.items() <= other.image_id.items():
                     return False
             else:
-                this_image = self.image_id.get(other.region) or self.image_id.get(None)
-                other_image = other.image_id.get(other.region) or other.image_id.get(
-                    None
-                )
+                this_image = (self.image_id.get(other.region) or
+                              self.image_id.get(None))
+                other_image = (other.image_id.get(other.region) or
+                               other.image_id.get(None))
                 if this_image != other_image:
                     return False
 
-        if (
-            self._instance_type is not None
-            and self._instance_type != other.instance_type
-        ):
+        if (self._instance_type is not None and
+                self._instance_type != other.instance_type):
             return False
         # self._instance_type <= other.instance_type
 
@@ -1034,10 +987,8 @@ class Resources:
                     return False
         # self.accelerators <= other.accelerators
 
-        if (
-            self.accelerator_args is not None
-            and self.accelerator_args != other.accelerator_args
-        ):
+        if (self.accelerator_args is not None and
+                self.accelerator_args != other.accelerator_args):
             return False
         # self.accelerator_args == other.accelerator_args
 
@@ -1048,8 +999,9 @@ class Resources:
             if other.disk_tier is None:
                 return False
             if self.disk_tier != other.disk_tier:
-                types = ["low", "medium", "high"]
-                return types.index(self.disk_tier) < types.index(other.disk_tier)
+                types = ['low', 'medium', 'high']
+                return types.index(self.disk_tier) < types.index(
+                    other.disk_tier)
 
         if check_ports:
             if self.ports is not None:
@@ -1063,72 +1015,68 @@ class Resources:
         # self <= other
         return True
 
-    def should_be_blocked_by(self, blocked: "Resources") -> bool:
+    def should_be_blocked_by(self, blocked: 'Resources') -> bool:
         """Whether this Resources matches the blocked Resources.
 
         If a field in `blocked` is None, it should be considered as a wildcard
         for that field.
         """
         is_matched = True
-        if blocked.cloud is not None and not self.cloud.is_same_cloud(blocked.cloud):
+        if (blocked.cloud is not None and
+                not self.cloud.is_same_cloud(blocked.cloud)):
             is_matched = False
-        if (
-            blocked.instance_type is not None
-            and self.instance_type != blocked.instance_type
-        ):
+        if (blocked.instance_type is not None and
+                self.instance_type != blocked.instance_type):
             is_matched = False
         if blocked.region is not None and self._region != blocked.region:
             is_matched = False
         if blocked.zone is not None and self._zone != blocked.zone:
             is_matched = False
-        if (
-            blocked.accelerators is not None
-            and self.accelerators != blocked.accelerators
-        ):
+        if (blocked.accelerators is not None and
+                self.accelerators != blocked.accelerators):
             is_matched = False
         return is_matched
 
     def is_empty(self) -> bool:
         """Is this Resources an empty request (all fields None)?"""
-        return all(
-            [
-                self.cloud is None,
-                self._instance_type is None,
-                self.cpus is None,
-                self.memory is None,
-                self.accelerators is None,
-                self.accelerator_args is None,
-                not self._use_spot_specified,
-                self.disk_size == _DEFAULT_DISK_SIZE_GB,
-                self.disk_tier is None,
-                self._image_id is None,
-                self.ports is None,
-                self._docker_login_config is None,
-            ]
-        )
-
-    def copy(self, **override) -> "Resources":
+        return all([
+            self.cloud is None,
+            self._instance_type is None,
+            self.cpus is None,
+            self.memory is None,
+            self.accelerators is None,
+            self.accelerator_args is None,
+            not self._use_spot_specified,
+            self.disk_size == _DEFAULT_DISK_SIZE_GB,
+            self.disk_tier is None,
+            self._image_id is None,
+            self.ports is None,
+            self._docker_login_config is None,
+        ])
+
+    def copy(self, **override) -> 'Resources':
         """Returns a copy of the given Resources."""
         use_spot = self.use_spot if self._use_spot_specified else None
         resources = Resources(
-            cloud=override.pop("cloud", self.cloud),
-            instance_type=override.pop("instance_type", self.instance_type),
-            cpus=override.pop("cpus", self.cpus),
-            memory=override.pop("memory", self.memory),
-            accelerators=override.pop("accelerators", self.accelerators),
-            accelerator_args=override.pop("accelerator_args", self.accelerator_args),
-            use_spot=override.pop("use_spot", use_spot),
-            spot_recovery=override.pop("spot_recovery", self.spot_recovery),
-            disk_size=override.pop("disk_size", self.disk_size),
-            region=override.pop("region", self.region),
-            zone=override.pop("zone", self.zone),
-            image_id=override.pop("image_id", self.image_id),
-            disk_tier=override.pop("disk_tier", self.disk_tier),
-            ports=override.pop("ports", self.ports),
-            _docker_login_config=override.pop(
-                "_docker_login_config", self._docker_login_config
-            ),
-            _is_image_managed=override.pop("_is_image_managed", self._is_image_managed),
+            cloud=override.pop('cloud', self.cloud),
+            instance_type=override.pop('instance_type', self.instance_type),
+            cpus=override.pop('cpus', self.cpus),
+            memory=override.pop('memory', self.memory),
+            accelerators=override.pop('accelerators', self.accelerators),
+            accelerator_args=override.pop('accelerator_args',
+                                          self.accelerator_args),
+            use_spot=override.pop('use_spot', use_spot),
+            spot_recovery=override.pop('spot_recovery', self.spot_recovery),
+            disk_size=override.pop('disk_size', self.disk_size),
+            region=override.pop('region', self.region),
+            zone=override.pop('zone', self.zone),
+            image_id=override.pop('image_id', self.image_id),
+            disk_tier=override.pop('disk_tier', self.disk_tier),
+            ports=override.pop('ports', self.ports),
+            _docker_login_config=override.pop('_docker_login_config',
+                                              self._docker_login_config),
+            _is_image_managed=override.pop('_is_image_managed',
+                                           self._is_image_managed),
         )
         assert len(override) == 0
         return resources
@@ -1144,7 +1092,8 @@ class Resources:
                 return False
         return True
 
-    def get_required_cloud_features(self) -> Set[clouds.CloudImplementationFeatures]:
+    def get_required_cloud_features(
+            self) -> Set[clouds.CloudImplementationFeatures]:
         """Returns the set of cloud features required by this Resources."""
         features = set()
         if self.use_spot:
@@ -1158,48 +1107,46 @@ class Resources:
         return features
 
     @classmethod
-    def from_yaml_config(cls, config: Optional[Dict[str, str]]) -> "Resources":
+    def from_yaml_config(cls, config: Optional[Dict[str, str]]) -> 'Resources':
         if config is None:
             return Resources()
 
-        common_utils.validate_schema(
-            config, schemas.get_resources_schema(), "Invalid resources YAML: "
-        )
+        common_utils.validate_schema(config, schemas.get_resources_schema(),
+                                     'Invalid resources YAML: ')
 
         resources_fields = {}
-        resources_fields["cloud"] = clouds.CLOUD_REGISTRY.from_str(
-            config.pop("cloud", None)
-        )
-        resources_fields["instance_type"] = config.pop("instance_type", None)
-        resources_fields["cpus"] = config.pop("cpus", None)
-        resources_fields["memory"] = config.pop("memory", None)
-        resources_fields["accelerators"] = config.pop("accelerators", None)
-        resources_fields["accelerator_args"] = config.pop("accelerator_args", None)
-        resources_fields["use_spot"] = config.pop("use_spot", None)
-        resources_fields["spot_recovery"] = config.pop("spot_recovery", None)
-        resources_fields["disk_size"] = config.pop("disk_size", None)
-        resources_fields["region"] = config.pop("region", None)
-        resources_fields["zone"] = config.pop("zone", None)
-        resources_fields["image_id"] = config.pop("image_id", None)
-        resources_fields["disk_tier"] = config.pop("disk_tier", None)
-        resources_fields["ports"] = config.pop("ports", None)
-        resources_fields["_docker_login_config"] = config.pop(
-            "_docker_login_config", None
-        )
-        resources_fields["_is_image_managed"] = config.pop("_is_image_managed", None)
-
-        if resources_fields["cpus"] is not None:
-            resources_fields["cpus"] = str(resources_fields["cpus"])
-        if resources_fields["memory"] is not None:
-            resources_fields["memory"] = str(resources_fields["memory"])
-        if resources_fields["accelerator_args"] is not None:
-            resources_fields["accelerator_args"] = dict(
-                resources_fields["accelerator_args"]
-            )
-        if resources_fields["disk_size"] is not None:
-            resources_fields["disk_size"] = int(resources_fields["disk_size"])
-
-        assert not config, f"Invalid resource args: {config.keys()}"
+        resources_fields['cloud'] = clouds.CLOUD_REGISTRY.from_str(
+            config.pop('cloud', None))
+        resources_fields['instance_type'] = config.pop('instance_type', None)
+        resources_fields['cpus'] = config.pop('cpus', None)
+        resources_fields['memory'] = config.pop('memory', None)
+        resources_fields['accelerators'] = config.pop('accelerators', None)
+        resources_fields['accelerator_args'] = config.pop(
+            'accelerator_args', None)
+        resources_fields['use_spot'] = config.pop('use_spot', None)
+        resources_fields['spot_recovery'] = config.pop('spot_recovery', None)
+        resources_fields['disk_size'] = config.pop('disk_size', None)
+        resources_fields['region'] = config.pop('region', None)
+        resources_fields['zone'] = config.pop('zone', None)
+        resources_fields['image_id'] = config.pop('image_id', None)
+        resources_fields['disk_tier'] = config.pop('disk_tier', None)
+        resources_fields['ports'] = config.pop('ports', None)
+        resources_fields['_docker_login_config'] = config.pop(
+            '_docker_login_config', None)
+        resources_fields['_is_image_managed'] = config.pop(
+            '_is_image_managed', None)
+
+        if resources_fields['cpus'] is not None:
+            resources_fields['cpus'] = str(resources_fields['cpus'])
+        if resources_fields['memory'] is not None:
+            resources_fields['memory'] = str(resources_fields['memory'])
+        if resources_fields['accelerator_args'] is not None:
+            resources_fields['accelerator_args'] = dict(
+                resources_fields['accelerator_args'])
+        if resources_fields['disk_size'] is not None:
+            resources_fields['disk_size'] = int(resources_fields['disk_size'])
+
+        assert not config, f'Invalid resource args: {config.keys()}'
         return Resources(**resources_fields)
 
     def to_yaml_config(self) -> Dict[str, Union[str, int]]:
@@ -1207,28 +1154,28 @@ class Resources:
         config = {}
 
         def add_if_not_none(key, value):
-            if value is not None and value != "None":
+            if value is not None and value != 'None':
                 config[key] = value
 
-        add_if_not_none("cloud", str(self.cloud))
-        add_if_not_none("instance_type", self.instance_type)
-        add_if_not_none("cpus", self.cpus)
-        add_if_not_none("memory", self.memory)
-        add_if_not_none("accelerators", self.accelerators)
-        add_if_not_none("accelerator_args", self.accelerator_args)
+        add_if_not_none('cloud', str(self.cloud))
+        add_if_not_none('instance_type', self.instance_type)
+        add_if_not_none('cpus', self.cpus)
+        add_if_not_none('memory', self.memory)
+        add_if_not_none('accelerators', self.accelerators)
+        add_if_not_none('accelerator_args', self.accelerator_args)
 
         if self._use_spot_specified:
-            add_if_not_none("use_spot", self.use_spot)
-        config["spot_recovery"] = self.spot_recovery
-        config["disk_size"] = self.disk_size
-        add_if_not_none("region", self.region)
-        add_if_not_none("zone", self.zone)
-        add_if_not_none("image_id", self.image_id)
-        add_if_not_none("disk_tier", self.disk_tier)
-        add_if_not_none("ports", self.ports)
-        add_if_not_none("_docker_login_config", self._docker_login_config)
+            add_if_not_none('use_spot', self.use_spot)
+        config['spot_recovery'] = self.spot_recovery
+        config['disk_size'] = self.disk_size
+        add_if_not_none('region', self.region)
+        add_if_not_none('zone', self.zone)
+        add_if_not_none('image_id', self.image_id)
+        add_if_not_none('disk_tier', self.disk_tier)
+        add_if_not_none('ports', self.ports)
+        add_if_not_none('_docker_login_config', self._docker_login_config)
         if self._is_image_managed is not None:
-            config["_is_image_managed"] = self._is_image_managed
+            config['_is_image_managed'] = self._is_image_managed
         return config
 
     def __setstate__(self, state):
@@ -1237,25 +1184,25 @@ class Resources:
 
         # TODO (zhwu): Design our persistent state format with `__getstate__`,
         # so that to get rid of the version tracking.
-        version = state.pop("_version", None)
+        version = state.pop('_version', None)
         # Handle old version(s) here.
         if version is None:
             version = -1
         if version < 0:
-            cloud = state.pop("cloud", None)
-            state["_cloud"] = cloud
+            cloud = state.pop('cloud', None)
+            state['_cloud'] = cloud
 
-            instance_type = state.pop("instance_type", None)
-            state["_instance_type"] = instance_type
+            instance_type = state.pop('instance_type', None)
+            state['_instance_type'] = instance_type
 
-            use_spot = state.pop("use_spot", False)
-            state["_use_spot"] = use_spot
+            use_spot = state.pop('use_spot', False)
+            state['_use_spot'] = use_spot
 
-            accelerator_args = state.pop("accelerator_args", None)
-            state["_accelerator_args"] = accelerator_args
+            accelerator_args = state.pop('accelerator_args', None)
+            state['_accelerator_args'] = accelerator_args
 
-            disk_size = state.pop("disk_size", _DEFAULT_DISK_SIZE_GB)
-            state["_disk_size"] = disk_size
+            disk_size = state.pop('disk_size', _DEFAULT_DISK_SIZE_GB)
+            state['_disk_size'] = disk_size
 
         if version < 2:
             self._region = None
@@ -1270,13 +1217,13 @@ class Resources:
             self._zone = None
 
         if version < 6:
-            accelerators = state.pop("_accelerators", None)
+            accelerators = state.pop('_accelerators', None)
             if accelerators is not None:
                 accelerators = {
-                    accelerator_registry.canonicalize_accelerator_name(acc): acc_count
-                    for acc, acc_count in accelerators.items()
+                    accelerator_registry.canonicalize_accelerator_name(acc):
+                    acc_count for acc, acc_count in accelerators.items()
                 }
-            state["_accelerators"] = accelerators
+            state['_accelerators'] = accelerators
 
         if version < 7:
             self._cpus = None
@@ -1284,9 +1231,9 @@ class Resources:
         if version < 8:
             self._memory = None
 
-        image_id = state.get("_image_id", None)
+        image_id = state.get('_image_id', None)
         if isinstance(image_id, str):
-            state["_image_id"] = {state.get("_region", None): image_id}
+            state['_image_id'] = {state.get('_region', None): image_id}
 
         if version < 9:
             self._disk_tier = None
@@ -1301,10 +1248,9 @@ class Resources:
             self._docker_login_config = None
 
         if version < 13:
-            original_ports = state.get("_ports", None)
+            original_ports = state.get('_ports', None)
             if original_ports is not None:
-                state["_ports"] = resources_utils.simplify_ports(
-                    [str(port) for port in original_ports]
-                )
+                state['_ports'] = resources_utils.simplify_ports(
+                    [str(port) for port in original_ports])
 
         self.__dict__.update(state)