settings.py

"""
DataJoint configuration system using pydantic-settings.

This module provides strongly-typed configuration with automatic loading
from environment variables, secrets directories, and JSON config files.

Configuration sources (highest precedence first):

1. Runtime assignment (``dj.config[...] = ...``)
2. Environment variables (``DJ_*``)
3. Project config file (``datajoint.json``, searched recursively up to ``.git/.hg``)
4. Secrets directories (``.secrets/`` in project, ``/run/secrets/datajoint/``)
5. Built-in defaults

A value set by a higher source is not overridden by a lower one. In particular
``datajoint.json`` takes precedence over ``.secrets/``: the secrets directory
only supplies values the config file and environment leave unset.

Examples
--------
>>> import datajoint as dj
>>> dj.config.database.host
'localhost'
>>> dj.config.database.backend
'mysql'
>>> dj.config.database.port  # Auto-detects: 3306 for MySQL, 5432 for PostgreSQL
3306
>>> with dj.config.override(safemode=False):
...     # dangerous operations here
...     pass

Project structure::

    myproject/
    ├── .git/
    ├── datajoint.json      # Project config (commit this)
    ├── .secrets/           # Local secrets (gitignore this)
    │   ├── database.password
    │   └── stores.main.secret_key   # one file per setting: database.* or stores.<name>.<attr>
    └── src/
        └── analysis.py     # Config found via parent search
"""

from __future__ import annotations

import copy
import json
import logging
import os
import warnings
from contextlib import contextmanager
from enum import Enum
from pathlib import Path
from typing import Any, Iterator, Literal

from pydantic import Field, SecretStr, field_validator, model_validator
from pydantic_settings import BaseSettings, SettingsConfigDict

from .errors import DataJointError

CONFIG_FILENAME = "datajoint.json"
SECRETS_DIRNAME = ".secrets"
SYSTEM_SECRETS_DIR = Path("/run/secrets/datajoint")
DEFAULT_SUBFOLDING = (2, 2)

# Mapping of config keys to environment variables
# Environment variables take precedence over config file values
ENV_VAR_MAPPING = {
    "database.host": "DJ_HOST",
    "database.user": "DJ_USER",
    "database.password": "DJ_PASS",
    "database.backend": "DJ_BACKEND",
    "database.port": "DJ_PORT",
    "database.name": "DJ_DATABASE_NAME",
    "database.database_prefix": "DJ_DATABASE_PREFIX",
    "database.create_tables": "DJ_CREATE_TABLES",
    "loglevel": "DJ_LOG_LEVEL",
    "display.diagram_direction": "DJ_DIAGRAM_DIRECTION",
    "display.diagram_theme": "DJ_DIAGRAM_THEME",
}

Role = Enum("Role", "manual lookup imported computed job")
role_to_prefix = {
    Role.manual: "",
    Role.lookup: "#",
    Role.imported: "_",
    Role.computed: "__",
    Role.job: "~",
}
prefix_to_role = dict(zip(role_to_prefix.values(), role_to_prefix))

logger = logging.getLogger(__name__.split(".")[0])


def find_config_file(start: Path | None = None) -> Path | None:
    """
    Search for datajoint.json in current and parent directories.

    Searches upward from ``start`` until finding the config file or hitting
    a project boundary (``.git``, ``.hg``) or filesystem root.

    Parameters
    ----------
    start : Path, optional
        Directory to start search from. Defaults to current working directory.

    Returns
    -------
    Path or None
        Path to config file if found, None otherwise.
    """
    current = (start or Path.cwd()).resolve()

    while True:
        config_path = current / CONFIG_FILENAME
        if config_path.is_file():
            return config_path

        # Stop at project/repo root
        if (current / ".git").exists() or (current / ".hg").exists():
            return None

        # Stop at filesystem root
        if current == current.parent:
            return None

        current = current.parent


def find_secrets_dir(config_path: Path | None = None) -> Path | None:
    """
    Find the secrets directory.

    Priority:

    1. ``.secrets/`` in same directory as datajoint.json (project secrets)
    2. ``/run/secrets/datajoint/`` (Docker/Kubernetes secrets)

    Parameters
    ----------
    config_path : Path, optional
        Path to datajoint.json if found.

    Returns
    -------
    Path or None
        Path to secrets directory if found, None otherwise.
    """
    # Check project secrets directory (next to config file)
    if config_path is not None:
        project_secrets = config_path.parent / SECRETS_DIRNAME
        if project_secrets.is_dir():
            return project_secrets

    # Check system secrets directory (Docker/Kubernetes)
    if SYSTEM_SECRETS_DIR.is_dir():
        return SYSTEM_SECRETS_DIR

    return None


def read_secret_file(secrets_dir: Path | None, name: str) -> str | None:
    """
    Read a secret value from a file in the secrets directory.

    Parameters
    ----------
    secrets_dir : Path or None
        Path to secrets directory.
    name : str
        Name of the secret file (e.g., ``'database.password'``).

    Returns
    -------
    str or None
        Secret value as string, or None if not found.
    """
    if secrets_dir is None:
        return None

    secret_path = secrets_dir / name
    if secret_path.is_file():
        return secret_path.read_text().strip()

    return None


class DatabaseSettings(BaseSettings):
    """Database connection settings."""

    model_config = SettingsConfigDict(
        env_prefix="DJ_",
        case_sensitive=False,
        extra="forbid",
        validate_assignment=True,
    )

    host: str = Field(default="localhost", validation_alias="DJ_HOST")
    user: str | None = Field(default=None, validation_alias="DJ_USER")
    password: SecretStr | None = Field(default=None, validation_alias="DJ_PASS")
    backend: Literal["mysql", "postgresql"] = Field(
        default="mysql",
        validation_alias="DJ_BACKEND",
        description="Database backend: 'mysql' or 'postgresql'",
    )
    port: int | None = Field(default=None, validation_alias="DJ_PORT")
    name: str | None = Field(
        default=None,
        validation_alias="DJ_DATABASE_NAME",
        description="Database name for PostgreSQL connections. Defaults to 'postgres' if not set.",
    )
    reconnect: bool = True
    use_tls: bool | None = Field(default=None, validation_alias="DJ_USE_TLS")
    database_prefix: str = Field(
        default="",
        validation_alias="DJ_DATABASE_PREFIX",
        description="Deprecated. Use database.name instead.",
    )
    create_tables: bool = Field(
        default=True,
        validation_alias="DJ_CREATE_TABLES",
        description="Default for Schema create_tables parameter. "
        "Set to False for production mode to prevent automatic table creation.",
    )

    @model_validator(mode="after")
    def set_default_port_from_backend(self) -> "DatabaseSettings":
        """Set default port based on backend if not explicitly provided."""
        if self.port is None:
            self.port = 5432 if self.backend == "postgresql" else 3306
        return self


class ConnectionSettings(BaseSettings):
    """Connection behavior settings."""

    model_config = SettingsConfigDict(extra="forbid", validate_assignment=True)

    charset: str = ""  # pymysql uses '' as default


class DisplaySettings(BaseSettings):
    """Display and preview settings."""

    model_config = SettingsConfigDict(extra="forbid", validate_assignment=True)

    limit: int = 12
    width: int = 14
    show_tuple_count: bool = True
    diagram_direction: Literal["TB", "LR"] = Field(
        default="LR",
        validation_alias="DJ_DIAGRAM_DIRECTION",
        description="Default diagram layout direction: 'TB' (top-to-bottom) or 'LR' (left-to-right)",
    )
    diagram_theme: Literal["light", "dark", "auto"] = Field(
        default="auto",
        validation_alias="DJ_DIAGRAM_THEME",
        description=(
            "Default diagram color theme: 'auto' (single SVG that adapts to the viewer's light/dark "
            "mode; default), 'light', or 'dark' (dark background with adjusted palette)"
        ),
    )


