[AIRFLOW-6140] Add missing types for some core classes.

.pre-commit-config.yaml

@@ -262,7 +262,7 @@ repos:
         language: system
         entry: "./scripts/ci/pre_commit_pylint_main.sh"
         files: \.py$
-        exclude: ^tests/.*\.py$|^airflow/_vendor/.*$
+        exclude: ^tests/.*\.py$|^airflow/_vendor/.*|^scripts/.*\.py$
         pass_filenames: true
         require_serial: true  # Pylint tests should be run in one chunk to detect all cycles
       - id: pylint-tests

airflow/__init__.py

@@ -31,14 +31,13 @@
 # pylint: disable=wrong-import-position
 from typing import Callable, Optional
 
-# noinspection PyUnresolvedReferences
 from airflow import utils
 from airflow import settings
 from airflow import version
 from airflow.utils.log.logging_mixin import LoggingMixin
 from airflow.configuration import conf
 from airflow.exceptions import AirflowException
-from airflow.models import DAG
+from airflow.models.dag import DAG
 
 __version__ = version.version
 

airflow/config_templates/airflow_local_settings.py

@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
 #
 # Licensed to the Apache Software Foundation (ASF) under one
 # or more contributor license agreements.  See the NOTICE file
@@ -19,43 +18,42 @@
 """Airflow logging settings"""
 
 import os
-from typing import Any, Dict
+from typing import Any, Dict, Union
 
-from airflow import AirflowException
-from airflow.configuration import conf
+from airflow import AirflowException, conf
 from airflow.utils.file import mkdirs
 
 # TODO: Logging format and level should be configured
 # 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 = conf.get('core', 'LOGGING_LEVEL').upper()
+LOG_LEVEL: str = conf.get('core', 'LOGGING_LEVEL').upper()
 
 
 # Flask appbuilder's info level log is very verbose,
 # so it's set to 'WARN' by default.
-FAB_LOG_LEVEL = conf.get('core', 'FAB_LOGGING_LEVEL').upper()
+FAB_LOG_LEVEL: str = conf.get('core', 'FAB_LOGGING_LEVEL').upper()
 
-LOG_FORMAT = conf.get('core', 'LOG_FORMAT')
+LOG_FORMAT: str = conf.get('core', 'LOG_FORMAT')
 
-COLORED_LOG_FORMAT = conf.get('core', 'COLORED_LOG_FORMAT')
+COLORED_LOG_FORMAT: str = conf.get('core', 'COLORED_LOG_FORMAT')
 
-COLORED_LOG = conf.getboolean('core', 'COLORED_CONSOLE_LOG')
+COLORED_LOG: bool = conf.getboolean('core', 'COLORED_CONSOLE_LOG')
 
-COLORED_FORMATTER_CLASS = conf.get('core', 'COLORED_FORMATTER_CLASS')
+COLORED_FORMATTER_CLASS: str = conf.get('core', 'COLORED_FORMATTER_CLASS')
 
-BASE_LOG_FOLDER = conf.get('core', 'BASE_LOG_FOLDER')
+BASE_LOG_FOLDER: str = conf.get('core', 'BASE_LOG_FOLDER')
 
-PROCESSOR_LOG_FOLDER = conf.get('scheduler', 'CHILD_PROCESS_LOG_DIRECTORY')
+PROCESSOR_LOG_FOLDER: str = conf.get('scheduler', 'CHILD_PROCESS_LOG_DIRECTORY')
 
-DAG_PROCESSOR_MANAGER_LOG_LOCATION = \
+DAG_PROCESSOR_MANAGER_LOG_LOCATION: str = \
     conf.get('core', 'DAG_PROCESSOR_MANAGER_LOG_LOCATION')
 
-FILENAME_TEMPLATE = conf.get('core', 'LOG_FILENAME_TEMPLATE')
+FILENAME_TEMPLATE: str = conf.get('core', 'LOG_FILENAME_TEMPLATE')
 
-PROCESSOR_FILENAME_TEMPLATE = conf.get('core', 'LOG_PROCESSOR_FILENAME_TEMPLATE')
+PROCESSOR_FILENAME_TEMPLATE: str = conf.get('core', 'LOG_PROCESSOR_FILENAME_TEMPLATE')
 
-DEFAULT_LOGGING_CONFIG = {
+DEFAULT_LOGGING_CONFIG: Dict[str, Any] = {
     'version': 1,
     'disable_existing_loggers': False,
     'formatters': {
@@ -107,9 +105,9 @@
         'handlers': ['console'],
         'level': LOG_LEVEL,
     }
-}  # type: Dict[str, Any]
+}
 
