Move Airflow Config Parser to shared library

airflow-core/pyproject.toml

@@ -152,6 +152,9 @@ dependencies = [
     "pygtrie>=2.5.0",
     "structlog>=25.4.0",
     # End of shared logging dependencies
+    # Start of shared configuration dependencies
+    "pyyaml>=6.0.2",
+    # End of shared configuration dependencies
 ]
 
 
@@ -218,6 +221,7 @@ exclude = [
 ]
 
 [tool.hatch.build.targets.sdist.force-include]
+"../shared/configuration/src/airflow_shared/configuration" = "src/airflow/_shared/configuration"
 "../shared/logging/src/airflow_shared/logging" = "src/airflow/_shared/logging"
 "../shared/secrets_masker/src/airflow_shared/secrets_masker" = "src/airflow/_shared/secrets_masker"
 "../shared/timezones/src/airflow_shared/timezones" = "src/airflow/_shared/timezones"
@@ -288,6 +292,7 @@ apache-airflow-devel-common = { workspace = true }
 
 [tool.airflow]
 shared_distributions = [
+    "apache-airflow-shared-configuration",
     "apache-airflow-shared-logging",
     "apache-airflow-shared-secrets-masker",
     "apache-airflow-shared-timezones",

airflow-core/src/airflow/_shared/configuration

@@ -0,0 +1 @@
+../../../../shared/configuration/src/airflow_shared/configuration

airflow-core/src/airflow/exceptions.py

@@ -31,6 +31,9 @@
     from airflow.sdk.definitions.asset import AssetNameRef, AssetUniqueKey, AssetUriRef
     from airflow.utils.state import DagRunState
 
+# Re exporting AirflowConfigException from shared configuration
+from airflow._shared.configuration.exceptions import AirflowConfigException as AirflowConfigException
+
 
 class AirflowException(Exception):
     """
@@ -58,10 +61,6 @@ class AirflowNotFoundException(AirflowException):
     status_code = HTTPStatus.NOT_FOUND
 
 
-class AirflowConfigException(AirflowException):
-    """Raise when there is configuration problem."""
-
-
 class AirflowSensorTimeout(AirflowException):
     """Raise when there is a timeout on sensor polling."""
 

airflow-core/tests/unit/core/test_configuration.py

@@ -399,6 +399,7 @@ def test_config_raise_exception_from_secret_backend_connection_error(self, mock_
 sql_alchemy_conn = airflow
 """
         test_conf = AirflowConfigParser(default_config=parameterized_config(test_config_default))
+        # Configure secrets backend on test_conf itself
         test_conf.read_string(test_config)
         test_conf.sensitive_config_values = test_conf.sensitive_config_values | {
             ("test", "sql_alchemy_conn"),

airflow-core/tests/unit/models/test_xcom.py

@@ -133,7 +133,10 @@ def test_resolve_xcom_class_fallback_to_basexcom(self):
 
     @conf_vars({("core", "xcom_backend"): "to be removed"})
     def test_resolve_xcom_class_fallback_to_basexcom_no_config(self):
+        from airflow.sdk.configuration import conf as sdk_conf
+
         conf.remove_option("core", "xcom_backend")
+        sdk_conf.remove_option("core", "xcom_backend")
         cls = resolve_xcom_backend()
         assert issubclass(cls, BaseXCom)
         assert cls.serialize_value([1]) == [1]

devel-common/src/tests_common/test_utils/config.py

@@ -23,42 +23,57 @@
 
 @contextlib.contextmanager
 def conf_vars(overrides):
+    """Automatically detects which config modules are loaded (Core, SDK, or both) and updates them accordingly temporarily."""
+    import sys
+
     from airflow import settings
-    from airflow.configuration import conf
 
-    original = {}
+    configs = []
+    if "airflow.configuration" in sys.modules:
+        from airflow.configuration import conf
+
+        configs.append(conf)
+    if "airflow.sdk.configuration" in sys.modules:
+        from airflow.sdk.configuration import conf
+
+        configs.append(conf)
+
+    originals = [{} for _ in configs]
     original_env_vars = {}
     for (section, key), value in overrides.items():
-        env = conf._env_var_name(section, key)
+        env = configs[0]._env_var_name(section, key)
         if env in os.environ:
             original_env_vars[env] = os.environ.pop(env)
 
-        if conf.has_option(section, key):
-            original[(section, key)] = conf.get(section, key)
-        else:
-            original[(section, key)] = None
-        if value is not None:
-            if not conf.has_section(section):
-                conf.add_section(section)
-            conf.set(section, key, value)
-        else:
-            if conf.has_section(section):
-                conf.remove_option(section, key)
-    settings.configure_vars()
-    try:
-        yield
-    finally:
-        for (section, key), value in original.items():
+        for i, conf in enumerate(configs):
+            originals[i][(section, key)] = conf.get(section, key) if conf.has_option(section, key) else None
             if value is not None:
                 if not conf.has_section(section):
                     conf.add_section(section)
                 conf.set(section, key, value)
-            else:
-                if conf.has_section(section):
+            elif conf.has_section(section) and conf.has_option(section, key):
+                conf.remove_option(section, key)
+
+    if "airflow.configuration" in sys.modules:
+        settings.configure_vars()
+
+    try:
+        yield
+    finally:
+        for i, conf in enumerate(configs):
+            for (section, key), value in originals[i].items():
+                if value is not None:
+                    if not conf.has_section(section):
+                        conf.add_section(section)
+                    conf.set(section, key, value)
+                elif conf.has_section(section) and conf.has_option(section, key):
                     conf.remove_option(section, key)
+
         for env, value in original_env_vars.items():
             os.environ[env] = value
-        settings.configure_vars()
+
+        if "airflow.configuration" in sys.modules:
+            settings.configure_vars()
 
 
 @contextlib.contextmanager

providers/apache/beam/tests/unit/apache/beam/hooks/test_beam.py

@@ -75,6 +75,12 @@
 except ImportError:
     APACHE_BEAM_VERSION = None
 
+try:
+    from airflow._shared.configuration import AirflowConfigException as ConfigException
+except ImportError:
+    # Compat for airflow < 3.2, where AirflowConfigException is in airflow.exceptions
+    ConfigException = AirflowException  # type: ignore[assignment,misc]
+
 
 class TestBeamHook:
     @mock.patch(BEAM_STRING.format("run_beam_command"))
@@ -347,7 +353,7 @@ def test_start_go_pipeline_without_go_installed_raises(self, mock_which):
             r"You need to have Go installed to run beam go pipeline\. See .* "
             "installation guide. If you are running airflow in Docker see more info at '.*'"
         )
-        with pytest.raises(AirflowException, match=error_message):
+        with pytest.raises(ConfigException, match=error_message):
             hook.start_go_pipeline(
                 go_file=GO_FILE,
                 variables=copy.deepcopy(BEAM_VARIABLES_GO),

pyproject.toml

@@ -1292,6 +1292,7 @@ dev = [
     "apache-airflow-task-sdk[all]",
     "apache-airflow-ctl",
     "apache-airflow-ctl-tests",
+    "apache-airflow-shared-configuration",
     "apache-airflow-shared-logging",
     "apache-airflow-shared-secrets-masker",
     "apache-airflow-shared-timezones",
@@ -1343,6 +1344,7 @@ apache-airflow-helm-tests = { workspace = true }
 apache-airflow-kubernetes-tests = { workspace = true }
 apache-airflow-providers = { workspace = true }
 apache-aurflow-docker-stack = { workspace = true }
+apache-airflow-shared-configuration = { workspace = true }
 apache-airflow-shared-logging = { workspace = true }
 apache-airflow-shared-secrets-masker = { workspace = true }
 apache-airflow-shared-timezones = { workspace = true }
@@ -1463,6 +1465,7 @@ members = [
     "task-sdk",
     "providers-summary-docs",
     "docker-stack-docs",
+    "shared/configuration",
     "shared/logging",
     "shared/secrets_masker",
     "shared/timezones",

shared/configuration/pyproject.toml

@@ -0,0 +1,55 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[project]
+name = "apache-airflow-shared-configuration"
+description = "Shared configuration parser code for Airflow distributions"
+version = "0.0"
+classifiers = [
+    "Private :: Do Not Upload",
+]
+
+dependencies = [
+    "packaging>=25.0",
+    "typing-extensions>=4.14.1",
+    "pendulum>=3.1.0",
+    "methodtools>=0.4.7",
+]
+
+[dependency-groups]
+dev = [
+    "apache-airflow-devel-common",
+]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/airflow_shared"]
+
+[tool.ruff]
+extend = "../../pyproject.toml"
+src = ["src"]
+
+[tool.ruff.lint.per-file-ignores]
+# Ignore Doc rules et al for anything outside of tests
+"!src/*" = ["D", "S101", "TRY002"]
+
+[tool.ruff.lint.flake8-tidy-imports]
+# Override the workspace level default
+ban-relative-imports = "parents"

shared/configuration/src/airflow_shared/configuration/__init__.py

@@ -0,0 +1,28 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Shared configuration parser for Airflow distributions."""
+
+from __future__ import annotations
+
+__all__ = [
+    "AirflowConfigException",
+    "AirflowConfigParser",
+]
+
+from .exceptions import AirflowConfigException
+from .parser import AirflowConfigParser

shared/configuration/src/airflow_shared/configuration/exceptions.py

@@ -0,0 +1,24 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Exceptions for configuration parsing."""
+
+from __future__ import annotations
+
+
+class AirflowConfigException(Exception):
+    """Raise when there is a configuration problem."""