You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by po...@apache.org on 2022/05/16 21:27:35 UTC
[airflow] branch v2-3-test updated: Add typing for airflow/configuration.py (#23716)
This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch v2-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v2-3-test by this push:
new 5e0c9896d2 Add typing for airflow/configuration.py (#23716)
5e0c9896d2 is described below
commit 5e0c9896d284355f0516aeb65a4d8c829c6e01b6
Author: Jarek Potiuk <ja...@polidea.com>
AuthorDate: Mon May 16 23:19:35 2022 +0200
Add typing for airflow/configuration.py (#23716)
* Add typing for airflow/configuration.py
The configuraiton.py did not have typing information and it made
it rather difficult to reason about it-especially that it went
a few changes in the past that made it rather complex to
understand.
This PR adds typing information all over the configuration file
(cherry picked from commit 71e4deb1b093b7ad9320eb5eb34eca8ea440a238)
---
airflow/api/client/__init__.py | 2 +-
airflow/config_templates/airflow_local_settings.py | 46 +--
airflow/config_templates/default_celery.py | 6 +-
airflow/configuration.py | 342 ++++++++++++++-------
airflow/dag_processing/manager.py | 5 +-
airflow/executors/executor_loader.py | 3 +-
airflow/jobs/scheduler_job.py | 7 +-
airflow/models/abstractoperator.py | 6 +-
airflow/models/dag.py | 6 +-
airflow/models/taskinstance.py | 2 +-
airflow/providers/apache/hive/operators/hive.py | 6 +-
.../providers/apache/spark/hooks/spark_submit.py | 2 +-
airflow/providers/qubole/hooks/qubole.py | 2 +-
airflow/security/kerberos.py | 20 +-
airflow/sensors/base.py | 2 +-
airflow/settings.py | 4 +-
airflow/utils/email.py | 8 +-
airflow/utils/file.py | 2 +-
airflow/utils/helpers.py | 2 +-
airflow/utils/mixins.py | 2 +-
airflow/utils/sqlalchemy.py | 2 +-
tests/cli/commands/test_dag_processor_command.py | 2 +-
tests/cli/commands/test_task_command.py | 2 +-
tests/test_utils/system_tests_class.py | 15 +-
24 files changed, 313 insertions(+), 183 deletions(-)
diff --git a/airflow/api/client/__init__.py b/airflow/api/client/__init__.py
index b7f9c78d30..49224b5336 100644
--- a/airflow/api/client/__init__.py
+++ b/airflow/api/client/__init__.py
@@ -26,7 +26,7 @@ from airflow.configuration import conf
def get_current_api_client() -> Client:
"""Return current API Client based on current Airflow configuration"""
- api_module = import_module(conf.get('cli', 'api_client')) # type: Any
+ api_module = import_module(conf.get_mandatory_value('cli', 'api_client')) # type: Any
auth_backends = api.load_auth()
session = None
for backend in auth_backends:
diff --git a/airflow/config_templates/airflow_local_settings.py b/airflow/config_templates/airflow_local_settings.py
index 14fa529991..b2752c2be7 100644
--- a/airflow/config_templates/airflow_local_settings.py
+++ b/airflow/config_templates/airflow_local_settings.py
@@ -19,7 +19,7 @@
import os
from pathlib import Path
-from typing import Any, Dict, Union
+from typing import Any, Dict, Optional, Union
from urllib.parse import urlparse
from airflow.configuration import conf
@@ -29,30 +29,32 @@ from airflow.exceptions import AirflowException
# in this file instead of from airflow.cfg. Currently
# there are other log format and level configurations in
# settings.py and cli.py. Please see AIRFLOW-1455.
-LOG_LEVEL: str = conf.get('logging', 'LOGGING_LEVEL').upper()
+LOG_LEVEL: str = conf.get_mandatory_value('logging', 'LOGGING_LEVEL').upper()
# Flask appbuilder's info level log is very verbose,
# so it's set to 'WARN' by default.
-FAB_LOG_LEVEL: str = conf.get('logging', 'FAB_LOGGING_LEVEL').upper()
+FAB_LOG_LEVEL: str = conf.get_mandatory_value('logging', 'FAB_LOGGING_LEVEL').upper()
-LOG_FORMAT: str = conf.get('logging', 'LOG_FORMAT')
+LOG_FORMAT: str = conf.get_mandatory_value('logging', 'LOG_FORMAT')
-COLORED_LOG_FORMAT: str = conf.get('logging', 'COLORED_LOG_FORMAT')
+COLORED_LOG_FORMAT: str = conf.get_mandatory_value('logging', 'COLORED_LOG_FORMAT')
COLORED_LOG: bool = conf.getboolean('logging', 'COLORED_CONSOLE_LOG')
-COLORED_FORMATTER_CLASS: str = conf.get('logging', 'COLORED_FORMATTER_CLASS')
+COLORED_FORMATTER_CLASS: str = conf.get_mandatory_value('logging', 'COLORED_FORMATTER_CLASS')
-BASE_LOG_FOLDER: str = conf.get('logging', 'BASE_LOG_FOLDER')
+BASE_LOG_FOLDER: str = conf.get_mandatory_value('logging', 'BASE_LOG_FOLDER')
-PROCESSOR_LOG_FOLDER: str = conf.get('scheduler', 'CHILD_PROCESS_LOG_DIRECTORY')
+PROCESSOR_LOG_FOLDER: str = conf.get_mandatory_value('scheduler', 'CHILD_PROCESS_LOG_DIRECTORY')
-DAG_PROCESSOR_MANAGER_LOG_LOCATION: str = conf.get('logging', 'DAG_PROCESSOR_MANAGER_LOG_LOCATION')
+DAG_PROCESSOR_MANAGER_LOG_LOCATION: str = conf.get_mandatory_value(
+ 'logging', 'DAG_PROCESSOR_MANAGER_LOG_LOCATION'
+)
-FILENAME_TEMPLATE: str = conf.get('logging', 'LOG_FILENAME_TEMPLATE')
+FILENAME_TEMPLATE: str = conf.get_mandatory_value('logging', 'LOG_FILENAME_TEMPLATE')
-PROCESSOR_FILENAME_TEMPLATE: str = conf.get('logging', 'LOG_PROCESSOR_FILENAME_TEMPLATE')
+PROCESSOR_FILENAME_TEMPLATE: str = conf.get_mandatory_value('logging', 'LOG_PROCESSOR_FILENAME_TEMPLATE')
DEFAULT_LOGGING_CONFIG: Dict[str, Any] = {
'version': 1,
@@ -116,7 +118,7 @@ DEFAULT_LOGGING_CONFIG: Dict[str, Any] = {
},
}
-EXTRA_LOGGER_NAMES: str = conf.get('logging', 'EXTRA_LOGGER_NAMES', fallback=None)
+EXTRA_LOGGER_NAMES: Optional[str] = conf.get('logging', 'EXTRA_LOGGER_NAMES', fallback=None)
if EXTRA_LOGGER_NAMES:
new_loggers = {
logger_name.strip(): {
@@ -171,7 +173,7 @@ REMOTE_LOGGING: bool = conf.getboolean('logging', 'remote_logging')
if REMOTE_LOGGING:
- ELASTICSEARCH_HOST: str = conf.get('elasticsearch', 'HOST')
+ ELASTICSEARCH_HOST: Optional[str] = conf.get('elasticsearch', 'HOST')
# Storage bucket URL for remote logging
# S3 buckets should start with "s3://"
@@ -179,7 +181,7 @@ if REMOTE_LOGGING:
# GCS buckets should start with "gs://"
# WASB buckets should start with "wasb"
# just to help Airflow select correct handler
- REMOTE_BASE_LOG_FOLDER: str = conf.get('logging', 'REMOTE_BASE_LOG_FOLDER')
+ REMOTE_BASE_LOG_FOLDER: str = conf.get_mandatory_value('logging', 'REMOTE_BASE_LOG_FOLDER')
if REMOTE_BASE_LOG_FOLDER.startswith('s3://'):
S3_REMOTE_HANDLERS: Dict[str, Dict[str, str]] = {
@@ -207,7 +209,7 @@ if REMOTE_LOGGING:
DEFAULT_LOGGING_CONFIG['handlers'].update(CLOUDWATCH_REMOTE_HANDLERS)
elif REMOTE_BASE_LOG_FOLDER.startswith('gs://'):
- key_path = conf.get('logging', 'GOOGLE_KEY_PATH', fallback=None)
+ key_path = conf.get_mandatory_value('logging', 'GOOGLE_KEY_PATH', fallback=None)
GCS_REMOTE_HANDLERS: Dict[str, Dict[str, str]] = {
'task': {
'class': 'airflow.providers.google.cloud.log.gcs_task_handler.GCSTaskHandler',
@@ -235,7 +237,7 @@ if REMOTE_LOGGING:
DEFAULT_LOGGING_CONFIG['handlers'].update(WASB_REMOTE_HANDLERS)
elif REMOTE_BASE_LOG_FOLDER.startswith('stackdriver://'):
- key_path = conf.get('logging', 'GOOGLE_KEY_PATH', fallback=None)
+ key_path = conf.get_mandatory_value('logging', 'GOOGLE_KEY_PATH', fallback=None)
# stackdriver:///airflow-tasks => airflow-tasks
log_name = urlparse(REMOTE_BASE_LOG_FOLDER).path[1:]
STACKDRIVER_REMOTE_HANDLERS = {
@@ -260,14 +262,14 @@ if REMOTE_LOGGING:
}
DEFAULT_LOGGING_CONFIG['handlers'].update(OSS_REMOTE_HANDLERS)
elif ELASTICSEARCH_HOST:
- ELASTICSEARCH_LOG_ID_TEMPLATE: str = conf.get('elasticsearch', 'LOG_ID_TEMPLATE')
- ELASTICSEARCH_END_OF_LOG_MARK: str = conf.get('elasticsearch', 'END_OF_LOG_MARK')
- ELASTICSEARCH_FRONTEND: str = conf.get('elasticsearch', 'frontend')
+ ELASTICSEARCH_LOG_ID_TEMPLATE: str = conf.get_mandatory_value('elasticsearch', 'LOG_ID_TEMPLATE')
+ ELASTICSEARCH_END_OF_LOG_MARK: str = conf.get_mandatory_value('elasticsearch', 'END_OF_LOG_MARK')
+ ELASTICSEARCH_FRONTEND: str = conf.get_mandatory_value('elasticsearch', 'frontend')
ELASTICSEARCH_WRITE_STDOUT: bool = conf.getboolean('elasticsearch', 'WRITE_STDOUT')
ELASTICSEARCH_JSON_FORMAT: bool = conf.getboolean('elasticsearch', 'JSON_FORMAT')
- ELASTICSEARCH_JSON_FIELDS: str = conf.get('elasticsearch', 'JSON_FIELDS')
- ELASTICSEARCH_HOST_FIELD: str = conf.get('elasticsearch', 'HOST_FIELD')
- ELASTICSEARCH_OFFSET_FIELD: str = conf.get('elasticsearch', 'OFFSET_FIELD')
+ ELASTICSEARCH_JSON_FIELDS: str = conf.get_mandatory_value('elasticsearch', 'JSON_FIELDS')
+ ELASTICSEARCH_HOST_FIELD: str = conf.get_mandatory_value('elasticsearch', 'HOST_FIELD')
+ ELASTICSEARCH_OFFSET_FIELD: str = conf.get_mandatory_value('elasticsearch', 'OFFSET_FIELD')
ELASTIC_REMOTE_HANDLERS: Dict[str, Dict[str, Union[str, bool]]] = {
'task': {
diff --git a/airflow/config_templates/default_celery.py b/airflow/config_templates/default_celery.py
index bbc8aa6a7f..9d81c6353f 100644
--- a/airflow/config_templates/default_celery.py
+++ b/airflow/config_templates/default_celery.py
@@ -59,14 +59,14 @@ except AirflowConfigException:
try:
if celery_ssl_active:
- if 'amqp://' in broker_url:
+ if broker_url and 'amqp://' in broker_url:
broker_use_ssl = {
'keyfile': conf.get('celery', 'SSL_KEY'),
'certfile': conf.get('celery', 'SSL_CERT'),
'ca_certs': conf.get('celery', 'SSL_CACERT'),
'cert_reqs': ssl.CERT_REQUIRED,
}
- elif 'redis://' in broker_url:
+ elif broker_url and 'redis://' in broker_url:
broker_use_ssl = {
'ssl_keyfile': conf.get('celery', 'SSL_KEY'),
'ssl_certfile': conf.get('celery', 'SSL_CERT'),
@@ -92,7 +92,7 @@ except Exception as e:
f'all necessary certs and key ({e}).'
)
-result_backend = DEFAULT_CELERY_CONFIG['result_backend']
+result_backend = str(DEFAULT_CELERY_CONFIG['result_backend'])
if 'amqp://' in result_backend or 'redis://' in result_backend or 'rpc://' in result_backend:
log.warning(
"You have configured a result_backend of %s, it is highly recommended "
diff --git a/airflow/configuration.py b/airflow/configuration.py
index 2b9bf6a6d2..91b8a4e94e 100644
--- a/airflow/configuration.py
+++ b/airflow/configuration.py
@@ -14,7 +14,6 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-
import datetime
import functools
import json
@@ -34,9 +33,12 @@ from collections import OrderedDict
from configparser import _UNSET, ConfigParser, NoOptionError, NoSectionError # type: ignore
from contextlib import suppress
from json.decoder import JSONDecodeError
-from typing import Any, Dict, List, Optional, Tuple, Union
+from re import Pattern
+from typing import IO, Any, Dict, Iterable, List, Optional, Set, Tuple, Union
from urllib.parse import urlparse
+from typing_extensions import overload
+
from airflow.exceptions import AirflowConfigException
from airflow.secrets import DEFAULT_SECRETS_SEARCH_PATH, BaseSecretsBackend
from airflow.utils import yaml
@@ -52,6 +54,11 @@ if not sys.warnoptions:
_SQLITE3_VERSION_PATTERN = re.compile(r"(?P<version>^\d+(?:\.\d+)*)\D?.*$")
+ConfigType = Union[str, int, float, bool]
+ConfigOptionsDictType = Dict[str, ConfigType]
+ConfigSectionSourcesType = Dict[str, Union[str, Tuple[str, str]]]
+ConfigSourcesType = Dict[str, ConfigSectionSourcesType]
+
def _parse_sqlite_version(s: str) -> Tuple[int, ...]:
match = _SQLITE3_VERSION_PATTERN.match(s)
@@ -60,7 +67,17 @@ def _parse_sqlite_version(s: str) -> Tuple[int, ...]:
return tuple(int(p) for p in match.group("version").split("."))
-def expand_env_var(env_var):
+@overload
+def expand_env_var(env_var: None) -> None:
+ ...
+
+
+@overload
+def expand_env_var(env_var: str) -> str:
+ ...
+
+
+def expand_env_var(env_var: Union[str, None]) -> Optional[Union[str, None]]:
"""
Expands (potentially nested) env vars by repeatedly applying
`expandvars` and `expanduser` until interpolation stops having
@@ -76,7 +93,7 @@ def expand_env_var(env_var):
env_var = interpolated
-def run_command(command):
+def run_command(command: str) -> str:
"""Runs command and returns stdout"""
process = subprocess.Popen(
shlex.split(command), stdout=subprocess.PIPE, stderr=subprocess.PIPE, close_fds=True
@@ -92,7 +109,7 @@ def run_command(command):
return output
-def _get_config_value_from_secret_backend(config_key):
+def _get_config_value_from_secret_backend(config_key: str) -> Optional[str]:
"""Get Config option values from Secret Backend"""
try:
secrets_client = get_custom_secret_backend()
@@ -108,7 +125,7 @@ def _get_config_value_from_secret_backend(config_key):
)
-def _default_config_file_path(file_name: str):
+def _default_config_file_path(file_name: str) -> str:
templates_dir = os.path.join(os.path.dirname(__file__), 'config_templates')
return os.path.join(templates_dir, file_name)
@@ -131,7 +148,7 @@ class AirflowConfigParser(ConfigParser):
# is to not store password on boxes in text files.
# These configs can also be fetched from Secrets backend
# following the "{section}__{name}__secret" pattern
- sensitive_config_values = {
+ sensitive_config_values: Set[Tuple[str, str]] = {
('database', 'sql_alchemy_conn'),
('core', 'fernet_key'),
('celery', 'broker_url'),
@@ -147,7 +164,7 @@ class AirflowConfigParser(ConfigParser):
# A mapping of (new section, new option) -> (old section, old option, since_version).
# When reading new option, the old option will be checked to see if it exists. If it does a
# DeprecationWarning will be issued and the old option will be used instead
- deprecated_options = {
+ deprecated_options: Dict[Tuple[str, str], Tuple[str, str, str]] = {
('celery', 'worker_precheck'): ('core', 'worker_precheck', '2.0.0'),
('logging', 'base_log_folder'): ('core', 'base_log_folder', '2.0.0'),
('logging', 'remote_logging'): ('core', 'remote_logging', '2.0.0'),
@@ -206,7 +223,7 @@ class AirflowConfigParser(ConfigParser):
# A mapping of old default values that we want to change and warn the user
# about. Mapping of section -> setting -> { old, replace, by_version }
- deprecated_values = {
+ deprecated_values: Dict[str, Dict[str, Tuple[Pattern, str, str]]] = {
'core': {
'hostname_callable': (re.compile(r':'), r'.', '2.1'),
},
@@ -225,7 +242,7 @@ class AirflowConfigParser(ConfigParser):
'log_filename_template': (
re.compile(re.escape("{{ ti.dag_id }}/{{ ti.task_id }}/{{ ts }}/{{ try_number }}.log")),
"XX-set-after-default-config-loaded-XX",
- 3.0,
+ '3.0',
),
},
'api': {
@@ -239,7 +256,7 @@ class AirflowConfigParser(ConfigParser):
'log_id_template': (
re.compile('^' + re.escape('{dag_id}-{task_id}-{run_id}-{try_number}') + '$'),
'{dag_id}-{task_id}-{run_id}-{map_index}-{try_number}',
- 3.0,
+ '3.0',
)
},
}
@@ -260,12 +277,12 @@ class AirflowConfigParser(ConfigParser):
"""Mapping of (section,option) to the old value that was upgraded"""
# This method transforms option names on every read, get, or set operation.
- # This changes from the default behaviour of ConfigParser from lowercasing
+ # This changes from the default behaviour of ConfigParser from lower-casing
# to instead be case-preserving
def optionxform(self, optionstr: str) -> str:
return optionstr
- def __init__(self, default_config=None, *args, **kwargs):
+ def __init__(self, default_config: Optional[str] = None, *args, **kwargs):
super().__init__(*args, **kwargs)
self.upgraded_values = {}
@@ -392,7 +409,7 @@ class AirflowConfigParser(ConfigParser):
from airflow.utils.docs import get_docs_url
- # Some of the features in storing rendered fields require sqlite version >= 3.15.0
+ # Some features in storing rendered fields require sqlite version >= 3.15.0
min_sqlite_version = (3, 15, 0)
if _parse_sqlite_version(sqlite3.sqlite_version) < min_sqlite_version:
min_sqlite_version_str = ".".join(str(s) for s in min_sqlite_version)
@@ -401,16 +418,16 @@ class AirflowConfigParser(ConfigParser):
f"See {get_docs_url('howto/set-up-database.html#setting-up-a-sqlite-database')}"
)
- def _using_old_value(self, old, current_value):
+ def _using_old_value(self, old: Pattern, current_value: str) -> bool:
return old.search(current_value) is not None
- def _update_env_var(self, section, name, new_value):
+ def _update_env_var(self, section: str, name: str, new_value: Union[str]):
env_var = self._env_var_name(section, name)
# Set it as an env var so that any subprocesses keep the same override!
os.environ[env_var] = new_value
@staticmethod
- def _create_future_warning(name, section, current_value, new_value, version):
+ def _create_future_warning(name: str, section: str, current_value: Any, new_value: Any, version: str):
warnings.warn(
f'The {name!r} setting in [{section}] has the old default value of {current_value!r}. '
f'This value has been changed to {new_value!r} in the running config, but '
@@ -423,7 +440,7 @@ class AirflowConfigParser(ConfigParser):
def _env_var_name(self, section: str, key: str) -> str:
return f'{self.ENV_VAR_PREFIX}{section.upper()}__{key.upper()}'
- def _get_env_var_option(self, section, key):
+ def _get_env_var_option(self, section: str, key: str):
# must have format AIRFLOW__{SECTION}__{KEY} (note double underscore)
env_var = self._env_var_name(section, key)
if env_var in os.environ:
@@ -442,7 +459,7 @@ class AirflowConfigParser(ConfigParser):
return _get_config_value_from_secret_backend(os.environ[env_var_secret_path])
return None
- def _get_cmd_option(self, section, key):
+ def _get_cmd_option(self, section: str, key: str):
fallback_key = key + '_cmd'
# if this is a valid command key...
if (section, key) in self.sensitive_config_values:
@@ -451,7 +468,7 @@ class AirflowConfigParser(ConfigParser):
return run_command(command)
return None
- def _get_secret_option(self, section, key):
+ def _get_secret_option(self, section: str, key: str) -> Optional[str]:
"""Get Config option values from Secret Backend"""
fallback_key = key + '_secret'
# if this is a valid secret key...
@@ -461,7 +478,13 @@ class AirflowConfigParser(ConfigParser):
return _get_config_value_from_secret_backend(secrets_path)
return None
- def get(self, section, key, **kwargs):
+ def get_mandatory_value(self, section: str, key: str, **kwargs) -> str:
+ value = self.get(section, key, **kwargs)
+ if value is None:
+ raise ValueError(f"The value {section}/{key} should be set!")
+ return value
+
+ def get(self, section: str, key: str, **kwargs) -> Optional[str]: # type: ignore[override]
section = str(section).lower()
key = str(key).lower()
@@ -487,7 +510,7 @@ class AirflowConfigParser(ConfigParser):
return self._get_option_from_default_config(section, key, **kwargs)
- def _get_option_from_default_config(self, section, key, **kwargs):
+ def _get_option_from_default_config(self, section: str, key: str, **kwargs) -> Optional[str]:
# ...then the default config
if self.airflow_defaults.has_option(section, key) or 'fallback' in kwargs:
return expand_env_var(self.airflow_defaults.get(section, key, **kwargs))
@@ -497,55 +520,68 @@ class AirflowConfigParser(ConfigParser):
raise AirflowConfigException(f"section/key [{section}/{key}] not found in config")
- def _get_option_from_secrets(self, deprecated_key, deprecated_section, key, section):
+ def _get_option_from_secrets(
+ self, deprecated_key: Optional[str], deprecated_section: Optional[str], key: str, section: str
+ ) -> Optional[str]:
# ...then from secret backends
option = self._get_secret_option(section, key)
if option:
return option
- if deprecated_section:
+ if deprecated_section and deprecated_key:
option = self._get_secret_option(deprecated_section, deprecated_key)
if option:
self._warn_deprecate(section, key, deprecated_section, deprecated_key)
return option
return None
- def _get_option_from_commands(self, deprecated_key, deprecated_section, key, section):
+ def _get_option_from_commands(
+ self, deprecated_key: Optional[str], deprecated_section: Optional[str], key: str, section: str
+ ) -> Optional[str]:
# ...then commands
option = self._get_cmd_option(section, key)
if option:
return option
- if deprecated_section:
+ if deprecated_section and deprecated_key:
option = self._get_cmd_option(deprecated_section, deprecated_key)
if option:
self._warn_deprecate(section, key, deprecated_section, deprecated_key)
return option
return None
- def _get_option_from_config_file(self, deprecated_key, deprecated_section, key, kwargs, section):
+ def _get_option_from_config_file(
+ self,
+ deprecated_key: Optional[str],
+ deprecated_section: Optional[str],
+ key: str,
+ kwargs: Dict[str, Any],
+ section: str,
+ ) -> Optional[str]:
# ...then the config file
if super().has_option(section, key):
# Use the parent's methods to get the actual config here to be able to
# separate the config from default config.
return expand_env_var(super().get(section, key, **kwargs))
- if deprecated_section:
+ if deprecated_section and deprecated_key:
if super().has_option(deprecated_section, deprecated_key):
self._warn_deprecate(section, key, deprecated_section, deprecated_key)
return expand_env_var(super().get(deprecated_section, deprecated_key, **kwargs))
return None
- def _get_environment_variables(self, deprecated_key, deprecated_section, key, section):
+ def _get_environment_variables(
+ self, deprecated_key: Optional[str], deprecated_section: Optional[str], key: str, section: str
+ ) -> Optional[str]:
# first check environment variables
option = self._get_env_var_option(section, key)
if option is not None:
return option
- if deprecated_section:
+ if deprecated_section and deprecated_key:
option = self._get_env_var_option(deprecated_section, deprecated_key)
if option is not None:
self._warn_deprecate(section, key, deprecated_section, deprecated_key)
return option
return None
- def getboolean(self, section, key, **kwargs):
+ def getboolean(self, section: str, key: str, **kwargs) -> bool: # type: ignore[override]
val = str(self.get(section, key, **kwargs)).lower().strip()
if '#' in val:
val = val.split('#')[0].strip()
@@ -559,9 +595,13 @@ class AirflowConfigParser(ConfigParser):
f'Current value: "{val}".'
)
- def getint(self, section, key, **kwargs):
+ def getint(self, section: str, key: str, **kwargs) -> int: # type: ignore[override]
val = self.get(section, key, **kwargs)
-
+ if val is None:
+ raise AirflowConfigException(
+ f'Failed to convert value None to int. '
+ f'Please check "{key}" key in "{section}" section is set.'
+ )
try:
return int(val)
except ValueError:
@@ -570,9 +610,13 @@ class AirflowConfigParser(ConfigParser):
f'Current value: "{val}".'
)
- def getfloat(self, section, key, **kwargs):
+ def getfloat(self, section: str, key: str, **kwargs) -> float: # type: ignore[override]
val = self.get(section, key, **kwargs)
-
+ if val is None:
+ raise AirflowConfigException(
+ f'Failed to convert value None to float. '
+ f'Please check "{key}" key in "{section}" section is set.'
+ )
try:
return float(val)
except ValueError:
@@ -581,7 +625,7 @@ class AirflowConfigParser(ConfigParser):
f'Current value: "{val}".'
)
- def getimport(self, section, key, **kwargs):
+ def getimport(self, section: str, key: str, **kwargs) -> Any:
"""
Reads options, imports the full qualified name, and returns the object.
@@ -602,7 +646,9 @@ class AirflowConfigParser(ConfigParser):
f'Current value: "{full_qualified_path}".'
)
- def getjson(self, section, key, fallback=_UNSET, **kwargs) -> Union[dict, list, str, int, float, None]:
+ def getjson(
+ self, section: str, key: str, fallback=_UNSET, **kwargs
+ ) -> Union[dict, list, str, int, float, None]:
"""
Return a config value parsed from a JSON string.
@@ -620,7 +666,7 @@ class AirflowConfigParser(ConfigParser):
except (NoSectionError, NoOptionError):
return default
- if len(data) == 0:
+ if not data:
return default if default is not _UNSET else None
try:
@@ -628,7 +674,9 @@ class AirflowConfigParser(ConfigParser):
except JSONDecodeError as e:
raise AirflowConfigException(f'Unable to parse [{section}] {key!r} as valid json') from e
- def gettimedelta(self, section, key, fallback=None, **kwargs) -> Optional[datetime.timedelta]:
+ def gettimedelta(
+ self, section: str, key: str, fallback: Any = None, **kwargs
+ ) -> Optional[datetime.timedelta]:
"""
Gets the config value for the given section and key, and converts it into datetime.timedelta object.
If the key is missing, then it is considered as `None`.
@@ -662,13 +710,26 @@ class AirflowConfigParser(ConfigParser):
return fallback
- def read(self, filenames, encoding=None):
+ def read(
+ self,
+ filenames: Union[
+ str,
+ bytes,
+ os.PathLike,
+ Iterable[Union[str, bytes, os.PathLike]],
+ ],
+ encoding=None,
+ ):
super().read(filenames=filenames, encoding=encoding)
- def read_dict(self, dictionary, source='<dict>'):
+ # The RawConfigParser defines "Mapping" from abc.collections is not subscriptable - so we have
+ # to use Dict here.
+ def read_dict( # type: ignore[override]
+ self, dictionary: Dict[str, Dict[str, Any]], source: str = '<dict>'
+ ):
super().read_dict(dictionary=dictionary, source=source)
- def has_option(self, section, option):
+ def has_option(self, section: str, option: str) -> bool:
try:
# Using self.get() to avoid reimplementing the priority order
# of config variables (env, config, cmd, defaults)
@@ -678,7 +739,7 @@ class AirflowConfigParser(ConfigParser):
except (NoOptionError, NoSectionError):
return False
- def remove_option(self, section, option, remove_default=True):
+ def remove_option(self, section: str, option: str, remove_default: bool = True):
"""
Remove an option if it exists in config from a file or
default config. If both of config have the same option, this removes
@@ -690,7 +751,7 @@ class AirflowConfigParser(ConfigParser):
if self.airflow_defaults.has_option(section, option) and remove_default:
self.airflow_defaults.remove_option(section, option)
- def getsection(self, section: str) -> Optional[Dict[str, Union[str, int, float, bool]]]:
+ def getsection(self, section: str) -> Optional[ConfigOptionsDictType]:
"""
Returns the section as a dict. Values are converted to int, float, bool
as required.
@@ -700,9 +761,8 @@ class AirflowConfigParser(ConfigParser):
"""
if not self.has_section(section) and not self.airflow_defaults.has_section(section):
return None
-
if self.airflow_defaults.has_section(section):
- _section = OrderedDict(self.airflow_defaults.items(section))
+ _section: ConfigOptionsDictType = OrderedDict(self.airflow_defaults.items(section))
else:
_section = OrderedDict()
@@ -719,40 +779,48 @@ class AirflowConfigParser(ConfigParser):
_section[key] = self._get_env_var_option(section, key)
for key, val in _section.items():
+ if val is None:
+ raise AirflowConfigException(
+ f'Failed to convert value automatically. '
+ f'Please check "{key}" key in "{section}" section is set.'
+ )
try:
- val = int(val)
+ _section[key] = int(val)
except ValueError:
try:
- val = float(val)
+ _section[key] = float(val)
except ValueError:
- if val.lower() in ('t', 'true'):
- val = True
- elif val.lower() in ('f', 'false'):
- val = False
- _section[key] = val
+ if isinstance(val, str) and val.lower() in ('t', 'true'):
+ _section[key] = True
+ elif isinstance(val, str) and val.lower() in ('f', 'false'):
+ _section[key] = False
return _section
- def write(self, fp, space_around_delimiters=True):
+ def write(self, fp: IO, space_around_delimiters: bool = True): # type: ignore[override]
# This is based on the configparser.RawConfigParser.write method code to add support for
# reading options from environment variables.
+ # Various type ignores below deal with less-than-perfect RawConfigParser superclass typing
if space_around_delimiters:
- delimiter = f" {self._delimiters[0]} "
+ delimiter = f" {self._delimiters[0]} " # type: ignore[attr-defined]
else:
- delimiter = self._delimiters[0]
- if self._defaults:
- self._write_section(fp, self.default_section, self._defaults.items(), delimiter)
- for section in self._sections:
- self._write_section(fp, section, self.getsection(section).items(), delimiter)
+ delimiter = self._delimiters[0] # type: ignore[attr-defined]
+ if self._defaults: # type: ignore
+ self._write_section( # type: ignore[attr-defined]
+ fp, self.default_section, self._defaults.items(), delimiter # type: ignore[attr-defined]
+ )
+ for section in self._sections: # type: ignore[attr-defined]
+ item_section: ConfigOptionsDictType = self.getsection(section) # type: ignore[assignment]
+ self._write_section(fp, section, item_section.items(), delimiter) # type: ignore[attr-defined]
def as_dict(
self,
- display_source=False,
- display_sensitive=False,
- raw=False,
- include_env=True,
- include_cmds=True,
- include_secret=True,
- ) -> Dict[str, Dict[str, str]]:
+ display_source: bool = False,
+ display_sensitive: bool = False,
+ raw: bool = False,
+ include_env: bool = True,
+ include_cmds: bool = True,
+ include_secret: bool = True,
+ ) -> ConfigSourcesType:
"""
Returns the current configuration as an OrderedDict of OrderedDicts.
@@ -785,7 +853,7 @@ class AirflowConfigParser(ConfigParser):
:return: Dictionary, where the key is the name of the section and the content is
the dictionary with the name of the parameter and its value.
"""
- config_sources: Dict[str, Dict[str, str]] = {}
+ config_sources: ConfigSourcesType = {}
configs = [
('default', self.airflow_defaults),
('airflow.cfg', self),
@@ -813,20 +881,34 @@ class AirflowConfigParser(ConfigParser):
return config_sources
- def _include_secrets(self, config_sources, display_sensitive, display_source, raw):
+ def _include_secrets(
+ self,
+ config_sources: ConfigSourcesType,
+ display_sensitive: bool,
+ display_source: bool,
+ raw: bool,
+ ):
for (section, key) in self.sensitive_config_values:
- opt = self._get_secret_option(section, key)
- if opt:
+ value: Optional[str] = self._get_secret_option(section, key)
+ if value:
if not display_sensitive:
- opt = '< hidden >'
+ value = '< hidden >'
if display_source:
- opt = (opt, 'secret')
+ opt: Union[str, Tuple[str, str]] = (value, 'secret')
elif raw:
- opt = opt.replace('%', '%%')
+ opt = value.replace('%', '%%')
+ else:
+ opt = value
config_sources.setdefault(section, OrderedDict()).update({key: opt})
del config_sources[section][key + '_secret']
- def _include_commands(self, config_sources, display_sensitive, display_source, raw):
+ def _include_commands(
+ self,
+ config_sources: ConfigSourcesType,
+ display_sensitive: bool,
+ display_source: bool,
+ raw: bool,
+ ):
for (section, key) in self.sensitive_config_values:
opt = self._get_cmd_option(section, key)
if not opt:
@@ -840,7 +922,13 @@ class AirflowConfigParser(ConfigParser):
config_sources.setdefault(section, OrderedDict()).update({key: opt})
del config_sources[section][key + '_cmd']
- def _include_envs(self, config_sources, display_sensitive, display_source, raw):
+ def _include_envs(
+ self,
+ config_sources: ConfigSourcesType,
+ display_sensitive: bool,
+ display_source: bool,
+ raw: bool,
+ ):
for env_var in [
os_environment for os_environment in os.environ if os_environment.startswith(self.ENV_VAR_PREFIX)
]:
@@ -868,7 +956,12 @@ class AirflowConfigParser(ConfigParser):
key = key.lower()
config_sources.setdefault(section, OrderedDict()).update({key: opt})
- def _filter_by_source(self, config_sources, display_source, getter_func):
+ def _filter_by_source(
+ self,
+ config_sources: ConfigSourcesType,
+ display_source: bool,
+ getter_func,
+ ):
"""
Deletes default configs from current configuration (an OrderedDict of
OrderedDicts) if it would conflict with special sensitive_config_values.
@@ -904,14 +997,20 @@ class AirflowConfigParser(ConfigParser):
continue
# Check to see if bare setting is the same as defaults
if display_source:
- opt, source = config_sources[section][key]
+ # when display_source = true, we know that the config_sources contains tuple
+ opt, source = config_sources[section][key] # type: ignore
else:
opt = config_sources[section][key]
if opt == self.airflow_defaults.get(section, key):
del config_sources[section][key]
@staticmethod
- def _replace_config_with_display_sources(config_sources, configs, display_source, raw):
+ def _replace_config_with_display_sources(
+ config_sources: ConfigSourcesType,
+ configs: Iterable[Tuple[str, ConfigParser]],
+ display_source: bool,
+ raw: bool,
+ ):
for (source_name, config) in configs:
for section in config.sections():
AirflowConfigParser._replace_section_config_with_display_sources(
@@ -920,13 +1019,19 @@ class AirflowConfigParser(ConfigParser):
@staticmethod
def _replace_section_config_with_display_sources(
- config, config_sources, display_source, raw, section, source_name
+ config: ConfigParser,
+ config_sources: ConfigSourcesType,
+ display_source: bool,
+ raw: bool,
+ section: str,
+ source_name: str,
):
sect = config_sources.setdefault(section, OrderedDict())
for (k, val) in config.items(section=section, raw=raw):
if display_source:
- val = (val, source_name)
- sect[k] = val
+ sect[k] = (val, source_name)
+ else:
+ sect[k] = val
def load_test_config(self):
"""
@@ -948,7 +1053,7 @@ class AirflowConfigParser(ConfigParser):
self.read(TEST_CONFIG_FILE)
@staticmethod
- def _warn_deprecate(section, key, deprecated_section, deprecated_name):
+ def _warn_deprecate(section: str, key: str, deprecated_section: str, deprecated_name: str):
if section == deprecated_section:
warnings.warn(
f'The {deprecated_name} option in [{section}] has been renamed to {key} - '
@@ -981,16 +1086,17 @@ class AirflowConfigParser(ConfigParser):
self.__dict__.update(state)
-def get_airflow_home():
+def get_airflow_home() -> str:
"""Get path to Airflow Home"""
return expand_env_var(os.environ.get('AIRFLOW_HOME', '~/airflow'))
-def get_airflow_config(airflow_home):
+def get_airflow_config(airflow_home) -> str:
"""Get Path to airflow.cfg path"""
- if 'AIRFLOW_CONFIG' not in os.environ:
+ airflow_config_var = os.environ.get('AIRFLOW_CONFIG')
+ if airflow_config_var is None:
return os.path.join(airflow_home, 'airflow.cfg')
- return expand_env_var(os.environ['AIRFLOW_CONFIG'])
+ return expand_env_var(airflow_config_var)
def _parameterized_config_from_template(filename) -> str:
@@ -1005,7 +1111,7 @@ def _parameterized_config_from_template(filename) -> str:
raise RuntimeError(f"Template marker not found in {path!r}")
-def parameterized_config(template):
+def parameterized_config(template) -> str:
"""
Generates a configuration from the provided template + variables defined in
current scope
@@ -1016,20 +1122,21 @@ def parameterized_config(template):
return template.format(**all_vars)
-def get_airflow_test_config(airflow_home):
+def get_airflow_test_config(airflow_home) -> str:
"""Get path to unittests.cfg"""
if 'AIRFLOW_TEST_CONFIG' not in os.environ:
return os.path.join(airflow_home, 'unittests.cfg')
- return expand_env_var(os.environ['AIRFLOW_TEST_CONFIG'])
+ # It will never return None
+ return expand_env_var(os.environ['AIRFLOW_TEST_CONFIG']) # type: ignore[return-value]
-def _generate_fernet_key():
+def _generate_fernet_key() -> str:
from cryptography.fernet import Fernet
return Fernet.generate_key().decode()
-def initialize_config():
+def initialize_config() -> AirflowConfigParser:
"""
Load the Airflow config files.
@@ -1039,9 +1146,9 @@ def initialize_config():
default_config = _parameterized_config_from_template('default_airflow.cfg')
- conf = AirflowConfigParser(default_config=default_config)
+ local_conf = AirflowConfigParser(default_config=default_config)
- if conf.getboolean('core', 'unit_test_mode'):
+ if local_conf.getboolean('core', 'unit_test_mode'):
# Load test config only
if not os.path.isfile(TEST_CONFIG_FILE):
from cryptography.fernet import Fernet
@@ -1055,7 +1162,7 @@ def initialize_config():
cfg = _parameterized_config_from_template('default_test.cfg')
file.write(cfg)
- conf.load_test_config()
+ local_conf.load_test_config()
else:
# Load normal config
if not os.path.isfile(AIRFLOW_CONFIG):
@@ -1071,9 +1178,9 @@ def initialize_config():
log.info("Reading the config from %s", AIRFLOW_CONFIG)
- conf.read(AIRFLOW_CONFIG)
+ local_conf.read(AIRFLOW_CONFIG)
- if conf.has_option('core', 'AIRFLOW_HOME'):
+ if local_conf.has_option('core', 'AIRFLOW_HOME'):
msg = (
'Specifying both AIRFLOW_HOME environment variable and airflow_home '
'in the config file is deprecated. Please use only the AIRFLOW_HOME '
@@ -1081,7 +1188,7 @@ def initialize_config():
)
if 'AIRFLOW_HOME' in os.environ:
warnings.warn(msg, category=DeprecationWarning)
- elif conf.get('core', 'airflow_home') == AIRFLOW_HOME:
+ elif local_conf.get('core', 'airflow_home') == AIRFLOW_HOME:
warnings.warn(
'Specifying airflow_home in the config file is deprecated. As you '
'have left it at the default value you should remove the setting '
@@ -1089,13 +1196,14 @@ def initialize_config():
category=DeprecationWarning,
)
else:
- AIRFLOW_HOME = conf.get('core', 'airflow_home')
+ # there
+ AIRFLOW_HOME = local_conf.get('core', 'airflow_home') # type: ignore[assignment]
warnings.warn(msg, category=DeprecationWarning)
# They _might_ have set unit_test_mode in the airflow.cfg, we still
# want to respect that and then load the unittests.cfg
- if conf.getboolean('core', 'unit_test_mode'):
- conf.load_test_config()
+ if local_conf.getboolean('core', 'unit_test_mode'):
+ local_conf.load_test_config()
# Make it no longer a proxy variable, just set it to an actual string
global WEBSERVER_CONFIG
@@ -1106,7 +1214,7 @@ def initialize_config():
log.info('Creating new FAB webserver config file in: %s', WEBSERVER_CONFIG)
shutil.copy(_default_config_file_path('default_webserver_config.py'), WEBSERVER_CONFIG)
- return conf
+ return local_conf
# Historical convenience functions to access config entries
@@ -1122,7 +1230,7 @@ def load_test_config():
conf.load_test_config()
-def get(*args, **kwargs):
+def get(*args, **kwargs) -> Optional[ConfigType]:
"""Historical get"""
warnings.warn(
"Accessing configuration method 'get' directly from the configuration module is "
@@ -1134,7 +1242,7 @@ def get(*args, **kwargs):
return conf.get(*args, **kwargs)
-def getboolean(*args, **kwargs):
+def getboolean(*args, **kwargs) -> bool:
"""Historical getboolean"""
warnings.warn(
"Accessing configuration method 'getboolean' directly from the configuration module is "
@@ -1146,7 +1254,7 @@ def getboolean(*args, **kwargs):
return conf.getboolean(*args, **kwargs)
-def getfloat(*args, **kwargs):
+def getfloat(*args, **kwargs) -> float:
"""Historical getfloat"""
warnings.warn(
"Accessing configuration method 'getfloat' directly from the configuration module is "
@@ -1158,7 +1266,7 @@ def getfloat(*args, **kwargs):
return conf.getfloat(*args, **kwargs)
-def getint(*args, **kwargs):
+def getint(*args, **kwargs) -> int:
"""Historical getint"""
warnings.warn(
"Accessing configuration method 'getint' directly from the configuration module is "
@@ -1170,7 +1278,7 @@ def getint(*args, **kwargs):
return conf.getint(*args, **kwargs)
-def getsection(*args, **kwargs):
+def getsection(*args, **kwargs) -> Optional[ConfigOptionsDictType]:
"""Historical getsection"""
warnings.warn(
"Accessing configuration method 'getsection' directly from the configuration module is "
@@ -1182,7 +1290,7 @@ def getsection(*args, **kwargs):
return conf.getsection(*args, **kwargs)
-def has_option(*args, **kwargs):
+def has_option(*args, **kwargs) -> bool:
"""Historical has_option"""
warnings.warn(
"Accessing configuration method 'has_option' directly from the configuration module is "
@@ -1194,7 +1302,7 @@ def has_option(*args, **kwargs):
return conf.has_option(*args, **kwargs)
-def remove_option(*args, **kwargs):
+def remove_option(*args, **kwargs) -> bool:
"""Historical remove_option"""
warnings.warn(
"Accessing configuration method 'remove_option' directly from the configuration module is "
@@ -1206,7 +1314,7 @@ def remove_option(*args, **kwargs):
return conf.remove_option(*args, **kwargs)
-def as_dict(*args, **kwargs):
+def as_dict(*args, **kwargs) -> ConfigSourcesType:
"""Historical as_dict"""
warnings.warn(
"Accessing configuration method 'as_dict' directly from the configuration module is "
@@ -1218,7 +1326,7 @@ def as_dict(*args, **kwargs):
return conf.as_dict(*args, **kwargs)
-def set(*args, **kwargs):
+def set(*args, **kwargs) -> None:
"""Historical set"""
warnings.warn(
"Accessing configuration method 'set' directly from the configuration module is "
@@ -1227,7 +1335,7 @@ def set(*args, **kwargs):
DeprecationWarning,
stacklevel=2,
)
- return conf.set(*args, **kwargs)
+ conf.set(*args, **kwargs)
def ensure_secrets_loaded() -> List[BaseSecretsBackend]:
@@ -1247,9 +1355,8 @@ def get_custom_secret_backend() -> Optional[BaseSecretsBackend]:
if secrets_backend_cls:
try:
- alternative_secrets_config_dict = json.loads(
- conf.get(section='secrets', key='backend_kwargs', fallback='{}')
- )
+ backends: Any = conf.get(section='secrets', key='backend_kwargs', fallback='{}')
+ alternative_secrets_config_dict = json.loads(backends)
except JSONDecodeError:
alternative_secrets_config_dict = {}
@@ -1277,14 +1384,14 @@ def initialize_secrets_backends() -> List[BaseSecretsBackend]:
@functools.lru_cache(maxsize=None)
-def _DEFAULT_CONFIG():
+def _DEFAULT_CONFIG() -> str:
path = _default_config_file_path('default_airflow.cfg')
with open(path) as fh:
return fh.read()
@functools.lru_cache(maxsize=None)
-def _TEST_CONFIG():
+def _TEST_CONFIG() -> str:
path = _default_config_file_path('default_test.cfg')
with open(path) as fh:
return fh.read()
@@ -1311,7 +1418,6 @@ def __getattr__(name):
# Setting AIRFLOW_HOME and AIRFLOW_CONFIG from environment variables, using
# "~/airflow" and "$AIRFLOW_HOME/airflow.cfg" respectively as defaults.
-
AIRFLOW_HOME = get_airflow_home()
AIRFLOW_CONFIG = get_airflow_config(AIRFLOW_HOME)
diff --git a/airflow/dag_processing/manager.py b/airflow/dag_processing/manager.py
index 4cb11cff82..63ab8e6e67 100644
--- a/airflow/dag_processing/manager.py
+++ b/airflow/dag_processing/manager.py
@@ -395,7 +395,10 @@ class DagFileProcessorManager(LoggingMixin):
os.set_blocking(self._direct_scheduler_conn.fileno(), False)
self._parallelism = conf.getint('scheduler', 'parsing_processes')
- if conf.get('database', 'sql_alchemy_conn').startswith('sqlite') and self._parallelism > 1:
+ if (
+ conf.get_mandatory_value('database', 'sql_alchemy_conn').startswith('sqlite')
+ and self._parallelism > 1
+ ):
self.log.warning(
"Because we cannot use more than 1 thread (parsing_processes = "
"%d) when using sqlite. So we set parallelism to 1.",
diff --git a/airflow/executors/executor_loader.py b/airflow/executors/executor_loader.py
index b98d2a8000..723060db0f 100644
--- a/airflow/executors/executor_loader.py
+++ b/airflow/executors/executor_loader.py
@@ -71,8 +71,7 @@ class ExecutorLoader:
from airflow.configuration import conf
- executor_name = conf.get('core', 'EXECUTOR')
-
+ executor_name = conf.get_mandatory_value('core', 'EXECUTOR')
cls._default_executor = cls.load_executor(executor_name)
return cls._default_executor
diff --git a/airflow/jobs/scheduler_job.py b/airflow/jobs/scheduler_job.py
index ac1d25833b..6f366b3987 100644
--- a/airflow/jobs/scheduler_job.py
+++ b/airflow/jobs/scheduler_job.py
@@ -142,7 +142,7 @@ class SchedulerJob(BaseJob):
self._log = log
# Check what SQL backend we use
- sql_conn: str = conf.get('database', 'sql_alchemy_conn').lower()
+ sql_conn: str = conf.get_mandatory_value('database', 'sql_alchemy_conn').lower()
self.using_sqlite = sql_conn.startswith('sqlite')
self.using_mysql = sql_conn.startswith('mysql')
# Dag Processor agent - not used in Dag Processor standalone mode.
@@ -152,7 +152,10 @@ class SchedulerJob(BaseJob):
if conf.getboolean('smart_sensor', 'use_smart_sensor'):
compatible_sensors = set(
- map(lambda l: l.strip(), conf.get('smart_sensor', 'sensors_enabled').split(','))
+ map(
+ lambda l: l.strip(),
+ conf.get_mandatory_value('smart_sensor', 'sensors_enabled').split(','),
+ )
)
docs_url = get_docs_url('concepts/smart-sensors.html#migrating-to-deferrable-operators')
warnings.warn(
diff --git a/airflow/models/abstractoperator.py b/airflow/models/abstractoperator.py
index 2c53a8b75d..8d2e06442a 100644
--- a/airflow/models/abstractoperator.py
+++ b/airflow/models/abstractoperator.py
@@ -57,17 +57,17 @@ if TYPE_CHECKING:
from airflow.models.operator import Operator
from airflow.models.taskinstance import TaskInstance
-DEFAULT_OWNER: str = conf.get("operators", "default_owner")
+DEFAULT_OWNER: str = conf.get_mandatory_value("operators", "default_owner")
DEFAULT_POOL_SLOTS: int = 1
DEFAULT_PRIORITY_WEIGHT: int = 1
-DEFAULT_QUEUE: str = conf.get("operators", "default_queue")
+DEFAULT_QUEUE: str = conf.get_mandatory_value("operators", "default_queue")
DEFAULT_RETRIES: int = conf.getint("core", "default_task_retries", fallback=0)
DEFAULT_RETRY_DELAY: datetime.timedelta = datetime.timedelta(seconds=300)
DEFAULT_WEIGHT_RULE: WeightRule = WeightRule(
conf.get("core", "default_task_weight_rule", fallback=WeightRule.DOWNSTREAM)
)
DEFAULT_TRIGGER_RULE: TriggerRule = TriggerRule.ALL_SUCCESS
-DEFAULT_TASK_EXECUTION_TIMEOUT: datetime.timedelta = conf.gettimedelta(
+DEFAULT_TASK_EXECUTION_TIMEOUT: Optional[datetime.timedelta] = conf.gettimedelta(
"core", "default_task_execution_timeout"
)
diff --git a/airflow/models/dag.py b/airflow/models/dag.py
index 25da5b9489..40788e68d6 100644
--- a/airflow/models/dag.py
+++ b/airflow/models/dag.py
@@ -326,8 +326,8 @@ class DAG(LoggingMixin):
sla_miss_callback: Optional[
Callable[["DAG", str, str, List["SlaMiss"], List[TaskInstance]], None]
] = None,
- default_view: str = conf.get('webserver', 'dag_default_view').lower(),
- orientation: str = conf.get('webserver', 'dag_orientation'),
+ default_view: str = conf.get_mandatory_value('webserver', 'dag_default_view').lower(),
+ orientation: str = conf.get_mandatory_value('webserver', 'dag_orientation'),
catchup: bool = conf.getboolean('scheduler', 'catchup_by_default'),
on_success_callback: Optional[DagStateChangeCallback] = None,
on_failure_callback: Optional[DagStateChangeCallback] = None,
@@ -2806,7 +2806,7 @@ class DagModel(Base):
have a value
"""
# This is for backwards-compatibility with old dags that don't have None as default_view
- return self.default_view or conf.get('webserver', 'dag_default_view').lower()
+ return self.default_view or conf.get_mandatory_value('webserver', 'dag_default_view').lower()
@property
def safe_dag_id(self):
diff --git a/airflow/models/taskinstance.py b/airflow/models/taskinstance.py
index 123b4ddea3..de85b4cf8c 100644
--- a/airflow/models/taskinstance.py
+++ b/airflow/models/taskinstance.py
@@ -2296,7 +2296,7 @@ class TaskInstance(Base, LoggingMixin):
def render(key: str, content: str) -> str:
if conf.has_option('email', key):
- path = conf.get('email', key)
+ path = conf.get_mandatory_value('email', key)
with open(path) as f:
content = f.read()
return render_template_to_string(jinja_env.from_string(content), jinja_context)
diff --git a/airflow/providers/apache/hive/operators/hive.py b/airflow/providers/apache/hive/operators/hive.py
index b4581aaea3..45cae0fa4e 100644
--- a/airflow/providers/apache/hive/operators/hive.py
+++ b/airflow/providers/apache/hive/operators/hive.py
@@ -100,11 +100,15 @@ class HiveOperator(BaseOperator):
self.mapred_queue = mapred_queue
self.mapred_queue_priority = mapred_queue_priority
self.mapred_job_name = mapred_job_name
- self.mapred_job_name_template = conf.get(
+
+ job_name_template = conf.get(
'hive',
'mapred_job_name_template',
fallback="Airflow HiveOperator task for {hostname}.{dag_id}.{task_id}.{execution_date}",
)
+ if job_name_template is None:
+ raise ValueError("Job name template should be set !")
+ self.mapred_job_name_template: str = job_name_template
# assigned lazily - just for consistency we can create the attribute with a
# `None` initial value, later it will be populated by the execute method.
diff --git a/airflow/providers/apache/spark/hooks/spark_submit.py b/airflow/providers/apache/spark/hooks/spark_submit.py
index bcbeb4b99c..0f5dc2f730 100644
--- a/airflow/providers/apache/spark/hooks/spark_submit.py
+++ b/airflow/providers/apache/spark/hooks/spark_submit.py
@@ -632,7 +632,7 @@ class SparkSubmitHook(BaseHook, LoggingMixin):
# we still attempt to kill the yarn application
renew_from_kt(self._principal, self._keytab, exit_on_fail=False)
env = os.environ.copy()
- env["KRB5CCNAME"] = airflow_conf.get('kerberos', 'ccache')
+ env["KRB5CCNAME"] = airflow_conf.get_mandatory_value('kerberos', 'ccache')
with subprocess.Popen(
kill_cmd, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE
diff --git a/airflow/providers/qubole/hooks/qubole.py b/airflow/providers/qubole/hooks/qubole.py
index a62243e9a0..7896fbd352 100644
--- a/airflow/providers/qubole/hooks/qubole.py
+++ b/airflow/providers/qubole/hooks/qubole.py
@@ -227,7 +227,7 @@ class QuboleHook(BaseHook):
"""
if fp is None:
iso = datetime.datetime.utcnow().isoformat()
- logpath = os.path.expanduser(conf.get('logging', 'BASE_LOG_FOLDER'))
+ logpath = os.path.expanduser(conf.get_mandatory_value('logging', 'BASE_LOG_FOLDER'))
resultpath = logpath + '/' + self.dag_id + '/' + self.task_id + '/results'
pathlib.Path(resultpath).mkdir(parents=True, exist_ok=True)
fp = open(resultpath + '/' + iso, 'wb')
diff --git a/airflow/security/kerberos.py b/airflow/security/kerberos.py
index ce8043b2eb..e8fc86af72 100644
--- a/airflow/security/kerberos.py
+++ b/airflow/security/kerberos.py
@@ -38,7 +38,7 @@ import socket
import subprocess
import sys
import time
-from typing import Optional
+from typing import List, Optional
from airflow.configuration import conf
@@ -59,7 +59,9 @@ def renew_from_kt(principal: Optional[str], keytab: str, exit_on_fail: bool = Tr
# minutes to give ourselves a large renewal buffer.
renewal_lifetime = f"{conf.getint('kerberos', 'reinit_frequency')}m"
- cmd_principal = principal or conf.get('kerberos', 'principal').replace("_HOST", socket.getfqdn())
+ cmd_principal = principal or conf.get_mandatory_value('kerberos', 'principal').replace(
+ "_HOST", socket.getfqdn()
+ )
if conf.getboolean('kerberos', 'forwardable'):
forwardable = '-f'
@@ -71,8 +73,8 @@ def renew_from_kt(principal: Optional[str], keytab: str, exit_on_fail: bool = Tr
else:
include_ip = '-A'
- cmdv = [
- conf.get('kerberos', 'kinit_path'),
+ cmdv: List[str] = [
+ conf.get_mandatory_value('kerberos', 'kinit_path'),
forwardable,
include_ip,
"-r",
@@ -81,7 +83,7 @@ def renew_from_kt(principal: Optional[str], keytab: str, exit_on_fail: bool = Tr
"-t",
keytab, # specify keytab
"-c",
- conf.get('kerberos', 'ccache'), # specify credentials cache
+ conf.get_mandatory_value('kerberos', 'ccache'), # specify credentials cache
cmd_principal,
]
log.info("Re-initialising kerberos from keytab: %s", " ".join(shlex.quote(f) for f in cmdv))
@@ -129,10 +131,10 @@ def perform_krb181_workaround(principal: str):
:param principal: principal name
:return: None
"""
- cmdv = [
- conf.get('kerberos', 'kinit_path'),
+ cmdv: List[str] = [
+ conf.get_mandatory_value('kerberos', 'kinit_path'),
"-c",
- conf.get('kerberos', 'ccache'),
+ conf.get_mandatory_value('kerberos', 'ccache'),
"-R",
] # Renew ticket_cache
@@ -162,7 +164,7 @@ def detect_conf_var() -> bool:
Sun Java Krb5LoginModule in Java6, so we need to take an action to work
around it.
"""
- ticket_cache = conf.get('kerberos', 'ccache')
+ ticket_cache = conf.get_mandatory_value('kerberos', 'ccache')
with open(ticket_cache, 'rb') as file:
# Note: this file is binary, so we check against a bytearray.
diff --git a/airflow/sensors/base.py b/airflow/sensors/base.py
index 5cbe009c82..7f1cd87c3d 100644
--- a/airflow/sensors/base.py
+++ b/airflow/sensors/base.py
@@ -143,7 +143,7 @@ class BaseSensorOperator(BaseOperator, SkipMixin):
self._validate_input_values()
self.sensor_service_enabled = conf.getboolean('smart_sensor', 'use_smart_sensor')
self.sensors_support_sensor_service = set(
- map(lambda l: l.strip(), conf.get('smart_sensor', 'sensors_enabled').split(','))
+ map(lambda l: l.strip(), conf.get_mandatory_value('smart_sensor', 'sensors_enabled').split(','))
)
def _validate_input_values(self) -> None:
diff --git a/airflow/settings.py b/airflow/settings.py
index 86b22fab93..a8394b1f9f 100644
--- a/airflow/settings.py
+++ b/airflow/settings.py
@@ -45,7 +45,7 @@ log = logging.getLogger(__name__)
TIMEZONE = pendulum.tz.timezone('UTC')
try:
- tz = conf.get("core", "default_timezone")
+ tz = conf.get_mandatory_value("core", "default_timezone")
if tz == "system":
TIMEZONE = pendulum.tz.local_timezone()
else:
@@ -77,7 +77,7 @@ SQL_ALCHEMY_CONN: Optional[str] = None
PLUGINS_FOLDER: Optional[str] = None
LOGGING_CLASS_PATH: Optional[str] = None
DONOT_MODIFY_HANDLERS: Optional[bool] = None
-DAGS_FOLDER: str = os.path.expanduser(conf.get('core', 'DAGS_FOLDER'))
+DAGS_FOLDER: str = os.path.expanduser(conf.get_mandatory_value('core', 'DAGS_FOLDER'))
engine: Engine
Session: Callable[..., SASession]
diff --git a/airflow/utils/email.py b/airflow/utils/email.py
index f9b8fe23ef..ec0095e983 100644
--- a/airflow/utils/email.py
+++ b/airflow/utils/email.py
@@ -94,9 +94,13 @@ def send_email_smtp(
"""
smtp_mail_from = conf.get('smtp', 'SMTP_MAIL_FROM')
- if smtp_mail_from:
+ if smtp_mail_from is not None:
mail_from = smtp_mail_from
else:
+ if from_email is None:
+ raise Exception(
+ "You should set from email - either by smtp/smtp_mail_from config or " "`from_email parameter"
+ )
mail_from = from_email
msg, recipients = build_mime_message(
@@ -188,7 +192,7 @@ def send_mime_email(
dryrun: bool = False,
) -> None:
"""Send MIME email."""
- smtp_host = conf.get('smtp', 'SMTP_HOST')
+ smtp_host = conf.get_mandatory_value('smtp', 'SMTP_HOST')
smtp_port = conf.getint('smtp', 'SMTP_PORT')
smtp_starttls = conf.getboolean('smtp', 'SMTP_STARTTLS')
smtp_ssl = conf.getboolean('smtp', 'SMTP_SSL')
diff --git a/airflow/utils/file.py b/airflow/utils/file.py
index 90be4ab172..89d0b7d9fc 100644
--- a/airflow/utils/file.py
+++ b/airflow/utils/file.py
@@ -247,7 +247,7 @@ def _find_path_from_directory(
def find_path_from_directory(
base_dir_path: str,
ignore_file_name: str,
- ignore_file_syntax: str = conf.get('core', 'DAG_IGNORE_FILE_SYNTAX', fallback="regexp"),
+ ignore_file_syntax: str = conf.get_mandatory_value('core', 'DAG_IGNORE_FILE_SYNTAX', fallback="regexp"),
) -> Generator[str, None, None]:
"""
Recursively search the base path and return the list of file paths that should not be ignored.
diff --git a/airflow/utils/helpers.py b/airflow/utils/helpers.py
index f4a1107bc3..e8366805b4 100644
--- a/airflow/utils/helpers.py
+++ b/airflow/utils/helpers.py
@@ -256,7 +256,7 @@ def build_airflow_url_with_query(query: Dict[str, Any]) -> str:
"""
import flask
- view = conf.get('webserver', 'dag_default_view').lower()
+ view = conf.get_mandatory_value('webserver', 'dag_default_view').lower()
return flask.url_for(f"Airflow.{view}", **query)
diff --git a/airflow/utils/mixins.py b/airflow/utils/mixins.py
index fbe0b24d56..4d0165e4df 100644
--- a/airflow/utils/mixins.py
+++ b/airflow/utils/mixins.py
@@ -30,7 +30,7 @@ class MultiprocessingStartMethodMixin:
mp_start_method is set in configs, else, it uses the OS default.
"""
if conf.has_option('core', 'mp_start_method'):
- return conf.get('core', 'mp_start_method')
+ return conf.get_mandatory_value('core', 'mp_start_method')
method = multiprocessing.get_start_method()
if not method:
diff --git a/airflow/utils/sqlalchemy.py b/airflow/utils/sqlalchemy.py
index de4ad01e69..6d751d106a 100644
--- a/airflow/utils/sqlalchemy.py
+++ b/airflow/utils/sqlalchemy.py
@@ -37,7 +37,7 @@ log = logging.getLogger(__name__)
utc = pendulum.tz.timezone('UTC')
-using_mysql = conf.get('database', 'sql_alchemy_conn').lower().startswith('mysql')
+using_mysql = conf.get_mandatory_value('database', 'sql_alchemy_conn').lower().startswith('mysql')
class UtcDateTime(TypeDecorator):
diff --git a/tests/cli/commands/test_dag_processor_command.py b/tests/cli/commands/test_dag_processor_command.py
index 39241a96de..21cec4cee2 100644
--- a/tests/cli/commands/test_dag_processor_command.py
+++ b/tests/cli/commands/test_dag_processor_command.py
@@ -43,7 +43,7 @@ class TestDagProcessorCommand(unittest.TestCase):
)
@mock.patch("airflow.cli.commands.dag_processor_command.DagFileProcessorManager")
@pytest.mark.skipif(
- conf.get('core', 'sql_alchemy_conn').lower().startswith('sqlite'),
+ conf.get_mandatory_value('database', 'sql_alchemy_conn').lower().startswith('sqlite'),
reason="Standalone Dag Processor doesn't support sqlite.",
)
def test_start_manager(
diff --git a/tests/cli/commands/test_task_command.py b/tests/cli/commands/test_task_command.py
index 481103f1f3..e539592a64 100644
--- a/tests/cli/commands/test_task_command.py
+++ b/tests/cli/commands/test_task_command.py
@@ -393,7 +393,7 @@ class TestLogsfromTaskRunCommand(unittest.TestCase):
self.execution_date = timezone.make_aware(datetime(2017, 1, 1))
self.execution_date_str = self.execution_date.isoformat()
self.task_args = ['tasks', 'run', self.dag_id, self.task_id, '--local', self.execution_date_str]
- self.log_dir = conf.get('logging', 'base_log_folder')
+ self.log_dir = conf.get_mandatory_value('logging', 'base_log_folder')
self.log_filename = f"dag_id={self.dag_id}/run_id={self.run_id}/task_id={self.task_id}/attempt=1.log"
self.ti_log_file_path = os.path.join(self.log_dir, self.log_filename)
self.parser = cli_parser.get_parser()
diff --git a/tests/test_utils/system_tests_class.py b/tests/test_utils/system_tests_class.py
index be9cf73486..846c34c3e0 100644
--- a/tests/test_utils/system_tests_class.py
+++ b/tests/test_utils/system_tests_class.py
@@ -19,6 +19,7 @@ import os
import shutil
from datetime import datetime
from pathlib import Path
+from typing import Optional
from unittest import TestCase
from airflow.configuration import AIRFLOW_HOME, AirflowConfigParser, get_airflow_config
@@ -31,6 +32,12 @@ from tests.test_utils.logging_command_executor import get_executor
DEFAULT_DAG_FOLDER = os.path.join(AIRFLOW_MAIN_FOLDER, "airflow", "example_dags")
+def get_default_logs_if_none(logs: Optional[str]) -> str:
+ if logs is None:
+ return os.path.join(AIRFLOW_HOME, 'logs')
+ return logs
+
+
def resolve_logs_folder() -> str:
"""
Returns LOGS folder specified in current Airflow config.
@@ -39,13 +46,13 @@ def resolve_logs_folder() -> str:
conf = AirflowConfigParser()
conf.read(config_file)
try:
- logs = conf.get("logging", "base_log_folder")
+ return get_default_logs_if_none(conf.get("logging", "base_log_folder"))
except AirflowException:
try:
- logs = conf.get("core", "base_log_folder")
+ return get_default_logs_if_none(conf.get("core", "base_log_folder"))
except AirflowException:
- logs = os.path.join(AIRFLOW_HOME, 'logs')
- return logs
+ pass
+ return get_default_logs_if_none(None)
class SystemTest(TestCase, LoggingMixin):