refactor: replace global config reads with connection-scoped config

All internal code now reads configuration from self.connection._config
instead of the global config singleton. This ensures thread-safe mode
works correctly: each Instance's connection carries its own config,
and tables/schemas/jobs access it through the connection.

Changes across 9 files:
- schemas.py: safemode, create_tables default
- table.py: safemode in delete/drop, config passed to declare()
- expression.py: loglevel in __repr__
- preview.py: display.* settings via query_expression.connection._config
- autopopulate.py: jobs.allow_new_pk_fields, jobs.auto_refresh
- jobs.py: jobs.default_priority, stale_timeout, keep_completed
- declare.py: jobs.add_job_metadata (config param threaded through)
- diagram.py: display.diagram_direction (connection stored on instance)
- staged_insert.py: get_store_spec()

Removed unused `from .settings import config` imports from 7 modules.

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

Author: dimitri-yatsenko

src/datajoint/autopopulate.py

@@ -146,10 +146,8 @@ def _declare_check(self, primary_key: list[str], fk_attribute_map: dict[str, tup
             If native (non-FK) PK attributes are found, unless bypassed via
             ``dj.config.jobs.allow_new_pk_fields_in_computed_tables = True``.
         """
-        from .settings import config
-
         # Check if validation is bypassed
-        if config.jobs.allow_new_pk_fields_in_computed_tables:
+        if self.connection._config.jobs.allow_new_pk_fields_in_computed_tables:
             return
 
         # Check for native (non-FK) primary key attributes
@@ -477,8 +475,6 @@ def _populate_distributed(
         """
         from tqdm import tqdm
 
-        from .settings import config
-
         # Define a signal handler for SIGTERM
         def handler(signum, frame):
             logger.info("Populate terminated by SIGTERM")
@@ -489,7 +485,7 @@ def handler(signum, frame):
         try:
             # Refresh job queue if configured
             if refresh is None:
-                refresh = config.jobs.auto_refresh
+                refresh = self.connection._config.jobs.auto_refresh
             if refresh:
                 # Use delay=-1 to ensure jobs are immediately schedulable
                 # (avoids race condition with scheduled_time <= CURRENT_TIMESTAMP(3) check)
@@ -659,7 +655,7 @@ def _populate1(
                     key,
                     start_time=datetime.datetime.fromtimestamp(start_time),
                     duration=duration,
-                    version=_get_job_version(),
+                    version=_get_job_version(self.connection._config),
                 )
 
             if jobs is not None:

src/datajoint/declare.py

@@ -15,7 +15,6 @@
 from .codecs import lookup_codec
 from .condition import translate_attribute
 from .errors import DataJointError
-from .settings import config
 
 # Core DataJoint types - scientist-friendly names that are fully supported
 # These are recorded in field comments using :type: syntax for reconstruction
@@ -401,7 +400,7 @@ def prepare_declare(
 
 
 def declare(
-    full_table_name: str, definition: str, context: dict, adapter
+    full_table_name: str, definition: str, context: dict, adapter, *, config=None
 ) -> tuple[str, list[str], list[str], dict[str, tuple[str, str]], list[str], list[str]]:
     r"""
     Parse a definition and generate SQL CREATE TABLE statement.
@@ -416,6 +415,8 @@ def declare(
         Namespace for resolving foreign key references.
     adapter : DatabaseAdapter
         Database adapter for backend-specific SQL generation.
+    config : Config, optional
+        Configuration object. If None, falls back to global config.
 
     Returns
     -------
@@ -464,6 +465,10 @@ def declare(
     ) = prepare_declare(definition, context, adapter)
 
     # Add hidden job metadata for Computed/Imported tables (not parts)
+    if config is None:
+        from .settings import config as _config
+
+        config = _config
     if config.jobs.add_job_metadata:
         # Check if this is a Computed (__) or Imported (_) table, but not a Part (contains __ in middle)
         is_computed = table_name.startswith("__") and "__" not in table_name[2:]

src/datajoint/diagram.py

@@ -16,7 +16,6 @@
 
 from .dependencies import topo_sort
 from .errors import DataJointError
-from .settings import config
 from .table import Table, lookup_class_name
 from .user_tables import Computed, Imported, Lookup, Manual, Part, _AliasNode, _get_tier
 
@@ -105,6 +104,7 @@ def __init__(self, source, context=None) -> None:
                 self.nodes_to_show = set(source.nodes_to_show)
                 self._expanded_nodes = set(source._expanded_nodes)
                 self.context = source.context
+                self._connection = source._connection
                 super().__init__(source)
                 return
 
@@ -126,6 +126,7 @@ def __init__(self, source, context=None) -> None:
                     raise DataJointError("Could not find database connection in %s" % repr(source[0]))
 
             # initialize graph from dependencies
+            self._connection = connection
             connection.dependencies.load()
             super().__init__(connection.dependencies)
 
@@ -584,7 +585,7 @@ def make_dot(self):
             Tables are grouped by schema, with the Python module name shown as the
             group label when available.
             """
-            direction = config.display.diagram_direction
+            direction = self._connection._config.display.diagram_direction
             graph = self._make_graph()
 
             # Apply collapse logic if needed
@@ -857,7 +858,7 @@ def make_mermaid(self) -> str:
                 Session --> Neuron
             """
             graph = self._make_graph()
-            direction = config.display.diagram_direction
+            direction = self._connection._config.display.diagram_direction
 
             # Apply collapse logic if needed
             graph, collapsed_counts = self._apply_collapse(graph)

src/datajoint/expression.py

@@ -20,7 +20,6 @@
 from .errors import DataJointError
 from .codecs import decode_attribute
 from .preview import preview, repr_html
-from .settings import config
 
 logger = logging.getLogger(__name__.split(".")[0])
 
@@ -1247,7 +1246,7 @@ def __repr__(self):
         str
             String representation of the QueryExpression.
         """
-        return super().__repr__() if config["loglevel"].lower() == "debug" else self.preview()
+        return super().__repr__() if self.connection._config["loglevel"].lower() == "debug" else self.preview()
 
     def preview(self, limit=None, width=None):
         """

src/datajoint/jobs.py

@@ -24,16 +24,22 @@
 logger = logging.getLogger(__name__.split(".")[0])
 
 
-def _get_job_version() -> str:
+def _get_job_version(config=None) -> str:
     """
     Get version string based on config settings.
 
+    Parameters
+    ----------
+    config : Config, optional
+        Configuration object. If None, falls back to global config.
+
     Returns
     -------
     str
         Version string, or empty string if version tracking disabled.
     """
-    from .settings import config
+    if config is None:
+        from .settings import config
 
     method = config.jobs.version_method
     if method is None or method == "none":
@@ -349,17 +355,15 @@ def refresh(
         3. Remove stale jobs: jobs older than stale_timeout whose keys not in key_source
         4. Remove orphaned jobs: reserved jobs older than orphan_timeout (if specified)
         """
-        from .settings import config
-
         # Ensure jobs table exists
         if not self.is_declared:
             self.declare()
 
         # Get defaults from config
         if priority is None:
-            priority = config.jobs.default_priority
+            priority = self.connection._config.jobs.default_priority
         if stale_timeout is None:
-            stale_timeout = config.jobs.stale_timeout
+            stale_timeout = self.connection._config.jobs.stale_timeout
 
         result = {"added": 0, "removed": 0, "orphaned": 0, "re_pended": 0}
 
@@ -392,7 +396,7 @@ def refresh(
                     pass  # Job already exists
 
         # 2. Re-pend success jobs if keep_completed=True
-        if config.jobs.keep_completed:
+        if self.connection._config.jobs.keep_completed:
             # Success jobs whose keys are in key_source but not in target
             # Disable semantic_check for Job table operations (job table PK has different lineage than target)
             success_to_repend = self.completed.restrict(key_source, semantic_check=False).restrict(
@@ -463,7 +467,7 @@ def reserve(self, key: dict) -> bool:
             "pid": os.getpid(),
             "connection_id": self.connection.connection_id,
             "user": self.connection.get_user(),
-            "version": _get_job_version(),
+            "version": _get_job_version(self.connection._config),
         }
 
         try:
@@ -490,9 +494,7 @@ def complete(self, key: dict, duration: float | None = None) -> None:
         - If True: updates status to ``'success'`` with completion time and duration
         - If False: deletes the job entry
         """
-        from .settings import config
-
-        if config.jobs.keep_completed:
+        if self.connection._config.jobs.keep_completed:
             # Use server time for completed_time
             server_now = self.connection.query("SELECT CURRENT_TIMESTAMP").fetchone()[0]
             pk = self._get_pk(key)
@@ -550,13 +552,11 @@ def ignore(self, key: dict) -> None:
         key : dict
             Primary key dict of the job.
         """
-        from .settings import config
-
         pk = self._get_pk(key)
         if pk in self:
             self.update1({**pk, "status": "ignore"})
         else:
-            priority = config.jobs.default_priority
+            priority = self.connection._config.jobs.default_priority
             self.insert1({**pk, "status": "ignore", "priority": priority})
 
     def progress(self) -> dict:

src/datajoint/preview.py

@@ -2,8 +2,6 @@
 
 import json
 
-from .settings import config
-
 
 def _format_object_display(json_data):
     """Format object metadata for display in query results."""
@@ -44,6 +42,7 @@ def _get_blob_placeholder(heading, field_name, html_escape=False):
 def preview(query_expression, limit, width):
     heading = query_expression.heading
     rel = query_expression.proj(*heading.non_blobs)
+    config = query_expression.connection._config
     # Object fields use codecs - not specially handled in simplified model
     object_fields = []
     if limit is None:
@@ -105,6 +104,7 @@ def get_display_value(tup, f, idx):
 def repr_html(query_expression):
     heading = query_expression.heading
     rel = query_expression.proj(*heading.non_blobs)
+    config = query_expression.connection._config
     # Object fields use codecs - not specially handled in simplified model
     object_fields = []
     tuples = rel.to_arrays(limit=config["display.limit"] + 1)

src/datajoint/schemas.py

@@ -23,7 +23,6 @@
     from .connection import Connection
 from .heading import Heading
 from .jobs import Job
-from .settings import config
 from .table import FreeTable, lookup_class_name
 from .user_tables import Computed, Imported, Lookup, Manual, Part, _get_tier
 from .utils import to_camel_case, user_choice
@@ -120,7 +119,7 @@ def __init__(
         self.database = None
         self.context = context
         self.create_schema = create_schema
-        self.create_tables = create_tables if create_tables is not None else config.database.create_tables
+        self.create_tables = create_tables  # None means "use connection config default"
         self.add_objects = add_objects
         self.declare_list = []
         if schema_name:
@@ -293,7 +292,10 @@ def _decorate_table(self, table_class: type, context: dict[str, Any], assert_dec
         # instantiate the class, declare the table if not already
         instance = table_class()
         is_declared = instance.is_declared
-        if not is_declared and not assert_declared and self.create_tables:
+        create_tables = (
+            self.create_tables if self.create_tables is not None else self.connection._config.database.create_tables
+        )
+        if not is_declared and not assert_declared and create_tables:
             instance.declare(context)
             self.connection.dependencies.clear()
         is_declared = is_declared or instance.is_declared
@@ -409,7 +411,7 @@ def drop(self, prompt: bool | None = None) -> None:
         AccessError
             If insufficient permissions to drop the schema.
         """
-        prompt = config["safemode"] if prompt is None else prompt
+        prompt = self.connection._config["safemode"] if prompt is None else prompt
 
         if not self.exists:
             logger.info("Schema named `{database}` does not exist. Doing nothing.".format(database=self.database))

src/datajoint/staged_insert.py

@@ -14,7 +14,6 @@
 import fsspec
 
 from .errors import DataJointError
-from .settings import config
 from .storage import StorageBackend, build_object_path
 
 
@@ -69,7 +68,7 @@ def _ensure_backend(self):
         """Ensure storage backend is initialized."""
         if self._backend is None:
             try:
-                spec = config.get_store_spec()  # Uses stores.default
+                spec = self._table.connection._config.get_store_spec()  # Uses stores.default
                 self._backend = StorageBackend(spec)
             except DataJointError:
                 raise DataJointError(
@@ -110,7 +109,7 @@ def _get_storage_path(self, field: str, ext: str = "") -> str:
             )
 
         # Get storage spec (uses stores.default)
-        spec = config.get_store_spec()
+        spec = self._table.connection._config.get_store_spec()
         partition_pattern = spec.get("partition_pattern")
         token_length = spec.get("token_length", 8)
 

src/datajoint/table.py

@@ -23,7 +23,6 @@
 )
 from .expression import QueryExpression
 from .heading import Heading
-from .settings import config
 from .staged_insert import staged_insert1 as _staged_insert1
 from .utils import get_master, is_camel_case, user_choice
 
@@ -153,7 +152,7 @@ def declare(self, context=None):
                 "Class names must be in CamelCase, starting with a capital letter."
             )
         sql, _external_stores, primary_key, fk_attribute_map, pre_ddl, post_ddl = declare(
-            self.full_table_name, self.definition, context, self.connection.adapter
+            self.full_table_name, self.definition, context, self.connection.adapter, config=self.connection._config
         )
 
         # Call declaration hook for validation (subclasses like AutoPopulate can override)
@@ -1119,7 +1118,7 @@ def strip_quotes(s):
                 raise DataJointError("Exceeded maximum number of delete attempts.")
             return delete_count
 
-        prompt = config["safemode"] if prompt is None else prompt
+        prompt = self.connection._config["safemode"] if prompt is None else prompt
 
         # Start transaction
         if transaction:
@@ -1227,7 +1226,7 @@ def drop(self, prompt: bool | None = None):
             raise DataJointError(
                 "A table with an applied restriction cannot be dropped. Call drop() on the unrestricted Table."
             )
-        prompt = config["safemode"] if prompt is None else prompt
+        prompt = self.connection._config["safemode"] if prompt is None else prompt
 
         self.connection.dependencies.load()
         do_drop = True