[AIRFLOW-6373] Make airflow/utils pylint compatible

airflow/task/task_runner/base_task_runner.py

@@ -63,9 +63,7 @@ def __init__(self, local_task_job):
             # want to have to specify them in the sudo call - they would show
             # up in `ps` that way! And run commands now, as the other user
             # might not be able to run the cmds to get credentials
-            cfg_path = tmp_configuration_copy(chmod=0o600,
-                                              include_env=True,
-                                              include_cmds=True)
+            cfg_path = tmp_configuration_copy(chmod=0o600)
 
             # Give ownership of file to user; only they can read and write
             subprocess.call(
@@ -85,9 +83,7 @@ def __init__(self, local_task_job):
             # we are running as the same user, and can pass through environment
             # variables then we don't need to include those in the config copy
             # - the runner can read/execute those values as it needs
-            cfg_path = tmp_configuration_copy(chmod=0o600,
-                                              include_env=False,
-                                              include_cmds=False)
+            cfg_path = tmp_configuration_copy(chmod=0o600)
 
         self._cfg_path = cfg_path
         self._command = popen_prepend + self._task_instance.command_as_list(

airflow/utils/cli_action_loggers.py

@@ -65,11 +65,11 @@ def on_pre_execution(**kwargs):
     :return: None
     """
     logging.debug("Calling callbacks: %s", __pre_exec_callbacks)
-    for cb in __pre_exec_callbacks:
+    for callback in __pre_exec_callbacks:
         try:
-            cb(**kwargs)
-        except Exception:
-            logging.exception('Failed on pre-execution callback using %s', cb)
+            callback(**kwargs)
+        except Exception:  # pylint: disable=broad-except
+            logging.exception('Failed on pre-execution callback using %s', callback)
 
 
 def on_post_execution(**kwargs):
@@ -83,11 +83,11 @@ def on_post_execution(**kwargs):
     :return: None
     """
     logging.debug("Calling callbacks: %s", __post_exec_callbacks)
-    for cb in __post_exec_callbacks:
+    for callback in __post_exec_callbacks:
         try:
-            cb(**kwargs)
-        except Exception:
-            logging.exception('Failed on post-execution callback using %s', cb)
+            callback(**kwargs)
+        except Exception:  # pylint: disable=broad-except
+            logging.exception('Failed on post-execution callback using %s', callback)
 
 
 def default_action_log(log, **_):
@@ -102,7 +102,7 @@ def default_action_log(log, **_):
     try:
         with create_session() as session:
             session.add(log)
-    except Exception as error:
+    except Exception as error:  # pylint: disable=broad-except
         logging.warning("Failed to log action with %s", error)
 
 

airflow/utils/configuration.py

@@ -24,7 +24,7 @@
 from airflow.configuration import conf
 
 
-def tmp_configuration_copy(chmod=0o600, include_env=True, include_cmds=True):
+def tmp_configuration_copy(chmod=0o600):
     """
     Returns a path for a temporary file including a full copy of the configuration
     settings.

airflow/utils/dates.py

@@ -20,7 +20,7 @@
 from datetime import datetime, timedelta
 
 from croniter import croniter
-from dateutil.relativedelta import relativedelta  # noqa: F401 for doctest
+from dateutil.relativedelta import relativedelta  # noqa: F401 for doctest  # pylint: disable=unused-import
 
 from airflow.utils import timezone
 
@@ -34,7 +34,7 @@
 }
 
 
-def date_range(start_date, end_date=None, num=None, delta=None):
+def date_range(start_date, end_date=None, num=None, delta=None):  # pylint: disable=too-many-branches
     """
     Get a set of dates as a list based on a start, end and delta, delta
     can be something that can be added to `datetime.datetime`
@@ -73,12 +73,12 @@ def date_range(start_date, end_date=None, num=None, delta=None):
         end_date = timezone.utcnow()
 
     delta_iscron = False
-    tz = start_date.tzinfo
+    time_zone = start_date.tzinfo
 
     if isinstance(delta, str):
         delta_iscron = True
         if timezone.is_localized(start_date):
-            start_date = timezone.make_naive(start_date, tz)
+            start_date = timezone.make_naive(start_date, time_zone)
         cron = croniter(delta, start_date)
     elif isinstance(delta, timedelta):
         delta = abs(delta)
@@ -88,10 +88,10 @@ def date_range(start_date, end_date=None, num=None, delta=None):
     dates = []
     if end_date:
         if timezone.is_naive(start_date) and not timezone.is_naive(end_date):
-            end_date = timezone.make_naive(end_date, tz)
+            end_date = timezone.make_naive(end_date, time_zone)
         while start_date <= end_date:
             if timezone.is_naive(start_date):
-                dates.append(timezone.make_aware(start_date, tz))
+                dates.append(timezone.make_aware(start_date, time_zone))
             else:
                 dates.append(start_date)
 
@@ -102,20 +102,19 @@ def date_range(start_date, end_date=None, num=None, delta=None):
     else:
         for _ in range(abs(num)):
             if timezone.is_naive(start_date):
-                dates.append(timezone.make_aware(start_date, tz))
+                dates.append(timezone.make_aware(start_date, time_zone))
             else:
                 dates.append(start_date)
 
-            if delta_iscron:
-                if num > 0:
-                    start_date = cron.get_next(datetime)
-                else:
-                    start_date = cron.get_prev(datetime)
+            if delta_iscron and num > 0:
+                start_date = cron.get_next(datetime)
+            elif delta_iscron:
+                start_date = cron.get_prev(datetime)
+            elif num > 0:
+                start_date += delta
             else:
-                if num > 0:
-                    start_date += delta
-                else:
-                    start_date -= delta
+                start_date -= delta
+
     return sorted(dates)
 
 
@@ -140,14 +139,14 @@ def round_time(dt, delta, start_date=timezone.make_aware(datetime.min)):
 
     if isinstance(delta, str):
         # It's cron based, so it's easy
-        tz = start_date.tzinfo
-        start_date = timezone.make_naive(start_date, tz)
+        time_zone = start_date.tzinfo
+        start_date = timezone.make_naive(start_date, time_zone)
         cron = croniter(delta, start_date)
         prev = cron.get_prev(datetime)
         if prev == start_date:
-            return timezone.make_aware(start_date, tz)
+            return timezone.make_aware(start_date, time_zone)
         else:
-            return timezone.make_aware(prev, tz)
+            return timezone.make_aware(prev, time_zone)
 
     # Ignore the microseconds of dt
     dt -= timedelta(microseconds=dt.microsecond)

airflow/utils/db.py

@@ -21,26 +21,32 @@
 
 from airflow import models, settings
 from airflow.configuration import conf
-from airflow.jobs.base_job import BaseJob  # noqa: F401
+from airflow.jobs.base_job import BaseJob  # noqa: F401  # pylint: disable=unused-import
 from airflow.models import Connection
 from airflow.models.pool import Pool
 # We need to add this model manually to get reset working well
-from airflow.models.serialized_dag import SerializedDagModel  # noqa: F401
+from airflow.models.serialized_dag import SerializedDagModel  # noqa: F401  # pylint: disable=unused-import
 from airflow.utils.log.logging_mixin import LoggingMixin
-from airflow.utils.session import create_session, provide_session  # noqa
+from airflow.utils.session import create_session, provide_session  # noqa  # pylint: disable=unused-import
 
 log = LoggingMixin().log
 
 
 @provide_session
 def merge_conn(conn, session=None):
+    """
+    Add new Connection.
+    """
     if not session.query(Connection).filter(Connection.conn_id == conn.conn_id).first():
         session.add(conn)
         session.commit()
 
 
 @provide_session
 def add_default_pool_if_not_exists(session=None):
+    """
+    Add default pool if it does not exist.
+    """
     if not Pool.get_pool(Pool.DEFAULT_POOL_NAME, session=session):
         default_pool = Pool(
             pool=Pool.DEFAULT_POOL_NAME,
@@ -54,6 +60,9 @@ def add_default_pool_if_not_exists(session=None):
 
 @provide_session
 def create_default_connections(session=None):
+    """
+    Create default Airflow connections.
+    """
     merge_conn(
         Connection(
             conn_id="airflow_db", conn_type="mysql", host="mysql", login="root", password="", schema="airflow"
@@ -242,7 +251,7 @@ def create_default_connections(session=None):
             conn_id="segment_default", conn_type="segment", extra='{"write_key": "my-segment-write-key"}'
         ),
         session,
-    ),
+    )
     merge_conn(
         Connection(
             conn_id="azure_data_lake_default",
@@ -278,6 +287,9 @@ def create_default_connections(session=None):
 
 
 def initdb():
+    """
+    Initialize Airflow database.
+    """
     upgradedb()
 
     create_default_connections()
@@ -290,10 +302,13 @@ def initdb():
     models.DAG.deactivate_unknown_dags(dagbag.dags.keys())
 
     from flask_appbuilder.models.sqla import Base
-    Base.metadata.create_all(settings.engine)
+    Base.metadata.create_all(settings.engine)  # pylint: disable=no-member
 
 
 def upgradedb():
+    """
+    Upgrade the database.
+    """
     # alembic adds significant import time, so we import it lazily
     from alembic import command
     from alembic.config import Config
@@ -322,11 +337,12 @@ def resetdb():
 
     connection = settings.engine.connect()
     models.base.Base.metadata.drop_all(connection)
-    mc = MigrationContext.configure(connection)
-    if mc._version.exists(connection):
-        mc._version.drop(connection)
+    migartion_ctx = MigrationContext.configure(connection)
+    version = migartion_ctx._version  # pylint: disable=protected-access
+    if version.exists(connection):
+        version.drop(connection)
 
     from flask_appbuilder.models.sqla import Base
-    Base.metadata.drop_all(connection)
+    Base.metadata.drop_all(connection)  # pylint: disable=no-member
 
     initdb()

airflow/utils/decorators.py

@@ -64,9 +64,7 @@ def wrapper(*args, **kwargs):
             dag_args = copy(dag.default_args) or {}
             dag_params = copy(dag.params) or {}
 
-        params = {}
-        if 'params' in kwargs:
-            params = kwargs['params']
+        params = kwargs.get('params', {})
         dag_params.update(params)
 
         default_args = {}

airflow/utils/email.py

@@ -92,37 +92,43 @@ def send_email_smtp(to, subject, html_content, files=None,
             part['Content-ID'] = '<%s>' % basename
             msg.attach(part)
 
-    send_MIME_email(smtp_mail_from, recipients, msg, dryrun)
+    send_mime_email(smtp_mail_from, recipients, msg, dryrun)
 
 
-def send_MIME_email(e_from, e_to, mime_msg, dryrun=False):
+def send_mime_email(e_from, e_to, mime_msg, dryrun=False):
+    """
+    Send MIME email.
+    """
     log = LoggingMixin().log
 
-    SMTP_HOST = conf.get('smtp', 'SMTP_HOST')
-    SMTP_PORT = conf.getint('smtp', 'SMTP_PORT')
-    SMTP_STARTTLS = conf.getboolean('smtp', 'SMTP_STARTTLS')
-    SMTP_SSL = conf.getboolean('smtp', 'SMTP_SSL')
-    SMTP_USER = None
-    SMTP_PASSWORD = None
+    smtp_host = conf.get('smtp', 'SMTP_HOST')
+    smtp_port = conf.getint('smtp', 'SMTP_PORT')
+    smtp_starttls = conf.getboolean('smtp', 'SMTP_STARTTLS')
+    smtp_ssl = conf.getboolean('smtp', 'SMTP_SSL')
+    smtp_user = None
+    smtp_password = None
 
     try:
-        SMTP_USER = conf.get('smtp', 'SMTP_USER')
-        SMTP_PASSWORD = conf.get('smtp', 'SMTP_PASSWORD')
+        smtp_user = conf.get('smtp', 'SMTP_USER')
+        smtp_password = conf.get('smtp', 'SMTP_PASSWORD')
     except AirflowConfigException:
         log.debug("No user/password found for SMTP, so logging in with no authentication.")
 
     if not dryrun:
-        s = smtplib.SMTP_SSL(SMTP_HOST, SMTP_PORT) if SMTP_SSL else smtplib.SMTP(SMTP_HOST, SMTP_PORT)
-        if SMTP_STARTTLS:
-            s.starttls()
-        if SMTP_USER and SMTP_PASSWORD:
-            s.login(SMTP_USER, SMTP_PASSWORD)
+        conn = smtplib.SMTP_SSL(smtp_host, smtp_port) if smtp_ssl else smtplib.SMTP(smtp_host, smtp_port)
+        if smtp_starttls:
+            conn.starttls()
+        if smtp_user and smtp_password:
+            conn.login(smtp_user, smtp_password)
         log.info("Sent an alert email to %s", e_to)
-        s.sendmail(e_from, e_to, mime_msg.as_string())
-        s.quit()
+        conn.sendmail(e_from, e_to, mime_msg.as_string())
+        conn.quit()
 
 
 def get_email_address_list(addresses: Union[str, Iterable[str]]) -> List[str]:
+    """
+    Get list of email addresses.
+    """
     if isinstance(addresses, str):
         return _get_email_list_from_str(addresses)
 

airflow/utils/file.py

@@ -25,7 +25,10 @@
 from airflow import LoggingMixin, conf
 
 
-def TemporaryDirectory(*args, **kwargs):
+def TemporaryDirectory(*args, **kwargs):  # pylint: disable=invalid-name
+    """
+    This function is deprecated. Please use `tempfile.TemporaryDirectory`
+    """
     import warnings
     from tempfile import TemporaryDirectory as TmpDir
     warnings.warn(