fix: check handle None later and add storage name consistency check

sky/cli.py

@@ -3525,11 +3525,10 @@ def storage_delete(names: List[str], all: bool, yes: bool):  # pylint: disable=r
     if sum([len(names) > 0, all]) != 1:
         raise click.UsageError('Either --all or a name must be specified.')
     if all:
-        storages = sky.storage_ls()
-        if not storages:
+        names = global_user_state.get_storage_names()
+        if not names:
             click.echo('No storage(s) to delete.')
             return
-        names = [s['name'] for s in storages]
     else:
         names = _get_glob_storages(names)
     if names:
@@ -3543,12 +3542,15 @@ def storage_delete(names: List[str], all: bool, yes: bool):  # pylint: disable=r
                 abort=True,
                 show_default=True)
 
-    results: List[Union[None, Exception]] = subprocess_utils.run_in_parallel(
-        sky.storage_delete, names, continue_on_error=True)
-
-    for idx, result in enumerate(results):
-        if isinstance(result, Exception):
-            click.secho(f'Failed to delete storage {names[idx]}: {result}', fg='red')
+    def delete_storage(name: str):
+        try:
+            sky.storage_delete(name)
+        except Exception as e:  # pylint: disable=broad-except
+            click.secho(f'Error deleting storage {name}: {e}', fg='red')
+
+    subprocess_utils.run_in_parallel(delete_storage,
+                                     names,
+                                     continue_on_error=True)
 
 
 @cli.group(cls=_NaturalOrderGroup)

sky/core.py

@@ -915,8 +915,11 @@ def storage_delete(name: str) -> None:
     handle = global_user_state.get_handle_from_storage_name(name)
     if handle is None:
         raise ValueError(f'Storage name {name!r} not found.')
-    else:
-        storage_object = data.Storage(name=handle.storage_name,
-                                      source=handle.source,
-                                      sync_on_reconstruction=False)
-        storage_object.delete()
+
+    assert handle.storage_name == name, (
+        f'In global_user_state, storage name {name!r} does not match '
+        f'handle.storage_name {handle.storage_name!r}')
+    storage_object = data.Storage(name=handle.storage_name,
+                                  source=handle.source,
+                                  sync_on_reconstruction=False)
+    storage_object.delete()

sky/global_user_state.py

@@ -826,8 +826,13 @@ def get_storage_names_start_with(starts_with: str) -> List[str]:
     return [row[0] for row in rows]
 
 
+def get_storage_names() -> List[str]:
+    rows = _DB.cursor.execute('SELECT name FROM storage')
+    return [row[0] for row in rows]
+
+
 def get_storage() -> List[Dict[str, Any]]:
-    rows = _DB.cursor.execute('select * from storage')
+    rows = _DB.cursor.execute('SELECT * FROM storage')
     records = []
     for name, launched_at, handle, last_use, status in rows:
         # TODO: use namedtuple instead of dict