validate resolved config without reloading sources

archivebox/config/collection.py

@@ -7,7 +7,7 @@ from collections.abc import Mapping
 from typing import Any
 
 from archivebox.config.constants import CONSTANTS
-from archivebox.config.configset import CaseConfigParser
+from archivebox.config.configset import CaseConfigParser, decode_config_inputs
 from archivebox.misc.logging import AttrDict
 
 
@@ -169,31 +169,8 @@ def _coerce_from_str_dict(file_config: dict[str, str]) -> dict[str, Any]:
     shape ``_coerce_to_str_dict`` writes them as.
     """
     from archivebox.config.common import ArchiveBoxConfig
-    from archivebox.config.configset import IniConfigSettingsSource
 
-    decoder = IniConfigSettingsSource(ArchiveBoxConfig)
-    decoded: dict[str, Any] = dict(file_config)
-    declared_fields = set(ArchiveBoxConfig.model_fields)
-    for field_name, field in ArchiveBoxConfig.model_fields.items():
-        if field_name not in decoded:
-            continue
-        raw = decoded[field_name]
-        if not isinstance(raw, str) or not raw:
-            continue
-        if decoder.field_is_complex(field):
-            decoded[field_name] = decoder.prepare_field_value(field_name, field, raw, True)
-    for key, raw in list(decoded.items()):
-        if key in declared_fields:
-            continue
-        if not isinstance(raw, str) or not raw:
-            continue
-        first = raw[:1]
-        if first not in ("{", "["):
-            continue
-        try:
-            decoded[key] = json.loads(raw)
-        except (TypeError, ValueError):
-            continue
+    decoded = decode_config_inputs(ArchiveBoxConfig, file_config, decode_unknown_json=True)
     if not str(decoded.get("BASE_URL") or "").strip():
         from archivebox.config.common import base_url_from_legacy_server_config
 

archivebox/config/common.py

@@ -22,7 +22,7 @@ from pydantic import BaseModel, Field, PrivateAttr, create_model, field_validato
 from pydantic_settings import SettingsConfigDict
 from rich.console import Console
 
-from archivebox.config.configset import COMPUTED_CONFIG_KEYS, BaseConfigSet, IniConfigSettingsSource
+from archivebox.config.configset import COMPUTED_CONFIG_KEYS, BaseConfigSet, decode_config_inputs
 
 from .constants import CONSTANTS
 from .ldap import LDAPConfig
@@ -1098,6 +1098,7 @@ def get_config(
     config_data: ConfigPayload = dict(defaults or {})
     config_data["PERSONAS_DIR"] = str(CONSTANTS.PERSONAS_DIR)
     base_config_payload: ConfigPayload = {}
+    file_config: dict[str, Any] | None = None
     base_config_model = ArchiveBoxConfig() if base_config is None else None
 
     if crawl_config_base:
@@ -1117,7 +1118,8 @@ def get_config(
         config_data.update(
             normalize_runtime_config(base_config_model.model_dump(mode="json"), exclude_runtime_derived=True, json_safe=False),
         )
-        legacy_config = {**BaseConfigSet.load_from_file(CONSTANTS.CONFIG_FILE), **os.environ}
+        file_config = BaseConfigSet.load_from_file(CONSTANTS.CONFIG_FILE)
+        legacy_config = {**file_config, **os.environ}
         legacy_permissions = permissions_from_legacy_public_flags(legacy_config)
         if legacy_permissions:
             config_data["PERMISSIONS"] = legacy_permissions
@@ -1191,7 +1193,8 @@ def get_config(
             },
         )
         if not crawl_config_base:
-            file_config = BaseConfigSet.load_from_file(CONSTANTS.CONFIG_FILE)
+            if file_config is None:
+                file_config = BaseConfigSet.load_from_file(CONSTANTS.CONFIG_FILE)
             explicit_plugin_enabled_keys.update(_explicit_plugin_enabled_keys(file_config))
             plugin_user_config = {
                 **_plugin_user_config(_plugin_input_config(file_config)),
@@ -1227,30 +1230,9 @@ def get_config(
             for enabled_key in _plugin_enabled_config_keys().values():
                 config_data.pop(enabled_key, None)
 
-    # Decode JSON-encoded complex values (dict/list fields) that came from
-    # string-only sources before validation. ``IniConfigSettingsSource`` does
-    # this for the ArchiveBox.conf path, but Machine.config (mirrored from the
-    # INI via ``_coerce_to_str_dict``) and plugin/env scope overrides bypass
-    # pydantic-settings sources entirely — they feed JSON strings directly
-    # into ``model_validate``, which rejects ``"{...}"`` for a ``dict[str, str]``
-    # field. Run pydantic-settings' own complex-value decoder here so every
-    # source converges on the same shape before validation.
-    _complex_decoder = IniConfigSettingsSource(ArchiveBoxConfig)
-    for _field_name, _field in ArchiveBoxConfig.model_fields.items():
-        if _field_name not in config_data:
-            continue
-        _raw = config_data[_field_name]
-        if not isinstance(_raw, str) or not _raw:
-            continue
-        if _complex_decoder.field_is_complex(_field):
-            config_data[_field_name] = _complex_decoder.prepare_field_value(
-                _field_name,
-                _field,
-                _raw,
-                True,
-            )
+    config_data = decode_config_inputs(ArchiveBoxConfig, config_data)
 
-    config = ArchiveBoxConfig.model_validate(config_data)
+    config = ArchiveBoxConfig.model_validate_resolved(config_data)
     if config.SERVER_SECURITY_MODE == "auto":
         base_host = (urlparse(config.BASE_URL if "://" in config.BASE_URL else f"//{config.BASE_URL}").hostname or "").lower()
         if base_host.endswith(".localhost"):

archivebox/config/configset.py

@@ -2,9 +2,11 @@
 
 __package__ = "archivebox.config"
 
-from pathlib import Path
-from typing import Any, ClassVar
+import json
+from collections.abc import Mapping
 from configparser import ConfigParser
+from pathlib import Path
+from typing import Any, ClassVar, Self
 
 from pydantic_settings import BaseSettings, PydanticBaseSettingsSource, SettingsConfigDict
 
@@ -61,20 +63,7 @@ class IniConfigSettingsSource(PydanticBaseSettingsSource):
         return field_value, field_name, value_is_complex
 
     def __call__(self) -> dict[str, Any]:
-        # Load the file once, then retain pydantic-settings' per-field complex
-        # value decoding without reopening the same file for every field.
-        result: dict[str, Any] = {}
-        config_vals = self._load_config_file()
-        for field_name, field in self.settings_cls.model_fields.items():
-            value = config_vals.get(field_name.upper())
-            if value is None:
-                continue
-            key = field_name
-            value_is_complex = self.field_is_complex(field)
-            prepared = self.prepare_field_value(field_name, field, value, value_is_complex)
-            if prepared is not None:
-                result[key] = prepared
-        return result
+        return decode_config_inputs(self.settings_cls, self._load_config_file())
 
     def _load_config_file(self) -> dict[str, Any]:
         try:
@@ -87,6 +76,34 @@ class IniConfigSettingsSource(PydanticBaseSettingsSource):
         return read_ini_config(config_path)
 
 
+def decode_config_inputs(
+    settings_cls: type[BaseSettings],
+    config: Mapping[str, Any],
+    *,
+    decode_unknown_json: bool = False,
+) -> dict[str, Any]:
+    """Decode string-only config values once at the boundary to typed config."""
+    decoder = IniConfigSettingsSource(settings_cls)
+    decoded: dict[str, Any] = dict(config)
+    for source_key, raw in list(decoded.items()):
+        if not isinstance(raw, str) or not raw:
+            continue
+        field_name = source_key if source_key in settings_cls.model_fields else source_key.upper()
+        field = settings_cls.model_fields.get(field_name)
+        if field is None:
+            if decode_unknown_json and raw[:1] in ("{", "["):
+                try:
+                    decoded[source_key] = json.loads(raw)
+                except (TypeError, ValueError):
+                    pass
+            continue
+        if decoder.field_is_complex(field):
+            decoded[field_name] = decoder.prepare_field_value(field_name, field, raw, True)
+            if source_key != field_name:
+                decoded.pop(source_key, None)
+    return decoded
+
+
 class BaseConfigSet(BaseSettings):
     """
     Base class for config sections.
