Handle exceptions raised during pkgs/ multiprocessor extraction

news/59-extraction-exceptions

@@ -0,0 +1,19 @@
+### Enhancements
+
+* <news item>
+
+### Bug fixes
+
+* Handle exceptions raised during `pkgs/` multiprocessor extraction. (#55 via #59)
+
+### Deprecations
+
+* <news item>
+
+### Docs
+
+* <news item>
+
+### Other
+
+* <news item>

src/entry_point.py

@@ -140,9 +140,9 @@ def __call__(self, parser, namespace, values, option_string=None):
 
 
 def _constructor_extract_conda_pkgs(prefix, max_workers=None):
-    from concurrent.futures import ProcessPoolExecutor
+    from concurrent.futures import ProcessPoolExecutor, as_completed
 
-    import tqdm
+    from tqdm.auto import tqdm
     from conda.base.constants import CONDA_PACKAGE_EXTENSIONS
     from conda_package_handling import api
 
@@ -155,10 +155,18 @@ def _constructor_extract_conda_pkgs(prefix, max_workers=None):
             if pkg.endswith(ext):
                 fn = os.path.join(os.getcwd(), pkg)
                 flist.append(fn)
-    with tqdm.tqdm(total=len(flist), leave=False) as t:
-        for fn, _ in zip(flist, executor.map(api.extract, flist)):
-            t.set_description("Extracting : %s" % os.path.basename(fn))
-            t.update()
+    with ProcessPoolExecutor(max_workers=max_workers) as executor:
+        futures = {executor.submit(api.extract, fn): fn for fn in flist}
+        with tqdm(total=len(flist), leave=False) as pbar:
+            for future in as_completed(futures):
+                fn = futures[future]
+                try:
+                    future.result()
+                except Exception as exc:
+                    raise RuntimeError(f"Failed to extract {fn}: {exc}") from exc
+                else:
+                    pbar.set_description(f"Extracting: {os.path.basename(fn)}")
+                    pbar.update()
 
 
 def _constructor_extract_tarball():