refactor: thread config through codec and hash_registry chain

Eliminate the last global config reads in the runtime path by threading
connection-scoped config through the codec encode/decode and
hash_registry layers.

Encode path: table.py adds _config to the context dict, codecs extract
it from key and pass to hash_registry/storage helpers.

Decode path: expression.py passes connection to decode_attribute(),
which builds a decode key with _config for codec.decode() calls.

GC path: scan()/collect() extract config from schemas[0].connection and
pass to list_stored_hashes/delete_path/delete_schema_path.

All functions accept config=None with lazy fallback to settings.config
for backward compatibility.

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

Author: dimitri-yatsenko

docs/design/thread-safe-mode.md

@@ -234,6 +234,14 @@ Every module that previously imported `from .settings import config` now reads c
 | `declare.py` | `config.jobs.add_job_metadata` | `config` param (threaded from `table.py`) |
 | `diagram.py` | `config.display.diagram_direction` | `self._connection._config.display.*` |
 | `staged_insert.py` | `config.get_store_spec()` | `self._table.connection._config.get_store_spec()` |
+| `hash_registry.py` | `config.get_store_spec()` in 5 functions | `config` kwarg (falls back to `settings.config`) |
+| `builtin_codecs/hash.py` | `config` via hash_registry | `_config` from key dict → `config` kwarg to hash_registry |
+| `builtin_codecs/attach.py` | `config.get("download_path")` | `_config` from key dict (falls back to `settings.config`) |
+| `builtin_codecs/filepath.py` | `config.get_store_spec()` | `_config` from key dict (falls back to `settings.config`) |
+| `builtin_codecs/schema.py` | `config.get_store_spec()` in helpers | `config` kwarg to `_build_path()`, `_get_backend()` |
+| `builtin_codecs/npy.py` | `config` via schema helpers | `_config` from key dict → `config` kwarg to helpers |
+| `builtin_codecs/object.py` | `config` via schema helpers | `_config` from key dict → `config` kwarg to helpers |
+| `gc.py` | `config` via hash_registry | `schemas[0].connection._config` → `config` kwarg |
 
 ### Functions that receive config as a parameter
 
@@ -243,8 +251,14 @@ Some module-level functions cannot access `self.connection`. Config is threaded
 |----------|--------|--------------------|
 | `declare()` in `declare.py` | `Table.declare()` in `table.py` | `config=self.connection._config` kwarg |
 | `_get_job_version()` in `jobs.py` | `AutoPopulate._make_tuples()`, `Job.reserve()` | `config=self.connection._config` positional arg |
-
-Both functions accept `config=None` and fall back to the global `settings.config` for backward compatibility.
+| `get_store_backend()` in `hash_registry.py` | codecs, gc.py | `config` kwarg from key dict or schema connection |
+| `get_store_subfolding()` in `hash_registry.py` | `put_hash()` | `config` kwarg chained from caller |
+| `put_hash()` in `hash_registry.py` | `HashCodec.encode()` | `config` kwarg from `_config` in key dict |
+| `get_hash()` in `hash_registry.py` | `HashCodec.decode()` | `config` kwarg from `_config` in key dict |
+| `delete_path()` in `hash_registry.py` | `gc.collect()` | `config` kwarg from `schemas[0].connection._config` |
+| `decode_attribute()` in `codecs.py` | `expression.py` fetch methods | `connection` kwarg → extracts `connection._config` |
+
+All functions accept `config=None` and fall back to the global `settings.config` for backward compatibility.
 
 ## Implementation
 

src/datajoint/builtin_codecs/attach.py

@@ -98,14 +98,15 @@ def decode(self, stored: bytes, *, key: dict | None = None) -> str:
         """
         from pathlib import Path
 
-        from ..settings import config
-
         # Split on first null byte
         null_pos = stored.index(b"\x00")
         filename = stored[:null_pos].decode("utf-8")
         contents = stored[null_pos + 1 :]
 
         # Write to download path
+        config = (key or {}).get("_config")
+        if config is None:
+            from ..settings import config
         download_path = Path(config.get("download_path", "."))
         download_path.mkdir(parents=True, exist_ok=True)
         local_path = download_path / filename

src/datajoint/builtin_codecs/filepath.py

@@ -98,9 +98,12 @@ 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
 
+        config = (key or {}).get("_config")
+        if config is None:
+            from ..settings import config
+
         path = str(value)
 
         # Get store spec to check prefix configuration
@@ -137,7 +140,7 @@ def encode(self, value: Any, *, key: dict | None = None, store_name: str | None
                 raise ValueError(f"<filepath@> must use prefix '{filepath_prefix}' (filepath_prefix). Got path: {path}")
 
         # Verify file exists
-        backend = get_store_backend(store_name)
+        backend = get_store_backend(store_name, config=config)
         if not backend.exists(path):
             raise FileNotFoundError(f"File not found in store '{store_name or 'default'}': {path}")
 
@@ -174,8 +177,9 @@ def decode(self, stored: dict, *, key: dict | None = None) -> Any:
         from ..objectref import ObjectRef
         from ..hash_registry import get_store_backend
 
+        config = (key or {}).get("_config")
         store_name = stored.get("store")
-        backend = get_store_backend(store_name)
+        backend = get_store_backend(store_name, config=config)
         return ObjectRef.from_json(stored, backend=backend)
 
     def validate(self, value: Any) -> None:

src/datajoint/builtin_codecs/hash.py

@@ -76,7 +76,8 @@ def encode(self, value: bytes, *, key: dict | None = None, store_name: str | Non
         from ..hash_registry import put_hash
 
         schema_name = (key or {}).get("_schema", "unknown")
-        return put_hash(value, schema_name=schema_name, store_name=store_name)
+        config = (key or {}).get("_config")
+        return put_hash(value, schema_name=schema_name, store_name=store_name, config=config)
 
     def decode(self, stored: dict, *, key: dict | None = None) -> bytes:
         """