@@ -111,6 +128,12 @@ class BaseConfigSet(BaseSettings):
     )
     computed_config_keys: ClassVar[tuple[str, ...]] = ()
 
+    @classmethod
+    def model_validate_resolved(cls, values: Mapping[str, Any]) -> Self:
+        """Validate merged values without running env and INI sources a second time."""
+        instance = cls.model_construct()
+        return cls.__pydantic_validator__.validate_python(values, self_instance=instance)
+
     @classmethod
     def settings_customise_sources(
         cls,

archivebox/tests/test_core_config.py

@@ -1,5 +1,72 @@
+import os
+
+import pytest
+from pydantic import ValidationError
+
 from archivebox.config import CONSTANTS
 
 
 def test_sonic_dir_is_allowed_inside_data_dir():
     assert "sonic" in CONSTANTS.ALLOWED_IN_DATA_DIR
+
+
+def test_resolved_config_validation_does_not_reload_environment():
+    from archivebox.config.common import ArchiveBoxConfig
+
+    previous_timeout = os.environ.get("TIMEOUT")
+    try:
+        os.environ["TIMEOUT"] = "1"
+        assert ArchiveBoxConfig().TIMEOUT == 1
+        assert ArchiveBoxConfig.model_validate_resolved({}).TIMEOUT != 1
+    finally:
+        if previous_timeout is None:
+            os.environ.pop("TIMEOUT", None)
+        else:
+            os.environ["TIMEOUT"] = previous_timeout
+
+
+def test_resolved_config_validation_matches_normal_validation():
+    from archivebox.config.common import ArchiveBoxConfig
+
+    payload = ArchiveBoxConfig().model_dump(mode="json")
+    payload.update(
+        {
+            "TIMEOUT": "17",
+            "CHROME_ARGS": ["--headless", "--no-sandbox"],
+            "ABXPKG_LIB_DIR": "./lib-from-resolved-validation",
+        },
+    )
+
+    normally_validated = ArchiveBoxConfig.model_validate(payload)
+    resolved_validated = ArchiveBoxConfig.model_validate_resolved(payload)
+
+    assert resolved_validated.model_dump(mode="json") == normally_validated.model_dump(mode="json")
+    assert resolved_validated.model_fields_set == normally_validated.model_fields_set
+
+
+def test_resolved_config_validation_preserves_validation_errors():
+    from archivebox.config.common import ArchiveBoxConfig
+
+    with pytest.raises(ValidationError) as normal_error:
+        ArchiveBoxConfig.model_validate({"TIMEOUT": "not-an-integer"})
+    with pytest.raises(ValidationError) as resolved_error:
+        ArchiveBoxConfig.model_validate_resolved({"TIMEOUT": "not-an-integer"})
+
+    assert resolved_error.value.errors(include_url=False) == normal_error.value.errors(include_url=False)
+
+
+def test_string_config_values_are_decoded_at_one_boundary():
+    from archivebox.config.common import ArchiveBoxConfig
+    from archivebox.config.configset import decode_config_inputs
+
+    decoded = decode_config_inputs(
+        ArchiveBoxConfig,
+        {
+            "CHROME_ARGS": '["--headless", "--no-sandbox"]',
+            "UNKNOWN_COMPLEX": '{"source": "plugin"}',
+        },
+        decode_unknown_json=True,
+    )
+
+    assert decoded["CHROME_ARGS"] == ["--headless", "--no-sandbox"]
+    assert decoded["UNKNOWN_COMPLEX"] == {"source": "plugin"}