Implement multi-store support for object type

- Add object@store_name syntax in table definitions (declare.py)
- Extract store name from attribute type in heading.py
- Add url and store fields to ObjectRef for metadata storage
- Add object_storage.stores.<name>.* configuration (settings.py)
- Add get_object_store_spec() method for named store configs
- Update table.py to use correct backend per attribute store
- Update fetch.py to use store from JSON metadata
- Rename ObjectRef.store property to fsmap to avoid conflict

This enables different object attributes to use different storage
backends (e.g., private vs public buckets) while maintaining OAS
(Object-Augmented Schema) principles for all stores.

Author: claude

src/datajoint/declare.py

@@ -65,7 +65,7 @@
         INTERNAL_ATTACH=r"attach$",
         EXTERNAL_ATTACH=r"attach@(?P<store>[a-z][\-\w]*)$",
         FILEPATH=r"filepath@(?P<store>[a-z][\-\w]*)$",
-        OBJECT=r"object$",  # managed object storage (files/folders)
+        OBJECT=r"object(@(?P<store>[a-z][\-\w]*))?$",  # managed object storage (files/folders)
         UUID=r"uuid$",
         ADAPTED=r"<.+>$",
     ).items()
@@ -469,6 +469,9 @@ def substitute_special_type(match, category, foreign_key_sql, context):
         match["type"] = "LONGBLOB"
     elif category == "OBJECT":
         # Object type stores metadata as JSON - no foreign key to external table
+        # Extract store name if present (object@store_name syntax)
+        if "@" in match["type"]:
+            match["store"] = match["type"].split("@", 1)[1]
         match["type"] = "JSON"
     elif category in EXTERNAL_TYPES:
         if category == "FILEPATH" and not _support_filepath_types():

src/datajoint/fetch.py

@@ -53,8 +53,10 @@ def _get(connection, attr, data, squeeze, download_path):
     if attr.is_object:
         # Object type - return ObjectRef handle
         json_data = json.loads(data) if isinstance(data, str) else data
+        # Get the correct backend based on store name in metadata
+        store_name = json_data.get("store")  # None for default store
         try:
-            spec = config.get_object_storage_spec()
+            spec = config.get_object_store_spec(store_name)
             backend = StorageBackend(spec)
         except DataJointError:
             backend = None

src/datajoint/heading.py

