Handle path conflict when creating leaf asset path

dandiapi/api/asset_paths.py

@@ -2,7 +2,7 @@
 
 import os
 
-from django.db import transaction
+from django.db import IntegrityError, transaction
 from django.db.models import Count, F, QuerySet, Sum
 from django.db.models.functions import Coalesce
 from tqdm import tqdm
@@ -101,8 +101,23 @@ def search_asset_paths(query: str, version: Version) -> QuerySet[AssetPath] | No
 
 def insert_asset_paths(asset: Asset, version: Version):
     """Add all intermediate paths from an asset and link them together."""
-    # Get or create leaf path
-    leaf, created = AssetPath.objects.get_or_create(path=asset.path, asset=asset, version=version)
+    try:
+        # Get or create leaf path
+        leaf, created = AssetPath.objects.get_or_create(
+            path=asset.path, asset=asset, version=version
+        )
+    except IntegrityError as e:
+        from dandiapi.api.services.asset.exceptions import AssetAlreadyExists
+
+        # If there are simultaneous requests to create the same asset, this check constraint can
+        # fail, and should be handled directly, rather than be allowed to bubble up
+        if 'unique-version-path' in str(e):
+            raise AssetAlreadyExists()
+
+        # Re-raise original exception otherwise
+        raise
+
+    # If the asset was not created, return early, as the work is already done
     if not created:
         return leaf
 

dandiapi/api/tests/test_asset_paths.py

@@ -11,6 +11,7 @@
 )
 from dandiapi.api.models import Asset, AssetPath, Version
 from dandiapi.api.models.asset_paths import AssetPathRelation
+from dandiapi.api.services.asset.exceptions import AssetAlreadyExists
 from dandiapi.api.tasks import publish_dandiset_task
 
 
@@ -96,6 +97,27 @@ def test_asset_path_add_asset_idempotent(draft_version_factory, asset_factory):
     assert path.aggregate_size == asset.size
 
 
+@pytest.mark.django_db
+def test_asset_path_add_asset_conflicting_path(draft_version_factory, asset_factory):
+    # Create asset with version
+    asset1: Asset = asset_factory()
+    asset2: Asset = asset_factory(path=asset1.path)
+    version: Version = draft_version_factory()
+    version.assets.add(asset1)
+    version.assets.add(asset2)
+
+    # Add asset1 paths
+    add_asset_paths(asset1, version)
+    assert version.asset_paths.filter(asset__isnull=False).count() == 1
+
+    # Ensure that adding asset2 raises the correct exception
+    with pytest.raises(AssetAlreadyExists):
+        add_asset_paths(asset2, version)
+
+    # Ensure that there no new asset paths created
+    assert version.asset_paths.filter(asset__isnull=False).count() == 1
+
+
 @pytest.mark.django_db
 def test_asset_path_add_version_asset_paths(draft_version_factory, asset_factory):
     # Create asset with version