feat: wire StorageAdapter delegation into StorageBackend

The three else-branches in StorageBackend (_create_filesystem,
_full_path, get_url) now query the adapter registry before falling
back to the built-in behaviour. Registered plugin adapters are called
for filesystem construction, path composition, and URL generation;
unknown protocols still raise DataJointError. Includes new test class
TestStorageBackendPluginDelegation.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: kushalbakshi

src/datajoint/storage.py

@@ -368,7 +368,14 @@ def _create_filesystem(self) -> fsspec.AbstractFileSystem:
             )
 
         else:
-            raise errors.DataJointError(f"Unsupported storage protocol: {self.protocol}")
+            from .storage_adapter import get_storage_adapter
+
+            adapter = get_storage_adapter(self.protocol)
+            if adapter is None:
+                raise errors.DataJointError(
+                    f"Unsupported storage protocol: {self.protocol}"
+                )
+            return adapter.create_filesystem(self.spec)
 
     def _full_path(self, path: str | PurePosixPath) -> str:
         """
@@ -398,7 +405,12 @@ def _full_path(self, path: str | PurePosixPath) -> str:
                 return f"{bucket}/{location}/{path}"
             return f"{bucket}/{path}"
         else:
-            # Local filesystem - prepend location if specified
+            from .storage_adapter import get_storage_adapter
+
+            adapter = get_storage_adapter(self.protocol)
+            if adapter is not None:
+                return adapter.full_path(self.spec, path)
+            # File-protocol fallback
             location = self.spec.get("location", "")
             if location:
                 return str(Path(location) / path)
@@ -448,7 +460,11 @@ def get_url(self, path: str | PurePosixPath) -> str:
         elif self.protocol == "azure":
             return f"az://{full_path}"
         else:
-            # Fallback: use protocol prefix
+            from .storage_adapter import get_storage_adapter
+
+            adapter = get_storage_adapter(self.protocol)
+            if adapter is not None:
+                return adapter.get_url(self.spec, full_path)
             return f"{self.protocol}://{full_path}"
 
     def put_file(self, local_path: str | Path, remote_path: str | PurePosixPath, metadata: dict | None = None) -> None:

tests/unit/test_storage_adapter.py

@@ -105,6 +105,79 @@ def test_default_url_format(self):
         assert self.adapter.get_url({}, "data/file.dat") == "dummy://data/file.dat"
 
 
+from datajoint.storage import StorageBackend
+
+
+class _FakeFS:
+    """Minimal fake fsspec filesystem for testing."""
+    protocol = "dummy"
+
+
+class _FSAdapter(StorageAdapter):
+    """Adapter that returns a fake filesystem."""
+
+    protocol = "testfs"
+    required_keys = ("protocol",)
+    allowed_keys = ("protocol",)
+
+    def create_filesystem(self, spec):
+        return _FakeFS()
+
+    def get_url(self, spec, path):
+        return f"https://test.example.com/{path}"
+
+
+class TestStorageBackendPluginDelegation:
+    """Tests for plugin delegation in StorageBackend methods."""
+
+    def setup_method(self):
+        import datajoint.storage_adapter as sa_mod
+
+        sa_mod._adapter_registry["testfs"] = _FSAdapter()
+
+    def teardown_method(self):
+        import datajoint.storage_adapter as sa_mod
+
+        sa_mod._adapter_registry.pop("testfs", None)
+
+    def test_create_filesystem_delegates_to_adapter(self):
+        backend = StorageBackend.__new__(StorageBackend)
+        backend.spec = {"protocol": "testfs"}
+        backend.protocol = "testfs"
+        backend._fs = None
+        fs = backend._create_filesystem()
+        assert isinstance(fs, _FakeFS)
+
+    def test_full_path_delegates_to_adapter(self):
+        backend = StorageBackend.__new__(StorageBackend)
+        backend.spec = {"protocol": "testfs", "location": "data"}
+        backend.protocol = "testfs"
+        result = backend._full_path("schema/ab/cd/hash123")
+        assert result == "data/schema/ab/cd/hash123"
+
+    def test_full_path_empty_location(self):
+        backend = StorageBackend.__new__(StorageBackend)
+        backend.spec = {"protocol": "testfs", "location": ""}
+        backend.protocol = "testfs"
+        result = backend._full_path("schema/ab/cd/hash123")
+        assert result == "schema/ab/cd/hash123"
+
+    def test_get_url_delegates_to_adapter(self):
+        backend = StorageBackend.__new__(StorageBackend)
+        backend.spec = {"protocol": "testfs", "location": ""}
+        backend.protocol = "testfs"
+        result = backend.get_url("schema/file.dat")
+        assert result == "https://test.example.com/schema/file.dat"
+
+    def test_unsupported_protocol_error(self):
+        backend = StorageBackend.__new__(StorageBackend)
+        backend.spec = {"protocol": "totally_unknown_xyz"}
+        backend.protocol = "totally_unknown_xyz"
+        backend._fs = None
+        with pytest.raises(DataJointError, match="Unsupported storage protocol"):
+            backend._create_filesystem()
+
+
 import datajoint as dj