@@ -331,6 +331,13 @@ def _init_from_database(self):
                         {env} = TRUE or upgrade datajoint.
                         """.format(env=FILEPATH_FEATURE_SWITCH)
                     )
+                # Extract store name for external types and object types with named stores
+                store = None
+                if category in EXTERNAL_TYPES:
+                    store = attr["type"].split("@")[1]
+                elif category == "OBJECT" and "@" in attr["type"]:
+                    store = attr["type"].split("@")[1]
+
                 attr.update(
                     unsupported=False,
                     is_attachment=category in ("INTERNAL_ATTACH", "EXTERNAL_ATTACH"),
@@ -340,7 +347,7 @@ def _init_from_database(self):
                     is_blob=category in ("INTERNAL_BLOB", "EXTERNAL_BLOB"),
                     uuid=category == "UUID",
                     is_external=category in EXTERNAL_TYPES,
-                    store=(attr["type"].split("@")[1] if category in EXTERNAL_TYPES else None),
+                    store=store,
                 )
 
             if attr["in_key"] and any(

src/datajoint/objectref.py

@@ -34,7 +34,9 @@ class ObjectRef:
     from the storage backend.
 
     Attributes:
-        path: Full path/key within storage backend (includes token)
+        path: Relative path within the store (includes token)
+        url: Full URI to the object (e.g., 's3://bucket/path/to/object.dat')
+        store: Store name (None for default store)
         size: Total size in bytes (sum for folders), or None if not computed.
             For large hierarchical data like Zarr stores, size computation can
             be expensive and is optional.
@@ -53,6 +55,8 @@ class ObjectRef:
     ext: str | None
     is_dir: bool
     timestamp: datetime
+    url: str | None = None
+    store: str | None = None
     mime_type: str | None = None
     item_count: int | None = None
     _backend: StorageBackend | None = None
@@ -80,6 +84,8 @@ def from_json(cls, json_data: dict | str, backend: StorageBackend | None = None)
 
         return cls(
             path=data["path"],
+            url=data.get("url"),
+            store=data.get("store"),
             size=data["size"],
             hash=data.get("hash"),
             ext=data.get("ext"),
@@ -105,6 +111,10 @@ def to_json(self) -> dict:
             "is_dir": self.is_dir,
             "timestamp": self.timestamp.isoformat() if self.timestamp else None,
         }
+        if self.url:
+            data["url"] = self.url
+        if self.store:
+            data["store"] = self.store
         if self.mime_type:
             data["mime_type"] = self.mime_type
         if self.item_count is not None:
@@ -121,7 +131,9 @@ def to_dict(self) -> dict:
 
         Returns:
             Dict containing the object metadata:
-                - path: Storage path
+                - path: Relative storage path within the store
+                - url: Full URI (e.g., 's3://bucket/path') (optional)
+                - store: Store name (optional, None for default store)
                 - size: File/folder size in bytes (or None)
                 - hash: Content hash (or None)
                 - ext: File extension (or None)
@@ -152,12 +164,15 @@ def fs(self) -> fsspec.AbstractFileSystem:
         return self._backend.fs
 
     @property
-    def store(self) -> fsspec.FSMap:
+    def fsmap(self) -> fsspec.FSMap:
         """
         Return FSMap suitable for Zarr/xarray.
 
         This provides a dict-like interface to the storage location,
         compatible with zarr.open() and xarray.open_zarr().
+
+        Example:
+            >>> z = zarr.open(obj_ref.fsmap, mode='r')
         """
         self._ensure_backend()
         full_path = self._backend._full_path(self.path)

src/datajoint/settings.py

@@ -215,6 +215,9 @@ class ObjectStorageSettings(BaseSettings):
     partition_pattern: str | None = Field(default=None, description="Path pattern with {attribute} placeholders")
     token_length: int = Field(default=8, ge=4, le=16, description="Random suffix length for filenames")
 
+    # Named stores configuration (object_storage.stores.<name>.*)
+    stores: dict[str, dict[str, Any]] = Field(default_factory=dict, description="Named object stores")
+
 
 class Config(BaseSettings):
     """
@@ -432,6 +435,88 @@ def get_object_storage_spec(self) -> dict[str, Any]:
 
         return spec
 
