Merge pull request #3174 from grondo/issue#3140

python: promote JobInfo and formatter class to API and add JobList class

mypy.ini

@@ -23,7 +23,7 @@ ignore_missing_imports = True
 follow_imports = silent
 
 # These are temporary while we find a way to generate stubs for flux.constants
-[mypy-flux-jobs]
+[mypy-flux.job.list]
 ignore_errors = True
 
 [mypy-python.t1000-service-add-remove]

src/bindings/python/flux/Makefile.am

@@ -21,6 +21,7 @@ nobase_fluxpy_PYTHON = \
 	job/kill.py \
 	job/kvs.py \
 	job/list.py \
+	job/info.py \
 	job/submit.py \
 	job/wait.py \
 	job/_wrapper.py

src/bindings/python/flux/future.py

@@ -87,8 +87,16 @@ def error_string(self):
         return errmsg.decode("utf-8") if errmsg else None
 
     def get_flux(self):
-        # pylint: disable=cyclic-import
-        flux_handle = self.pimpl.get_flux()
+        try:
+            # pylint: disable=cyclic-import
+            flux_handle = self.pimpl.get_flux()
+        except OSError as exc:
+            #  get_flux() throws OSError of EINVAL if !f->h, but this should
+            #   a valid return (i.e. no flux handle set yet)
+            if exc.errno == errno.EINVAL:
+                return None
+            else:
+                raise
         if flux_handle == ffi.NULL:
             return None
         handle = flux.core.handle.Flux(handle=flux_handle)
@@ -143,3 +151,31 @@ def get(self):
         until future is fulfilled and throws OSError on failure.
         """
         self.pimpl.flux_future_get(ffi.NULL)
+
+
+class WaitAllFuture(Future):
+    """Create a composite future which waits for all children to be fulfilled"""
+
+    def __init__(self, children=None):
+        self.children = children
+        if self.children is None:
+            self.children = []
+        future = raw.flux_future_wait_all_create()
+        super(WaitAllFuture, self).__init__(future)
+
+    def push(self, child, name=None):
+        if name is None:
+            name = ffi.NULL
+        #
+        #  if this future does not have a flux handle yet, attempt
+        #   to grab from the first pushed "child" future.
+        if self.get_flux() is None:
+            self.pimpl.set_flux(child.get_flux())
+
+        self.pimpl.push(name, child)
+        #
+        #  flux_future_push(3) "adopts" memory for child, so call
+        #   incref on child to avoid  calling flux_future_destroy(3) when
+        #   child goes out of scope in caller context.
+        child.pimpl.incref()
+        self.children.append(child)

src/bindings/python/flux/job/JobID.py

@@ -56,7 +56,9 @@ def __new__(cls, value):
             jobid = value
         else:
             jobid = id_parse(value)
-        return super(cls, cls).__new__(cls, jobid)
+        self = super(cls, cls).__new__(cls, jobid)
+        self.orig_str = str(value)
+        return self
 
     def encode(self, encoding="dec"):
         """Encode a JobID to alternate supported format"""
@@ -92,6 +94,11 @@ def kvs(self):
         """Return KVS directory path of a JobID"""
         return self.encode("kvs")
 
+    @property
+    def orig(self):
+        """Return the original string used to create the JobID"""
+        return self.orig_str
+
     def __str__(self):
         return self.encode("f58")
 

src/bindings/python/flux/job/__init__.py

@@ -13,7 +13,8 @@
 from flux.job.kvs import job_kvs, job_kvs_guest
 from flux.job.kill import kill_async, kill, cancel_async, cancel
 from flux.job.submit import submit_async, submit, submit_get_id
-from flux.job.list import job_list, job_list_inactive, job_list_id
+from flux.job.info import JobInfo, JobInfoFormat
+from flux.job.list import job_list, job_list_inactive, job_list_id, JobList
 from flux.job.wait import wait_async, wait, wait_get_status
 from flux.job.event import (
     event_watch_async,