[Python API] Return job_id on sky.jobs.launch

sky/jobs/core.py

@@ -2,7 +2,7 @@
 import os
 import tempfile
 import typing
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, Dict, List, Optional, Tuple, Union
 import uuid
 
 import colorama
@@ -37,13 +37,13 @@
 @timeline.event
 @usage_lib.entrypoint
 def launch(
-        task: Union['sky.Task', 'sky.Dag'],
-        name: Optional[str] = None,
-        stream_logs: bool = True,
-        detach_run: bool = False,
-        # TODO(cooperc): remove fast arg before 0.8.0
-        fast: bool = True,  # pylint: disable=unused-argument for compatibility
-) -> None:
+    task: Union['sky.Task', 'sky.Dag'],
+    name: Optional[str] = None,
+    stream_logs: bool = True,
+    detach_run: bool = False,
+    # TODO(cooperc): remove fast arg before 0.8.0
+    fast: bool = True,  # pylint: disable=unused-argument for compatibility
+) -> Tuple[Optional[int], Optional[backends.ResourceHandle]]:
     # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
     """Launch a managed job.
 
@@ -61,6 +61,13 @@ def launch(
         ValueError: cluster does not exist. Or, the entrypoint is not a valid
             chain dag.
         sky.exceptions.NotSupportedError: the feature is not supported.
+
+    Returns:
+      job_id: Optional[int]; the job ID of the submitted job. None if the
+        backend is not CloudVmRayBackend, or no job is submitted to
+        the cluster.
+      handle: Optional[backends.ResourceHandle]; handle to the controller VM.
+        None if dryrun.
     """
     entrypoint = task
     dag_uuid = str(uuid.uuid4().hex[:4])
@@ -140,15 +147,15 @@ def launch(
             f'{colorama.Fore.YELLOW}'
             f'Launching managed job {dag.name!r} from jobs controller...'
             f'{colorama.Style.RESET_ALL}')
-        sky.launch(task=controller_task,
-                   stream_logs=stream_logs,
-                   cluster_name=controller_name,
-                   detach_run=detach_run,
-                   idle_minutes_to_autostop=skylet_constants.
-                   CONTROLLER_IDLE_MINUTES_TO_AUTOSTOP,
-                   retry_until_up=True,
-                   fast=True,
-                   _disable_controller_check=True)
+        return sky.launch(task=controller_task,
+                          stream_logs=stream_logs,
+                          cluster_name=controller_name,
+                          detach_run=detach_run,
+                          idle_minutes_to_autostop=skylet_constants.
+                          CONTROLLER_IDLE_MINUTES_TO_AUTOSTOP,
+                          retry_until_up=True,
+                          fast=True,
+                          _disable_controller_check=True)
 
 
 def queue_from_kubernetes_pod(