feat: add StorageAdapter plugin system for third-party storage protocols

Adds a StorageAdapter ABC and entry-point plugin system (datajoint.storage
group) so third-party packages can register new storage protocols without
modifying DataJoint internals. Built-in protocols (file, s3, gcs, azure)
remain hardcoded; full unification is tracked in #1440 with Phase 0
(per-protocol unit-test scaffolding) as a hard prerequisite, gated on the
atomicity contract in safe_write/safe_copy and recursive-op semantics.

- StorageAdapter ABC with four extension points: create_filesystem,
  validate_spec, full_path, get_url
- Lazy entry-point discovery via _discover_adapters; adapters auto-load
  when their protocol is referenced in dj.config.stores
- _require_adapter helper provides symmetric missing-adapter errors
  across _create_filesystem, _full_path, and get_url; _full_path now
  reaches file-protocol logic only via an explicit elif, not as a
  catch-all else, so unknown protocols can no longer silently take it
- _apply_common_store_defaults keeps built-in and plugin paths in sync
  on shared defaults; the location default is intentionally not applied
  to plugins so adapters can declare it in required_keys
- 23 unit tests covering the registry, validation defaults, backend
  delegation, symmetric error handling on unknown protocols, entry-point
  discovery, and graceful failure of a bad entry point

Author: kushalbakshi

src/datajoint/settings.py

@@ -420,24 +420,29 @@ def get_store_spec(self, store: str | None = None, *, use_filepath_default: bool
 
         spec = dict(self.stores[store])
 
-        # Set defaults for optional fields (common to all protocols)
-        spec.setdefault("subfolding", None)  # No subfolding by default
-        spec.setdefault("partition_pattern", None)  # No partitioning by default
-        spec.setdefault("token_length", 8)  # Default token length
-
-        # Set defaults for storage section prefixes
-        spec.setdefault("hash_prefix", "_hash")  # Hash-addressed storage section
-        spec.setdefault("schema_prefix", "_schema")  # Schema-addressed storage section
-        spec.setdefault("filepath_prefix", None)  # Filepath storage (unrestricted by default)
+        self._apply_common_store_defaults(spec)
 
         # Validate protocol
         protocol = spec.get("protocol", "").lower()
         supported_protocols = ("file", "s3", "gcs", "azure")
         if protocol not in supported_protocols:
-            raise DataJointError(
-                f'Missing or invalid protocol in config.stores["{store}"]. '
-                f"Supported protocols: {', '.join(supported_protocols)}"
+            from .storage_adapter import get_storage_adapter
+
+            adapter = get_storage_adapter(protocol)
+            if adapter is None:
+                raise DataJointError(
+                    f'Unknown protocol "{protocol}" in config.stores["{store}"]. '
+                    f"Built-in: {', '.join(supported_protocols)}. "
+                    f"Install a plugin package for additional protocols."
+                )
+            adapter.validate_spec(spec)
+            self._validate_prefix_separation(
+                store_name=store,
+                hash_prefix=spec.get("hash_prefix"),
+                schema_prefix=spec.get("schema_prefix"),
+                filepath_prefix=spec.get("filepath_prefix"),
             )
+            return spec
 
         # Set protocol-specific defaults
         if protocol == "s3":
@@ -582,6 +587,16 @@ def normalize(p: str) -> str:
                         f"Storage section prefixes must be mutually exclusive."
                     )
 
+    @staticmethod
+    def _apply_common_store_defaults(spec: dict[str, Any]) -> None:
+        """Apply defaults shared by every store protocol (built-in and plugin)."""
+        spec.setdefault("subfolding", None)
+        spec.setdefault("partition_pattern", None)
+        spec.setdefault("token_length", 8)
+        spec.setdefault("hash_prefix", "_hash")
+        spec.setdefault("schema_prefix", "_schema")
+        spec.setdefault("filepath_prefix", None)
+
     def load(self, filename: str | Path) -> None:
         """
         Load settings from a JSON file.

src/datajoint/storage.py

@@ -330,6 +330,15 @@ def fs(self) -> fsspec.AbstractFileSystem:
             self._fs = self._create_filesystem()
         return self._fs
 
+    def _require_adapter(self):
+        """Look up a registered storage adapter, raising if none is registered."""
+        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
+
     def _create_filesystem(self) -> fsspec.AbstractFileSystem:
         """Create fsspec filesystem based on protocol."""
         if self.protocol == "file":
@@ -368,7 +377,7 @@ def _create_filesystem(self) -> fsspec.AbstractFileSystem:
             )
 
         else:
-            raise errors.DataJointError(f"Unsupported storage protocol: {self.protocol}")
+            return self._require_adapter().create_filesystem(self.spec)
 
     def _full_path(self, path: str | PurePosixPath) -> str:
         """
@@ -397,12 +406,13 @@ def _full_path(self, path: str | PurePosixPath) -> str:
             if location:
                 return f"{bucket}/{location}/{path}"
             return f"{bucket}/{path}"
-        else:
-            # Local filesystem - prepend location if specified
+        elif self.protocol == "file":
             location = self.spec.get("location", "")
             if location:
                 return str(Path(location) / path)
             return path
+        else:
+            return self._require_adapter().full_path(self.spec, path)
 
     def get_url(self, path: str | PurePosixPath) -> str:
         """
@@ -448,8 +458,7 @@ def get_url(self, path: str | PurePosixPath) -> str:
         elif self.protocol == "azure":
             return f"az://{full_path}"
         else:
-            # Fallback: use protocol prefix
-            return f"{self.protocol}://{full_path}"
+            return self._require_adapter().get_url(self.spec, full_path)
 
     def put_file(self, local_path: str | Path, remote_path: str | PurePosixPath, metadata: dict | None = None) -> None:
         """

src/datajoint/storage_adapter.py

@@ -0,0 +1,109 @@
+"""Plugin system for third-party storage protocols.
+
+Third-party packages register adapters via entry points::
+
+    [project.entry-points."datajoint.storage"]
+    myprotocol = "my_package:MyStorageAdapter"
+
+The adapter is auto-discovered when DataJoint encounters the protocol name
+in a store configuration. No explicit import is needed.
+"""
+
+from abc import ABC, abstractmethod
+from typing import Any
+import logging
+
+import fsspec
+
+from . import errors
+
+logger = logging.getLogger(__name__)
+
+
+class StorageAdapter(ABC):
+    """Base class for storage protocol adapters.
+
+    Subclass this and declare an entry point to add a new storage protocol
+    to DataJoint. At minimum, implement ``create_filesystem`` and set
+    ``protocol``, ``required_keys``, and ``allowed_keys``.
+    """
+
+    protocol: str
+    required_keys: tuple[str, ...] = ()
+    allowed_keys: tuple[str, ...] = ()
+
+    @abstractmethod
+    def create_filesystem(self, spec: dict[str, Any]) -> fsspec.AbstractFileSystem:
+        """Return an fsspec filesystem instance for this protocol."""
+        ...
+
+    def validate_spec(self, spec: dict[str, Any]) -> None:
+        """Validate protocol-specific config fields."""
+        missing = [k for k in self.required_keys if k not in spec]
+        if missing:
+            raise errors.DataJointError(f'{self.protocol} store is missing: {", ".join(missing)}')
+        all_allowed = set(self.allowed_keys) | _COMMON_STORE_KEYS
+        invalid = [k for k in spec if k not in all_allowed]
+        if invalid:
+            raise errors.DataJointError(f'Invalid key(s) for {self.protocol}: {", ".join(invalid)}')
+
+    def full_path(self, spec: dict[str, Any], relpath: str) -> str:
+        """Construct storage path from a relative path."""
+        location = spec.get("location", "")
+        return f"{location}/{relpath}" if location else relpath
+
+    def get_url(self, spec: dict[str, Any], path: str) -> str:
+        """Return a display URL for the stored object."""
+        return f"{self.protocol}://{path}"
+
+
+_COMMON_STORE_KEYS = frozenset(
+    {
+        "protocol",
+        "location",
+        "subfolding",
+        "partition_pattern",
+        "token_length",
+        "hash_prefix",
+        "schema_prefix",
+        "filepath_prefix",
+        "stage",
+    }
+)
+
+_adapter_registry: dict[str, StorageAdapter] = {}
+_adapters_loaded: bool = False
+
+
+def get_storage_adapter(protocol: str) -> StorageAdapter | None:
+    """Look up a registered storage adapter by protocol name."""
+    global _adapters_loaded
+    if not _adapters_loaded:
+        _discover_adapters()
+        _adapters_loaded = True
+    return _adapter_registry.get(protocol)
+
+
+def _discover_adapters() -> None:
+    """Load storage adapters from datajoint.storage entry points."""
+    try:
+        from importlib.metadata import entry_points
+    except ImportError:
+        logger.debug("importlib.metadata not available, skipping adapter discovery")
+        return
+
+    try:
+        eps = entry_points(group="datajoint.storage")
+    except TypeError:
+        eps = entry_points().get("datajoint.storage", [])
+
+    for ep in eps:
+        if ep.name in _adapter_registry:
+            continue
+        try:
+            adapter_cls = ep.load()
+            adapter = adapter_cls()
+            _adapter_registry[adapter.protocol] = adapter
+            logger.debug(f"Loaded storage adapter: {adapter.protocol}")
+        except Exception as e:
+            logger.warning(f"Failed to load storage adapter '{ep.name}': {e}")