-DEFAULT_DAG_PARSING_LOGGING_CONFIG = {
+DEFAULT_DAG_PARSING_LOGGING_CONFIG: Dict[str, Dict[str, Dict[str, Any]]] = {
     'handlers': {
         'processor_manager': {
             'class': 'logging.handlers.RotatingFileHandler',
@@ -140,58 +138,58 @@
 
     # Manually create log directory for processor_manager handler as RotatingFileHandler
     # will only create file but not the directory.
-    processor_manager_handler_config = DEFAULT_DAG_PARSING_LOGGING_CONFIG['handlers'][
-        'processor_manager']
-    directory = os.path.dirname(processor_manager_handler_config['filename'])
+    processor_manager_handler_config: Dict[str, Any] = \
+        DEFAULT_DAG_PARSING_LOGGING_CONFIG['handlers']['processor_manager']
+    directory: str = os.path.dirname(processor_manager_handler_config['filename'])
     mkdirs(directory, 0o755)
 
 ##################
 # Remote logging #
 ##################
 
-REMOTE_LOGGING = conf.getboolean('core', 'remote_logging')
+REMOTE_LOGGING: bool = conf.getboolean('core', 'remote_logging')
 
 if REMOTE_LOGGING:
 
-    ELASTICSEARCH_HOST = conf.get('elasticsearch', 'HOST')
+    ELASTICSEARCH_HOST: str = conf.get('elasticsearch', 'HOST')
 
     # Storage bucket URL for remote logging
     # S3 buckets should start with "s3://"
     # GCS buckets should start with "gs://"
     # WASB buckets should start with "wasb"
     # just to help Airflow select correct handler
-    REMOTE_BASE_LOG_FOLDER = conf.get('core', 'REMOTE_BASE_LOG_FOLDER')
+    REMOTE_BASE_LOG_FOLDER: str = conf.get('core', 'REMOTE_BASE_LOG_FOLDER')
 
     if REMOTE_BASE_LOG_FOLDER.startswith('s3://'):
-        S3_REMOTE_HANDLERS = {
+        S3_REMOTE_HANDLERS: Dict[str, Dict[str, str]] = {
             'task': {
                 'class': 'airflow.utils.log.s3_task_handler.S3TaskHandler',
                 'formatter': 'airflow',
-                'base_log_folder': os.path.expanduser(BASE_LOG_FOLDER),
+                'base_log_folder': str(os.path.expanduser(BASE_LOG_FOLDER)),
                 's3_log_folder': REMOTE_BASE_LOG_FOLDER,
                 'filename_template': FILENAME_TEMPLATE,
             },
         }
 
         DEFAULT_LOGGING_CONFIG['handlers'].update(S3_REMOTE_HANDLERS)
     elif REMOTE_BASE_LOG_FOLDER.startswith('gs://'):
-        GCS_REMOTE_HANDLERS = {
+        GCS_REMOTE_HANDLERS: Dict[str, Dict[str, str]] = {
             'task': {
                 'class': 'airflow.utils.log.gcs_task_handler.GCSTaskHandler',
                 'formatter': 'airflow',
-                'base_log_folder': os.path.expanduser(BASE_LOG_FOLDER),
+                'base_log_folder': str(os.path.expanduser(BASE_LOG_FOLDER)),
                 'gcs_log_folder': REMOTE_BASE_LOG_FOLDER,
                 'filename_template': FILENAME_TEMPLATE,
             },
         }
 
         DEFAULT_LOGGING_CONFIG['handlers'].update(GCS_REMOTE_HANDLERS)
     elif REMOTE_BASE_LOG_FOLDER.startswith('wasb'):
-        WASB_REMOTE_HANDLERS = {
+        WASB_REMOTE_HANDLERS: Dict[str, Dict[str, Union[str, bool]]] = {
             'task': {
                 'class': 'airflow.utils.log.wasb_task_handler.WasbTaskHandler',
                 'formatter': 'airflow',
-                'base_log_folder': os.path.expanduser(BASE_LOG_FOLDER),
+                'base_log_folder': str(os.path.expanduser(BASE_LOG_FOLDER)),
                 'wasb_log_folder': REMOTE_BASE_LOG_FOLDER,
                 'wasb_container': 'airflow-logs',
                 'filename_template': FILENAME_TEMPLATE,
@@ -201,17 +199,17 @@
 
         DEFAULT_LOGGING_CONFIG['handlers'].update(WASB_REMOTE_HANDLERS)
     elif ELASTICSEARCH_HOST:
-        ELASTICSEARCH_LOG_ID_TEMPLATE = conf.get('elasticsearch', 'LOG_ID_TEMPLATE')
-        ELASTICSEARCH_END_OF_LOG_MARK = conf.get('elasticsearch', 'END_OF_LOG_MARK')
-        ELASTICSEARCH_WRITE_STDOUT = conf.get('elasticsearch', 'WRITE_STDOUT')
-        ELASTICSEARCH_JSON_FORMAT = conf.get('elasticsearch', 'JSON_FORMAT')
-        ELASTICSEARCH_JSON_FIELDS = conf.get('elasticsearch', 'JSON_FIELDS')
+        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_WRITE_STDOUT: str = conf.get('elasticsearch', 'WRITE_STDOUT')
+        ELASTICSEARCH_JSON_FORMAT: str = conf.get('elasticsearch', 'JSON_FORMAT')
+        ELASTICSEARCH_JSON_FIELDS: str = conf.get('elasticsearch', 'JSON_FIELDS')
 
-        ELASTIC_REMOTE_HANDLERS = {
+        ELASTIC_REMOTE_HANDLERS: Dict[str, Dict[str, str]] = {
             'task': {
                 'class': 'airflow.utils.log.es_task_handler.ElasticsearchTaskHandler',
                 'formatter': 'airflow',
-                'base_log_folder': os.path.expanduser(BASE_LOG_FOLDER),
+                'base_log_folder': str(os.path.expanduser(BASE_LOG_FOLDER)),
                 'log_id_template': ELASTICSEARCH_LOG_ID_TEMPLATE,
                 'filename_template': FILENAME_TEMPLATE,
                 'end_of_log_mark': ELASTICSEARCH_END_OF_LOG_MARK,

airflow/configuration.py

@@ -1,5 +1,4 @@
-# -*- coding: utf-8 -*-
-#
+
 # Licensed to the Apache Software Foundation (ASF) under one
 # or more contributor license agreements.  See the NOTICE file
 # distributed with this work for additional information

airflow/macros/__init__.py

@@ -16,9 +16,7 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-
-# pylint: disable=missing-docstring
-
+"""Macros."""
 import time  # noqa
 import uuid  # noqa
 from datetime import datetime, timedelta