Cache file digests and check for change in digest when uploading

dandi/dandiapi.py

@@ -8,7 +8,6 @@
 from .consts import MAX_CHUNK_SIZE, known_instances_rev
 from .girder import keyring_lookup
 from . import get_logger
-from .support.digests import Digester
 
 lgr = get_logger()
 
@@ -348,9 +347,7 @@ def _migrate_dandiset_metadata(cls, dandiset):
             dandiset["metadata"] = dandiset_metadata.pop("dandiset")
         return dandiset
 
-    def upload(
-        self, dandiset_id, version_id, asset_metadata, filepath, sha256_digest=None
-    ):
+    def upload(self, dandiset_id, version_id, asset_metadata, filepath):
         """
         Parameters
         ----------
@@ -364,22 +361,12 @@ def upload(
           the server.
         filepath: str or PathLike
           the path to the local file to upload
-        sha256_digest: str, optional
-          a precalculated SHA 256 digest for the file
         """
-        for r in self.iter_upload(
-            dandiset_id,
-            version_id,
-            asset_metadata,
-            filepath,
-            sha256_digest=sha256_digest,
-        ):
+        for r in self.iter_upload(dandiset_id, version_id, asset_metadata, filepath):
             if r["status"] == "validating":
                 sleep(0.1)
 
-    def iter_upload(
-        self, dandiset_id, version_id, asset_metadata, filepath, sha256_digest=None
-    ):
+    def iter_upload(self, dandiset_id, version_id, asset_metadata, filepath):
         """
         Parameters
         ----------
@@ -393,21 +380,24 @@ def iter_upload(
           the server.
         filepath: str or PathLike
           the path to the local file to upload
-        sha256_digest: str, optional
-          a precalculated SHA 256 digest for the file
 
         Returns
         -------
         a generator of `dict`s containing at least a ``"status"`` key
         """
-        if sha256_digest is not None:
-            filehash = sha256_digest
-            lgr.debug(
-                "Using precalculated sha256 digest of %s for %s", filehash, filepath
+        from .support.digests import get_digest
+
+        filehash = get_digest(filepath)
+        lgr.debug("Calculated sha256 digest of %s for %s", filehash, filepath)
+        if (
+            asset_metadata.get("digest") is not None
+            and asset_metadata.get("digest_type") == "SHA256"
+            and asset_metadata["digest"] != filehash
+        ):
+            raise RuntimeError(
+                "File digest changed; was originally {asset_metadata['digest']}"
+                " but is now {filehash}"
             )
-        else:
-            filehash = Digester(["sha256"])(filepath)["sha256"]
-            lgr.debug("Calculated sha256 digest of %s for %s", filehash, filepath)
         try:
             self.post("/uploads/validate/", json={"sha256": filehash})
         except requests.HTTPError as e:

dandi/support/digests.py

@@ -12,6 +12,7 @@
 import hashlib
 import logging
 
+from .cache import PersistentCache
 from ..utils import auto_repr
 
 lgr = logging.getLogger("dandi.support.digests")
@@ -69,3 +70,11 @@ def __call__(self, fpath):
                 [d.update(block) for d in digests]
 
         return {n: d.hexdigest() for n, d in zip(self.digests, digests)}
+
+
+checksums = PersistentCache(name="checksums")
+
+
+@checksums.memoize_path
+def get_digest(filepath, digest="sha256"):
+    return Digester([digest])(filepath)[digest]

dandi/upload.py

@@ -610,7 +610,7 @@ def _new_upload(
 ):
     from .dandiapi import DandiAPIClient
     from .dandiset import APIDandiset
-    from .support.digests import Digester
+    from .support.digests import get_digest
 
     client = DandiAPIClient(api_url)
     client.dandi_authenticate()
@@ -703,15 +703,11 @@ def process_path(path, relpath):
                 return
 
             #
-            # Compute checksums and possible other digests (e.g. for s3, ipfs - TODO)
+            # Compute checksums
             #
             yield {"status": "digesting"}
             try:
-                # TODO: in theory we could also cache the result, but since it is
-                # critical to get correct checksums, safer to just do it all the time.
-                # Should typically be faster than upload itself ;-)
-                digester = Digester(["sha256"])
-                sha256_digest = digester(path)["sha256"]
+                sha256_digest = get_digest(path)
             except Exception as exc:
                 yield skip_file("failed to compute digests: %s" % str(exc))
                 return
@@ -810,9 +806,7 @@ def process_path(path, relpath):
                 lgr.info("Replacing asset %s", extant["uuid"])
                 client.delete_asset(ds_identifier, "draft", extant["uuid"])
             yield {"status": "uploading"}
-            for r in client.iter_upload(
-                ds_identifier, "draft", metadata, str(path), sha256_digest=sha256_digest
-            ):
+            for r in client.iter_upload(ds_identifier, "draft", metadata, str(path)):
                 if r["status"] == "uploading":
                     uploaded_paths[str(path)]["size"] = r["current"]
                 yield r