@@ -96,7 +97,8 @@ def decode(self, stored: dict, *, key: dict | None = None) -> bytes:
         """
         from ..hash_registry import get_hash
 
-        return get_hash(stored)
+        config = (key or {}).get("_config")
+        return get_hash(stored, config=config)
 
     def validate(self, value: Any) -> None:
         """Validate that value is bytes."""

src/datajoint/builtin_codecs/npy.py

@@ -336,17 +336,18 @@ def encode(
 
         # Extract context using inherited helper
         schema, table, field, primary_key = self._extract_context(key)
+        config = (key or {}).get("_config")
 
         # Build schema-addressed storage path
-        path, _ = self._build_path(schema, table, field, primary_key, ext=".npy", store_name=store_name)
+        path, _ = self._build_path(schema, table, field, primary_key, ext=".npy", store_name=store_name, config=config)
 
         # Serialize to .npy format
         buffer = io.BytesIO()
         np.save(buffer, value, allow_pickle=False)
         npy_bytes = buffer.getvalue()
 
         # Upload to storage using inherited helper
-        backend = self._get_backend(store_name)
+        backend = self._get_backend(store_name, config=config)
         backend.put_buffer(npy_bytes, path)
 
         # Return metadata (includes numpy-specific shape/dtype)
@@ -373,5 +374,6 @@ def decode(self, stored: dict, *, key: dict | None = None) -> NpyRef:
         NpyRef
             Lazy array reference with metadata access and numpy integration.
         """
-        backend = self._get_backend(stored.get("store"))
+        config = (key or {}).get("_config")
+        backend = self._get_backend(stored.get("store"), config=config)
         return NpyRef(stored, backend)

src/datajoint/builtin_codecs/object.py

