[AIRFLOW-2407] Resolve Python undefined names

airflow/contrib/auth/backends/kerberos_auth.py

@@ -17,8 +17,9 @@
 # specific language governing permissions and limitations
 # under the License.
 
+import logging
 import flask_login
-from flask_login import login_required, current_user, logout_user
+from flask_login import current_user
 from flask import flash
 from wtforms import (
     Form, PasswordField, StringField)

airflow/contrib/hooks/aws_hook.py

@@ -20,6 +20,7 @@
 
 import boto3
 import configparser
+import logging
 
 from airflow.exceptions import AirflowException
 from airflow.hooks.base_hook import BaseHook

airflow/contrib/operators/datastore_export_operator.py

@@ -19,6 +19,7 @@
 #
 from airflow.contrib.hooks.datastore_hook import DatastoreHook
 from airflow.contrib.hooks.gcs_hook import GoogleCloudStorageHook
+from airflow.exceptions import AirflowException
 from airflow.models import BaseOperator
 from airflow.utils.decorators import apply_defaults
 

airflow/contrib/operators/datastore_import_operator.py

@@ -18,6 +18,7 @@
 # under the License.
 #
 from airflow.contrib.hooks.datastore_hook import DatastoreHook
+from airflow.exceptions import AirflowException
 from airflow.models import BaseOperator
 from airflow.utils.decorators import apply_defaults
 
@@ -88,11 +89,9 @@ def execute(self, context):
         result = ds_hook.poll_operation_until_done(operation_name,
                                                    self.polling_interval_in_seconds)
 
-
         state = result['metadata']['common']['state']
         if state != 'SUCCESSFUL':
             raise AirflowException('Operation failed: result={}'.format(result))
 
         if self.xcom_push:
             return result
-

airflow/contrib/sensors/qubole_sensor.py

@@ -56,6 +56,7 @@ def __init__(self, data, qubole_conn_id="qubole_default", *args, **kwargs):
         super(QuboleSensor, self).__init__(*args, **kwargs)
 
     def poke(self, context):
+        global this  # apache/incubator-airflow/pull/3297#issuecomment-385988083
         conn = BaseHook.get_connection(self.qubole_conn_id)
         Qubole.configure(api_token=conn.password, api_url=conn.host)
 
@@ -65,7 +66,7 @@ def poke(self, context):
         try:
             status = self.sensor_class.check(self.data)
         except Exception as e:
-            logging.exception(e)
+            this.log.exception(e)
             status = False
 
         this.log.info('Status of this Poke: %s', status)

tests/operators/test_virtualenv_operator.py

@@ -163,6 +163,7 @@ def f(a):
 
     def test_string_args(self):
         def f():
+            global virtualenv_string_args
             print(virtualenv_string_args)
             if virtualenv_string_args[0] != virtualenv_string_args[2]:
                 raise Exception
@@ -197,4 +198,3 @@ def test_context(self):
         def f(**kwargs):
             return kwargs['templates_dict']['ds']
         self._run_as_operator(f, templates_dict={'ds': '{{ ds }}'})
-