Support job operations via job-URI (#992)

CHANGELOG.D/988.feature

@@ -0,0 +1 @@
+Support job operations via job-URI (e.g., `neuro status job://owner-name/job-name`).

neuromation/api/url_utils.py

@@ -1,7 +1,7 @@
 import re
 import sys
 from pathlib import Path
-from typing import Sequence
+from typing import Sequence, Union
 
 from yarl import URL
 
@@ -51,7 +51,8 @@ def normalize_storage_path_uri(uri: URL, username: str) -> URL:
     return _normalize_uri(uri, username)
 
 
-def _normalize_uri(uri: URL, username: str) -> URL:
+def _normalize_uri(resource: Union[URL, str], username: str) -> URL:
+    uri = resource if isinstance(resource, URL) else URL(resource)
     path = uri.path
     if not uri.host:
         if path.startswith("~"):

neuromation/cli/job.py

@@ -334,7 +334,7 @@ async def exec(
     neuro exec --no-tty my-job ls -l
     """
     cmd = shlex.split(" ".join(cmd))
-    id = await resolve_job(root.client, job)
+    id = await resolve_job(job, client=root.client, default_user=root.username)
     retcode = await root.client.jobs.exec(
         id,
         cmd,
@@ -376,7 +376,7 @@ async def port_forward(
     neuro job port-forward my-job- 2080:80 2222:22 2000:100
 
     """
-    job_id = await resolve_job(root.client, job)
+    job_id = await resolve_job(job, client=root.client, default_user=root.username)
     async with AsyncExitStack() as stack:
         for local_port, job_port in local_remote_port:
             click.echo(
@@ -404,7 +404,7 @@ async def logs(root: Root, job: str) -> None:
     """
     Print the logs for a container.
     """
-    id = await resolve_job(root.client, job)
+    id = await resolve_job(job, client=root.client, default_user=root.username)
     await _print_logs(root, id)
 
 
@@ -503,7 +503,7 @@ async def status(root: Root, job: str) -> None:
     """
     Display status of a job.
     """
-    id = await resolve_job(root.client, job)
+    id = await resolve_job(job, client=root.client, default_user=root.username)
     res = await root.client.jobs.status(id)
     click.echo(JobStatusFormatter()(res))
 
@@ -515,7 +515,7 @@ async def browse(root: Root, job: str) -> None:
     """
     Opens a job's URL in a web browser.
     """
-    id = await resolve_job(root.client, job)
+    id = await resolve_job(job, client=root.client, default_user=root.username)
     res = await root.client.jobs.status(id)
     await browse_job(root, res)
 
@@ -528,7 +528,7 @@ async def top(root: Root, job: str) -> None:
     Display GPU/CPU/Memory usage.
     """
     formatter = JobTelemetryFormatter()
-    id = await resolve_job(root.client, job)
+    id = await resolve_job(job, client=root.client, default_user=root.username)
     print_header = True
     async for res in root.client.jobs.top(id):
         if print_header:
@@ -552,7 +552,7 @@ async def save(root: Root, job: str, image: RemoteImage) -> None:
     neuro job save my-favourite-job image://~/ubuntu-patched:v1
     neuro job save my-favourite-job image://bob/ubuntu-patched
     """
-    id = await resolve_job(root.client, job)
+    id = await resolve_job(job, client=root.client, default_user=root.username)
     progress = DockerImageProgress.create(tty=root.tty, quiet=root.quiet)
     with contextlib.closing(progress):
         await root.client.jobs.save(id, image, progress=progress)
@@ -568,7 +568,9 @@ async def kill(root: Root, jobs: Sequence[str]) -> None:
     """
     errors = []
     for job in jobs:
-        job_resolved = await resolve_job(root.client, job)
+        job_resolved = await resolve_job(
+            job, client=root.client, default_user=root.username
+        )
         try:
             await root.client.jobs.kill(job_resolved)
             # TODO (ajuszkowski) printing should be on the cli level

neuromation/cli/utils.py

@@ -41,7 +41,7 @@
 )
 from neuromation.api.config import _CookieSession, _PyPIVersion
 from neuromation.api.parsing_utils import _ImageNameParser
-from neuromation.api.url_utils import uri_from_cli
+from neuromation.api.url_utils import _normalize_uri, uri_from_cli
 
 from .asyncio_utils import run
 from .parse_utils import to_megabytes
@@ -436,13 +436,36 @@ def volume_to_verbose_str(volume: Volume) -> str:
     )
 
 
-async def resolve_job(client: Client, id_or_name: str) -> str:
+async def resolve_job(
+    id_or_name_or_uri: str, *, client: Client, default_user: str
+) -> str:
+    if id_or_name_or_uri.startswith("job:"):
+        uri = _normalize_uri(id_or_name_or_uri, username=default_user)
+        id_or_name = uri.path.lstrip("/")
+        owner = uri.host or default_user
+        if not id_or_name:
+            raise ValueError(
+                f"Invalid job URI: owner='{owner}', missing job-id or job-name"
+            )
+    else:
+        id_or_name = id_or_name_or_uri
+        owner = default_user
+
     jobs: List[JobDescription] = []
+    details = f"name={id_or_name}, owner={owner}"
     try:
-        jobs = await client.jobs.list(name=id_or_name)
+        jobs = await client.jobs.list(name=id_or_name, owners={owner})
     except Exception as e:
-        log.error(f"Failed to resolve job-name '{id_or_name}' to a job-ID: {e}")
+        log.error(
+            f"Failed to resolve job-name {id_or_name_or_uri} resolved as "
+            f"{details} to a job-ID: {e}"
+        )
     if jobs:
+        if len(jobs) > 1:
+            log.warning(
+                f"Found {len(jobs)} jobs matching {details}: "
+                ", ".join(job.id for job in jobs)
+            )
         job_id = jobs[-1].id
         log.debug(f"Job name '{id_or_name}' resolved to job ID '{job_id}'")
     else: