[SYNPY-1488] Patch nested tqdm progress bars and messages to logger (#1177)

* Patch tqdm progress bars for nested bars, and migrating print/sysout messages to logger

Author: BryanFauble

synapseclient/__main__.py

@@ -359,12 +359,12 @@ def show(args, syn):
 
     ent = syn.get(args.id, downloadFile=False)
     syn.printEntity(ent)
-    sys.stdout.write("Provenance:\n")
+    syn.logger.info("Provenance:")
     try:
         prov = syn.getProvenance(ent)
         syn.logger.info(prov)
     except SynapseHTTPError:
-        syn.logger.error("  No Activity specified.\n")
+        syn.logger.exception("No Activity specified.")
 
 
 def delete(args, syn):
@@ -1785,11 +1785,11 @@ def perform_main(args, syn):
             )
         try:
             args.func(args, syn)
-        except Exception as ex:
+        except Exception:
             if args.debug:
                 raise
             else:
-                sys.stderr.write(utils._synapse_error_msg(ex))
+                syn.logger.exception(f"Failed to perform command: {args.func}")
                 sys.exit(1)
     else:
         # if no command provided print out help and quit

synapseclient/client.py

@@ -43,6 +43,7 @@
 from opentelemetry.instrumentation.threading import ThreadingInstrumentor
 from opentelemetry.instrumentation.urllib import URLLibInstrumentor
 from opentelemetry.trace import Span
+from tqdm import tqdm
 
 import synapseclient
 import synapseclient.core.multithread_download as multithread_download
@@ -5129,22 +5130,27 @@ def _waitForAsync(self, uri, request, endpoint=None):
         # https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/asynch/AsynchronousJobStatus.html
         sleep = self.table_query_sleep
         start_time = time.time()
-        lastMessage, lastProgress, lastTotal, progressed = "", 0, 1, False
+        lastMessage, lastProgress, lastTotal = "", 0, 1
+        progress_bar = tqdm(desc=uri, unit_scale=True, smoothing=0, leave=None)
         while time.time() - start_time < self.table_query_timeout:
             result = self.restGET(
                 uri + "/get/%s" % async_job_id["token"], endpoint=endpoint
             )
             if result.get("jobState", None) == "PROCESSING":
-                progressed = True
                 message = result.get("progressMessage", lastMessage)
                 progress = result.get("progressCurrent", lastProgress)
                 total = result.get("progressTotal", lastTotal)
+                if total and progress_bar.total != total:
+                    progress_bar.total = total
+                    progress_bar.refresh()
+
                 if message != "":
-                    self._print_transfer_progress(
-                        progress, total, message, isBytes=False
-                    )
+                    progress_bar.desc = message
+                    progress_bar.refresh()
+
                 # Reset the time if we made progress (fix SYNPY-214)
                 if message != lastMessage or lastProgress != progress:
+                    progress_bar.update(progress)
                     start_time = time.time()
                     lastMessage, lastProgress, lastTotal = message, progress, total
                 sleep = min(
@@ -5165,8 +5171,10 @@ def _waitForAsync(self, uri, request, endpoint=None):
                 + result.get("errorDetails", None),
                 asynchronousJobStatus=result,
             )
-        if progressed:
-            self._print_transfer_progress(total, total, message, isBytes=False)
+        if progress_bar.total and progress_bar.n:
+            progress_bar.update(progress_bar.total - progress_bar.n)
+        progress_bar.refresh()
+        progress_bar.close()
         return result
 
     def getColumn(self, id):

synapseclient/core/cumulative_transfer_progress.py

@@ -1,8 +1,9 @@
-import sys
 import threading
 import time
 from contextlib import contextmanager
 
+from tqdm import tqdm
+
 from synapseclient.core import utils
 
 # Defines a mechanism for printing file transfer progress that includes potentially multiple file transfers to the
@@ -42,7 +43,7 @@ def __init__(self, label, start=None):
         self._lock = threading.Lock()
         self._label = label
 
-        self._spinner = utils.Spinner()
+        self._progress_bar: tqdm = None
         self._start = start if start is not None else time.time()
 
         self._total_transferred = 0
@@ -61,51 +62,47 @@ def accumulate_progress(self):
 
     def printTransferProgress(
         self,
-        transferred,
-        toBeTransferred,
-        prefix="",
-        postfix="",
-        isBytes=True,
-        dt=None,
-        previouslyTransferred=0,
-    ):
+        transferred: int,
+        toBeTransferred: int,
+        prefix: str = "",
+        postfix: str = "",
+        isBytes: bool = True,
+        dt: None = None,
+        previouslyTransferred: int = 0,
+    ) -> None:
         """
         Parameters match those of synapseclient.core.utils.printTransferProgress.
-        """
-
-        if not sys.stdout.isatty():
-            return
 
