feat: Add configurable prefixes for storage sections

Allow users to configure custom prefixes for hash-addressed, schema-addressed,
and filepath storage sections per store. This enables mapping DataJoint to
existing storage layouts without restructuring.

Configuration:
- hash_prefix (default: '_hash') - Hash-addressed storage section
- schema_prefix (default: '_schema') - Schema-addressed storage section
- filepath_prefix (default: None) - Optional filepath restriction

Features:
- Validates prefixes don't overlap (mutual exclusion)
- FilepathCodec enforces dynamic reserved prefixes
- Optional filepath_prefix to restrict filepath paths
- Backwards compatible defaults

Examples:
{
  "stores": {
    "legacy": {
      "protocol": "file",
      "location": "/data/existing",
      "hash_prefix": "content_addressed",
      "schema_prefix": "structured_data",
      "filepath_prefix": "raw_files"
    }
  }
}

Changes:
- settings.py: Add prefix fields, validation logic
- builtin_codecs.py: Dynamic prefix checking in FilepathCodec
- test_settings.py: 7 new tests for prefix validation
- test_codecs.py: 2 new tests for custom prefixes

Author: dimitri-yatsenko

src/datajoint/builtin_codecs.py

@@ -809,18 +809,45 @@ def encode(self, value: Any, *, key: dict | None = None, store_name: str | None
         """
         from datetime import datetime, timezone
 
+        from . import config
         from .hash_registry import get_store_backend
 
         path = str(value)
 
-        # Validate path doesn't use reserved sections
-        path_normalized = path.lstrip('/')
-        if path_normalized.startswith('_hash/') or path_normalized.startswith('_schema/'):
-            raise ValueError(
-                f"<filepath@> cannot use reserved sections '_hash/' or '_schema/'. "
-                f"These sections are managed by DataJoint. "
-                f"Got path: {path}"
-            )
+        # Get store spec to check prefix configuration
+        spec = config.get_store_spec(store_name)
+
+        # Validate path doesn't use reserved sections (hash and schema)
+        path_normalized = path.lstrip("/")
+        reserved_prefixes = []
+
+        hash_prefix = spec.get("hash_prefix")
+        if hash_prefix:
+            reserved_prefixes.append(("hash_prefix", hash_prefix))
+
+        schema_prefix = spec.get("schema_prefix")
+        if schema_prefix:
+            reserved_prefixes.append(("schema_prefix", schema_prefix))
+
+        # Check if path starts with any reserved prefix
+        for prefix_name, prefix_value in reserved_prefixes:
+            prefix_normalized = prefix_value.strip("/") + "/"
+            if path_normalized.startswith(prefix_normalized):
+                raise ValueError(
+                    f"<filepath@> cannot use reserved section '{prefix_value}' ({prefix_name}). "
+                    f"This section is managed by DataJoint. "
+                    f"Got path: {path}"
+                )
+
+        # If filepath_prefix is configured, enforce it
+        filepath_prefix = spec.get("filepath_prefix")
+        if filepath_prefix:
+            filepath_prefix_normalized = filepath_prefix.strip("/") + "/"
+            if not path_normalized.startswith(filepath_prefix_normalized):
+                raise ValueError(
+                    f"<filepath@> must use prefix '{filepath_prefix}' (filepath_prefix). "
+                    f"Got path: {path}"
+                )
 
         # Verify file exists
         backend = get_store_backend(store_name)

src/datajoint/settings.py

@@ -373,6 +373,11 @@ def get_store_spec(self, store: str | None = None) -> dict[str, Any]:
         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)
+
         # Validate protocol
         protocol = spec.get("protocol", "").lower()
         supported_protocols = ("file", "s3", "gcs", "azure")
@@ -394,7 +399,17 @@ def get_store_spec(self, store: str | None = None) -> dict[str, Any]:
             "azure": ("protocol", "container", "location"),
         }
         allowed_keys: dict[str, tuple[str, ...]] = {
-            "file": ("protocol", "location", "subfolding", "partition_pattern", "token_length", "stage"),
+            "file": (
+                "protocol",
+                "location",
+                "subfolding",
+                "partition_pattern",
+                "token_length",
+                "hash_prefix",
+                "schema_prefix",
+                "filepath_prefix",
+                "stage",
+            ),
             "s3": (
                 "protocol",
                 "endpoint",
@@ -406,6 +421,9 @@ def get_store_spec(self, store: str | None = None) -> dict[str, Any]:
                 "subfolding",
                 "partition_pattern",
                 "token_length",
+                "hash_prefix",
+                "schema_prefix",
+                "filepath_prefix",
                 "stage",
                 "proxy_server",
             ),
@@ -418,6 +436,9 @@ def get_store_spec(self, store: str | None = None) -> dict[str, Any]:
                 "subfolding",
                 "partition_pattern",
                 "token_length",
+                "hash_prefix",
+                "schema_prefix",
+                "filepath_prefix",
                 "stage",
             ),
             "azure": (
@@ -430,6 +451,9 @@ def get_store_spec(self, store: str | None = None) -> dict[str, Any]:
                 "subfolding",
                 "partition_pattern",
                 "token_length",
+                "hash_prefix",
+                "schema_prefix",
+                "filepath_prefix",
                 "stage",
             ),
         }
@@ -444,8 +468,67 @@ def get_store_spec(self, store: str | None = None) -> dict[str, Any]:
         if invalid:
             raise DataJointError(f'Invalid key(s) in config.stores["{store}"]: {", ".join(invalid)}')
 
+        # Validate prefix separation to prevent overlap
+        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
 
+    def _validate_prefix_separation(
+        self,
+        store_name: str,
+        hash_prefix: str | None,
+        schema_prefix: str | None,
+        filepath_prefix: str | None,
+    ) -> None:
+        """
+        Validate that storage section prefixes don't overlap.
+
+        Parameters
+        ----------
+        store_name : str
+            Name of the store being validated (for error messages).
+        hash_prefix : str or None
+            Prefix for hash-addressed storage.
+        schema_prefix : str or None
+            Prefix for schema-addressed storage.
+        filepath_prefix : str or None
+            Prefix for filepath storage (None means unrestricted).
+
+        Raises
+        ------
+        DataJointError
+            If any prefixes overlap (one is a parent/child of another).
+        """
+        # Collect non-null prefixes with their names
+        prefixes = []
+        if hash_prefix:
+            prefixes.append(("hash_prefix", hash_prefix))
+        if schema_prefix:
+            prefixes.append(("schema_prefix", schema_prefix))
+        if filepath_prefix:
+            prefixes.append(("filepath_prefix", filepath_prefix))
+
+        # Normalize prefixes: remove leading/trailing slashes, ensure trailing slash for comparison
+        def normalize(p: str) -> str:
+            return p.strip("/") + "/"
+
+        normalized = [(name, normalize(prefix)) for name, prefix in prefixes]
+
+        # Check each pair for overlap
+        for i, (name1, p1) in enumerate(normalized):
+            for j, (name2, p2) in enumerate(normalized[i + 1 :], start=i + 1):
+                # Check if one prefix is a parent of another
+                if p1.startswith(p2) or p2.startswith(p1):
+                    raise DataJointError(
+                        f'config.stores["{store_name}"]: {name1}="{prefixes[i][1]}" and '
+                        f'{name2}="{prefixes[j][1]}" overlap. '
+                        f"Storage section prefixes must be mutually exclusive."
+                    )
 
     def load(self, filename: str | Path) -> None:
         """