%px: show outputs on error

always show outputs on completion, on success or failure

- simplify conditions a bit to reduce repetition
- display_outputs shows captured outputs, even on failure

Author: minrk

ipyparallel/client/asyncresult.py

@@ -990,7 +990,7 @@ def display_outputs(self, groupby="type", result_only=False):
         stderrs = self.stderr
         execute_results = self.execute_result
         output_lists = self.outputs
-        results = self.get()
+        results = self.get(return_exceptions=True)
 
         targets = self.engine_id
 

ipyparallel/client/magics.py

@@ -330,6 +330,10 @@ def result(self, line=''):
         if self.last_result is None:
             raise UsageError(NO_LAST_RESULT)
 
+        if args.save_name:
+            self.shell.user_ns[args.save_name] = self.last_result
+            return
+
         self.last_result.get()
         self.last_result.display_outputs(groupby=args.groupby)
 
@@ -402,37 +406,26 @@ def parallel_execute(
                         # wait for 'quick' results before showing progress
                         tic = time.perf_counter()
                         deadline = tic + progress_after
-                        try:
-                            result.get(timeout=progress_after)
-                            remaining = max(deadline - time.perf_counter(), 0)
-                            result.wait_for_output(timeout=remaining)
-                        except TimeoutError:
-                            pass
-                        except error.CompositeError as e:
-                            if stream_output:
-                                # already streamed, show an abbreviated result
-                                raise error.AlreadyDisplayedError(e) from None
-                            else:
-                                raise
-                        else:
-                            finished_waiting = True
+                        result.wait(timeout=progress_after)
+                        remaining = max(deadline - time.perf_counter(), 0)
+                        result.wait_for_output(timeout=remaining)
+                        finished_waiting = result.done()
 
                     if not finished_waiting:
                         if progress_after >= 0:
                             # not an immediate result, start interactive progress
                             result.wait_interactive()
-                        result.wait_for_output()
-                        try:
-                            result.get()
-                        except error.CompositeError as e:
-                            if stream_output:
-                                # already streamed, show an abbreviated result
-                                raise error.AlreadyDisplayedError(e) from None
-                            else:
-                                raise
+                            result.wait_for_output(1)
+
+                    try:
+                        result.get()
+                    except error.CompositeError as e:
+                        if stream_output:
+                            # already streamed, show an abbreviated result
+                            raise error.AlreadyDisplayedError(e) from None
+                        else:
+                            raise
                 # Skip redisplay if streaming output
-                if not stream_output:
-                    result.display_outputs(groupby)
             except KeyboardInterrupt:
                 if signal_on_interrupt is not None:
                     print(
@@ -444,6 +437,14 @@ def parallel_execute(
                     )
                 else:
                     raise
+            finally:
+                # always redisplay outputs if not streaming,
+                # on both success and error
+
+                if not stream_output:
+                    # wait for at most 1 second for output to be complete
+                    result.wait_for_output(1)
+                    result.display_outputs(groupby)
         else:
             # return AsyncResult only on non-blocking submission
             return result

ipyparallel/tests/test_asyncresult.py

@@ -288,6 +288,20 @@ def test_display_empty_streams_engine(self):
         self.assertEqual(io.stderr, '')
         self.assertEqual(io.stdout, '')
 
+    def test_display_output_error(self):
+        """display_outputs shows output on error"""
+        self.minimum_engines(1)
+
+        v = self.client[-1]
+        ar = v.execute("print (5555)\n1/0")
+        ar.get(5, return_exceptions=True)
+        ar.wait_for_output(5)
+        with capture_output() as io:
+            ar.display_outputs()
+        self.assertEqual(io.stderr, '')
+        self.assertEqual('5555\n', io.stdout)
+        assert 'ZeroDivisionError' not in io.stdout
+
     def test_await_data(self):
         """asking for ar.data flushes outputs"""
         self.minimum_engines(1)