+        Arguments:
+            transferred: The number of bytes transferred in the current transfer.
+            toBeTransferred: The total number of bytes to be transferred in the current
+                transfer.
+            prefix: A string to prepend to the progress bar.
+            postfix: A string to append to the progress bar.
+            isBytes: If True, the progress bar will display bytes. If False, the
+                progress bar will display the unit of the transferred data.
+            dt: Deprecated.
+            previouslyTransferred: Deprecated.
+
+        Returns:
+            None
+        """
         with self._lock:
-            if toBeTransferred == 0 or float(transferred) / toBeTransferred >= 1:
-                # if the individual transfer is complete then we pass through the print
-                # to the underlying utility method which will print a complete 100%
-                # progress bar on a newline.
-                utils.printTransferProgress(
-                    transferred,
-                    toBeTransferred,
-                    prefix=prefix,
+            if not self._progress_bar:
+                self._progress_bar = tqdm(
+                    desc=prefix if prefix else "Transfer Progress",
+                    unit_scale=True,
+                    total=toBeTransferred,
+                    smoothing=0,
                     postfix=postfix,
-                    isBytes=isBytes,
-                    dt=dt,
-                    previouslyTransferred=previouslyTransferred,
+                    unit="B" if isBytes else None,
+                    leave=None,
                 )
-
             # in order to know how much of the transferred data is newly transferred
             # we subtract the previously reported amount. this assumes that the printing
             # of the progress for any particular transfer is always conducted by the same
             # thread, which is true for all current transfer implementations.
             self._total_transferred += transferred - _thread_local.thread_transferred
             _thread_local.thread_transferred = transferred
 
-            cumulative_dt = time.time() - self._start
-            rate = self._total_transferred / float(cumulative_dt)
-            rate = "(%s/s)" % utils.humanizeBytes(rate) if isBytes else rate
-
-            # we print a rotating tick with each update
-            self._spinner.print_tick()
-
-            sys.stdout.write(
-                f"{self._label} {utils.humanizeBytes(self._total_transferred)} {rate}"
-            )
-            sys.stdout.flush()
+            self._progress_bar.update(transferred - _thread_local.thread_transferred)

synapseclient/core/remote_file_storage_wrappers.py

@@ -215,6 +215,7 @@ def upload_file(
                 unit_scale=True,
                 postfix=filename,
                 smoothing=0,
+                leave=None,
             )
             progress_callback = S3ClientWrapper._create_progress_callback_func(
                 progress_bar
@@ -291,6 +292,7 @@ def upload_file(
             unit_scale=True,
             smoothing=0,
             postfix=filepath,
+            leave=None,
         )
 
         def progress_callback(*args, **kwargs) -> None:

synapseclient/core/transfer_bar.py

@@ -159,6 +159,7 @@ def get_or_create_download_progress_bar(
             unit_scale=True,
             smoothing=0,
             postfix=postfix,
+            leave=None,
         )
         _thread_local.progress_bar_download = progress_bar
     else:

synapseclient/core/upload/multipart_upload.py

@@ -16,6 +16,7 @@
 
 import requests
 from opentelemetry import trace
+from tqdm import tqdm
 
 from synapseclient.core import pool_provider
 from synapseclient.core.constants import concrete_types
@@ -33,7 +34,7 @@
     get_file_chunk,
     get_part_size,
 )
-from synapseclient.core.utils import MB, Spinner, md5_fn, md5_for_file
+from synapseclient.core.utils import MB, md5_fn, md5_for_file
 
 # AWS limits
 MAX_NUMBER_OF_PARTS = 10000
@@ -468,8 +469,21 @@ def multipart_upload_file(
         mime_type, _ = mimetypes.guess_type(file_path, strict=False)
         content_type = mime_type or "application/octet-stream"
 
-    callback_func = Spinner().print_tick if not syn.silent else None
-    md5_hex = md5 or md5_for_file(file_path, callback=callback_func).hexdigest()
+    if md5:
+        md5_hex = md5
+    else:
+        progress_bar = (
+            tqdm(
+                desc=f"Calculating MD5: {os.path.basename(file_path)}",
+                unit="B",
+                unit_scale=True,
+                total=file_size,
+                leave=None,
+            )
+            if not syn.silent
+            else None
+        )
+        md5_hex = md5_for_file(file_path, progress_bar=progress_bar).hexdigest()
 
     part_size = get_part_size(
         part_size or DEFAULT_PART_SIZE,

synapseclient/core/upload/multipart_upload_async.py

@@ -357,6 +357,7 @@ async def _upload_parts(
                     unit_scale=True,
                     postfix=self._dest_file_name,
                     smoothing=0,
+                    leave=None,
                 )
                 self._progress_bar.update(completed_part_count)
             else:
@@ -374,6 +375,7 @@ async def _upload_parts(
                     unit_scale=True,
                     postfix=self._dest_file_name,
                     smoothing=0,
+                    leave=None,
                 )
                 self._progress_bar.update(previously_transferred)
 

synapseclient/core/utils.py

@@ -26,10 +26,12 @@
 import zipfile
 from dataclasses import asdict, fields, is_dataclass
 from email.message import Message
-from typing import TYPE_CHECKING, List, TypeVar
+from typing import TYPE_CHECKING, List, Optional, TypeVar
 
 import requests
+from deprecated import deprecated
 from opentelemetry import trace
+from tqdm import tqdm
 
 from synapseclient.core.logging_setup import DEFAULT_LOGGER_NAME
 
@@ -57,7 +59,10 @@
 
 
 def md5_for_file(
-    filename: str, block_size: int = 2 * MB, callback: typing.Callable = None
+    filename: str,
+    block_size: int = 2 * MB,
+    callback: typing.Callable = None,
+    progress_bar: Optional[tqdm] = None,
 ):
     """
     Calculates the MD5 of the given file.