+    def get_object_store_spec(self, store_name: str | None = None) -> dict[str, Any]:
+        """
+        Get validated configuration for a specific object store.
+
+        Args:
+            store_name: Name of the store (None for default store)
+
+        Returns:
+            Object store configuration dict
+
+        Raises:
+            DataJointError: If store is not configured or has invalid config
+        """
+        if store_name is None:
+            # Return default store spec
+            return self.get_object_storage_spec()
+
+        os_settings = self.object_storage
+
+        # Check if named store exists
+        if store_name not in os_settings.stores:
+            raise DataJointError(
+                f"Object store '{store_name}' is not configured. "
+                f"Add object_storage.stores.{store_name}.* settings to datajoint.json"
+            )
+
+        store_config = os_settings.stores[store_name]
+        protocol = store_config.get("protocol", "").lower()
+
+        supported_protocols = ("file", "s3", "gcs", "azure")
+        if protocol not in supported_protocols:
+            raise DataJointError(
+                f"Invalid protocol for store '{store_name}': {protocol}. "
+                f'Supported protocols: {", ".join(supported_protocols)}'
+            )
+
+        # Use project_name from default config if not specified in store
+        project_name = store_config.get("project_name") or os_settings.project_name
+        if not project_name:
+            raise DataJointError(
+                f"project_name is required for object store '{store_name}'. "
+                "Set object_storage.project_name or object_storage.stores.{store_name}.project_name"
+            )
+
+        # Build spec dict
+        spec = {
+            "project_name": project_name,
+            "protocol": protocol,
+            "location": store_config.get("location", ""),
+            "partition_pattern": store_config.get("partition_pattern") or os_settings.partition_pattern,
+            "token_length": store_config.get("token_length") or os_settings.token_length,
+            "store_name": store_name,
+        }
+
+        # Add protocol-specific settings
+        if protocol == "s3":
+            endpoint = store_config.get("endpoint")
+            bucket = store_config.get("bucket")
+            if not endpoint or not bucket:
+                raise DataJointError(f"endpoint and bucket are required for S3 store '{store_name}'")
+            spec.update(
+                {
+                    "endpoint": endpoint,
+                    "bucket": bucket,
+                    "access_key": store_config.get("access_key"),
+                    "secret_key": store_config.get("secret_key"),
+                    "secure": store_config.get("secure", True),
+                }
+            )
+        elif protocol == "gcs":
+            bucket = store_config.get("bucket")
+            if not bucket:
+                raise DataJointError(f"bucket is required for GCS store '{store_name}'")
+            spec["bucket"] = bucket
+        elif protocol == "azure":
+            container = store_config.get("container")
+            if not container:
+                raise DataJointError(f"container is required for Azure store '{store_name}'")
+            spec["container"] = container
+
+        return spec
+
     def load(self, filename: str | Path) -> None:
         """
         Load settings from a JSON file.
@@ -464,6 +549,13 @@ def _update_from_flat_dict(self, data: dict[str, Any]) -> None:
                     group_obj = getattr(self, group)
                     if hasattr(group_obj, attr):
                         setattr(group_obj, attr, value)
+            elif len(parts) == 4:
+                # Handle object_storage.stores.<name>.<attr> pattern
+                group, subgroup, store_name, attr = parts
+                if group == "object_storage" and subgroup == "stores":
+                    if store_name not in self.object_storage.stores:
+                        self.object_storage.stores[store_name] = {}
+                    self.object_storage.stores[store_name][attr] = value
 
     def _load_secrets(self, secrets_dir: Path) -> None:
         """Load secrets from a secrets directory."""

src/datajoint/table.py

@@ -275,32 +275,49 @@ def external(self):
 
     @property
     def object_storage(self) -> StorageBackend | None:
-        """Get the object storage backend for this table."""
-        if not hasattr(self, "_object_storage"):
+        """Get the default object storage backend for this table."""
+        return self.get_object_storage()
+
+    def get_object_storage(self, store_name: str | None = None) -> StorageBackend | None:
+        """
+        Get the object storage backend for a specific store.
+
+        Args:
+            store_name: Name of the store (None for default store)
+
+        Returns:
+            StorageBackend instance or None if not configured
+        """
+        cache_key = f"_object_storage_{store_name or 'default'}"
+        if not hasattr(self, cache_key):
             try:
-                spec = config.get_object_storage_spec()
-                self._object_storage = StorageBackend(spec)
+                spec = config.get_object_store_spec(store_name)
+                backend = StorageBackend(spec)
                 # Verify/create store metadata on first use
-                verify_or_create_store_metadata(self._object_storage, spec)
+                verify_or_create_store_metadata(backend, spec)
+                setattr(self, cache_key, backend)
             except DataJointError:
-                self._object_storage = None
-        return self._object_storage
+                setattr(self, cache_key, None)
+        return getattr(self, cache_key)
 
-    def _process_object_value(self, name: str, value, row: dict) -> str:
+    def _process_object_value(self, name: str, value, row: dict, store_name: str | None = None) -> str:
         """
         Process an object attribute value for insert.
 
         Args:
             name: Attribute name
             value: Input value (file path, folder path, or (ext, stream) tuple)
             row: The full row dict (needed for primary key values)