class StoresSettings(BaseSettings):
    """
    Unified object storage configuration.

    Stores configuration supports both hash-addressed and schema-addressed storage
    using the same named stores with _hash and _schema sections.
    """

    model_config = SettingsConfigDict(
        case_sensitive=False,
        extra="allow",  # Allow dynamic store names
        validate_assignment=True,
    )

    default: str | None = Field(default=None, description="Name of the default store")

    # Named stores are added dynamically as stores.<name>.*
    # Structure: stores.<name>.protocol, stores.<name>.location, etc.


class JobsSettings(BaseSettings):
    """Job queue configuration for AutoPopulate 2.0."""

    model_config = SettingsConfigDict(
        env_prefix="DJ_JOBS_",
        case_sensitive=False,
        extra="forbid",
        validate_assignment=True,
    )

    auto_refresh: bool = Field(default=True, description="Auto-refresh jobs queue on populate")
    keep_completed: bool = Field(default=False, description="Keep success records in jobs table")
    stale_timeout: int = Field(default=3600, ge=0, description="Seconds before pending job is checked for staleness")
    default_priority: int = Field(default=5, ge=0, le=255, description="Default priority for new jobs (lower = more urgent)")
    version_method: Literal["git", "none"] | None = Field(
        default=None, description="Method to obtain version: 'git' (commit hash), 'none' (empty), or None (disabled)"
    )
    allow_new_pk_fields_in_computed_tables: bool = Field(
        default=False,
        description="Allow native (non-FK) primary key fields in Computed/Imported tables. "
        "When True, bypasses the FK-only PK validation. Job granularity will be degraded for such tables.",
    )
    add_job_metadata: bool = Field(
        default=False,
        description="Add hidden job metadata attributes (_job_start_time, _job_duration, _job_version) "
        "to Computed and Imported tables during declaration. Tables created without this setting "
        "will not receive metadata updates during populate.",
    )


class ProvenanceSettings(BaseSettings):
    """Extrinsic provenance capture at pipeline entry points."""

    model_config = SettingsConfigDict(
        env_prefix="DJ_PROVENANCE_",
        case_sensitive=False,
        extra="forbid",
        validate_assignment=True,
    )

    capture: bool = Field(
        default=False,
        description="Add the hidden `_prov` attribute to Entry (dj.Manual) tables at declaration "
        "and fill it on insert. Off by default, matching jobs.add_job_metadata: enabling it changes "
        "the DDL of every Entry table declared afterwards, which is a deployment's decision rather "
        "than a library default. Tables declared while this is False never receive the column; "
        "use datajoint.deploy.add_prov_column to add it to an existing table.",
    )
    source: dict[str, Any] = Field(
        default_factory=dict,
        description="External source identity recorded on every row this process enters, e.g. "
        "{'system': 'PyRat', 'endpoint': 'https://...'}. Set per deployment via "
        "DJ_PROVENANCE_SOURCE (as JSON), the config file, or the secrets directory. "
        "No author supplies this at the insert call site.",
    )

    @field_validator("source")
    @classmethod
    def _source_must_be_json_serializable(cls, value: dict) -> dict:
        """Reject a source that cannot be recorded, at the point it is set.

        Every insert into an Entry table serializes this. Without the check the
        failure surfaces from inside an unrelated insert, naming neither
        provenance nor the setting responsible.
        """
        try:
            json.dumps(value, default=str)
        except (TypeError, ValueError) as error:
            raise ValueError(
                f"provenance.source must be JSON-serializable; it is recorded on every row "
                f"entering an Entry table. {error.__class__.__name__}: {error}"
            ) from error
        return value


