Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1#
2# Licensed to the Apache Software Foundation (ASF) under one
3# or more contributor license agreements. See the NOTICE file
4# distributed with this work for additional information
5# regarding copyright ownership. The ASF licenses this file
6# to you under the Apache License, Version 2.0 (the
7# "License"); you may not use this file except in compliance
8# with the License. You may obtain a copy of the License at
9#
10# http://www.apache.org/licenses/LICENSE-2.0
11#
12# Unless required by applicable law or agreed to in writing,
13# software distributed under the License is distributed on an
14# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15# KIND, either express or implied. See the License for the
16# specific language governing permissions and limitations
17# under the License.
18"""Base configuration parser with pure parsing logic."""
20from __future__ import annotations
22import contextlib
23import datetime
24import functools
25import itertools
26import json
27import logging
28import os
29import shlex
30import subprocess
31import sys
32import warnings
33from collections.abc import Callable, Generator, Iterable
34from configparser import ConfigParser, NoOptionError, NoSectionError
35from contextlib import contextmanager
36from copy import deepcopy
37from enum import Enum
38from json.decoder import JSONDecodeError
39from re import Pattern
40from typing import IO, TYPE_CHECKING, Any, TypeVar, overload
42from .exceptions import AirflowConfigException
44log = logging.getLogger(__name__)
47def _build_kwarg_env_prefix(section: str, kwargs_key: str) -> str:
48 """
49 Build env prefix for per-key backend kwargs.
51 ("secrets", "backend_kwargs") -> "AIRFLOW__SECRETS__BACKEND_KWARG__"
52 ("workers", "secrets_backend_kwargs") -> "AIRFLOW__WORKERS__SECRETS_BACKEND_KWARG__"
53 """
54 singular_key = kwargs_key.replace("_kwargs", "_kwarg")
55 return f"{ENV_VAR_PREFIX}{section.upper()}__{singular_key.upper()}__"
58def _collect_kwarg_env_vars(prefix: str) -> dict[str, str]:
59 """
60 Scan os.environ for per-key secrets backend kwargs.
62 AIRFLOW__SECRETS__BACKEND_KWARG__ROLE_ID -> {"role_id": value}
63 Values are raw strings (not JSON-parsed).
64 Empty keys (trailing __ with no suffix) are ignored.
65 """
66 overrides: dict[str, str] = {}
67 for env_var, value in os.environ.items():
68 if env_var.startswith(prefix):
69 kwarg_key = env_var[len(prefix) :].lower()
70 if kwarg_key:
71 overrides[kwarg_key] = value
72 return overrides
75ConfigType = str | int | float | bool
76ConfigOptionsDictType = dict[str, ConfigType]
77ConfigSectionSourcesType = dict[str, str | tuple[str, str]]
78ConfigSourcesType = dict[str, ConfigSectionSourcesType]
79ENV_VAR_PREFIX = "AIRFLOW__"
80# Separates the team name from the base section name in a team scoped config file section.
81TEAM_SECTION_SEPARATOR = "="
84def team_section_name(team_name: str, section: str) -> str:
85 """
86 Build the config file section name that holds the team scoped overrides of ``section``.
88 :param team_name: name of the team the overrides belong to
89 :param section: base section name that is being overridden
90 :return: the team scoped section name, e.g. ``team_a=celery``
91 """
92 return f"{team_name}{TEAM_SECTION_SEPARATOR}{section}"
95def base_section_name(section: str) -> str:
96 """
97 Return the base section name of a possibly team scoped config file section.
99 Team scoped sections are built by :func:`team_section_name`. Base section names never contain
100 the separator, so the name is split on the last one - that way the base section is recovered
101 even for a team name that contains the separator itself.
103 :param section: section name, either a base one or a team scoped one
104 :return: the base section name, which is ``section`` itself when it is not team scoped
105 """
106 _, separator, base_section = section.rpartition(TEAM_SECTION_SEPARATOR)
107 return base_section if separator else section
110if TYPE_CHECKING:
111 from airflow.providers_manager import ProvidersManager
112 from airflow.sdk.providers_manager_runtime import ProvidersManagerTaskRuntime
115class ValueNotFound:
116 """Object of this is raised when a configuration value cannot be found."""
118 pass
121VALUE_NOT_FOUND_SENTINEL = ValueNotFound()
124@overload
125def expand_env_var(env_var: None) -> None: ...
126@overload
127def expand_env_var(env_var: str) -> str: ...
130def expand_env_var(env_var: str | None) -> str | None:
131 """
132 Expand (potentially nested) env vars.
134 Repeat and apply `expandvars` and `expanduser` until
135 interpolation stops having any effect.
136 """
137 if not env_var or not isinstance(env_var, str):
138 return env_var
139 while True:
140 interpolated = os.path.expanduser(os.path.expandvars(str(env_var)))
141 if interpolated == env_var:
142 return interpolated
143 env_var = interpolated
146def run_command(command: str) -> str:
147 """Run command and returns stdout."""
148 process = subprocess.Popen(
149 shlex.split(command), stdout=subprocess.PIPE, stderr=subprocess.PIPE, close_fds=True
150 )
151 output, stderr = (stream.decode(sys.getdefaultencoding(), "ignore") for stream in process.communicate())
153 if process.returncode != 0:
154 raise AirflowConfigException(
155 f"Cannot execute {command}. Error code is: {process.returncode}. "
156 f"Output: {output}, Stderr: {stderr}"
157 )
159 return output
162def _is_template(configuration_description: dict[str, dict[str, Any]], section: str, key: str) -> bool:
163 """
164 Check if the config is a template.
166 :param configuration_description: description of configuration
167 :param section: section
168 :param key: key
169 :return: True if the config is a template
170 """
171 return configuration_description.get(section, {}).get(key, {}).get("is_template", False)
174def configure_parser_from_configuration_description(
175 parser: ConfigParser,
176 configuration_description: dict[str, dict[str, Any]],
177 all_vars: dict[str, Any],
178) -> None:
179 """
180 Configure a ConfigParser based on configuration description.
182 :param parser: ConfigParser to configure
183 :param configuration_description: configuration description from config.yml
184 """
185 for section, section_desc in configuration_description.items():
186 parser.add_section(section)
187 options = section_desc["options"]
188 for key in options:
189 default_value = options[key]["default"]
190 is_template = options[key].get("is_template", False)
191 if (default_value is not None) and not (
192 options[key].get("version_deprecated") or options[key].get("deprecation_reason")
193 ):
194 if is_template or not isinstance(default_value, str):
195 parser.set(section, key, str(default_value))
196 else:
197 try:
198 parser.set(section, key, default_value.format(**all_vars))
199 except (KeyError, ValueError):
200 parser.set(section, key, default_value)
203def create_provider_cfg_config_fallback_defaults(
204 provider_config_fallback_defaults_cfg_path: str,
205) -> ConfigParser:
206 """
207 Create fallback defaults for configuration.
209 This parser contains provider defaults for Airflow configuration, containing fallback default values
210 that might be needed when provider classes are being imported - before provider's configuration
211 is loaded.
213 Unfortunately airflow currently performs a lot of stuff during importing and some of that might lead
214 to retrieving provider configuration before the defaults for the provider are loaded.
216 Those are only defaults, so if you have "real" values configured in your configuration (.cfg file or
217 environment variables) those will be used as usual.
219 NOTE!! Do NOT attempt to remove those default fallbacks thinking that they are unnecessary duplication,
220 at least not until we fix the way how airflow imports "do stuff". This is unlikely to succeed.
222 You've been warned!
224 :param provider_config_fallback_defaults_cfg_path: path to the provider config fallback defaults .cfg file
225 """
226 config_parser = ConfigParser()
227 config_parser.read(provider_config_fallback_defaults_cfg_path)
228 return config_parser
231class AirflowConfigParser(ConfigParser):
232 """
233 Base configuration parser with pure parsing logic.
235 This class provides the core parsing methods that work with:
236 - configuration_description: dict describing config options (required in __init__)
237 - _default_values: ConfigParser with default values (required in __init__)
238 - deprecated_options: class attribute mapping new -> old options
239 - deprecated_sections: class attribute mapping new -> old sections
240 """
242 # A mapping of section -> setting -> { old, replace } for deprecated default values.
243 # Subclasses can override this to define deprecated values that should be upgraded.
244 deprecated_values: dict[str, dict[str, tuple[Pattern, str]]] = {}
246 # A mapping of (new section, new option) -> (old section, old option, since_version).
247 # When reading new option, the old option will be checked to see if it exists. If it does a
248 # DeprecationWarning will be issued and the old option will be used instead
249 deprecated_options: dict[tuple[str, str], tuple[str, str, str]] = {
250 ("dag_processor", "dag_file_processor_timeout"): ("core", "dag_file_processor_timeout", "3.0"),
251 ("dag_processor", "refresh_interval"): ("scheduler", "dag_dir_list_interval", "3.0"),
252 ("api", "base_url"): ("webserver", "base_url", "3.0"),
253 ("api", "host"): ("webserver", "web_server_host", "3.0"),
254 ("api", "port"): ("webserver", "web_server_port", "3.0"),
255 ("api", "workers"): ("webserver", "workers", "3.0"),
256 ("api", "worker_timeout"): ("webserver", "web_server_worker_timeout", "3.0"),
257 ("api", "ssl_cert"): ("webserver", "web_server_ssl_cert", "3.0"),
258 ("api", "ssl_key"): ("webserver", "web_server_ssl_key", "3.0"),
259 ("api", "access_logfile"): ("webserver", "access_logfile", "3.0"),
260 ("triggerer", "capacity"): ("triggerer", "default_capacity", "3.0"),
261 ("api", "expose_config"): ("webserver", "expose_config", "3.0.1"),
262 ("fab", "access_denied_message"): ("webserver", "access_denied_message", "3.0.2"),
263 ("fab", "expose_hostname"): ("webserver", "expose_hostname", "3.0.2"),
264 ("fab", "navbar_color"): ("webserver", "navbar_color", "3.0.2"),
265 ("fab", "navbar_text_color"): ("webserver", "navbar_text_color", "3.0.2"),
266 ("fab", "navbar_hover_color"): ("webserver", "navbar_hover_color", "3.0.2"),
267 ("fab", "navbar_text_hover_color"): ("webserver", "navbar_text_hover_color", "3.0.2"),
268 ("api", "secret_key"): ("webserver", "secret_key", "3.0.2"),
269 ("api", "enable_swagger_ui"): ("webserver", "enable_swagger_ui", "3.0.2"),
270 ("dag_processor", "parsing_pre_import_modules"): ("scheduler", "parsing_pre_import_modules", "3.0.4"),
271 ("api", "grid_view_sorting_order"): ("webserver", "grid_view_sorting_order", "3.1.0"),
272 ("api", "log_fetch_timeout_sec"): ("webserver", "log_fetch_timeout_sec", "3.1.0"),
273 ("api", "hide_paused_dags_by_default"): ("webserver", "hide_paused_dags_by_default", "3.1.0"),
274 ("core", "num_dag_runs_to_retain_rendered_fields"): (
275 "core",
276 "max_num_rendered_ti_fields_per_task",
277 "3.2.0",
278 ),
279 ("api", "page_size"): ("webserver", "page_size", "3.1.0"),
280 ("api", "default_wrap"): ("webserver", "default_wrap", "3.1.0"),
281 ("api", "auto_refresh_interval"): ("webserver", "auto_refresh_interval", "3.1.0"),
282 ("api", "require_confirmation_dag_change"): ("webserver", "require_confirmation_dag_change", "3.1.0"),
283 ("api", "instance_name"): ("webserver", "instance_name", "3.1.0"),
284 ("api", "log_config"): ("api", "access_logfile", "3.1.0"),
285 ("scheduler", "ti_metrics_interval"): ("scheduler", "running_metrics_interval", "3.2.0"),
286 ("api", "fallback_page_limit"): ("api", "page_size", "3.2.0"),
287 ("workers", "missing_dag_retries"): ("workers", "missing_dag_retires", "3.1.8"),
288 ("core", "execution_api_server_url"): ("workers", "execution_api_server_url", "3.0"),
289 ("database", "sql_alchemy_conn"): ("core", "sql_alchemy_conn", "3.0"),
290 }
292 # A mapping of new section -> (old section, since_version).
293 deprecated_sections: dict[str, tuple[str, str]] = {}
295 @property
296 def _lookup_sequence(self) -> list[Callable]:
297 """
298 Define the sequence of lookup methods for get(). The definition here does not have provider lookup.
300 Subclasses can override this to customise lookup order.
301 """
302 lookup_methods = [
303 self._get_environment_variables,
304 self._get_option_from_config_file,
305 self._get_option_from_commands,
306 self._get_option_from_secrets,
307 self._get_option_from_defaults,
308 ]
309 if self._use_providers_configuration:
310 # Provider fallback lookups are last so they have the lowest priority in the lookup sequence.
311 lookup_methods += [
312 self._get_option_from_provider_metadata_config_fallbacks,
313 self._get_option_from_provider_cfg_config_fallbacks,
314 ]
315 return lookup_methods
317 @functools.cached_property
318 def configuration_description(self) -> dict[str, dict[str, Any]]:
319 """
320 Return configuration description from multiple sources.
322 Respects the ``_use_providers_configuration`` flag to decide whether to include
323 provider configuration.
325 The merged description is built as follows:
327 1. Start from the base configuration description provided in ``__init__``, usually
328 loaded from ``config.yml`` in core. Values defined here are never overridden.
329 2. Merge provider metadata from ``_provider_metadata_configuration_description``,
330 loaded from provider packages' ``get_provider_info`` method. Only adds missing
331 sections/options; does not overwrite existing entries from the base configuration.
332 3. Merge default values from ``_provider_cfg_config_fallback_default_values``,
333 loaded from ``provider_config_fallback_defaults.cfg``. Only sets ``"default"``
334 (and heuristically ``"sensitive"``) for options that do not already define them.
336 Base configuration takes precedence, then provider metadata fills in missing
337 descriptions/options, and finally cfg-based fallbacks provide defaults only where
338 none are defined.
340 We use ``cached_property`` to cache the merged result; clear this cache (via
341 ``invalidate_cache``) when toggling ``_use_providers_configuration``.
342 """
343 if not self._use_providers_configuration:
344 return self._configuration_description
346 merged_description: dict[str, dict[str, Any]] = deepcopy(self._configuration_description)
348 # Merge full provider config descriptions (with metadata like sensitive, description, etc.)
349 # from provider packages' get_provider_info method, reusing the cached raw dict.
350 for section, section_content in self._provider_metadata_configuration_description.items():
351 if section not in merged_description:
352 merged_description[section] = deepcopy(section_content)
353 else:
354 existing_options = merged_description[section].setdefault("options", {})
355 for option, option_content in section_content.get("options", {}).items():
356 if option not in existing_options:
357 existing_options[option] = deepcopy(option_content)
359 # Merge default values from cfg-based fallbacks (key=value only, no metadata).
360 # Uses setdefault so provider metadata values above take priority.
361 cfg = self._provider_cfg_config_fallback_default_values
362 for section in cfg.sections():
363 section_options = merged_description.setdefault(section, {"options": {}}).setdefault(
364 "options", {}
365 )
366 for option in cfg.options(section):
367 opt_dict = section_options.setdefault(option, {})
368 opt_dict.setdefault("default", cfg.get(section, option))
369 # For cfg-only options with no provider metadata, infer sensitivity from name.
370 if "sensitive" not in opt_dict and option.endswith(("password", "secret")):
371 opt_dict["sensitive"] = True
373 return merged_description
375 @property
376 def _config_sources_for_as_dict(self) -> list[tuple[str, ConfigParser]]:
377 """Override the base method to add provider fallbacks when providers are loaded."""
378 sources: list[tuple[str, ConfigParser]] = []
379 if self._use_providers_configuration:
380 # Provider fallback defaults are listed first so they have the lowest priority
381 # in as_dict()'s "last source wins" semantics.
382 sources += [
383 ("provider-cfg-fallback-defaults", self._provider_cfg_config_fallback_default_values),
384 (
385 "provider-metadata-fallback-defaults",
386 self._provider_metadata_config_fallback_default_values,
387 ),
388 ]
389 sources += [
390 ("default", self._default_values),
391 ("airflow.cfg", self),
392 ]
393 return sources
395 def _get_option_from_provider_cfg_config_fallbacks(
396 self,
397 deprecated_key: str | None,
398 deprecated_section: str | None,
399 key: str,
400 section: str,
401 issue_warning: bool = True,
402 extra_stacklevel: int = 0,
403 **kwargs,
404 ) -> str | ValueNotFound:
405 """Get config option from provider fallback defaults."""
406 value = self.get_from_provider_cfg_config_fallback_defaults(section, key, **kwargs)
407 if value is not VALUE_NOT_FOUND_SENTINEL:
408 return value
409 return VALUE_NOT_FOUND_SENTINEL
411 def _get_option_from_provider_metadata_config_fallbacks(
412 self,
413 deprecated_key: str | None,
414 deprecated_section: str | None,
415 key: str,
416 section: str,
417 issue_warning: bool = True,
418 extra_stacklevel: int = 0,
419 **kwargs,
420 ) -> str | ValueNotFound:
421 """Get config option from provider metadata fallback defaults."""
422 value = self.get_from_provider_metadata_config_fallback_defaults(section, key, **kwargs)
423 if value is not VALUE_NOT_FOUND_SENTINEL:
424 return value
425 return VALUE_NOT_FOUND_SENTINEL
427 def get_from_provider_cfg_config_fallback_defaults(self, section: str, key: str, **kwargs) -> Any:
428 """Get provider config fallback default values."""
429 raw = kwargs.get("raw", False)
430 vars_ = kwargs.get("vars")
431 return self._provider_cfg_config_fallback_default_values.get(
432 section, key, fallback=VALUE_NOT_FOUND_SENTINEL, raw=raw, vars=vars_
433 )
435 @functools.cached_property
436 def _provider_metadata_configuration_description(self) -> dict[str, dict[str, Any]]:
437 """Raw provider configuration descriptions with full metadata (sensitive, description, etc.)."""
438 result: dict[str, dict[str, Any]] = {}
439 for _, config in self._provider_manager_type().provider_configs:
440 result.update(config)
441 return result
443 @functools.cached_property
444 def _provider_metadata_config_fallback_default_values(self) -> ConfigParser:
445 """Return Provider metadata config fallback default values."""
446 return self._create_default_config_parser_callable(self._provider_metadata_configuration_description)
448 def get_from_provider_metadata_config_fallback_defaults(self, section: str, key: str, **kwargs) -> Any:
449 """Get provider metadata config fallback default values."""
450 raw = kwargs.get("raw", False)
451 vars_ = kwargs.get("vars")
452 return self._provider_metadata_config_fallback_default_values.get(
453 section, key, fallback=VALUE_NOT_FOUND_SENTINEL, raw=raw, vars=vars_
454 )
456 @property
457 def _validators(self) -> list[Callable[[], None]]:
458 """
459 Return list of validators defined on a config parser class. Base class will return an empty list.
461 Subclasses can override this to customize the validators that are run during validation on the
462 config parser instance.
463 """
464 return []
466 def validate(self) -> None:
467 """Run all registered validators."""
468 for validator in self._validators:
469 validator()
470 self.is_validated = True
472 def _validate_deprecated_values(self) -> None:
473 """Validate and upgrade deprecated default values."""
474 for section, replacement in self.deprecated_values.items():
475 for name, info in replacement.items():
476 old, new = info
477 current_value = self.get(section, name, fallback="")
478 if self._using_old_value(old, current_value):
479 self.upgraded_values[(section, name)] = current_value
480 new_value = old.sub(new, current_value)
481 self._update_env_var(section=section, name=name, new_value=new_value)
482 self._create_future_warning(
483 name=name,
484 section=section,
485 current_value=current_value,
486 new_value=new_value,
487 )
489 def _using_old_value(self, old: Pattern, current_value: str) -> bool:
490 """Check if current_value matches the old pattern."""
491 return old.search(current_value) is not None
493 def _update_env_var(self, section: str, name: str, new_value: str) -> None:
494 """Update environment variable with new value."""
495 env_var = self._env_var_name(section, name)
496 # Set it as an env var so that any subprocesses keep the same override!
497 os.environ[env_var] = new_value
499 @staticmethod
500 def _create_future_warning(name: str, section: str, current_value: Any, new_value: Any) -> None:
501 """Create a FutureWarning for deprecated default values."""
502 warnings.warn(
503 f"The {name!r} setting in [{section}] has the old default value of {current_value!r}. "
504 f"This value has been changed to {new_value!r} in the running config, but please update your config.",
505 FutureWarning,
506 stacklevel=3,
507 )
509 def __init__(
510 self,
511 configuration_description: dict[str, dict[str, Any]],
512 _default_values: ConfigParser,
513 provider_manager_type: type[ProvidersManager] | type[ProvidersManagerTaskRuntime],
514 create_default_config_parser_callable: Callable[[dict[str, dict[str, Any]]], ConfigParser],
515 provider_config_fallback_defaults_cfg_path: str,
516 *args,
517 **kwargs,
518 ):
519 """
520 Initialize the parser.
522 :param configuration_description: Description of configuration options
523 :param _default_values: ConfigParser with default values
524 :param provider_manager_type: Either ProvidersManager or ProvidersManagerTaskRuntime, depending on the context of the caller.
525 :param create_default_config_parser_callable: The `create_default_config_parser` function from core or SDK, depending on the context of the caller.
526 :param provider_config_fallback_defaults_cfg_path: Path to the `provider_config_fallback_defaults.cfg` file.
527 """
528 super().__init__(*args, **kwargs)
529 self._configuration_description = configuration_description
530 self._default_values = _default_values
531 self._provider_manager_type = provider_manager_type
532 self._create_default_config_parser_callable = create_default_config_parser_callable
533 self._provider_cfg_config_fallback_default_values = create_provider_cfg_config_fallback_defaults(
534 provider_config_fallback_defaults_cfg_path
535 )
536 self._suppress_future_warnings = False
537 self.upgraded_values: dict[tuple[str, str], str] = {}
538 # The _use_providers_configuration flag will always be True unless we call `write(include_providers=False)` or `with self.make_sure_configuration_loaded(with_providers=False)`.
539 # Even when we call those methods, the flag will be set back to True after the method is done, so it only affects the current call to `as_dict()` and does not have any effect on subsequent calls.
540 self._use_providers_configuration = True
542 def invalidate_cache(self) -> None:
543 """
544 Clear all ``functools.cached_property`` entries on this instance.
546 Call this after mutating class-level attributes (e.g. ``deprecated_options``)
547 so that derived cached properties are recomputed on next access.
548 """
549 for attr_name in (
550 name
551 for name in dir(type(self))
552 if isinstance(getattr(type(self), name, None), functools.cached_property)
553 ):
554 self.__dict__.pop(attr_name, None)
556 def _invalidate_provider_flag_caches(self) -> None:
557 """Invalidate caches related to provider configuration flags."""
558 self.__dict__.pop("configuration_description", None)
559 self.__dict__.pop("sensitive_config_values", None)
561 @functools.cached_property
562 def inversed_deprecated_options(self):
563 """Build inverse mapping from old options to new options."""
564 return {(sec, name): key for key, (sec, name, ver) in self.deprecated_options.items()}
566 @functools.cached_property
567 def inversed_deprecated_sections(self):
568 """Build inverse mapping from old sections to new sections."""
569 return {
570 old_section: new_section for new_section, (old_section, ver) in self.deprecated_sections.items()
571 }
573 @functools.cached_property
574 def sensitive_config_values(self) -> set[tuple[str, str]]:
575 """Get set of sensitive config values that should be masked."""
576 flattened = {
577 (s, k): item
578 for s, s_c in self.configuration_description.items()
579 for k, item in s_c.get("options", {}).items()
580 }
581 sensitive = {
582 (section.lower(), key.lower())
583 for (section, key), v in flattened.items()
584 if v.get("sensitive") is True
585 }
586 depr_option = {self.deprecated_options[x][:-1] for x in sensitive if x in self.deprecated_options}
587 depr_section = {
588 (self.deprecated_sections[s][0], k) for s, k in sensitive if s in self.deprecated_sections
589 }
590 sensitive.update(depr_section, depr_option)
591 return sensitive
593 def _names_sensitive_team_env_var(self, env_var: str) -> bool:
594 """
595 Check whether an environment variable name is a team scoped override of a sensitive option.
597 Team scoped variables are named ``AIRFLOW__<TEAM>___<SECTION>__<KEY>`` (see
598 :meth:`_env_var_name`). A team name may contain underscores itself, so the team name is not
599 parsed out of the variable name; the name is matched against the tail that each option
600 registered as sensitive contributes instead. The ``_CMD`` / ``_SECRET`` fallbacks are
601 matched too, because - unlike for a base section - they are never resolved into their value.
603 :param env_var: environment variable name
604 :return: True if the variable holds a team scoped value of an option registered as sensitive
605 """
606 env_var = env_var.upper()
607 if not env_var.startswith(ENV_VAR_PREFIX):
608 return False
609 # Every tail matched below starts with the ``___`` separating the team name from the
610 # section, so a name not containing it cannot be a team scoped one.
611 if "___" not in env_var:
612 return False
613 for section, key in self.sensitive_config_values:
614 option_tail = self._env_var_name(section, key).removeprefix(ENV_VAR_PREFIX)
615 for tail in (f"___{option_tail}", f"___{option_tail}_CMD", f"___{option_tail}_SECRET"):
616 # The team name sits between the prefix and the tail, so it must not be empty.
617 if env_var.endswith(tail) and len(env_var) > len(ENV_VAR_PREFIX) + len(tail):
618 return True
619 return False
621 def is_sensitive_option(self, section: str, key: str) -> bool:
622 """
623 Check whether the value of ``key`` in ``section`` is registered as sensitive.
625 Options are registered as sensitive under their base section name, while a team scoped
626 override of the very same option is held by a ``<team name>=<section>`` config file section
627 or by an ``AIRFLOW__<TEAM>___<SECTION>__<KEY>`` environment variable. Both of those
628 spellings are resolved back to the base option here, so that a team scoped value is treated
629 exactly like the base one. A name that does not resolve to a registered option is not
630 sensitive - so this only ever recognises more options as sensitive, never fewer.
632 :param section: section name, either a base one or a team scoped one
633 :param key: option name
634 :return: True if the value of the option should be treated as sensitive
635 """
636 section = section.lower()
637 key = key.lower()
638 if (section, key) in self.sensitive_config_values:
639 return True
640 base_section = base_section_name(section)
641 if base_section != section:
642 if (base_section, key) in self.sensitive_config_values:
643 return True
644 # A team scoped ``_cmd`` / ``_secret`` fallback is not resolved into its value, so it
645 # stays in the output as configured and has to be recognised on its own.
646 for fallback_suffix in ("_cmd", "_secret"):
647 if not key.endswith(fallback_suffix):
648 continue
649 if (base_section, key.removesuffix(fallback_suffix)) in self.sensitive_config_values:
650 return True
651 # A team scoped environment variable is reported under the section and key its name splits
652 # into, which is neither the base nor the team scoped section name.
653 return self._names_sensitive_team_env_var(self._env_var_name(section, key))
655 def _update_defaults_from_string(self, config_string: str) -> None:
656 """
657 Update the defaults in _default_values based on values in config_string ("ini" format).
659 Override shared parser's method to add validation for template variables.
660 Note that those values are not validated and cannot contain variables because we are using
661 regular config parser to load them. This method is used to test the config parser in unit tests.
663 :param config_string: ini-formatted config string
664 """
665 parser = ConfigParser()
666 parser.read_string(config_string)
667 for section in parser.sections():
668 if section not in self._default_values.sections():
669 self._default_values.add_section(section)
670 errors = False
671 for key, value in parser.items(section):
672 if not self.is_template(section, key) and "{" in value:
673 errors = True
674 log.error(
675 "The %s.%s value %s read from string contains variable. This is not supported",
676 section,
677 key,
678 value,
679 )
680 self._default_values.set(section, key, value)
681 if errors:
682 raise AirflowConfigException(
683 f"The string config passed as default contains variables. "
684 f"This is not supported. String config: {config_string}"
685 )
687 def get_default_value(self, section: str, key: str, fallback: Any = None, raw=False, **kwargs) -> Any:
688 """
689 Retrieve default value from default config parser, including provider fallbacks.
691 This will retrieve the default value from the core default config parser first. If not found
692 and providers configuration is loaded, it also checks provider fallback defaults.
693 Optionally a raw, stored value can be retrieved by setting skip_interpolation to True.
694 This is useful for example when we want to write the default value to a file, and we don't
695 want the interpolation to happen as it is going to be done later when the config is read.
697 :param section: section of the config
698 :param key: key to use
699 :param fallback: fallback value to use
700 :param raw: if raw, then interpolation will be reversed
701 :param kwargs: other args
702 :return:
703 """
704 value = self._default_values.get(section, key, fallback=VALUE_NOT_FOUND_SENTINEL, **kwargs)
705 # Provider metadata has higher priority than cfg fallback — check it first.
706 if value is VALUE_NOT_FOUND_SENTINEL and self._use_providers_configuration:
707 value = self._provider_metadata_config_fallback_default_values.get(
708 section, key, fallback=VALUE_NOT_FOUND_SENTINEL, **kwargs
709 )
710 if value is VALUE_NOT_FOUND_SENTINEL and self._use_providers_configuration:
711 value = self._provider_cfg_config_fallback_default_values.get(
712 section, key, fallback=VALUE_NOT_FOUND_SENTINEL, **kwargs
713 )
714 if value is VALUE_NOT_FOUND_SENTINEL:
715 value = fallback
716 if raw and isinstance(value, str):
717 return value.replace("%", "%%")
718 return value
720 def _get_custom_secret_backend(self, worker_mode: bool = False) -> Any | None:
721 """
722 Get Secret Backend if defined in airflow.cfg.
724 Conditionally selects the section, key and kwargs key based on whether it is called from worker or not.
725 """
726 section = "workers" if worker_mode else "secrets"
727 key = "secrets_backend" if worker_mode else "backend"
728 kwargs_key = "secrets_backend_kwargs" if worker_mode else "backend_kwargs"
730 secrets_backend_cls = self.getimport(section=section, key=key)
732 if not secrets_backend_cls:
733 if worker_mode:
734 # if we find no secrets backend for worker, return that of secrets backend
735 secrets_backend_cls = self.getimport(section="secrets", key="backend")
736 if not secrets_backend_cls:
737 return None
738 # When falling back to secrets backend, use its kwargs
739 kwargs_key = "backend_kwargs"
740 section = "secrets"
741 else:
742 return None
744 try:
745 backend_kwargs = self.getjson(section=section, key=kwargs_key)
746 if not backend_kwargs:
747 backend_kwargs = {}
748 elif not isinstance(backend_kwargs, dict):
749 raise ValueError("not a dict")
750 except AirflowConfigException:
751 log.warning("Failed to parse [%s] %s as JSON, defaulting to no kwargs.", section, kwargs_key)
752 backend_kwargs = {}
753 except ValueError:
754 log.warning("Failed to parse [%s] %s into a dict, defaulting to no kwargs.", section, kwargs_key)
755 backend_kwargs = {}
757 # Collect per-key overrides; they take precedence over the JSON blob.
758 env_prefix = _build_kwarg_env_prefix(section, kwargs_key)
759 backend_kwargs.update(_collect_kwarg_env_vars(env_prefix))
761 return secrets_backend_cls(**backend_kwargs)
763 def _get_config_value_from_secret_backend(self, config_key: str) -> str | None:
764 """
765 Get Config option values from Secret Backend.
767 Called by the shared parser's _get_secret_option() method as part of the lookup chain.
768 Uses _get_custom_secret_backend() to get the backend instance.
770 :param config_key: the config key to retrieve
771 :return: config value or None
772 """
773 try:
774 secrets_client = self._get_custom_secret_backend()
775 if not secrets_client:
776 return None
777 return secrets_client.get_config(config_key)
778 except Exception as e:
779 raise AirflowConfigException(
780 "Cannot retrieve config from alternative secrets backend. "
781 "Make sure it is configured properly and that the Backend "
782 "is accessible.\n"
783 f"{e}"
784 )
786 def _get_cmd_option_from_config_sources(
787 self, config_sources: ConfigSourcesType, section: str, key: str
788 ) -> str | None:
789 fallback_key = key + "_cmd"
790 if (section, key) in self.sensitive_config_values:
791 section_dict = config_sources.get(section)
792 if section_dict is not None:
793 command_value = section_dict.get(fallback_key)
794 if command_value is not None:
795 if isinstance(command_value, str):
796 command = command_value
797 else:
798 command = command_value[0]
799 return run_command(command)
800 return None
802 def _get_secret_option_from_config_sources(
803 self, config_sources: ConfigSourcesType, section: str, key: str
804 ) -> str | None:
805 fallback_key = key + "_secret"
806 if (section, key) in self.sensitive_config_values:
807 section_dict = config_sources.get(section)
808 if section_dict is not None:
809 secrets_path_value = section_dict.get(fallback_key)
810 if secrets_path_value is not None:
811 if isinstance(secrets_path_value, str):
812 secrets_path = secrets_path_value
813 else:
814 secrets_path = secrets_path_value[0]
815 return self._get_config_value_from_secret_backend(secrets_path)
816 return None
818 def _include_secrets(
819 self,
820 config_sources: ConfigSourcesType,
821 display_sensitive: bool,
822 display_source: bool,
823 raw: bool,
824 ):
825 for section, key in self.sensitive_config_values:
826 value: str | None = self._get_secret_option_from_config_sources(config_sources, section, key)
827 if value:
828 if not display_sensitive:
829 value = "< hidden >"
830 if display_source:
831 opt: str | tuple[str, str] = (value, "secret")
832 elif raw:
833 opt = value.replace("%", "%%")
834 else:
835 opt = value
836 config_sources.setdefault(section, {}).update({key: opt})
837 del config_sources[section][key + "_secret"]
839 def _include_commands(
840 self,
841 config_sources: ConfigSourcesType,
842 display_sensitive: bool,
843 display_source: bool,
844 raw: bool,
845 ):
846 for section, key in self.sensitive_config_values:
847 opt = self._get_cmd_option_from_config_sources(config_sources, section, key)
848 if not opt:
849 continue
850 opt_to_set: str | tuple[str, str] | None = opt
851 if not display_sensitive:
852 opt_to_set = "< hidden >"
853 if display_source:
854 opt_to_set = (str(opt_to_set), "cmd")
855 elif raw:
856 opt_to_set = str(opt_to_set).replace("%", "%%")
857 if opt_to_set is not None:
858 dict_to_update: dict[str, str | tuple[str, str]] = {key: opt_to_set}
859 config_sources.setdefault(section, {}).update(dict_to_update)
860 del config_sources[section][key + "_cmd"]
862 def _include_envs(
863 self,
864 config_sources: ConfigSourcesType,
865 display_sensitive: bool,
866 display_source: bool,
867 raw: bool,
868 ):
869 for env_var in [
870 os_environment for os_environment in os.environ if os_environment.startswith(ENV_VAR_PREFIX)
871 ]:
872 try:
873 _, section, key = env_var.split("__", 2)
874 opt = self._get_env_var_option(section, key)
875 except ValueError:
876 continue
877 if opt is None:
878 log.warning("Ignoring unknown env var '%s'", env_var)
879 continue
880 if not display_sensitive and env_var != self._env_var_name("core", "unit_test_mode"):
881 if self._names_sensitive_team_env_var(env_var):
882 # Covers the cmd/secret variants too; see is_sensitive_option.
883 opt = "< hidden >"
884 # Don't hide cmd/secret values here
885 elif not env_var.lower().endswith(("cmd", "secret")):
886 if (section, key) in self.sensitive_config_values:
887 opt = "< hidden >"
888 elif raw:
889 opt = opt.replace("%", "%%")
890 if display_source:
891 opt = (opt, "env var")
893 section = section.lower()
894 key = key.lower()
895 config_sources.setdefault(section, {}).update({key: opt})
897 def _filter_by_source(
898 self,
899 config_sources: ConfigSourcesType,
900 display_source: bool,
901 getter_func,
902 ):
903 """
904 Delete default configs from current configuration.
906 An OrderedDict of OrderedDicts, if it would conflict with special sensitive_config_values.
908 This is necessary because bare configs take precedence over the command
909 or secret key equivalents so if the current running config is
910 materialized with Airflow defaults they in turn override user set
911 command or secret key configs.
913 :param config_sources: The current configuration to operate on
914 :param display_source: If False, configuration options contain raw
915 values. If True, options are a tuple of (option_value, source).
916 Source is either 'airflow.cfg', 'default', 'env var', or 'cmd'.
917 :param getter_func: A callback function that gets the user configured
918 override value for a particular sensitive_config_values config.
919 :return: None, the given config_sources is filtered if necessary,
920 otherwise untouched.
921 """
922 for section, key in self.sensitive_config_values:
923 # Don't bother if we don't have section / key
924 if section not in config_sources or key not in config_sources[section]:
925 continue
926 # Check that there is something to override defaults
927 try:
928 getter_opt = getter_func(section, key)
929 except ValueError:
930 continue
931 if not getter_opt:
932 continue
933 # Check to see that there is a default value
934 if self.get_default_value(section, key) is None:
935 continue
936 # Check to see if bare setting is the same as defaults
937 if display_source:
938 # when display_source = true, we know that the config_sources contains tuple
939 opt, source = config_sources[section][key] # type: ignore
940 else:
941 opt = config_sources[section][key] # type: ignore[assignment]
942 if opt == self.get_default_value(section, key):
943 del config_sources[section][key]
945 @staticmethod
946 def _deprecated_value_is_set_in_config(
947 deprecated_section: str,
948 deprecated_key: str,
949 configs: Iterable[tuple[str, ConfigParser]],
950 ) -> bool:
951 for config_type, config in configs:
952 if config_type != "default":
953 with contextlib.suppress(NoSectionError):
954 deprecated_section_array = config.items(section=deprecated_section, raw=True)
955 if any(key == deprecated_key for key, _ in deprecated_section_array):
956 return True
957 return False
959 @staticmethod
960 def _deprecated_variable_is_set(deprecated_section: str, deprecated_key: str) -> bool:
961 return (
962 os.environ.get(f"{ENV_VAR_PREFIX}{deprecated_section.upper()}__{deprecated_key.upper()}")
963 is not None
964 )
966 @staticmethod
967 def _deprecated_command_is_set_in_config(
968 deprecated_section: str,
969 deprecated_key: str,
970 configs: Iterable[tuple[str, ConfigParser]],
971 ) -> bool:
972 return AirflowConfigParser._deprecated_value_is_set_in_config(
973 deprecated_section=deprecated_section, deprecated_key=deprecated_key + "_cmd", configs=configs
974 )
976 @staticmethod
977 def _deprecated_variable_command_is_set(deprecated_section: str, deprecated_key: str) -> bool:
978 return (
979 os.environ.get(f"{ENV_VAR_PREFIX}{deprecated_section.upper()}__{deprecated_key.upper()}_CMD")
980 is not None
981 )
983 @staticmethod
984 def _deprecated_secret_is_set_in_config(
985 deprecated_section: str,
986 deprecated_key: str,
987 configs: Iterable[tuple[str, ConfigParser]],
988 ) -> bool:
989 return AirflowConfigParser._deprecated_value_is_set_in_config(
990 deprecated_section=deprecated_section, deprecated_key=deprecated_key + "_secret", configs=configs
991 )
993 @staticmethod
994 def _deprecated_variable_secret_is_set(deprecated_section: str, deprecated_key: str) -> bool:
995 return (
996 os.environ.get(f"{ENV_VAR_PREFIX}{deprecated_section.upper()}__{deprecated_key.upper()}_SECRET")
997 is not None
998 )
1000 @staticmethod
1001 def _replace_config_with_display_sources(
1002 config_sources: ConfigSourcesType,
1003 configs: Iterable[tuple[str, ConfigParser]],
1004 configuration_description: dict[str, dict[str, Any]],
1005 display_source: bool,
1006 raw: bool,
1007 deprecated_options: dict[tuple[str, str], tuple[str, str, str]],
1008 include_env: bool,
1009 include_cmds: bool,
1010 include_secret: bool,
1011 ):
1012 for source_name, config in configs:
1013 sections = config.sections()
1014 for section in sections:
1015 AirflowConfigParser._replace_section_config_with_display_sources(
1016 config,
1017 config_sources,
1018 configuration_description,
1019 display_source,
1020 raw,
1021 section,
1022 source_name,
1023 deprecated_options,
1024 configs,
1025 include_env=include_env,
1026 include_cmds=include_cmds,
1027 include_secret=include_secret,
1028 )
1030 @staticmethod
1031 def _replace_section_config_with_display_sources(
1032 config: ConfigParser,
1033 config_sources: ConfigSourcesType,
1034 configuration_description: dict[str, dict[str, Any]],
1035 display_source: bool,
1036 raw: bool,
1037 section: str,
1038 source_name: str,
1039 deprecated_options: dict[tuple[str, str], tuple[str, str, str]],
1040 configs: Iterable[tuple[str, ConfigParser]],
1041 include_env: bool,
1042 include_cmds: bool,
1043 include_secret: bool,
1044 ):
1045 sect = config_sources.setdefault(section, {})
1046 if isinstance(config, AirflowConfigParser):
1047 with config.suppress_future_warnings():
1048 items: Iterable[tuple[str, Any]] = config.items(section=section, raw=raw)
1049 else:
1050 items = config.items(section=section, raw=raw)
1051 for k, val in items:
1052 deprecated_section, deprecated_key, _ = deprecated_options.get((section, k), (None, None, None))
1053 if deprecated_section and deprecated_key:
1054 if source_name == "default":
1055 # If deprecated entry has some non-default value set for any of the sources requested,
1056 # We should NOT set default for the new entry (because it will override anything
1057 # coming from the deprecated ones)
1058 if AirflowConfigParser._deprecated_value_is_set_in_config(
1059 deprecated_section, deprecated_key, configs
1060 ):
1061 continue
1062 if include_env and AirflowConfigParser._deprecated_variable_is_set(
1063 deprecated_section, deprecated_key
1064 ):
1065 continue
1066 if include_cmds and (
1067 AirflowConfigParser._deprecated_variable_command_is_set(
1068 deprecated_section, deprecated_key
1069 )
1070 or AirflowConfigParser._deprecated_command_is_set_in_config(
1071 deprecated_section, deprecated_key, configs
1072 )
1073 ):
1074 continue
1075 if include_secret and (
1076 AirflowConfigParser._deprecated_variable_secret_is_set(
1077 deprecated_section, deprecated_key
1078 )
1079 or AirflowConfigParser._deprecated_secret_is_set_in_config(
1080 deprecated_section, deprecated_key, configs
1081 )
1082 ):
1083 continue
1084 if display_source:
1085 updated_source_name = source_name
1086 if source_name == "default":
1087 # defaults can come from other sources (default-<PROVIDER>) that should be used here
1088 source_description_section = configuration_description.get(section, {})
1089 source_description_key = source_description_section.get("options", {}).get(k, {})
1090 if source_description_key is not None:
1091 updated_source_name = source_description_key.get("source", source_name)
1092 sect[k] = (val, updated_source_name)
1093 else:
1094 sect[k] = val
1096 def _warn_deprecate(
1097 self, section: str, key: str, deprecated_section: str, deprecated_name: str, extra_stacklevel: int
1098 ):
1099 """Warn about deprecated config option usage."""
1100 if section == deprecated_section:
1101 warnings.warn(
1102 f"The {deprecated_name} option in [{section}] has been renamed to {key} - "
1103 f"the old setting has been used, but please update your config.",
1104 DeprecationWarning,
1105 stacklevel=4 + extra_stacklevel,
1106 )
1107 else:
1108 warnings.warn(
1109 f"The {deprecated_name} option in [{deprecated_section}] has been moved to the {key} option "
1110 f"in [{section}] - the old setting has been used, but please update your config.",
1111 DeprecationWarning,
1112 stacklevel=4 + extra_stacklevel,
1113 )
1115 @contextmanager
1116 def suppress_future_warnings(self):
1117 """
1118 Context manager to temporarily suppress future warnings.
1120 This is a stub used by the shared parser's lookup methods when checking deprecated options.
1121 Subclasses can override this to customize warning suppression behavior.
1123 :return: context manager that suppresses future warnings
1124 """
1125 suppress_future_warnings = self._suppress_future_warnings
1126 self._suppress_future_warnings = True
1127 yield self
1128 self._suppress_future_warnings = suppress_future_warnings
1130 def _env_var_name(self, section: str, key: str, team_name: str | None = None) -> str:
1131 """Generate environment variable name for a config option."""
1132 team_component: str = f"{team_name.upper()}___" if team_name else ""
1133 return f"{ENV_VAR_PREFIX}{team_component}{section.replace('.', '_').upper()}__{key.upper()}"
1135 def _get_env_var_option(self, section: str, key: str, team_name: str | None = None):
1136 """Get config option from environment variable."""
1137 env_var: str = self._env_var_name(section, key, team_name=team_name)
1138 if env_var in os.environ:
1139 return expand_env_var(os.environ[env_var])
1140 # alternatively AIRFLOW__{SECTION}__{KEY}_CMD (for a command)
1141 env_var_cmd = env_var + "_CMD"
1142 if env_var_cmd in os.environ:
1143 # if this is a valid command key...
1144 if (section, key) in self.sensitive_config_values:
1145 return run_command(os.environ[env_var_cmd])
1146 # alternatively AIRFLOW__{SECTION}__{KEY}_SECRET (to get from Secrets Backend)
1147 env_var_secret_path = env_var + "_SECRET"
1148 if env_var_secret_path in os.environ:
1149 # if this is a valid secret path...
1150 if (section, key) in self.sensitive_config_values:
1151 return self._get_config_value_from_secret_backend(os.environ[env_var_secret_path])
1152 return None
1154 def _get_cmd_option(self, section: str, key: str):
1155 """Get config option from command execution."""
1156 fallback_key = key + "_cmd"
1157 if (section, key) in self.sensitive_config_values:
1158 if super().has_option(section, fallback_key):
1159 command = super().get(section, fallback_key)
1160 try:
1161 cmd_output = run_command(command)
1162 except AirflowConfigException as e:
1163 raise e
1164 except Exception as e:
1165 raise AirflowConfigException(
1166 f"Cannot run the command for the config section [{section}]{fallback_key}_cmd."
1167 f" Please check the {fallback_key} value."
1168 ) from e
1169 return cmd_output
1170 return None
1172 def _get_secret_option(self, section: str, key: str) -> str | None:
1173 """Get Config option values from Secret Backend."""
1174 fallback_key = key + "_secret"
1175 if (section, key) in self.sensitive_config_values:
1176 if super().has_option(section, fallback_key):
1177 secrets_path = super().get(section, fallback_key)
1178 return self._get_config_value_from_secret_backend(secrets_path)
1179 return None
1181 def _get_environment_variables(
1182 self,
1183 deprecated_key: str | None,
1184 deprecated_section: str | None,
1185 key: str,
1186 section: str,
1187 issue_warning: bool = True,
1188 extra_stacklevel: int = 0,
1189 **kwargs,
1190 ) -> str | ValueNotFound:
1191 """Get config option from environment variables."""
1192 team_name = kwargs.get("team_name", None)
1193 option = self._get_env_var_option(section, key, team_name=team_name)
1194 if option is not None:
1195 return option
1196 if deprecated_section and deprecated_key:
1197 with self.suppress_future_warnings():
1198 option = self._get_env_var_option(deprecated_section, deprecated_key, team_name=team_name)
1199 if option is not None:
1200 if issue_warning:
1201 self._warn_deprecate(section, key, deprecated_section, deprecated_key, extra_stacklevel)
1202 return option
1203 return VALUE_NOT_FOUND_SENTINEL
1205 def _get_option_from_config_file(
1206 self,
1207 deprecated_key: str | None,
1208 deprecated_section: str | None,
1209 key: str,
1210 section: str,
1211 issue_warning: bool = True,
1212 extra_stacklevel: int = 0,
1213 **kwargs,
1214 ) -> str | ValueNotFound:
1215 """Get config option from config file."""
1216 if team_name := kwargs.get("team_name", None):
1217 section = team_section_name(team_name, section)
1218 # since this is the last lookup that supports team_name, pop it
1219 kwargs.pop("team_name")
1220 if super().has_option(section, key):
1221 return expand_env_var(super().get(section, key, **kwargs))
1222 if deprecated_section and deprecated_key:
1223 if super().has_option(deprecated_section, deprecated_key):
1224 if issue_warning:
1225 self._warn_deprecate(section, key, deprecated_section, deprecated_key, extra_stacklevel)
1226 with self.suppress_future_warnings():
1227 return expand_env_var(super().get(deprecated_section, deprecated_key, **kwargs))
1228 return VALUE_NOT_FOUND_SENTINEL
1230 def _get_option_from_commands(
1231 self,
1232 deprecated_key: str | None,
1233 deprecated_section: str | None,
1234 key: str,
1235 section: str,
1236 issue_warning: bool = True,
1237 extra_stacklevel: int = 0,
1238 **kwargs,
1239 ) -> str | ValueNotFound:
1240 """Get config option from command execution."""
1241 if kwargs.get("team_name", None):
1242 # Commands based team config fetching is not currently supported
1243 return VALUE_NOT_FOUND_SENTINEL
1244 option = self._get_cmd_option(section, key)
1245 if option:
1246 return option
1247 if deprecated_section and deprecated_key:
1248 with self.suppress_future_warnings():
1249 option = self._get_cmd_option(deprecated_section, deprecated_key)
1250 if option:
1251 if issue_warning:
1252 self._warn_deprecate(section, key, deprecated_section, deprecated_key, extra_stacklevel)
1253 return option
1254 return VALUE_NOT_FOUND_SENTINEL
1256 def _get_option_from_secrets(
1257 self,
1258 deprecated_key: str | None,
1259 deprecated_section: str | None,
1260 key: str,
1261 section: str,
1262 issue_warning: bool = True,
1263 extra_stacklevel: int = 0,
1264 **kwargs,
1265 ) -> str | ValueNotFound:
1266 """Get config option from secrets backend."""
1267 if kwargs.get("team_name", None):
1268 # Secrets based team config fetching is not currently supported
1269 return VALUE_NOT_FOUND_SENTINEL
1270 option = self._get_secret_option(section, key)
1271 if option:
1272 return option
1273 if deprecated_section and deprecated_key:
1274 with self.suppress_future_warnings():
1275 option = self._get_secret_option(deprecated_section, deprecated_key)
1276 if option:
1277 if issue_warning:
1278 self._warn_deprecate(section, key, deprecated_section, deprecated_key, extra_stacklevel)
1279 return option
1280 return VALUE_NOT_FOUND_SENTINEL
1282 def _get_option_from_defaults(
1283 self,
1284 deprecated_key: str | None,
1285 deprecated_section: str | None,
1286 key: str,
1287 section: str,
1288 issue_warning: bool = True,
1289 extra_stacklevel: int = 0,
1290 team_name: str | None = None,
1291 **kwargs,
1292 ) -> str | ValueNotFound:
1293 """Get config option from default values."""
1294 if self.get_default_value(section, key) is not None or "fallback" in kwargs:
1295 return expand_env_var(self.get_default_value(section, key, **kwargs))
1296 return VALUE_NOT_FOUND_SENTINEL
1298 def _resolve_deprecated_lookup(
1299 self,
1300 section: str,
1301 key: str,
1302 lookup_from_deprecated: bool,
1303 extra_stacklevel: int = 0,
1304 ) -> tuple[str, str, str | None, str | None, bool]:
1305 """
1306 Resolve deprecated section/key mappings and determine deprecated values.
1308 :param section: Section name (will be lowercased)
1309 :param key: Key name (will be lowercased)
1310 :param lookup_from_deprecated: Whether to lookup from deprecated options
1311 :param extra_stacklevel: Extra stack level for warnings
1312 :return: Tuple of (resolved_section, resolved_key, deprecated_section, deprecated_key, warning_emitted)
1313 """
1314 section = section.lower()
1315 key = key.lower()
1316 warning_emitted = False
1317 deprecated_section: str | None = None
1318 deprecated_key: str | None = None
1320 if not lookup_from_deprecated:
1321 return section, key, deprecated_section, deprecated_key, warning_emitted
1323 option_description = self.configuration_description.get(section, {}).get("options", {}).get(key, {})
1324 if option_description.get("deprecated"):
1325 deprecation_reason = option_description.get("deprecation_reason", "")
1326 warnings.warn(
1327 f"The '{key}' option in section {section} is deprecated. {deprecation_reason}",
1328 DeprecationWarning,
1329 stacklevel=2 + extra_stacklevel,
1330 )
1331 # For the cases in which we rename whole sections
1332 if section in self.inversed_deprecated_sections:
1333 deprecated_section, deprecated_key = (section, key)
1334 section = self.inversed_deprecated_sections[section]
1335 if not self._suppress_future_warnings:
1336 warnings.warn(
1337 f"The config section [{deprecated_section}] has been renamed to "
1338 f"[{section}]. Please update your `conf.get*` call to use the new name",
1339 FutureWarning,
1340 stacklevel=2 + extra_stacklevel,
1341 )
1342 # Don't warn about individual rename if the whole section is renamed
1343 warning_emitted = True
1344 elif (section, key) in self.inversed_deprecated_options:
1345 # Handle using deprecated section/key instead of the new section/key
1346 new_section, new_key = self.inversed_deprecated_options[(section, key)]
1347 if not self._suppress_future_warnings and not warning_emitted:
1348 warnings.warn(
1349 f"section/key [{section}/{key}] has been deprecated, you should use"
1350 f"[{new_section}/{new_key}] instead. Please update your `conf.get*` call to use the "
1351 "new name",
1352 FutureWarning,
1353 stacklevel=2 + extra_stacklevel,
1354 )
1355 warning_emitted = True
1356 deprecated_section, deprecated_key = section, key
1357 section, key = (new_section, new_key)
1358 elif section in self.deprecated_sections:
1359 # When accessing the new section name, make sure we check under the old config name
1360 deprecated_key = key
1361 deprecated_section = self.deprecated_sections[section][0]
1362 else:
1363 deprecated_section, deprecated_key, _ = self.deprecated_options.get(
1364 (section, key), (None, None, None)
1365 )
1367 return section, key, deprecated_section, deprecated_key, warning_emitted
1369 def load_providers_configuration(self) -> None:
1370 """
1371 Load configuration for providers.
1373 .. deprecated:: 3.2.0
1374 Provider configuration is now loaded lazily via the ``configuration_description``
1375 cached property. This method is kept for backwards compatibility and will be
1376 removed in a future version.
1377 """
1378 warnings.warn(
1379 "load_providers_configuration() is deprecated. "
1380 "Provider configuration is now loaded lazily via the "
1381 "`configuration_description` cached property.",
1382 DeprecationWarning,
1383 stacklevel=2,
1384 )
1385 self._use_providers_configuration = True
1386 self._invalidate_provider_flag_caches()
1388 def restore_core_default_configuration(self) -> None:
1389 """
1390 Restore the parser state before provider-contributed sections were loaded.
1392 .. deprecated:: 3.2.0
1393 Use ``make_sure_configuration_loaded(with_providers=False)`` context manager
1394 instead. This method is kept for backwards compatibility and will be removed
1395 in a future version.
1396 """
1397 warnings.warn(
1398 "restore_core_default_configuration() is deprecated. "
1399 "Use `make_sure_configuration_loaded(with_providers=False)` instead.",
1400 DeprecationWarning,
1401 stacklevel=2,
1402 )
1403 self._use_providers_configuration = False
1404 self._invalidate_provider_flag_caches()
1406 @overload # type: ignore[override]
1407 def get(self, section: str, key: str, fallback: str = ..., **kwargs) -> str: ...
1409 @overload # type: ignore[override]
1410 def get(self, section: str, key: str, **kwargs) -> str | None: ...
1412 def get( # type: ignore[misc, override]
1413 self,
1414 section: str,
1415 key: str,
1416 suppress_warnings: bool = False,
1417 lookup_from_deprecated: bool = True,
1418 _extra_stacklevel: int = 0,
1419 team_name: str | None = None,
1420 **kwargs,
1421 ) -> str | None:
1422 """
1423 Get config value by iterating through lookup sequence.
1425 Priority order is defined by _lookup_sequence property.
1426 """
1427 section, key, deprecated_section, deprecated_key, warning_emitted = self._resolve_deprecated_lookup(
1428 section=section,
1429 key=key,
1430 lookup_from_deprecated=lookup_from_deprecated,
1431 extra_stacklevel=_extra_stacklevel,
1432 )
1434 if team_name is not None:
1435 kwargs["team_name"] = team_name
1437 for lookup_method in self._lookup_sequence:
1438 value = lookup_method(
1439 deprecated_key=deprecated_key,
1440 deprecated_section=deprecated_section,
1441 key=key,
1442 section=section,
1443 issue_warning=not warning_emitted,
1444 extra_stacklevel=_extra_stacklevel,
1445 **kwargs,
1446 )
1447 if value is not VALUE_NOT_FOUND_SENTINEL:
1448 return value
1450 # Check if fallback was explicitly provided (even if None)
1451 if "fallback" in kwargs:
1452 return kwargs["fallback"]
1454 if not suppress_warnings:
1455 log.warning("section/key [%s/%s] not found in config", section, key)
1457 raise AirflowConfigException(f"section/key [{section}/{key}] not found in config")
1459 def getboolean(self, section: str, key: str, **kwargs) -> bool: # type: ignore[override]
1460 """Get config value as boolean."""
1461 val = str(self.get(section, key, _extra_stacklevel=1, **kwargs)).lower().strip()
1462 if "#" in val:
1463 val = val.split("#")[0].strip()
1464 if val in ("t", "true", "1"):
1465 return True
1466 if val in ("f", "false", "0"):
1467 return False
1468 raise AirflowConfigException(
1469 f'Failed to convert value to bool. Please check "{key}" key in "{section}" section. '
1470 f'Current value: "{val}".'
1471 )
1473 def getint(self, section: str, key: str, **kwargs) -> int: # type: ignore[override]
1474 """Get config value as integer."""
1475 val = self.get(section, key, _extra_stacklevel=1, **kwargs)
1476 if val is None:
1477 raise AirflowConfigException(
1478 f"Failed to convert value None to int. "
1479 f'Please check "{key}" key in "{section}" section is set.'
1480 )
1481 try:
1482 return int(val)
1483 except ValueError:
1484 try:
1485 if (float_val := float(val)) != (int_val := int(float_val)):
1486 raise ValueError
1487 return int_val
1488 except (ValueError, OverflowError):
1489 raise AirflowConfigException(
1490 f'Failed to convert value to int. Please check "{key}" key in "{section}" section. '
1491 f'Current value: "{val}".'
1492 )
1494 def getfloat(self, section: str, key: str, **kwargs) -> float: # type: ignore[override]
1495 """Get config value as float."""
1496 val = self.get(section, key, _extra_stacklevel=1, **kwargs)
1497 if val is None:
1498 raise AirflowConfigException(
1499 f"Failed to convert value None to float. "
1500 f'Please check "{key}" key in "{section}" section is set.'
1501 )
1502 try:
1503 return float(val)
1504 except ValueError:
1505 raise AirflowConfigException(
1506 f'Failed to convert value to float. Please check "{key}" key in "{section}" section. '
1507 f'Current value: "{val}".'
1508 )
1510 def getlist(self, section: str, key: str, delimiter=",", **kwargs):
1511 """Get config value as list."""
1512 val = self.get(section, key, **kwargs)
1514 if isinstance(val, list) or val is None:
1515 # `get` will always return a (possibly-empty) string, so the only way we can
1516 # have these types is with `fallback=` was specified. So just return it.
1517 return val
1519 if val == "":
1520 return []
1522 try:
1523 return [item.strip() for item in val.split(delimiter)]
1524 except Exception:
1525 raise AirflowConfigException(
1526 f'Failed to parse value to a list. Please check "{key}" key in "{section}" section. '
1527 f'Current value: "{val}".'
1528 )
1530 E = TypeVar("E", bound=Enum)
1532 def getenum(self, section: str, key: str, enum_class: type[E], **kwargs) -> E:
1533 """Get config value as enum."""
1534 val = self.get(section, key, **kwargs)
1535 enum_names = [enum_item.name for enum_item in enum_class]
1537 if val is None:
1538 raise AirflowConfigException(
1539 f'Failed to convert value. Please check "{key}" key in "{section}" section. '
1540 f'Current value: "{val}" and it must be one of {", ".join(enum_names)}'
1541 )
1543 try:
1544 return enum_class[val]
1545 except KeyError:
1546 if "fallback" in kwargs and kwargs["fallback"] in enum_names:
1547 return enum_class[kwargs["fallback"]]
1548 raise AirflowConfigException(
1549 f'Failed to convert value. Please check "{key}" key in "{section}" section. '
1550 f"the value must be one of {', '.join(enum_names)}"
1551 )
1553 def getenumlist(self, section: str, key: str, enum_class: type[E], delimiter=",", **kwargs) -> list[E]:
1554 """Get config value as list of enums."""
1555 kwargs.setdefault("fallback", [])
1556 string_list = self.getlist(section, key, delimiter, **kwargs)
1558 enum_names = [enum_item.name for enum_item in enum_class]
1559 enum_list = []
1561 for val in string_list:
1562 try:
1563 enum_list.append(enum_class[val])
1564 except KeyError:
1565 log.warning(
1566 "Failed to convert value %r. Please check %s key in %s section. "
1567 "it must be one of %s, if not the value is ignored",
1568 val,
1569 key,
1570 section,
1571 ", ".join(enum_names),
1572 )
1574 return enum_list
1576 def getimport(self, section: str, key: str, **kwargs) -> Any:
1577 """
1578 Read options, import the full qualified name, and return the object.
1580 In case of failure, it throws an exception with the key and section names
1582 :return: The object or None, if the option is empty
1583 """
1584 # Fixed: use self.get() instead of conf.get()
1585 full_qualified_path = self.get(section=section, key=key, **kwargs)
1586 if not full_qualified_path:
1587 return None
1589 try:
1590 # Import here to avoid circular dependency
1591 from ..module_loading import import_string
1593 return import_string(full_qualified_path)
1594 except ImportError as e:
1595 log.warning(e)
1596 raise AirflowConfigException(
1597 f'The object could not be loaded. Please check "{key}" key in "{section}" section. '
1598 f'Current value: "{full_qualified_path}".'
1599 )
1601 def getjson(
1602 self, section: str, key: str, fallback=None, **kwargs
1603 ) -> dict | list | str | int | float | None:
1604 """
1605 Return a config value parsed from a JSON string.
1607 ``fallback`` is *not* JSON parsed but used verbatim when no config value is given.
1608 """
1609 try:
1610 data = self.get(section=section, key=key, fallback=None, _extra_stacklevel=1, **kwargs)
1611 except (NoSectionError, NoOptionError):
1612 data = None
1614 if data is None or data == "":
1615 return fallback
1617 try:
1618 return json.loads(data)
1619 except JSONDecodeError as e:
1620 raise AirflowConfigException(f"Unable to parse [{section}] {key!r} as valid json") from e
1622 def gettimedelta(
1623 self, section: str, key: str, fallback: Any = None, **kwargs
1624 ) -> datetime.timedelta | None:
1625 """
1626 Get the config value for the given section and key, and convert it into datetime.timedelta object.
1628 If the key is missing, then it is considered as `None`.
1630 :param section: the section from the config
1631 :param key: the key defined in the given section
1632 :param fallback: fallback value when no config value is given, defaults to None
1633 :raises AirflowConfigException: raised because ValueError or OverflowError
1634 :return: datetime.timedelta(seconds=<config_value>) or None
1635 """
1636 val = self.get(section, key, fallback=fallback, _extra_stacklevel=1, **kwargs)
1638 if val:
1639 # the given value must be convertible to integer
1640 try:
1641 int_val = int(val)
1642 except ValueError:
1643 raise AirflowConfigException(
1644 f'Failed to convert value to int. Please check "{key}" key in "{section}" section. '
1645 f'Current value: "{val}".'
1646 )
1648 try:
1649 return datetime.timedelta(seconds=int_val)
1650 except OverflowError as err:
1651 raise AirflowConfigException(
1652 f"Failed to convert value to timedelta in `seconds`. "
1653 f"{err}. "
1654 f'Please check "{key}" key in "{section}" section. Current value: "{val}".'
1655 )
1657 return fallback
1659 def get_mandatory_value(self, section: str, key: str, **kwargs) -> str:
1660 """Get mandatory config value, raising ValueError if not found."""
1661 value = self.get(section, key, _extra_stacklevel=1, **kwargs)
1662 if value is None:
1663 raise ValueError(f"The value {section}/{key} should be set!")
1664 return value
1666 def get_mandatory_list_value(self, section: str, key: str, **kwargs) -> list[str]:
1667 """Get mandatory config value as list, raising ValueError if not found."""
1668 value = self.getlist(section, key, **kwargs)
1669 if value is None:
1670 raise ValueError(f"The value {section}/{key} should be set!")
1671 return value
1673 def read( # type: ignore[override]
1674 self,
1675 filenames: str | bytes | os.PathLike | Iterable[str | bytes | os.PathLike],
1676 encoding: str | None = None,
1677 ) -> list[str]:
1678 return super().read(filenames=filenames, encoding=encoding) # type: ignore[arg-type,return-value]
1680 def read_dict( # type: ignore[override]
1681 self, dictionary: dict[str, dict[str, Any]], source: str = "<dict>"
1682 ) -> None:
1683 """
1684 We define a different signature here to add better type hints and checking.
1686 :param dictionary: dictionary to read from
1687 :param source: source to be used to store the configuration
1688 :return:
1689 """
1690 super().read_dict(dictionary=dictionary, source=source)
1692 def _has_section_in_any_defaults(self, section: str) -> bool:
1693 """Check if section exists in core defaults or provider fallback defaults."""
1694 if self._default_values.has_section(section):
1695 return True
1696 if self._use_providers_configuration:
1697 if self._provider_cfg_config_fallback_default_values.has_section(section):
1698 return True
1699 if self._provider_metadata_config_fallback_default_values.has_section(section):
1700 return True
1701 return False
1703 def get_sections_including_defaults(self) -> list[str]:
1704 """
1705 Retrieve all sections from the configuration parser, including sections defined by built-in defaults.
1707 :return: list of section names
1708 """
1709 sections_from_config = self.sections()
1710 sections_from_description = list(self.configuration_description.keys())
1711 return list(dict.fromkeys(itertools.chain(sections_from_description, sections_from_config)))
1713 def get_options_including_defaults(self, section: str) -> list[str]:
1714 """
1715 Retrieve all possible options from the configuration parser for the section given.
1717 Includes options defined by built-in defaults.
1719 :param section: section name
1720 :return: list of option names for the section given
1721 """
1722 my_own_options = self.options(section) if self.has_section(section) else []
1723 all_options_from_defaults = list(
1724 self.configuration_description.get(section, {}).get("options", {}).keys()
1725 )
1726 return list(dict.fromkeys(itertools.chain(all_options_from_defaults, my_own_options)))
1728 def has_option( # type: ignore[override]
1729 self, section: str, option: str, lookup_from_deprecated: bool = True, **kwargs
1730 ) -> bool:
1731 """
1732 Check if option is defined.
1734 Uses self.get() to avoid reimplementing the priority order of config variables
1735 (env, config, cmd, defaults).
1737 :param section: section to get option from
1738 :param option: option to get
1739 :param lookup_from_deprecated: If True, check if the option is defined in deprecated sections
1740 :param kwargs: additional keyword arguments to pass to get(), such as team_name
1741 :return:
1742 """
1743 try:
1744 value = self.get(
1745 section,
1746 option,
1747 fallback=VALUE_NOT_FOUND_SENTINEL,
1748 _extra_stacklevel=1,
1749 suppress_warnings=True,
1750 lookup_from_deprecated=lookup_from_deprecated,
1751 **kwargs,
1752 )
1753 if value is VALUE_NOT_FOUND_SENTINEL:
1754 return False
1755 return True
1756 except (NoOptionError, NoSectionError, AirflowConfigException):
1757 return False
1759 def set(self, section: str, option: str, value: str | None = None) -> None: # type: ignore[override]
1760 """
1761 Set an option to the given value.
1763 This override just makes sure the section and option are lower case, to match what we do in `get`.
1764 """
1765 section = section.lower()
1766 option = option.lower()
1767 defaults = self.configuration_description or {}
1768 if not self.has_section(section) and section in defaults:
1769 # Trying to set a key in a section that exists in default, but not in the user config;
1770 # automatically create it
1771 self.add_section(section)
1772 super().set(section, option, value)
1774 def remove_option(self, section: str, option: str, remove_default: bool = True): # type: ignore[override]
1775 """
1776 Remove an option if it exists in config from a file or default config.
1778 If both of config have the same option, this removes the option
1779 in both configs unless remove_default=False.
1780 """
1781 section = section.lower()
1782 option = option.lower()
1783 if super().has_option(section, option):
1784 super().remove_option(section, option)
1786 if remove_default and self._default_values.has_option(section, option):
1787 self._default_values.remove_option(section, option)
1789 def optionxform(self, optionstr: str) -> str:
1790 """
1791 Transform option names on every read, get, or set operation.
1793 This changes from the default behaviour of ConfigParser from lower-casing
1794 to instead be case-preserving.
1796 :param optionstr:
1797 :return:
1798 """
1799 return optionstr
1801 def as_dict(
1802 self,
1803 display_source: bool = False,
1804 display_sensitive: bool = False,
1805 raw: bool = False,
1806 include_env: bool = True,
1807 include_cmds: bool = True,
1808 include_secret: bool = True,
1809 ) -> ConfigSourcesType:
1810 """
1811 Return the current configuration as an OrderedDict of OrderedDicts.
1813 When materializing current configuration Airflow defaults are
1814 materialized along with user set configs. If any of the `include_*`
1815 options are False then the result of calling command or secret key
1816 configs do not override Airflow defaults and instead are passed through.
1817 In order to then avoid Airflow defaults from overwriting user set
1818 command or secret key configs we filter out bare sensitive_config_values
1819 that are set to Airflow defaults when command or secret key configs
1820 produce different values.
1822 :param display_source: If False, the option value is returned. If True,
1823 a tuple of (option_value, source) is returned. Source is either
1824 'airflow.cfg', 'default', 'env var', or 'cmd'.
1825 :param display_sensitive: If True, the values of options set by env
1826 vars and bash commands will be displayed. If False, those options
1827 are shown as '< hidden >'
1828 :param raw: Should the values be output as interpolated values, or the
1829 "raw" form that can be fed back in to ConfigParser
1830 :param include_env: Should the value of configuration from AIRFLOW__
1831 environment variables be included or not
1832 :param include_cmds: Should the result of calling any ``*_cmd`` config be
1833 set (True, default), or should the _cmd options be left as the
1834 command to run (False)
1835 :param include_secret: Should the result of calling any ``*_secret`` config be
1836 set (True, default), or should the _secret options be left as the
1837 path to get the secret from (False)
1838 :return: Dictionary, where the key is the name of the section and the content is
1839 the dictionary with the name of the parameter and its value.
1840 """
1841 if not display_sensitive:
1842 # We want to hide the sensitive values at the appropriate methods
1843 # since envs from cmds, secrets can be read at _include_envs method
1844 if not all([include_env, include_cmds, include_secret]):
1845 raise ValueError(
1846 "If display_sensitive is false, then include_env, "
1847 "include_cmds, include_secret must all be set as True"
1848 )
1850 config_sources: ConfigSourcesType = {}
1852 # We check sequentially all those sources and the last one we saw it in will "win"
1853 configs = self._config_sources_for_as_dict
1855 self._replace_config_with_display_sources(
1856 config_sources,
1857 configs,
1858 self.configuration_description,
1859 display_source,
1860 raw,
1861 self.deprecated_options,
1862 include_cmds=include_cmds,
1863 include_env=include_env,
1864 include_secret=include_secret,
1865 )
1867 # add env vars and overwrite because they have priority
1868 if include_env:
1869 self._include_envs(config_sources, display_sensitive, display_source, raw)
1870 else:
1871 self._filter_by_source(config_sources, display_source, self._get_env_var_option)
1873 # add bash commands
1874 if include_cmds:
1875 self._include_commands(config_sources, display_sensitive, display_source, raw)
1876 else:
1877 self._filter_by_source(config_sources, display_source, self._get_cmd_option)
1879 # add config from secret backends
1880 if include_secret:
1881 self._include_secrets(config_sources, display_sensitive, display_source, raw)
1882 else:
1883 self._filter_by_source(config_sources, display_source, self._get_secret_option)
1885 if not display_sensitive:
1886 # This ensures the ones from config file is hidden too
1887 # if they are not provided through env, cmd and secret
1888 # The collected options are walked (rather than the registered sensitive ones) so that
1889 # team scoped sections are covered as well - they are named after the team, not after
1890 # the base section the option is registered under.
1891 hidden = "< hidden >"
1892 for section, options in config_sources.items():
1893 for key, value in list(options.items()):
1894 if not value or not self.is_sensitive_option(section, key):
1895 continue
1896 if display_source:
1897 source = value[1]
1898 options[key] = (hidden, source)
1899 else:
1900 options[key] = hidden
1902 return config_sources
1904 def _write_option_header(
1905 self,
1906 file: IO[str],
1907 option: str,
1908 extra_spacing: bool,
1909 include_descriptions: bool,
1910 include_env_vars: bool,
1911 include_examples: bool,
1912 include_sources: bool,
1913 section_config_description: dict[str, dict[str, Any]],
1914 section_to_write: str,
1915 sources_dict: ConfigSourcesType,
1916 ) -> tuple[bool, bool]:
1917 """
1918 Write header for configuration option.
1920 Returns tuple of (should_continue, needs_separation) where needs_separation should be
1921 set if the option needs additional separation to visually separate it from the next option.
1922 """
1923 option_config_description = (
1924 section_config_description.get("options", {}).get(option, {})
1925 if section_config_description
1926 else {}
1927 )
1928 description = option_config_description.get("description")
1929 needs_separation = False
1930 if description and include_descriptions:
1931 for line in description.splitlines():
1932 file.write(f"# {line}\n")
1933 needs_separation = True
1934 example = option_config_description.get("example")
1935 if example is not None and include_examples:
1936 if extra_spacing:
1937 file.write("#\n")
1938 example_lines = example.splitlines()
1939 example = "\n# ".join(example_lines)
1940 file.write(f"# Example: {option} = {example}\n")
1941 needs_separation = True
1942 if include_sources and sources_dict:
1943 sources_section = sources_dict.get(section_to_write)
1944 value_with_source = sources_section.get(option) if sources_section else None
1945 if value_with_source is None:
1946 file.write("#\n# Source: not defined\n")
1947 else:
1948 file.write(f"#\n# Source: {value_with_source[1]}\n")
1949 needs_separation = True
1950 if include_env_vars:
1951 file.write(f"#\n# Variable: AIRFLOW__{section_to_write.upper()}__{option.upper()}\n")
1952 if extra_spacing:
1953 file.write("#\n")
1954 needs_separation = True
1955 return True, needs_separation
1957 def is_template(self, section: str, key) -> bool:
1958 """
1959 Return whether the value is templated.
1961 :param section: section of the config
1962 :param key: key in the section
1963 :return: True if the value is templated
1964 """
1965 return _is_template(self.configuration_description, section, key)
1967 def getsection(self, section: str, team_name: str | None = None) -> ConfigOptionsDictType | None:
1968 """
1969 Return the section as a dict.
1971 Values are converted to int, float, bool as required.
1973 :param section: section from the config
1974 :param team_name: optional team name for team-specific configuration lookup
1975 """
1976 # Handle team-specific section lookup for config file
1977 config_section = team_section_name(team_name, section) if team_name else section
1979 if not self.has_section(config_section) and not self._has_section_in_any_defaults(config_section):
1980 return None
1981 if self._default_values.has_section(config_section):
1982 _section: ConfigOptionsDictType = dict(self._default_values.items(config_section))
1983 else:
1984 _section = {}
1986 if self.has_section(config_section):
1987 _section.update(self.items(config_section))
1989 # Use section (not config_section) for env var lookup - team_name is handled by _env_var_name
1990 section_prefix = self._env_var_name(section, "", team_name=team_name)
1991 for env_var in sorted(os.environ.keys()):
1992 if env_var.startswith(section_prefix):
1993 key = env_var.replace(section_prefix, "")
1994 if key.endswith("_CMD"):
1995 key = key[:-4]
1996 key = key.lower()
1997 _section[key] = self._get_env_var_option(section, key, team_name=team_name)
1999 for key, val in _section.items():
2000 if val is None:
2001 raise AirflowConfigException(
2002 f"Failed to convert value automatically. "
2003 f'Please check "{key}" key in "{section}" section is set.'
2004 )
2005 try:
2006 _section[key] = int(val)
2007 except ValueError:
2008 try:
2009 _section[key] = float(val)
2010 except ValueError:
2011 if isinstance(val, str) and val.lower() in ("t", "true"):
2012 _section[key] = True
2013 elif isinstance(val, str) and val.lower() in ("f", "false"):
2014 _section[key] = False
2015 return _section
2017 @staticmethod
2018 def _write_section_header(
2019 file: IO[str],
2020 include_descriptions: bool,
2021 section_config_description: dict[str, str],
2022 section_to_write: str,
2023 ) -> None:
2024 """Write header for configuration section."""
2025 file.write(f"[{section_to_write}]\n")
2026 section_description = section_config_description.get("description")
2027 if section_description and include_descriptions:
2028 for line in section_description.splitlines():
2029 file.write(f"# {line}\n")
2030 file.write("\n")
2032 def _write_value(
2033 self,
2034 file: IO[str],
2035 option: str,
2036 comment_out_everything: bool,
2037 needs_separation: bool,
2038 only_defaults: bool,
2039 section_to_write: str,
2040 hide_sensitive: bool,
2041 is_sensitive: bool,
2042 show_values: bool = False,
2043 ):
2044 default_value = self.get_default_value(section_to_write, option, raw=True)
2045 if only_defaults:
2046 value = default_value
2047 else:
2048 value = self.get(section_to_write, option, fallback=default_value, raw=True)
2049 if not show_values:
2050 file.write(f"# {option} = \n")
2051 else:
2052 if hide_sensitive and is_sensitive:
2053 value = "< hidden >"
2054 else:
2055 pass
2056 if value is None:
2057 file.write(f"# {option} = \n")
2058 else:
2059 if comment_out_everything:
2060 value_lines = value.splitlines()
2061 value = "\n# ".join(value_lines)
2062 file.write(f"# {option} = {value}\n")
2063 else:
2064 if "\n" in value:
2065 try:
2066 value = json.dumps(json.loads(value), indent=4)
2067 value = value.replace(
2068 "\n", "\n "
2069 ) # indent multi-line JSON to satisfy configparser format
2070 except JSONDecodeError:
2071 pass
2072 file.write(f"{option} = {value}\n")
2073 if needs_separation:
2074 file.write("\n")
2076 def write( # type: ignore[override]
2077 self,
2078 file: IO[str],
2079 section: str | None = None,
2080 include_examples: bool = True,
2081 include_descriptions: bool = True,
2082 include_sources: bool = True,
2083 include_env_vars: bool = True,
2084 include_providers: bool = True,
2085 comment_out_everything: bool = False,
2086 hide_sensitive: bool = False,
2087 extra_spacing: bool = True,
2088 only_defaults: bool = False,
2089 show_values: bool = False,
2090 **kwargs: Any,
2091 ) -> None:
2092 """
2093 Write configuration with comments and examples to a file.
2095 :param file: file to write to
2096 :param section: section of the config to write, defaults to all sections
2097 :param include_examples: Include examples in the output
2098 :param include_descriptions: Include descriptions in the output
2099 :param include_sources: Include the source of each config option
2100 :param include_env_vars: Include environment variables corresponding to each config option
2101 :param include_providers: Include providers configuration
2102 :param comment_out_everything: Comment out all values
2103 :param hide_sensitive_values: Include sensitive values in the output
2104 :param extra_spacing: Add extra spacing before examples and after variables
2105 :param only_defaults: Only include default values when writing the config, not the actual values
2106 """
2107 with self.make_sure_configuration_loaded(with_providers=include_providers):
2108 sources_dict = {}
2109 if include_sources:
2110 sources_dict = self.as_dict(display_source=True)
2111 for section_to_write in self.get_sections_including_defaults():
2112 section_config_description = self.configuration_description.get(section_to_write, {})
2113 if section_to_write != section and section is not None:
2114 continue
2115 if self._has_section_in_any_defaults(section_to_write) or self.has_section(section_to_write):
2116 self._write_section_header(
2117 file, include_descriptions, section_config_description, section_to_write
2118 )
2119 for option in self.get_options_including_defaults(section_to_write):
2120 should_continue, needs_separation = self._write_option_header(
2121 file=file,
2122 option=option,
2123 extra_spacing=extra_spacing,
2124 include_descriptions=include_descriptions,
2125 include_env_vars=include_env_vars,
2126 include_examples=include_examples,
2127 include_sources=include_sources,
2128 section_config_description=section_config_description,
2129 section_to_write=section_to_write,
2130 sources_dict=sources_dict,
2131 )
2132 is_sensitive = self.is_sensitive_option(section_to_write, option)
2133 self._write_value(
2134 file=file,
2135 option=option,
2136 comment_out_everything=comment_out_everything,
2137 needs_separation=needs_separation,
2138 only_defaults=only_defaults,
2139 section_to_write=section_to_write,
2140 hide_sensitive=hide_sensitive,
2141 is_sensitive=is_sensitive,
2142 show_values=show_values,
2143 )
2144 if include_descriptions and not needs_separation:
2145 # extra separation between sections in case last option did not need it
2146 file.write("\n")
2148 @contextmanager
2149 def make_sure_configuration_loaded(self, with_providers: bool) -> Generator[None, None, None]:
2150 """
2151 Make sure configuration is loaded with or without providers.
2153 The context manager will only toggle the `self._use_providers_configuration` flag if `with_providers` is False, and will reset `self._use_providers_configuration` to True after the context block.
2154 Nop for `with_providers=True` as the configuration already loads providers configuration by default.
2156 :param with_providers: whether providers should be loaded
2157 """
2158 if not with_providers:
2159 self._use_providers_configuration = False
2160 # Only invalidate cached properties that depend on _use_providers_configuration.
2161 # Do NOT use invalidate_cache() here — it would also evict expensive provider-discovery
2162 # caches (_provider_metadata_configuration_description, _provider_metadata_config_fallback_default_values)
2163 # that don't depend on this flag.
2164 self._invalidate_provider_flag_caches()
2165 try:
2166 yield
2167 finally:
2168 if not with_providers:
2169 self._use_providers_configuration = True
2170 self._invalidate_provider_flag_caches()