+            store_name: Name of the object store (None for default store)
 
         Returns:
             JSON string for database storage
         """
-        if self.object_storage is None:
+        backend = self.get_object_storage(store_name)
+        if backend is None:
+            store_desc = f"'{store_name}'" if store_name else "default"
             raise DataJointError(
-                "Object storage is not configured. Set object_storage settings in datajoint.json "
+                f"Object storage ({store_desc}) is not configured. Set object_storage settings in datajoint.json "
                 "or DJ_OBJECT_STORAGE_* environment variables."
             )
 
@@ -339,7 +356,7 @@ def _process_object_value(self, name: str, value, row: dict) -> str:
             )
 
         # Get storage spec for path building
-        spec = config.get_object_storage_spec()
+        spec = config.get_object_store_spec(store_name)
         partition_pattern = spec.get("partition_pattern")
         token_length = spec.get("token_length", 8)
         location = spec.get("location", "")
@@ -362,24 +379,33 @@ def _process_object_value(self, name: str, value, row: dict) -> str:
         manifest = None
         if source_path:
             if is_dir:
-                manifest = self.object_storage.put_folder(source_path, full_storage_path)
+                manifest = backend.put_folder(source_path, full_storage_path)
                 size = manifest["total_size"]
             else:
-                self.object_storage.put_file(source_path, full_storage_path)
+                backend.put_file(source_path, full_storage_path)
         elif stream:
-            self.object_storage.put_buffer(content, full_storage_path)
+            backend.put_buffer(content, full_storage_path)
+
+        # Build full URL for the object
+        url = self._build_object_url(spec, full_storage_path)
 
         # Build JSON metadata
         timestamp = datetime.now(timezone.utc).isoformat()
         metadata = {
-            "path": relative_path,
+            "path": full_storage_path,
             "size": size,
             "hash": None,  # Hash is optional, not computed by default
             "ext": ext,
             "is_dir": is_dir,
             "timestamp": timestamp,
         }
 
+        # Add URL and store name
+        if url:
+            metadata["url"] = url
+        if store_name:
+            metadata["store"] = store_name
+
         # Add mime_type for files
         if not is_dir and ext:
             mime_type, _ = mimetypes.guess_type(f"file{ext}")
@@ -392,6 +418,34 @@ def _process_object_value(self, name: str, value, row: dict) -> str:
 
         return json.dumps(metadata)
 
+    def _build_object_url(self, spec: dict, path: str) -> str | None:
+        """
+        Build a full URL for an object based on the storage spec.
+
+        Args:
+            spec: Storage configuration dict
+            path: Path within the storage
+
+        Returns:
+            Full URL string or None for local storage
+        """
+        protocol = spec.get("protocol", "")
+        if protocol == "s3":
+            bucket = spec.get("bucket", "")
+            return f"s3://{bucket}/{path}"
+        elif protocol == "gcs":
+            bucket = spec.get("bucket", "")
+            return f"gs://{bucket}/{path}"
+        elif protocol == "azure":
+            container = spec.get("container", "")
+            return f"az://{container}/{path}"
+        elif protocol == "file":
+            # For local storage, return file:// URL
+            location = spec.get("location", "")
+            full_path = f"{location}/{path}" if location else path
+            return f"file://{full_path}"
+        return None
+
     def update1(self, row):
         """
         ``update1`` updates one existing entry in the table.
@@ -912,7 +966,7 @@ def __make_placeholder(self, name, value, ignore_extra_fields=False, row=None):
                     raise DataJointError(
                         f"Object attribute {name} requires full row context for insert. " "This is an internal error."
                     )
-                value = self._process_object_value(name, value, row)
+                value = self._process_object_value(name, value, row, store_name=attr.store)
             elif attr.numeric:
                 value = str(int(value) if isinstance(value, bool) else value)
             elif attr.json: