Refactor retry_on_timeout for reuse.

Author: jmchilton

planemo/galaxy/activity.py

@@ -36,12 +36,12 @@
     unicodify,
 )
 from pathvalidate import sanitize_filename
-from requests.exceptions import (
-    HTTPError,
-    RequestException,
-)
+from requests.exceptions import HTTPError
 
-from planemo.galaxy.api import summarize_history
+from planemo.galaxy.api import (
+    retry_on_timeouts,
+    summarize_history,
+)
 from planemo.io import wait_on
 from planemo.runnable import (
     ErrorRunResponse,
@@ -811,30 +811,11 @@ def wait_for_invocation_and_jobs(
 
 def _wait_for_invocation(ctx, gi, invocation_id, polling_backoff=0):
     def state_func():
-        return _retry_on_timeouts(ctx, gi, lambda gi: gi.invocations.show_invocation(invocation_id))
+        return retry_on_timeouts(ctx, gi, lambda gi: gi.invocations.show_invocation(invocation_id))
 
     return _wait_on_state(state_func, polling_backoff)
 
 
-def _retry_on_timeouts(ctx, gi, f):
-    gi.timeout = 60
-    try_count = 5
-    try:
-        for try_num in range(try_count):
-            start_time = time.time()
-            try:
-                return f(gi)
-            except RequestException:
-                end_time = time.time()
-                if end_time - start_time > 45 and (try_num + 1) < try_count:
-                    ctx.vlog("Galaxy seems to have timed out, retrying to fetch status.")
-                    continue
-                else:
-                    raise
-    finally:
-        gi.timeout = None
-
-
 def has_jobs_in_states(ctx, gi, history_id, states):
     params = {"history_id": history_id}
     jobs_url = gi.url + "/jobs"
@@ -849,7 +830,7 @@ def _wait_for_history(ctx, gi, history_id, polling_backoff=0):
     # no need to wait for active jobs anymore I think.
 
     def state_func():
-        return _retry_on_timeouts(ctx, gi, lambda gi: gi.histories.show_history(history_id))
+        return retry_on_timeouts(ctx, gi, lambda gi: gi.histories.show_history(history_id))
 
     return _wait_on_state(state_func, polling_backoff)
 
@@ -862,7 +843,7 @@ def _wait_for_invocation_jobs(ctx, gi, invocation_id, polling_backoff=0):
     ctx.log(f"waiting for invocation {invocation_id}")
 
     def state_func():
-        return _retry_on_timeouts(ctx, gi, lambda gi: gi.jobs.get_jobs(invocation_id=invocation_id))
+        return retry_on_timeouts(ctx, gi, lambda gi: gi.jobs.get_jobs(invocation_id=invocation_id))
 
     return _wait_on_state(state_func, polling_backoff)
 

planemo/galaxy/api.py

@@ -1,9 +1,11 @@
 """A high-level interface to local Galaxy instances using bioblend."""
 
+import time
 from io import StringIO
 from typing import Optional
 
 from bioblend.galaxy import GalaxyInstance
+from requests.exceptions import RequestException
 
 DEFAULT_ADMIN_API_KEY = "test_key"
 
@@ -136,6 +138,25 @@ def _dataset_provenance(gi, history_id, id):
     return provenance
 
 
+def retry_on_timeouts(ctx, gi, f):
+    gi.timeout = 60
+    try_count = 5
+    try:
+        for try_num in range(try_count):
+            start_time = time.time()
+            try:
+                return f(gi)
+            except RequestException:
+                end_time = time.time()
+                if end_time - start_time > 45 and (try_num + 1) < try_count:
+                    ctx.vlog("Galaxy seems to have timed out, retrying to fetch status.")
+                    continue
+                else:
+                    raise
+    finally:
+        gi.timeout = None
+
+
 __all__ = (
     "DEFAULT_ADMIN_API_KEY",
     "gi",