Add get_asset_metadata() function

dandi/metadata.py

@@ -660,6 +660,28 @@ def process_ndtypes(asset, nd_types):
     return asset
 
 
+def get_asset_metadata(
+    filepath, relpath, digest=None, digest_type=None, allow_any_path=True
+) -> models.BareAsset:
+    try:
+        metadata = nwb2asset(filepath, digest=digest, digest_type=digest_type)
+    except Exception as e:
+        lgr.info(
+            "Failed to extract NWB metadata from %s: %s: %s",
+            filepath,
+            type(e).__name__,
+            str(e),
+        )
+        if allow_any_path:
+            metadata = get_default_metadata(
+                filepath, digest=digest, digest_type=digest_type
+            )
+        else:
+            raise
+    metadata.path = str(relpath)
+    return metadata
+
+
 def nwb2asset(
     nwb_path, digest=None, digest_type=None, schema_version=None
 ) -> models.BareAsset:

dandi/upload.py

@@ -54,7 +54,7 @@ def upload(
             f"convention {dandiset_identifier_regex!r}."
         )
 
-    from .metadata import get_default_metadata, nwb2asset
+    from .metadata import get_asset_metadata
     from .pynwb_utils import ignore_benign_pynwb_warnings
     from .support.pyout import naturalsize
     from .utils import find_dandi_files, find_files, path_is_subpath
@@ -242,21 +242,15 @@ def process_path(path, relpath):
             # ad-hoc for dandiset.yaml for now
             yield {"status": "extracting metadata"}
             try:
-                asset_metadata = nwb2asset(
-                    path, digest=file_etag, digest_type="dandi_etag"
-                )
-            except Exception as exc:
-                lgr.exception("Failed to extract metadata from %s", path)
-                if allow_any_path:
-                    yield {"status": "failed to extract metadata"}
-                    asset_metadata = get_default_metadata(
-                        path, digest=file_etag, digest_type="dandi_etag"
-                    )
-                else:
-                    yield skip_file("failed to extract metadata: %s" % str(exc))
-                    return
-            metadata = asset_metadata.json_dict()
-            metadata["path"] = str(relpath)
+                metadata = get_asset_metadata(
+                    path,
+                    relpath,
+                    digest=file_etag,
+                    digest_type="dandi_etag",
+                    allow_any_path=allow_any_path,
+                ).json_dict()
+            except Exception as e:
+                yield skip_file("failed to extract metadata: %s" % str(e))
 
             #
             # Upload file