[AIRFLOW-6664] Introduce implicit packages

.github/boring-cyborg.yml

@@ -20,46 +20,24 @@
 
 labelPRBasedOnFilePath:
   provider:GCP:
-    - airflow/**/gcp/*
-    - airflow/gcp/*
-    - airflow/gcp/**/*
     - airflow/providers/google/cloud/**/*
-    - airflow/**/gcp_*.py
-    - airflow/**/gcs_*.py
-    - airflow/**/bigquery_*.py
-    - tests/gcp/**/*
-    - tests/gcp/*
     - tests/providers/google/cloud/**/*
     - tests/providers/google/cloud/*
     - docs/howto/connection/gcp.rst
     - docs/howto/connection/gcp_sql.rst
     - docs/howto/operator/gcp/*
 
   provider:AWS:
-    - airflow/**/aws/*
     - airflow/providers/amazon/aws/**/*
-    - airflow/**/aws_*.py
-    - airflow/**/ecs_*.py
-    - airflow/**/emr_*.py
-    - airflow/**/s3_*.py
-    - airflow/**/sagemaker_*.py
     - tests/providers/amazon/aws/**/*
     - tests/providers/amazon/aws/*
-    - tests/**/aws_*.py
-    - tests/**/s3_*.py
-    - tests/**/ecs_*.py
-    - tests/**/emr_*.py
     - docs/howto/connection/aws.rst
     - docs/howto/operator/amazon/aws/*
 
   provider:Azure:
-    - airflow/**/azure/*
-    - airflow/**/azure_*.py
-    - airflow/**/adls_*.py
-    - airflow/**/wasb_*.py
-    - tests/**/wasb_*.py
-    - tests/**/adls_*.py
-    - tests/**/azure_*.py
+    - airflow/providers/microsoft/azure/**/*
+    - tests/providers/microsoft/azure/**/*
+    - tests/providers/microsoft/azure/*
 
   provider:Apache:
     - airflow/providers/apache/**/*
@@ -69,7 +47,7 @@ labelPRBasedOnFilePath:
   k8s:
     - airflow/**/kubernetes_*.py
     - airflow/kubernetes/**/*
