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