class Config(BaseSettings):
    """
    Main DataJoint configuration.

    Settings are loaded from (in priority order):

    1. Environment variables (``DJ_*``)
    2. Secrets directory (``.secrets/`` or ``/run/secrets/datajoint/``)
    3. Config file (``datajoint.json``, searched in parent directories)
    4. Default values

    Examples
    --------
    Access settings via attributes:

    >>> config.database.host
    >>> config.safemode

    Override temporarily with context manager:

    >>> with config.override(safemode=False):
    ...     pass
    """

    model_config = SettingsConfigDict(
        env_prefix="DJ_",
        case_sensitive=False,
        extra="forbid",
        validate_assignment=True,
    )

    # Nested settings groups
    database: DatabaseSettings = Field(default_factory=DatabaseSettings)
    connection: ConnectionSettings = Field(default_factory=ConnectionSettings)
    display: DisplaySettings = Field(default_factory=DisplaySettings)
    jobs: JobsSettings = Field(default_factory=JobsSettings)
    provenance: ProvenanceSettings = Field(default_factory=ProvenanceSettings)

    # Unified stores configuration (replaces external and object_storage)
    # ``validation_alias`` redirects pydantic-settings' env source away from the
    # natural ``DJ_STORES`` so it doesn't auto-parse on Config() construction.
    # ``DJ_STORES`` is handled by ``_apply_stores_env`` after the config file
    # load so env-var precedence is honored. *New in 2.2.3.*
    stores: dict[str, Any] = Field(
        default_factory=dict,
        validation_alias="_DJ_STORES_PYDANTIC_DISABLED",
        description="Unified object storage configuration. "
        "Use stores.default to designate default store. "
        "Configure named stores as stores.<name>.protocol, stores.<name>.location, etc. "
        "Set via DJ_STORES (JSON object) or in datajoint.json. *New in 2.2.3* for "
        "DJ_STORES env-var support.",
    )

    # Top-level settings
    loglevel: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = Field(default="INFO", validation_alias="DJ_LOG_LEVEL")
    safemode: bool = True

    ignore_config_file: bool = Field(
        default=False,
        validation_alias="DJ_IGNORE_CONFIG_FILE",
        description="If True, skip loading datajoint.json and the secrets directory. "
        "Intended for env-var-only deployments (e.g. the DataJoint platform). "
        "*New in 2.2.3.*",
    )

    # Cache path for query results
    query_cache: Path | None = None

    # Download path for attachments and filepaths
    download_path: str = "."

    # Internal: track where config was loaded from
    _config_path: Path | None = None
    _secrets_dir: Path | None = None

    @field_validator("loglevel", mode="after")
    @classmethod
    def set_logger_level(cls, v: str) -> str:
        """Update logger level when loglevel changes."""
        logger.setLevel(v)
        return v

    @field_validator("query_cache", mode="before")
    @classmethod
    def convert_path(cls, v: Any) -> Path | None:
        """Convert string paths to Path objects."""
        if v is None:
            return None
        return Path(v) if not isinstance(v, Path) else v

    def resolve_store_name(self, store: str | None = None, *, use_filepath_default: bool = False) -> str:
        """
        Name the configured store that a store reference resolves to.

        An empty or missing name means the default store: ``stores.default``,
        or ``stores.filepath_default`` for filepath references. Codecs record
        the resolved name in their metadata, so a stored object stays tied to
        the store it was written to even if the default is later re-pointed.

        Parameters
        ----------
        store : str, optional
            Store name as declared. ``""`` (the bare ``@`` form) and None both
            mean the default store.
        use_filepath_default : bool, optional
            If True, resolve the default to stores.filepath_default instead of
            stores.default. Default: False.

        Returns
        -------
        str
            Name of a key in ``stores``.

        Raises
        ------
        DataJointError
            If the default store is needed but not configured.
        """
        if store:
            return store
        # An empty name is the bare ``@`` form (``<npy@>``, ``<object@>``, ...),
        # which the declaration keeps as "" so the codec can tell "store
        # requested" from "no store"; it means the default store.
        if use_filepath_default:
            # Filepath references use separate default (not part of OAS)
            if "filepath_default" not in self.stores:
                raise DataJointError(
                    "stores.filepath_default is not configured. "
                    "Set stores.filepath_default or specify store explicitly with <filepath@store>"
                )
            store = self.stores["filepath_default"]
        else:
            # Integrated storage (hash, schema) uses stores.default
            if "default" not in self.stores:
                raise DataJointError("stores.default is not configured")
            store = self.stores["default"]

        if not isinstance(store, str):
            default_key = "filepath_default" if use_filepath_default else "default"
            raise DataJointError(f"stores.{default_key} must be a string")
        return store

    def get_store_spec(self, store: str | None = None, *, use_filepath_default: bool = False) -> dict[str, Any]:
        """
        Get configuration for a storage store.

        Parameters
        ----------
        store : str, optional
            Name of the store to retrieve. If None, uses the appropriate default.
        use_filepath_default : bool, optional
            If True and store is None, uses stores.filepath_default instead of
            stores.default. Use for filepath references which are not part of OAS.
            Default: False (use stores.default for integrated storage).

        Returns
        -------
        dict[str, Any]
            Store configuration dict with validated fields.

        Raises
        ------
        DataJointError
            If store is not configured or has invalid config.
        """
        store = self.resolve_store_name(store, use_filepath_default=use_filepath_default)

        # Check store exists
        if store not in self.stores:
            raise DataJointError(f"Storage '{store}' is requested but not configured in stores")

        spec = dict(self.stores[store])

        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:
            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":
            spec.setdefault("secure", True)  # HTTPS by default for S3

        # Define required and allowed keys by protocol
        required_keys: dict[str, tuple[str, ...]] = {
            "file": ("protocol", "location"),
            "s3": ("protocol", "endpoint", "bucket", "location"),
            "gcs": ("protocol", "bucket", "location"),
            "azure": ("protocol", "container", "location"),
        }
        allowed_keys: dict[str, tuple[str, ...]] = {
            "file": (
                "protocol",
                "location",
                "subfolding",
                "partition_pattern",
                "token_length",
                "hash_prefix",
                "schema_prefix",
                "filepath_prefix",
                "stage",
            ),
            "s3": (
                "protocol",
                "endpoint",
                "bucket",
                "access_key",
                "secret_key",
                "location",
                "secure",
                "subfolding",
                "partition_pattern",
                "token_length",
                "hash_prefix",
                "schema_prefix",
                "filepath_prefix",
                "stage",
                "proxy_server",
            ),
            "gcs": (
                "protocol",
                "bucket",
                "location",
                "token",
                "project",
                "subfolding",
                "partition_pattern",
                "token_length",
                "hash_prefix",
                "schema_prefix",
                "filepath_prefix",
                "stage",
            ),
            "azure": (
                "protocol",
                "container",
                "location",
                "account_name",
                "account_key",
                "connection_string",
                "subfolding",
                "partition_pattern",
                "token_length",
                "hash_prefix",
                "schema_prefix",
                "filepath_prefix",
                "stage",
            ),
        }

        # Check required keys
        missing = [k for k in required_keys[protocol] if k not in spec]
        if missing:
            raise DataJointError(f'config.stores["{store}"] is missing: {", ".join(missing)}')

        # Check for invalid keys
        invalid = [k for k in spec if k not in allowed_keys[protocol]]
        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.

        These keys **control the store layout** (see
        how-to/configure-storage in the documentation): ``hash_prefix``
        (default ``_hash``) is where hash-addressed objects are written and
        scanned; ``schema_prefix`` (default ``_schema``) is where
        schema-addressed objects are written; ``filepath_prefix`` (default
        unrestricted) is the namespace ``<filepath@>`` paths must stay in.
        Writers, garbage collection, and ``<filepath@>`` validation all
        consume the same values, so the sections may be relocated per store
        without the components drifting apart. Changing a prefix on a store
        that already holds data is not recommended: existing objects remain
        readable (their metadata stores full paths) but reclamation scans
        only the currently configured sections.

        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."
                    )

    @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.

        Parameters
        ----------
        filename : str or Path
            Path to load configuration from.
        """
        filepath = Path(filename)
        if not filepath.exists():
            raise FileNotFoundError(f"Config file not found: {filepath}")

        logger.info(f"Loading configuration from {filepath.absolute()}")

        with open(filepath) as f:
            data = json.load(f)

        self._update_from_flat_dict(data)
        self._config_path = filepath

    def _update_from_flat_dict(self, data: dict[str, Any]) -> None:
        """
        Update settings from a dict (flat dot-notation or nested).

        Environment variables take precedence over config file values.
        If an env var is set for a setting, the file value is skipped.
        """
        for key, value in data.items():
            # Special handling for stores - accept nested dict directly
            if key == "stores" and isinstance(value, dict):
                # Merge stores dict
                for store_key, store_value in value.items():
                    self.stores[store_key] = store_value
                continue

            # Handle nested dicts by recursively updating
            if isinstance(value, dict) and hasattr(self, key):
                group_obj = getattr(self, key)
                for nested_key, nested_value in value.items():
                    if hasattr(group_obj, nested_key):
                        # Check if env var is set for this nested key
                        full_key = f"{key}.{nested_key}"
                        env_var = ENV_VAR_MAPPING.get(full_key)
                        if env_var and os.environ.get(env_var):
                            logger.debug(f"Skipping {full_key} from file (env var {env_var} takes precedence)")
                            continue
                        setattr(group_obj, nested_key, nested_value)
                continue

            # Handle flat dot-notation keys
            parts = key.split(".")
            if len(parts) == 1:
                if hasattr(self, key) and not key.startswith("_"):
                    # Check if env var is set for this key
                    env_var = ENV_VAR_MAPPING.get(key)
                    if env_var and os.environ.get(env_var):
                        logger.debug(f"Skipping {key} from file (env var {env_var} takes precedence)")
                        continue
                    setattr(self, key, value)
            elif len(parts) == 2:
                group, attr = parts
                if hasattr(self, group):
                    group_obj = getattr(self, group)
                    if hasattr(group_obj, attr):
                        # Check if env var is set for this key
                        env_var = ENV_VAR_MAPPING.get(key)
                        if env_var and os.environ.get(env_var):
                            logger.debug(f"Skipping {key} from file (env var {env_var} takes precedence)")
                            continue
                        setattr(group_obj, attr, value)
            elif len(parts) == 3:
                # Handle stores.<name>.<attr> pattern
                group, store_name, attr = parts
                if group == "stores":
                    if store_name not in self.stores:
                        self.stores[store_name] = {}
                    self.stores[store_name][attr] = value

    def _load_secrets(self, secrets_dir: Path) -> None:
        """Load secrets from a secrets directory."""
        self._secrets_dir = secrets_dir

        # Load database secrets
        db_user = read_secret_file(secrets_dir, "database.user")
        if db_user is not None and self.database.user is None:
            self.database.user = db_user
            logger.debug(f"Loaded database.user from {secrets_dir}")

        db_password = read_secret_file(secrets_dir, "database.password")
        if db_password is not None and self.database.password is None:
            self.database.password = db_password
            logger.debug(f"Loaded database.password from {secrets_dir}")

        # Load per-store secrets from any stores.<name>.<attr> file.
        # The attr name is recorded as-is on stores.<name>; this lets
        # plugin-registered adapters define their own secret fields
        # (e.g. a Bearer ``token`` for HTTP-based protocols) without
        # forcing AWS-style ``access_key`` / ``secret_key`` naming.
        if secrets_dir.is_dir():
            for secret_file in secrets_dir.iterdir():
                if not secret_file.is_file() or secret_file.name.startswith("."):
                    continue

                parts = secret_file.name.split(".")
                if len(parts) == 3 and parts[0] == "stores":
                    store_name, attr = parts[1], parts[2]
                    value = secret_file.read_text().strip()
                    # Initialize store dict if needed
                    if store_name not in self.stores:
                        self.stores[store_name] = {}
                    # Only set if not already present (config / env vars win)
                    if attr not in self.stores[store_name]:
                        self.stores[store_name][attr] = value
                        logger.debug(f"Loaded stores.{store_name}.{attr} from {secrets_dir}")

    def _apply_stores_env(self) -> None:
        """Replace ``self.stores`` from the ``DJ_STORES`` env var if set.

        ``DJ_STORES`` holds a JSON object in the same shape as the ``stores``
        block of ``datajoint.json``. This lets env-var-only deployments
        configure plugin-registered storage adapters with arbitrary attr
        names (e.g. a Bearer ``token`` field) without negotiating an env-var
        naming scheme per attr.

        *New in 2.2.3.*
        """
        raw = os.environ.get("DJ_STORES")
        if not raw:
            return
        try:
            data = json.loads(raw)
        except json.JSONDecodeError as e:
            raise ValueError(f"DJ_STORES contains invalid JSON: {e}") from e
        if not isinstance(data, dict):
            raise ValueError(f"DJ_STORES must be a JSON object, got {type(data).__name__}")
        self.stores = data
        logger.debug("Loaded stores from DJ_STORES env var")

    @contextmanager
    def override(self, **kwargs: Any) -> Iterator["Config"]:
        """
        Temporarily override configuration values.

        Parameters
        ----------
        **kwargs : Any
            Settings to override. Use double underscore for nested settings
            (e.g., ``database__host="localhost"``).

        Yields
        ------
        Config
            The config instance with overridden values.

        Examples
        --------
        >>> with config.override(safemode=False, database__host="test"):
        ...     # config.safemode is False here
        ...     pass
        >>> # config.safemode is restored
        """
        # Store original values
        backup = {}

        # Convert double underscore to nested access
        converted = {}
        for key, value in kwargs.items():
            if "__" in key:
                parts = key.split("__")
                converted[tuple(parts)] = value
            else:
                converted[(key,)] = value

        try:
            # Save originals and apply overrides
            for key_parts, value in converted.items():
                if len(key_parts) == 1:
                    key = key_parts[0]
                    if hasattr(self, key):
                        backup[key_parts] = copy.deepcopy(getattr(self, key))
                        setattr(self, key, value)
                elif len(key_parts) == 2:
                    group, attr = key_parts
                    if hasattr(self, group):
                        group_obj = getattr(self, group)
                        if hasattr(group_obj, attr):
                            backup[key_parts] = copy.deepcopy(getattr(group_obj, attr))
                            setattr(group_obj, attr, value)

            yield self

        finally:
            # Restore original values
            for key_parts, original in backup.items():
                if len(key_parts) == 1:
                    setattr(self, key_parts[0], original)
                elif len(key_parts) == 2:
                    group, attr = key_parts
                    setattr(getattr(self, group), attr, original)

    @staticmethod
    def save_template(
        path: str | Path = "datajoint.json",
        minimal: bool = True,
        create_secrets_dir: bool = True,
    ) -> Path:
        """
        Create a template datajoint.json configuration file.

        Credentials should NOT be stored in datajoint.json. Instead, use either:

        - Environment variables (``DJ_USER``, ``DJ_PASS``, ``DJ_HOST``,
          ``DJ_STORES`` for JSON-encoded store configs, etc.)
        - The ``.secrets/`` directory (created alongside datajoint.json)

        Set ``DJ_IGNORE_CONFIG_FILE=true`` to skip both ``datajoint.json`` and
        the secrets directory entirely (env-var-only configuration).

        Parameters
        ----------
        path : str or Path, optional
            Where to save the template. Default ``'datajoint.json'``.
        minimal : bool, optional
            If True (default), create minimal template with just database settings.
            If False, create full template with all available settings.
        create_secrets_dir : bool, optional
            If True (default), also create a ``.secrets/`` directory with
            template files for credentials.

        Returns
        -------
        Path
            Absolute path to the created config file.

        Raises
        ------
        FileExistsError
            If config file already exists (won't overwrite).

        Examples
        --------
        >>> import datajoint as dj
        >>> dj.config.save_template()  # Creates minimal template + .secrets/
        >>> dj.config.save_template("full-config.json", minimal=False)
        """
        filepath = Path(path)
        if filepath.exists():
            raise FileExistsError(f"File already exists: {filepath}. Remove it first or choose a different path.")

        if minimal:
            template = {
                "database": {
                    "host": "localhost",
                    "port": 3306,
                },
            }
        else:
            template = {
                "database": {
                    "host": "localhost",
                    "port": 3306,
                    "reconnect": True,
                    "use_tls": None,
                },
                "connection": {
                    "charset": "",
                },
                "display": {
                    "limit": 12,
                    "width": 14,
                    "show_tuple_count": True,
                },
                "stores": {
                    "default": "main",
                    "filepath_default": "raw_data",
                    "main": {
                        "protocol": "file",