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# Licensed to the Apache Software Foundation (ASF) under one
2# or more contributor license agreements. See the NOTICE file
3# distributed with this work for additional information
4# regarding copyright ownership. The ASF licenses this file
5# to you under the Apache License, Version 2.0 (the
6# "License"); you may not use this file except in compliance
7# with the License. You may obtain a copy of the License at
8#
9# http://www.apache.org/licenses/LICENSE-2.0
10#
11# Unless required by applicable law or agreed to in writing,
12# software distributed under the License is distributed on an
13# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14# KIND, either express or implied. See the License for the
15# specific language governing permissions and limitations
16# under the License.
17from __future__ import annotations
19import logging
20import os
21import pathlib
22import re
23import shlex
24import stat
25import subprocess
26import sys
27import warnings
28from base64 import b64encode
29from collections.abc import Callable
30from configparser import ConfigParser
31from importlib.util import find_spec
32from inspect import ismodule
33from io import StringIO
34from re import Pattern
35from typing import IO, TYPE_CHECKING, Any
36from urllib.parse import urlsplit
38from typing_extensions import overload
40from airflow._shared.configuration.parser import (
41 AirflowConfigParser as _SharedAirflowConfigParser,
42 configure_parser_from_configuration_description,
43)
44from airflow._shared.module_loading import import_string
45from airflow.exceptions import AirflowConfigException, RemovedInAirflow4Warning
46from airflow.secrets import DEFAULT_SECRETS_SEARCH_PATH
47from airflow.task.weight_rule import WeightRule
48from airflow.utils import yaml
50if TYPE_CHECKING:
51 from airflow.api_fastapi.auth.managers.base_auth_manager import BaseAuthManager
52 from airflow.secrets import BaseSecretsBackend
54log = logging.getLogger(__name__)
56# show Airflow's deprecation warnings
57if not sys.warnoptions:
58 warnings.filterwarnings(action="default", category=DeprecationWarning, module="airflow")
59 warnings.filterwarnings(action="default", category=PendingDeprecationWarning, module="airflow")
61ConfigType = str | int | float | bool
62ConfigOptionsDictType = dict[str, ConfigType]
63ConfigSectionSourcesType = dict[str, str | tuple[str, str]]
64ConfigSourcesType = dict[str, ConfigSectionSourcesType]
66ENV_VAR_PREFIX = "AIRFLOW__"
69class _SecretKeys:
70 """Holds the secret keys used in Airflow during runtime."""
72 fernet_key: str = "" # Set only if needed when generating a new file
73 jwt_secret_key: str = ""
76class ConfigModifications:
77 """
78 Holds modifications to be applied when writing out the config.
80 :param rename: Mapping from (old_section, old_option) to (new_section, new_option)
81 :param remove: Set of (section, option) to remove
82 :param default_updates: Mapping from (section, option) to new default value
83 """
85 def __init__(self) -> None:
86 self.rename: dict[tuple[str, str], tuple[str, str]] = {}
87 self.remove: set[tuple[str, str]] = set()
88 self.default_updates: dict[tuple[str, str], str] = {}
90 def add_rename(self, old_section: str, old_option: str, new_section: str, new_option: str) -> None:
91 self.rename[(old_section, old_option)] = (new_section, new_option)
93 def add_remove(self, section: str, option: str) -> None:
94 self.remove.add((section, option))
96 def add_default_update(self, section: str, option: str, new_default: str) -> None:
97 self.default_updates[(section, option)] = new_default
100@overload
101def expand_env_var(env_var: None) -> None: ...
104@overload
105def expand_env_var(env_var: str) -> str: ...
108def expand_env_var(env_var: str | None) -> str | None:
109 """
110 Expand (potentially nested) env vars.
112 Repeat and apply `expandvars` and `expanduser` until
113 interpolation stops having any effect.
114 """
115 if not env_var or not isinstance(env_var, str):
116 return env_var
117 while True:
118 interpolated = os.path.expanduser(os.path.expandvars(str(env_var)))
119 if interpolated == env_var:
120 return interpolated
121 env_var = interpolated
124def run_command(command: str) -> str:
125 """Run command and returns stdout."""
126 process = subprocess.Popen(
127 shlex.split(command), stdout=subprocess.PIPE, stderr=subprocess.PIPE, close_fds=True
128 )
129 output, stderr = (stream.decode(sys.getdefaultencoding(), "ignore") for stream in process.communicate())
131 if process.returncode != 0:
132 raise AirflowConfigException(
133 f"Cannot execute {command}. Error code is: {process.returncode}. "
134 f"Output: {output}, Stderr: {stderr}"
135 )
137 return output
140def _default_config_file_path(file_name: str) -> str:
141 templates_dir = os.path.join(os.path.dirname(__file__), "config_templates")
142 return os.path.join(templates_dir, file_name)
145def retrieve_configuration_description(
146 include_airflow: bool = True,
147 include_providers: bool = True,
148 selected_provider: str | None = None,
149) -> dict[str, dict[str, Any]]:
150 """
151 Read Airflow configuration description from YAML file.
153 :param include_airflow: Include Airflow configs
154 :param include_providers: Include provider configs
155 :param selected_provider: If specified, include selected provider only
156 :return: Python dictionary containing configs & their info
157 """
158 base_configuration_description: dict[str, dict[str, Any]] = {}
159 if include_airflow:
160 with open(_default_config_file_path("config.yml")) as config_file:
161 base_configuration_description.update(yaml.safe_load(config_file))
162 if include_providers:
163 from airflow.providers_manager import ProvidersManager
165 for provider, config in ProvidersManager().provider_configs:
166 if not selected_provider or provider == selected_provider:
167 base_configuration_description.update(config)
168 return base_configuration_description
171class AirflowConfigParser(_SharedAirflowConfigParser):
172 """
173 Custom Airflow Configparser supporting defaults and deprecated options.
175 This is a subclass of the shared AirflowConfigParser that adds Core-specific initialization
176 and functionality (providers, validation, writing, etc.).
178 The defaults are stored in the ``_default_values``. The configuration description keeps
179 description of all the options available in Airflow (description follow config.yaml.schema).
181 :param default_config: default configuration (in the form of ini file).
182 :param configuration_description: description of configuration to use
183 """
185 def __init__(
186 self,
187 default_config: str | None = None,
188 *args,
189 **kwargs,
190 ):
191 _configuration_description = retrieve_configuration_description(include_providers=False)
192 # For those who would like to use a different data structure to keep defaults:
193 # We have to keep the default values in a ConfigParser rather than in any other
194 # data structure, because the values we have might contain %% which are ConfigParser
195 # interpolation placeholders. The _default_values config parser will interpolate them
196 # properly when we call get() on it.
197 _default_values = create_default_config_parser(_configuration_description)
198 from airflow.providers_manager import ProvidersManager
200 super().__init__(
201 _configuration_description,
202 _default_values,
203 ProvidersManager,
204 create_default_config_parser,
205 _default_config_file_path("provider_config_fallback_defaults.cfg"),
206 *args,
207 **kwargs,
208 )
209 self._configuration_description = _configuration_description
210 self._default_values = _default_values
211 if default_config is not None:
212 self._update_defaults_from_string(default_config)
213 self._update_logging_deprecated_template_to_one_from_defaults()
214 self.is_validated = False
215 self._suppress_future_warnings = False
217 @property
218 def _validators(self) -> list[Callable[[], None]]:
219 """Overring _validators from shared base class to add core-specific validators."""
220 return [
221 self._validate_sqlite3_version,
222 self._validate_enums,
223 self._validate_deprecated_values,
224 self._upgrade_postgres_metastore_conn,
225 ]
227 def _update_logging_deprecated_template_to_one_from_defaults(self):
228 default = self.get_default_value("logging", "log_filename_template")
229 if default:
230 # Tuple does not support item assignment, so we have to create a new tuple and replace it
231 original_replacement = self.deprecated_values["logging"]["log_filename_template"]
232 self.deprecated_values["logging"]["log_filename_template"] = (
233 original_replacement[0],
234 default,
235 )
237 # A mapping of old default values that we want to change and warn the user
238 # about. Mapping of section -> setting -> { old, replace }
239 deprecated_values: dict[str, dict[str, tuple[Pattern, str]]] = {
240 "logging": {
241 "log_filename_template": (
242 re.compile(
243 re.escape(
244 "dag_id={{ ti.dag_id }}/run_id={{ ti.run_id }}/task_id={{ ti.task_id }}/{% if ti.map_index >= 0 %}map_index={{ ti.map_index }}/{% endif %}attempt={{ try_number }}.log"
245 )
246 ),
247 # The actual replacement value will be updated after defaults are loaded from config.yml
248 "XX-set-after-default-config-loaded-XX",
249 ),
250 },
251 "core": {
252 "executor": (re.compile(re.escape("SequentialExecutor")), "LocalExecutor"),
253 },
254 }
256 _available_logging_levels = ["CRITICAL", "FATAL", "ERROR", "WARN", "WARNING", "INFO", "DEBUG"]
257 enums_options = {
258 ("core", "default_task_weight_rule"): sorted(WeightRule.all_weight_rules()),
259 ("core", "dag_ignore_file_syntax"): ["regexp", "glob"],
260 ("dag_processor", "file_parsing_sort_mode"): [
261 "modified_time",
262 "random_seeded_by_host",
263 "alphabetical",
264 ],
265 ("logging", "logging_level"): _available_logging_levels,
266 ("logging", "fab_logging_level"): _available_logging_levels,
267 # celery_logging_level can be empty, which uses logging_level as fallback
268 ("logging", "celery_logging_level"): [*_available_logging_levels, ""],
269 # uvicorn and gunicorn logging levels for web servers
270 ("logging", "uvicorn_logging_level"): _available_logging_levels,
271 ("logging", "gunicorn_logging_level"): _available_logging_levels,
272 ("webserver", "analytical_tool"): ["google_analytics", "metarouter", "segment", "matomo", ""],
273 ("api", "grid_view_sorting_order"): ["topological", "hierarchical_alphabetical"],
274 ("logging", "dag_processor_log_target"): ["file", "stdout"],
275 }
277 upgraded_values: dict[tuple[str, str], str]
278 """Mapping of (section,option) to the old value that was upgraded"""
280 def write_custom_config(
281 self,
282 file: IO[str],
283 comment_out_defaults: bool = True,
284 include_descriptions: bool = True,
285 extra_spacing: bool = True,
286 modifications: ConfigModifications | None = None,
287 ) -> None:
288 """
289 Write a configuration file using a ConfigModifications object.
291 This method includes only options from the current airflow.cfg. For each option:
292 - If it's marked for removal, omit it.
293 - If renamed, output it under its new name and add a comment indicating its original location.
294 - If a default update is specified, apply the new default and output the option as a commented line.
295 - Otherwise, if the current value equals the default and comment_out_defaults is True, output it as a comment.
296 Options absent from the current airflow.cfg are omitted.
298 :param file: File to write the configuration.
299 :param comment_out_defaults: If True, options whose value equals the default are written as comments.
300 :param include_descriptions: Whether to include section descriptions.
301 :param extra_spacing: Whether to insert an extra blank line after each option.
302 :param modifications: ConfigModifications instance with rename, remove, and default updates.
303 """
304 modifications = modifications or ConfigModifications()
305 output: dict[str, list[tuple[str, str, bool, str]]] = {}
307 for section in self._sections: # type: ignore[attr-defined] # accessing _sections from ConfigParser
308 for option, orig_value in self._sections[section].items(): # type: ignore[attr-defined]
309 key = (section.lower(), option.lower())
310 if key in modifications.remove:
311 continue
313 mod_comment = ""
314 if key in modifications.rename:
315 new_sec, new_opt = modifications.rename[key]
316 effective_section = new_sec
317 effective_option = new_opt
318 mod_comment += f"# Renamed from {section}.{option}\n"
319 else:
320 effective_section = section
321 effective_option = option
323 value = orig_value
324 if key in modifications.default_updates:
325 mod_comment += (
326 f"# Default updated from {orig_value} to {modifications.default_updates[key]}\n"
327 )
328 value = modifications.default_updates[key]
330 default_value = self.get_default_value(effective_section, effective_option, fallback="")
331 is_default = str(value) == str(default_value)
332 output.setdefault(effective_section.lower(), []).append(
333 (effective_option, str(value), is_default, mod_comment)
334 )
336 for section, options in output.items():
337 section_buffer = StringIO()
338 section_buffer.write(f"[{section}]\n")
339 if include_descriptions:
340 description = self.configuration_description.get(section, {}).get("description", "")
341 if description:
342 for line in description.splitlines():
343 section_buffer.write(f"# {line}\n")
344 section_buffer.write("\n")
345 for option, value_str, is_default, mod_comment in options:
346 key = (section.lower(), option.lower())
347 if key in modifications.default_updates and comment_out_defaults:
348 section_buffer.write(f"# {option} = {value_str}\n")
349 else:
350 if mod_comment:
351 section_buffer.write(mod_comment)
352 if is_default and comment_out_defaults:
353 section_buffer.write(f"# {option} = {value_str}\n")
354 else:
355 section_buffer.write(f"{option} = {value_str}\n")
356 if extra_spacing:
357 section_buffer.write("\n")
358 content = section_buffer.getvalue().strip()
359 if content:
360 file.write(f"{content}\n\n")
362 def _upgrade_postgres_metastore_conn(self):
363 """
364 Upgrade SQL schemas.
366 As of SQLAlchemy 1.4, schemes `postgres+psycopg2` and `postgres`
367 must be replaced with `postgresql+psycopg` if the psycopg (v3) driver
368 is installed, or `postgresql+psycopg2` otherwise. The bare `postgresql`
369 scheme is upgraded the same way to make the driver explicit.
370 """
371 section, key = "database", "sql_alchemy_conn"
372 old_value = self.get(section, key, _extra_stacklevel=1)
373 bad_schemes = ["postgres+psycopg2", "postgres", "postgresql"]
374 # The provider-side hooks (common.sql / postgres / amazon) also gate psycopg (v3) on an
375 # ``_is_sqlalchemy_2()`` check, because they support Airflow 2.11 on SQLAlchemy 1.4, which has
376 # no native ``postgresql+psycopg`` dialect. airflow-core pins ``sqlalchemy>=2.0``, so that
377 # dialect is always present when the package is importable — ``find_spec`` alone suffices here.
378 good_scheme = "postgresql+psycopg" if find_spec("psycopg") is not None else "postgresql+psycopg2"
379 parsed = urlsplit(old_value)
380 if parsed.scheme in bad_schemes:
381 warnings.warn(
382 f"Bad scheme in Airflow configuration [database] sql_alchemy_conn: `{parsed.scheme}`. "
383 "As of SQLAlchemy 1.4 (adopted in Airflow 2.3) this is no longer supported. You must "
384 f"change to `{good_scheme}` before the next Airflow release.",
385 FutureWarning,
386 stacklevel=1,
387 )
388 self.upgraded_values[(section, key)] = old_value
389 new_value = re.sub("^" + re.escape(f"{parsed.scheme}://"), f"{good_scheme}://", old_value)
390 self._update_env_var(section=section, name=key, new_value=new_value)
392 # if the old value is set via env var, we need to wipe it
393 # otherwise, it'll "win" over our adjusted value
394 old_env_var = self._env_var_name("core", key)
395 os.environ.pop(old_env_var, None)
397 def _validate_enums(self):
398 """Validate that enum type config has an accepted value."""
399 for (section_key, option_key), enum_options in self.enums_options.items():
400 if self.has_option(section_key, option_key):
401 value = self.get(section_key, option_key, fallback=None)
402 if value and value not in enum_options:
403 raise AirflowConfigException(
404 f"`[{section_key}] {option_key}` should not be "
405 f"{value!r}. Possible values: {', '.join(enum_options)}."
406 )
408 def _validate_sqlite3_version(self):
409 """
410 Validate SQLite version.
412 Some features in storing rendered fields require SQLite >= 3.15.0.
413 """
414 if "sqlite" not in self.get("database", "sql_alchemy_conn"):
415 return
417 import sqlite3
419 min_sqlite_version = (3, 15, 0)
420 if sqlite3.sqlite_version_info >= min_sqlite_version:
421 return
423 from airflow.utils.docs import get_docs_url
425 min_sqlite_version_str = ".".join(str(s) for s in min_sqlite_version)
426 raise AirflowConfigException(
427 f"error: SQLite C library too old (< {min_sqlite_version_str}). "
428 f"See {get_docs_url('howto/set-up-database.html#setting-up-a-sqlite-database')}"
429 )
431 def _get_custom_secret_backend(self, worker_mode: bool | None = None) -> Any | None:
432 return super()._get_custom_secret_backend(
433 worker_mode=worker_mode if worker_mode is not None else False
434 )
436 def mask_secrets(self):
437 from airflow._shared.configuration.parser import _build_kwarg_env_prefix, _collect_kwarg_env_vars
438 from airflow._shared.secrets_masker import mask_secret as mask_secret_core
439 from airflow.sdk.log import mask_secret as mask_secret_sdk
441 for section, key in self.sensitive_config_values:
442 try:
443 with self.suppress_future_warnings():
444 value = self.get(section, key, suppress_warnings=True)
445 except AirflowConfigException:
446 log.debug(
447 "Could not retrieve value from section %s, for key %s. Skipping redaction of this conf.",
448 section,
449 key,
450 )
451 continue
452 mask_secret_core(value)
453 mask_secret_sdk(value)
455 # Mask per-key backend kwarg env vars (AIRFLOW__SECRETS__BACKEND_KWARG__* etc.).
456 # These are not in sensitive_config_values but may contain sensitive values.
457 for _section, _kwargs_key in [
458 ("secrets", "backend_kwargs"),
459 ("workers", "secrets_backend_kwargs"),
460 ]:
461 _prefix = _build_kwarg_env_prefix(_section, _kwargs_key)
462 for _value in _collect_kwarg_env_vars(_prefix).values():
463 mask_secret_core(_value)
464 mask_secret_sdk(_value)
466 def load_test_config(self):
467 """
468 Use test configuration rather than the configuration coming from airflow defaults.
470 When running tests we use special the unit_test configuration to avoid accidental modifications and
471 different behaviours when running the tests. Values for those test configuration are stored in
472 the "unit_tests.cfg" configuration file in the ``airflow/config_templates`` folder
473 and you need to change values there if you want to make some specific configuration to be used
474 """
475 from cryptography.fernet import Fernet
477 unit_test_config_file = pathlib.Path(__file__).parent / "config_templates" / "unit_tests.cfg"
478 unit_test_config = unit_test_config_file.read_text()
479 self.remove_all_read_configurations()
480 with StringIO(unit_test_config) as test_config_file:
481 self.read_file(test_config_file)
483 # We need those globals before we run "get_all_expansion_variables" because this is where
484 # the variables are expanded from in the configuration - set to random values for tests
485 _SecretKeys.fernet_key = Fernet.generate_key().decode()
486 _SecretKeys.jwt_secret_key = b64encode(os.urandom(16)).decode("utf-8")
487 self.expand_all_configuration_values()
488 log.info("Unit test configuration loaded from 'config_unit_tests.cfg'")
490 def expand_all_configuration_values(self):
491 """Expand all configuration values using global and local variables defined in this module."""
492 all_vars = get_all_expansion_variables()
493 for section in self.sections():
494 for key, value in self.items(section):
495 if value is not None:
496 if self.has_option(section, key):
497 self.remove_option(section, key, remove_default=False)
498 if self.is_template(section, key) or not isinstance(value, str):
499 self.set(section, key, value)
500 else:
501 self.set(section, key, value.format(**all_vars))
503 def remove_all_read_configurations(self):
504 """Remove all read configurations, leaving only default values in the config."""
505 for section in self.sections():
506 self.remove_section(section)
508 def _get_config_value_from_secret_backend(self, config_key: str) -> str | None:
509 """
510 Override to use module-level function that reads from global conf.
512 This ensures as_dict() and other methods use the same secrets backend
513 configuration as the global conf instance (set via conf_vars in tests).
514 """
515 secrets_client = get_custom_secret_backend()
516 if not secrets_client:
517 return None
518 try:
519 return secrets_client.get_config(config_key)
520 except Exception as e:
521 raise AirflowConfigException(
522 "Cannot retrieve config from alternative secrets backend. "
523 "Make sure it is configured properly and that the Backend "
524 "is accessible.\n"
525 f"{e}"
526 )
528 def __getstate__(self) -> dict[str, Any]:
529 """Return the state of the object as a dictionary for pickling."""
530 return {
531 name: getattr(self, name)
532 for name in [
533 "_sections",
534 "is_validated",
535 "configuration_description",
536 "upgraded_values",
537 "_default_values",
538 ]
539 }
541 def __setstate__(self, state) -> None:
542 """Restore the state of the object from a dictionary representation."""
543 self.__init__() # type: ignore[misc]
544 config = state.pop("_sections")
545 self.read_dict(config)
546 self.__dict__.update(state)
549def get_airflow_home() -> str:
550 """Get path to Airflow Home."""
551 return expand_env_var(os.environ.get("AIRFLOW_HOME", "~/airflow"))
554def get_airflow_config(airflow_home: str) -> str:
555 """Get Path to airflow.cfg path."""
556 airflow_config_var = os.environ.get("AIRFLOW_CONFIG")
557 if airflow_config_var is None:
558 return os.path.join(airflow_home, "airflow.cfg")
559 return expand_env_var(airflow_config_var)
562def get_all_expansion_variables() -> dict[str, Any]:
563 return {
564 "FERNET_KEY": _SecretKeys.fernet_key,
565 "JWT_SECRET_KEY": _SecretKeys.jwt_secret_key,
566 **{
567 k: v
568 for k, v in globals().items()
569 if not k.startswith("_") and not callable(v) and not ismodule(v)
570 },
571 }
574def _generate_fernet_key() -> str:
575 from cryptography.fernet import Fernet
577 return Fernet.generate_key().decode()
580def create_default_config_parser(configuration_description: dict[str, dict[str, Any]]) -> ConfigParser:
581 """
582 Create default config parser based on configuration description.
584 It creates ConfigParser with all default values retrieved from the configuration description and
585 expands all the variables from the global and local variables defined in this module.
587 :param configuration_description: configuration description - retrieved from config.yaml files
588 following the schema defined in "config.yml.schema.json" in the config_templates folder.
589 :return: Default Config Parser that can be used to read configuration values from.
590 """
591 parser = ConfigParser()
592 all_vars = get_all_expansion_variables()
593 configure_parser_from_configuration_description(parser, configuration_description, all_vars)
594 return parser
597def write_default_airflow_configuration_if_needed() -> AirflowConfigParser:
598 airflow_config = pathlib.Path(AIRFLOW_CONFIG)
599 if airflow_config.is_dir():
600 msg = (
601 "Airflow config expected to be a path to the configuration file, "
602 f"but got a directory {airflow_config.__fspath__()!r}."
603 )
604 raise IsADirectoryError(msg)
605 if not airflow_config.exists():
606 log.debug("Creating new Airflow config file in: %s", airflow_config.__fspath__())
607 config_directory = airflow_config.parent
608 if not config_directory.exists():
609 if not config_directory.is_relative_to(AIRFLOW_HOME):
610 msg = (
611 f"Config directory {config_directory.__fspath__()!r} not exists "
612 f"and it is not relative to AIRFLOW_HOME {AIRFLOW_HOME!r}. "
613 "Please create this directory first."
614 )
615 raise FileNotFoundError(msg) from None
616 log.debug("Create directory %r for Airflow config", config_directory.__fspath__())
617 config_directory.mkdir(parents=True, exist_ok=True)
618 if not conf.get("core", "fernet_key"):
619 # We know that fernet_key is not set, so we can generate it, set as global key
620 # and also write it to the config file so that same key will be used next time
621 _SecretKeys.fernet_key = _generate_fernet_key()
622 conf._configuration_description["core"]["options"]["fernet_key"]["default"] = (
623 _SecretKeys.fernet_key
624 )
625 conf._default_values.set("core", "fernet_key", _SecretKeys.fernet_key)
627 _SecretKeys.jwt_secret_key = b64encode(os.urandom(16)).decode("utf-8")
628 conf._configuration_description["api_auth"]["options"]["jwt_secret"]["default"] = (
629 _SecretKeys.jwt_secret_key
630 )
631 conf._default_values.set("api_auth", "jwt_secret", _SecretKeys.jwt_secret_key)
632 # Invalidate cached configuration_description so it recomputes with the updated base
633 conf.invalidate_cache()
634 pathlib.Path(airflow_config.__fspath__()).touch()
635 make_group_other_inaccessible(airflow_config.__fspath__())
636 with open(airflow_config, "w") as file:
637 conf.write(
638 file,
639 include_sources=False,
640 include_env_vars=True,
641 include_providers=True,
642 extra_spacing=True,
643 only_defaults=True,
644 show_values=True,
645 )
646 return conf
649def load_standard_airflow_configuration(airflow_config_parser: AirflowConfigParser):
650 """
651 Load standard airflow configuration.
653 In case it finds that the configuration file is missing, it will create it and write the default
654 configuration values there, based on defaults passed, and will add the comments and examples
655 from the default configuration.
657 :param airflow_config_parser: parser to which the configuration will be loaded
659 """
660 global AIRFLOW_HOME # to be cleaned in Airflow 4.0
661 log.info("Reading the config from %s", AIRFLOW_CONFIG)
662 airflow_config_parser.read(AIRFLOW_CONFIG)
663 if airflow_config_parser.has_option("core", "AIRFLOW_HOME"):
664 msg = (
665 "Specifying both AIRFLOW_HOME environment variable and airflow_home "
666 "in the config file is deprecated. Please use only the AIRFLOW_HOME "
667 "environment variable and remove the config file entry."
668 )
669 if "AIRFLOW_HOME" in os.environ:
670 warnings.warn(msg, category=RemovedInAirflow4Warning, stacklevel=1)
671 elif airflow_config_parser.get("core", "airflow_home") == AIRFLOW_HOME:
672 warnings.warn(
673 "Specifying airflow_home in the config file is deprecated. As you "
674 "have left it at the default value you should remove the setting "
675 "from your airflow.cfg and suffer no change in behaviour.",
676 category=RemovedInAirflow4Warning,
677 stacklevel=1,
678 )
679 else:
680 AIRFLOW_HOME = airflow_config_parser.get("core", "airflow_home")
681 warnings.warn(msg, category=RemovedInAirflow4Warning, stacklevel=1)
684def initialize_config() -> AirflowConfigParser:
685 """
686 Load the Airflow config files.
688 Called for you automatically as part of the Airflow boot process.
689 """
690 airflow_config_parser = AirflowConfigParser()
691 if airflow_config_parser.getboolean("core", "unit_test_mode"):
692 airflow_config_parser.load_test_config()
693 else:
694 load_standard_airflow_configuration(airflow_config_parser)
695 # If the user set unit_test_mode in the airflow.cfg, we still
696 # want to respect that and then load the default unit test configuration
697 # file on top of it.
698 if airflow_config_parser.getboolean("core", "unit_test_mode"):
699 airflow_config_parser.load_test_config()
700 return airflow_config_parser
703def make_group_other_inaccessible(file_path: str):
704 try:
705 permissions = os.stat(file_path)
706 os.chmod(file_path, permissions.st_mode & (stat.S_IRUSR | stat.S_IWUSR))
707 except Exception as e:
708 log.warning(
709 "Could not change permissions of config file to be group/other inaccessible. "
710 "Continuing with original permissions: %s",
711 e,
712 )
715def ensure_secrets_loaded(
716 default_backends: list[str] = DEFAULT_SECRETS_SEARCH_PATH,
717) -> list[BaseSecretsBackend]:
718 """
719 Ensure that all secrets backends are loaded.
721 If the secrets_backend_list contains only 2 default backends, reload it.
722 """
723 # Check if the secrets_backend_list contains only 2 default backends.
725 # Check if we are loading the backends for worker too by checking if the default_backends is equal
726 # to DEFAULT_SECRETS_SEARCH_PATH.
727 if len(secrets_backend_list) == 2 or default_backends != DEFAULT_SECRETS_SEARCH_PATH:
728 return initialize_secrets_backends(default_backends=default_backends)
729 return secrets_backend_list
732def get_custom_secret_backend(worker_mode: bool = False) -> BaseSecretsBackend | None:
733 """
734 Get Secret Backend if defined in airflow.cfg.
736 Conditionally selects the section, key and kwargs key based on whether it is called from worker or not.
738 This is a convenience function that calls conf._get_custom_secret_backend().
739 """
740 return conf._get_custom_secret_backend(worker_mode=worker_mode)
743def initialize_secrets_backends(
744 default_backends: list[str] = DEFAULT_SECRETS_SEARCH_PATH,
745) -> list[BaseSecretsBackend]:
746 """
747 Initialize secrets backend.
749 * import secrets backend classes
750 * instantiate them and return them in a list
751 """
752 backend_list = []
753 worker_mode = False
754 if default_backends != DEFAULT_SECRETS_SEARCH_PATH:
755 worker_mode = True
757 custom_secret_backend = get_custom_secret_backend(worker_mode)
759 if custom_secret_backend is not None:
760 from airflow.models import Connection
762 custom_secret_backend._set_connection_class(Connection)
763 backend_list.append(custom_secret_backend)
765 for class_name in default_backends:
766 from airflow.models import Connection
768 secrets_backend_cls = import_string(class_name)
769 backend = secrets_backend_cls()
770 backend._set_connection_class(Connection)
771 backend_list.append(backend)
773 return backend_list
776def initialize_auth_manager() -> BaseAuthManager:
777 """
778 Initialize auth manager.
780 * import user manager class
781 * instantiate it and return it
782 """
783 auth_manager_cls = conf.getimport(section="core", key="auth_manager")
785 if not auth_manager_cls:
786 raise AirflowConfigException(
787 "No auth manager defined in the config. Please specify one using section/key [core/auth_manager]."
788 )
790 return auth_manager_cls()
793# Setting AIRFLOW_HOME and AIRFLOW_CONFIG from environment variables, using
794# "~/airflow" and "$AIRFLOW_HOME/airflow.cfg" respectively as defaults.
795AIRFLOW_HOME = get_airflow_home()
796AIRFLOW_CONFIG = get_airflow_config(AIRFLOW_HOME)
798# Set up dags folder for unit tests
799# this directory won't exist if users install via pip
800_TEST_DAGS_FOLDER = os.path.join(
801 os.path.dirname(os.path.dirname(os.path.realpath(__file__))), "tests", "dags"
802)
803if os.path.exists(_TEST_DAGS_FOLDER):
804 TEST_DAGS_FOLDER = _TEST_DAGS_FOLDER
805else:
806 TEST_DAGS_FOLDER = os.path.join(AIRFLOW_HOME, "dags")
808# Set up plugins folder for unit tests
809_TEST_PLUGINS_FOLDER = os.path.join(
810 os.path.dirname(os.path.dirname(os.path.realpath(__file__))), "tests", "plugins"
811)
812if os.path.exists(_TEST_PLUGINS_FOLDER):
813 TEST_PLUGINS_FOLDER = _TEST_PLUGINS_FOLDER
814else:
815 TEST_PLUGINS_FOLDER = os.path.join(AIRFLOW_HOME, "plugins")
817SECRET_KEY = b64encode(os.urandom(16)).decode("utf-8")
819conf: AirflowConfigParser = initialize_config()
820secrets_backend_list = initialize_secrets_backends()
821conf.validate()