-    - tests/integration/kubernetes/*
+    - tests/runtime/kubernetes/*
     - docs/kubernetes.rst
 
   area:dev:

.pre-commit-config.yaml

@@ -259,6 +259,13 @@ repos:
         entry: "(airflow\\.){0,1}utils\\.dates\\.days_ago"
         files: \.py$
         pass_filenames: true
+      - id: remove-empty-init-py
+        name: Removes empty __init__.py files
+        language: system
+        entry: "./scripts/ci/pre_commit_remove_empty_init_py.sh"
+        files: __init__\.py$
+        exclude: ^airflow/_vendor/.*$|node_modules|^pylint_plugins|^tests/__init__.py$
+        pass_filenames: true
       - id: build
         name: Check if image build is needed
         entry: ./scripts/ci/pre_commit_ci_build.sh
@@ -286,22 +293,26 @@ repos:
         entry: "./scripts/ci/pre_commit_mypy.sh"
         files: \.py$
         exclude: ^airflow/_vendor/.*$
-        pass_filenames: true
+        # MyPy does not work well when you have multiple files with same name (even in different packages)
+        # Passed as parameters - this is a known problem and it can be solved by running mypy on whole
+        # packages instead - MyPy is fast enough so this is not a problem to run it for all files
+        # even if few files change. See https://github.com/pre-commit/mirrors-mypy/issues/5
+        pass_filenames: false
+        require_serial: true
       - id: pylint
         name: Run pylint for main sources
         language: system
         entry: "./scripts/ci/pre_commit_pylint_main.sh"
         files: \.py$
         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
+        require_serial: false
       - id: pylint-tests
         name: Run pylint for tests
         language: system
         entry: "./scripts/ci/pre_commit_pylint_tests.sh"
         files: ^tests/.*\.py$
         pass_filenames: true
-        require_serial: true
       - id: flake8
         name: Run flake8
         language: system

airflow/api/auth/backend/kerberos_auth.py

@@ -45,9 +45,12 @@
 from functools import wraps
 from socket import getfqdn
 
-import kerberos
 # noinspection PyProtectedMember
 from flask import Response, _request_ctx_stack as stack, g, make_response, request  # type: ignore
+from kerberos import (  # pylint: disable=no-name-in-module
+    AUTH_GSS_COMPLETE, AUTH_GSS_CONTINUE, GSSError, KrbError, authGSSServerClean, authGSSServerInit,
+    authGSSServerResponse, authGSSServerStep, authGSSServerUserName, getServerPrincipalDetails,
+)
 from requests_kerberos import HTTPKerberosAuth
 
 from airflow.configuration import conf
@@ -85,8 +88,8 @@ def init_app(app):
 
     try:
         log.info("Kerberos init: %s %s", service, hostname)
-        principal = kerberos.getServerPrincipalDetails(service, hostname)
-    except kerberos.KrbError as err:
+        principal = getServerPrincipalDetails(service, hostname)
+    except KrbError as err:
         log.warning("Kerberos: %s", err)
     else:
         log.info("Kerberos API: server is %s", principal)
@@ -108,22 +111,22 @@ def _gssapi_authenticate(token):
     state = None
     ctx = stack.top
     try:
-        return_code, state = kerberos.authGSSServerInit(_KERBEROS_SERVICE.service_name)
-        if return_code != kerberos.AUTH_GSS_COMPLETE:
+        return_code, state = authGSSServerInit(_KERBEROS_SERVICE.service_name)
+        if return_code != AUTH_GSS_COMPLETE:
             return None
-        return_code = kerberos.authGSSServerStep(state, token)
-        if return_code == kerberos.AUTH_GSS_COMPLETE:
-            ctx.kerberos_token = kerberos.authGSSServerResponse(state)
-            ctx.kerberos_user = kerberos.authGSSServerUserName(state)
+        return_code = authGSSServerStep(state, token)
+        if return_code == AUTH_GSS_COMPLETE:
+            ctx.kerberos_token = authGSSServerResponse(state)
+            ctx.kerberos_user = authGSSServerUserName(state)
             return return_code
-        if return_code == kerberos.AUTH_GSS_CONTINUE:
-            return kerberos.AUTH_GSS_CONTINUE
+        if return_code == AUTH_GSS_CONTINUE:
+            return AUTH_GSS_CONTINUE
         return None
-    except kerberos.GSSError:
+    except GSSError:
         return None
     finally:
         if state:
-            kerberos.authGSSServerClean(state)
+            authGSSServerClean(state)
 
 
 def requires_authentication(function):
@@ -135,7 +138,7 @@ def decorated(*args, **kwargs):
             ctx = stack.top
             token = ''.join(header.split()[1:])
             return_code = _gssapi_authenticate(token)
-            if return_code == kerberos.AUTH_GSS_COMPLETE:
+            if return_code == AUTH_GSS_COMPLETE:
                 g.user = ctx.kerberos_user
                 response = function(*args, **kwargs)
                 response = make_response(response)
@@ -144,7 +147,7 @@ def decorated(*args, **kwargs):
                                                                      ctx.kerberos_token])
 
                 return response
-            if return_code != kerberos.AUTH_GSS_CONTINUE:
+            if return_code != AUTH_GSS_CONTINUE:
                 return _forbidden()
         return _unauthorized()
     return decorated

airflow/api/common/experimental/mark_tasks.py

@@ -22,7 +22,7 @@
 
 from sqlalchemy import or_
 
-from airflow.jobs import BackfillJob
+from airflow.jobs.backfill_job import BackfillJob
 from airflow.models import DagRun, TaskInstance
 from airflow.models.baseoperator import BaseOperator
 from airflow.operators.subdag_operator import SubDagOperator

airflow/cli/commands/kerberos_command.py

@@ -20,7 +20,7 @@
 from daemon.pidfile import TimeoutPIDLockFile
 
 from airflow import settings
-from airflow.security import kerberos as krb
+from airflow.security import kerberos_security as krb
 from airflow.utils import cli as cli_utils
 from airflow.utils.cli import setup_locations
 

airflow/config_templates/config.yml

@@ -948,7 +948,7 @@
       version_added: ~
       type: string
       example: ~
-      default: "airflow.utils.email.send_email_smtp"
+      default: "airflow.utils.email_utils.send_email_smtp"
     - name: default_email_on_retry
       description: |
         Whether email alerts should be sent when a task is retried
@@ -963,11 +963,10 @@
       type: boolean
       example: ~
       default: "True"
-
 - name: smtp
   description: |
     If you want airflow to send emails on retries, failure, and you want to use
-    the airflow.utils.email.send_email_smtp function, you have to configure an
+    the airflow.utils.email_utils.send_email_smtp function, you have to configure an
     smtp server here
   options:
     - name: smtp_host

airflow/config_templates/default_airflow.cfg

@@ -454,7 +454,7 @@ session_lifetime_days = 30
 # Configuration email backend and whether to
 # send email alerts on retry or failure
 # Email backend to use
-email_backend = airflow.utils.email.send_email_smtp
+email_backend = airflow.utils.email_utils.send_email_smtp
 
 # Whether email alerts should be sent when a task is retried
 default_email_on_retry = True
@@ -465,7 +465,7 @@ default_email_on_failure = True
 [smtp]
 
 # If you want airflow to send emails on retries, failure, and you want to use
-# the airflow.utils.email.send_email_smtp function, you have to configure an
+# the airflow.utils.email_utils.send_email_smtp function, you have to configure an
 # smtp server here
 smtp_host = localhost
 smtp_starttls = True

airflow/config_templates/default_test.cfg

@@ -79,7 +79,7 @@ hide_paused_dags_by_default = False
 page_size = 100
 
 [email]
-email_backend = airflow.utils.email.send_email_smtp
+email_backend = airflow.utils.email_utils.send_email_smtp
 
 [smtp]
 smtp_host = localhost

airflow/api/auth/backend/__init__.py → airflow/contrib/auth/backends/authentication_error.py

@@ -1,4 +1,3 @@
-#
 # 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
@@ -15,3 +14,7 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+
+
+class AuthenticationError(Exception):
+    """Error returned on authentication problems"""

airflow/contrib/auth/backends/github_enterprise_auth.py

@@ -26,6 +26,7 @@
 
 from airflow import models
 from airflow.configuration import AirflowConfigException, conf
+from airflow.contrib.auth.backends.authentication_error import AuthenticationError
 from airflow.utils.session import provide_session
 
 log = logging.getLogger(__name__)
@@ -68,10 +69,6 @@ def is_superuser(self):
         return True
 
 
-class AuthenticationError(Exception):
-    pass
-
-
 class GHEAuthBackend:
 
     def __init__(self):