[AIRFLOW-3606] Fix Flake8 test & fix the Flake8 errors introduced sin…

…ce Flake8 test was broken (apache#4415)

The flake8 test in the Travis CI was broken since apache#4361
(apache@7a6acbf )

And some Flake8 errors (code style/quality issues. found in 10 files) were introduce since flake8 test was broken.

.flake8

@@ -1,4 +1,3 @@
 [flake8]
 max-line-length = 110
 ignore = E731
-exclude = .git, env, venv, .venv, VENV, .tox, .eggs, .*

airflow/bin/cli.py

@@ -69,7 +69,6 @@
 from airflow.www_rbac.app import create_app as create_app_rbac
 from airflow.www_rbac.app import cached_appbuilder
 
-from sqlalchemy import func
 from sqlalchemy.orm import exc
 
 api.load_auth()

airflow/migrations/versions/a56c9515abdc_remove_dag_stat_table.py

@@ -6,9 +6,9 @@
 # to you under the Apache License, Version 2.0 (the
 # "License"); you may not use this file except in compliance
 # with the License.  You may obtain a copy of the License at
-# 
+#
 #   http://www.apache.org/licenses/LICENSE-2.0
-# 
+#
 # Unless required by applicable law or agreed to in writing,
 # software distributed under the License is distributed on an
 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
@@ -24,15 +24,15 @@
 
 """
 
+from alembic import op
+import sqlalchemy as sa
+
 # revision identifiers, used by Alembic.
 revision = 'a56c9515abdc'
 down_revision = 'c8ffec048a3b'
 branch_labels = None
 depends_on = None
 
-from alembic import op
-import sqlalchemy as sa
-
 
 def upgrade():
     op.drop_table("dag_stats")

airflow/migrations/versions/c8ffec048a3b_add_fields_to_dag.py

@@ -6,9 +6,9 @@
 # to you under the Apache License, Version 2.0 (the
 # "License"); you may not use this file except in compliance
 # with the License.  You may obtain a copy of the License at
-# 
+#
 #   http://www.apache.org/licenses/LICENSE-2.0
-# 
+#
 # Unless required by applicable law or agreed to in writing,
 # software distributed under the License is distributed on an
 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
@@ -24,15 +24,15 @@
 
 """
 
+from alembic import op
+import sqlalchemy as sa
+
 # revision identifiers, used by Alembic.
 revision = 'c8ffec048a3b'
 down_revision = '41f5f12752f8'
 branch_labels = None
 depends_on = None
 
-from alembic import op
-import sqlalchemy as sa
-
 
 def upgrade():
     op.add_column('dag', sa.Column('description', sa.Text(), nullable=True))

airflow/models/__init__.py

@@ -43,7 +43,6 @@
 import getpass
 import imp
 import importlib
-import itertools
 import zipfile
 import jinja2
 import json

airflow/www/views.py

@@ -554,7 +554,9 @@ def dag_stats(self, session=None):
         dm = models.DagModel
         dag_ids = session.query(dm.dag_id)
 
-        dag_state_stats = session.query(dr.dag_id, dr.state, sqla.func.count(dr.state)).group_by(dr.dag_id, dr.state)
+        dag_state_stats = (
+            session.query(dr.dag_id, dr.state, sqla.func.count(dr.state)).group_by(dr.dag_id, dr.state)
+        )
 
         data = {}
         for (dag_id, ) in dag_ids:

airflow/www_rbac/views.py

@@ -281,7 +281,8 @@ def dag_stats(self, session=None):
 
         filter_dag_ids = appbuilder.sm.get_accessible_dag_ids()
 
-        dag_state_stats = session.query(dr.dag_id, dr.state, sqla.func.count(dr.state)).group_by(dr.dag_id, dr.state)
+        dag_state_stats = session.query(dr.dag_id, dr.state, sqla.func.count(dr.state))\
+            .group_by(dr.dag_id, dr.state)
 
         payload = {}
         if filter_dag_ids:

tests/contrib/executors/test_kubernetes_executor.py

@@ -243,7 +243,8 @@ def test_worker_git_dags(self):
         self.assertTrue(dag_volume_mount[0]['readOnly'])
 
         init_container = worker_config._get_init_containers(volume_mounts)[0]
-        init_container_volume_mount = [mount for mount in init_container['volumeMounts'] if mount['name'] == 'airflow-dags']
+        init_container_volume_mount = [mount for mount in init_container['volumeMounts']
+                                       if mount['name'] == 'airflow-dags']
 
         self.assertEqual('git-sync-clone', init_container['name'])
         self.assertEqual('gcr.io/google-containers/git-sync-amd64:v2.0.5', init_container['image'])

tests/models.py

@@ -747,7 +747,8 @@ def test_sync_to_db(self, mock_now):
         self.assertEqual(orm_dag.last_scheduler_run, now)
         self.assertTrue(orm_dag.is_active)
         self.assertIsNone(orm_dag.default_view)
-        self.assertEqual(orm_dag.get_default_view(), configuration.conf.get('webserver', 'dag_default_view').lower())
+        self.assertEqual(orm_dag.get_default_view(),
+                         configuration.conf.get('webserver', 'dag_default_view').lower())
 
         orm_subdag = session.query(DagModel).filter(
             DagModel.dag_id == 'dag.subtask').one()

tests/www/test_views.py

@@ -30,7 +30,6 @@
 
 from urllib.parse import quote_plus
 from werkzeug.test import Client
-from sqlalchemy import func
 
 from airflow import models, configuration
 from airflow.config_templates.airflow_local_settings import DEFAULT_LOGGING_CONFIG

tests/www_rbac/test_views.py

@@ -1448,7 +1448,7 @@ def test_trigger_dag_button(self):
         self.session.query(DR).delete()
         self.session.commit()
 
-        resp = self.client.get('trigger?dag_id={}'.format(test_dag_id))
+        self.client.get('trigger?dag_id={}'.format(test_dag_id))
 
         run = self.session.query(DR).filter(DR.dag_id == test_dag_id).first()
         self.assertIsNotNone(run)