@@ -82,8 +87,14 @@ def md5_for_file(
                 callback()
             data = f.read(block_size)
             if not data:
+                if progress_bar:
+                    progress_bar.update(progress_bar.total - progress_bar.n)
+                    progress_bar.refresh()
+                    progress_bar.close()
                 break
             md5.update(data)
+            if progress_bar:
+                progress_bar.update(len(data))
             del data
             # Garbage collect every 100 iterations
             if loop_iteration % 100 == 0:
@@ -617,23 +628,36 @@ def make_bogus_binary_file(
     Returns:
         The name of the file
     """
-
+    progress_bar = (
+        tqdm(
+            desc=f"Generating {filepath}",
+            unit_scale=True,
+            total=n,
+            smoothing=0,
+            unit="B",
+            leave=None,
+        )
+        if printprogress
+        else None
+    )
     with (
         open(filepath, "wb")
         if filepath
         else tempfile.NamedTemporaryFile(mode="wb", suffix=".dat", delete=False)
     ) as f:
         if not filepath:
             filepath = f.name
-        progress = 0
         remaining = n
         while remaining > 0:
             buff_size = int(min(remaining, 1 * KB))
             f.write(os.urandom(buff_size))
+            if progress_bar:
+                progress_bar.update(buff_size)
             remaining -= buff_size
-            if printprogress:
-                progress += buff_size
-                printTransferProgress(progress, n, "Generated ", filepath)
+        if progress_bar:
+            progress_bar.update(progress_bar.total - progress_bar.n)
+            progress_bar.refresh()
+            progress_bar.close()
         return normalize_path(filepath)
 
 
@@ -1361,6 +1385,11 @@ def wrapper(*args, **kwargs):
         return wrapper
 
 
+@deprecated(
+    version="4.8.0",
+    reason="To be removed in 5.0.0. "
+    "This is removed in favor of using the tqdm library for progress bars.",
+)
 class Spinner:
     def __init__(self, msg=""):
         self._tick = 0

synapseclient/models/mixins/table_operator.py

@@ -2262,6 +2262,7 @@ async def main():
                 desc="Querying & Updating rows",
                 unit_scale=True,
                 smoothing=0,
+                leave=None,
             )
             for individual_chunk in chunk_list:
                 select_statement = self._construct_select_statement_for_upsert(
@@ -3116,6 +3117,7 @@ async def _chunk_and_upload_csv(
                 unit_scale=True,
                 smoothing=0,
                 unit="B",
+                leave=None,
             )
             # The original file is read twice, the reason is that on the first pass we
             # are calculating the size of the chunks that we will be uploading and the
@@ -3316,6 +3318,7 @@ async def _chunk_and_upload_df(
             unit_scale=True,
             smoothing=0,
             unit="B",
+            leave=None,
         )
 
         changes = []