@@ -104,6 +104,7 @@ def encode(
 
         # Extract context using inherited helper
         schema, table, field, primary_key = self._extract_context(key)
+        config = (key or {}).get("_config")
 
         # Check for pre-computed metadata (from staged insert)
         if isinstance(value, dict) and "path" in value:
@@ -145,10 +146,10 @@ def encode(
             raise TypeError(f"<object> expects bytes or path, got {type(value).__name__}")
 
         # Build storage path using inherited helper
-        path, token = self._build_path(schema, table, field, primary_key, ext=ext, store_name=store_name)
+        path, token = self._build_path(schema, table, field, primary_key, ext=ext, store_name=store_name, config=config)
 
         # Get storage backend using inherited helper
-        backend = self._get_backend(store_name)
+        backend = self._get_backend(store_name, config=config)
 
         # Upload content
         if is_dir:
@@ -192,7 +193,8 @@ def decode(self, stored: dict, *, key: dict | None = None) -> Any:
         """
         from ..objectref import ObjectRef
 
-        backend = self._get_backend(stored.get("store"))
+        config = (key or {}).get("_config")
+        backend = self._get_backend(stored.get("store"), config=config)
         return ObjectRef.from_json(stored, backend=backend)
 
     def validate(self, value: Any) -> None:

src/datajoint/builtin_codecs/schema.py

@@ -108,6 +108,7 @@ def _build_path(
         primary_key: dict,
         ext: str | None = None,
         store_name: str | None = None,
+        config=None,
     ) -> tuple[str, str]:
         """
         Build schema-addressed storage path.
@@ -131,6 +132,8 @@ def _build_path(
             File extension (e.g., ".npy", ".zarr").
         store_name : str, optional
             Store name for retrieving partition configuration.
+        config : Config, optional
+            Config instance. If None, falls back to global settings.config.
 
         Returns
         -------
@@ -139,7 +142,9 @@ def _build_path(
             is a unique identifier.
         """
         from ..storage import build_object_path
-        from .. import config
+
+        if config is None:
+            from ..settings import config
 
         # Get store configuration for partition_pattern and token_length
         spec = config.get_store_spec(store_name)
@@ -156,14 +161,16 @@ def _build_path(
             token_length=token_length,
         )
 
-    def _get_backend(self, store_name: str | None = None):
+    def _get_backend(self, store_name: str | None = None, config=None):
         """
         Get storage backend by name.
 
         Parameters
         ----------
         store_name : str, optional
             Store name. If None, returns default store.
+        config : Config, optional
+            Config instance. If None, falls back to global settings.config.
 
         Returns
         -------
@@ -172,4 +179,4 @@ def _get_backend(self, store_name: str | None = None):
         """
         from ..hash_registry import get_store_backend
 
-        return get_store_backend(store_name)
+        return get_store_backend(store_name, config=config)

src/datajoint/codecs.py

@@ -515,7 +515,7 @@ def lookup_codec(codec_spec: str) -> tuple[Codec, str | None]:
 # =============================================================================
 
 
-def decode_attribute(attr, data, squeeze: bool = False):
+def decode_attribute(attr, data, squeeze: bool = False, connection=None):
     """
     Decode raw database value using attribute's codec or native type handling.
 
@@ -528,6 +528,8 @@ def decode_attribute(attr, data, squeeze: bool = False):
         attr: Attribute from the table's heading.
         data: Raw value fetched from the database.
         squeeze: If True, remove singleton dimensions from numpy arrays.
+        connection: Connection instance for config access. If provided,
+            ``connection._config`` is passed to codecs via the key dict.
 
     Returns:
         Decoded Python value.
@@ -560,9 +562,14 @@ def decode_attribute(attr, data, squeeze: bool = False):
         elif final_dtype.lower() == "binary(16)":
             data = uuid_module.UUID(bytes=data)
 
+        # Build decode key with config if connection is available
+        decode_key = None
+        if connection is not None:
+            decode_key = {"_config": connection._config}
+
         # Apply decoders in reverse order: innermost first, then outermost
         for codec in reversed(type_chain):
-            data = codec.decode(data, key=None)
+            data = codec.decode(data, key=decode_key)
 
         # Squeeze arrays if requested
         if squeeze and isinstance(data, np.ndarray):

src/datajoint/expression.py

@@ -715,7 +715,7 @@ def fetch(
         import warnings
 
         warnings.warn(
-            "fetch() is deprecated in DataJoint 2.0. " "Use to_dicts(), to_pandas(), to_arrays(), or keys() instead.",
+            "fetch() is deprecated in DataJoint 2.0. Use to_dicts(), to_pandas(), to_arrays(), or keys() instead.",
             DeprecationWarning,
             stacklevel=2,
         )
@@ -817,7 +817,10 @@ def fetch1(self, *attrs, squeeze=False):
             row = cursor.fetchone()
             if not row or cursor.fetchone():
                 raise DataJointError("fetch1 requires exactly one tuple in the input set.")
-            return {name: decode_attribute(heading[name], row[name], squeeze=squeeze) for name in heading.names}
+            return {
+                name: decode_attribute(heading[name], row[name], squeeze=squeeze, connection=self.connection)
+                for name in heading.names
+            }
         else:
             # Handle "KEY" specially - it means primary key columns
             def is_key(attr):
@@ -892,7 +895,10 @@ def to_dicts(self, order_by=None, limit=None, offset=None, squeeze=False):
         expr = self._apply_top(order_by, limit, offset)
         cursor = expr.cursor(as_dict=True)
         heading = expr.heading
-        return [{name: decode_attribute(heading[name], row[name], squeeze) for name in heading.names} for row in cursor]
+        return [
+            {name: decode_attribute(heading[name], row[name], squeeze, connection=expr.connection) for name in heading.names}
+            for row in cursor
+        ]
 
     def to_pandas(self, order_by=None, limit=None, offset=None, squeeze=False):
         """
@@ -1063,7 +1069,7 @@ def to_arrays(self, *attrs, include_key=False, order_by=None, limit=None, offset
             return result_arrays[0] if len(attrs) == 1 else tuple(result_arrays)
         else:
             # Fetch all columns as structured array
-            get = partial(decode_attribute, squeeze=squeeze)
+            get = partial(decode_attribute, squeeze=squeeze, connection=expr.connection)
             cursor = expr.cursor(as_dict=False)
             rows = list(cursor.fetchall())
 
@@ -1217,7 +1223,10 @@ def __iter__(self):
         cursor = self.cursor(as_dict=True)
         heading = self.heading
         for row in cursor:
-            yield {name: decode_attribute(heading[name], row[name], squeeze=False) for name in heading.names}
+            yield {
+                name: decode_attribute(heading[name], row[name], squeeze=False, connection=self.connection)
+                for name in heading.names
+            }
 
     def cursor(self, as_dict=False):
         """