[AIRFLOW-4386] Remove urlparse and replace it with urllib.parse (#5154)

* [AIRFLOW-4386] Remove urlparse and replace it with urllib.parse

* Replace remaining imports

* Update test_utils.py

airflow/contrib/example_dags/example_gcp_sql.py

@@ -29,6 +29,7 @@
 """
 
 import os
+from urllib.parse import urlsplit
 
 import airflow
 from airflow import models
@@ -41,8 +42,6 @@
     GoogleCloudStorageBucketCreateAclEntryOperator, \
     GoogleCloudStorageObjectCreateAclEntryOperator
 
-from six.moves.urllib.parse import urlsplit
-
 # [START howto_operator_cloudsql_arguments]
 GCP_PROJECT_ID = os.environ.get('GCP_PROJECT_ID', 'example-project')
 INSTANCE_NAME = os.environ.get('GCSQL_MYSQL_INSTANCE_NAME', 'test-mysql')

airflow/contrib/example_dags/example_gcp_sql_query.py

@@ -41,8 +41,7 @@
 import os
 import subprocess
 from os.path import expanduser
-
-from six.moves.urllib.parse import quote_plus
+from urllib.parse import quote_plus
 
 import airflow
 from airflow import models

airflow/contrib/hooks/gcp_sql_hook.py

@@ -33,7 +33,7 @@
 
 from googleapiclient.errors import HttpError
 from subprocess import Popen, PIPE
-from six.moves.urllib.parse import quote_plus
+from urllib.parse import quote_plus
 
 import requests
 from googleapiclient.discovery import build

airflow/contrib/hooks/gcs_hook.py

@@ -22,6 +22,7 @@
 import shutil
 
 from google.cloud import storage
+from urllib.parse import urlparse
 
 from airflow.contrib.hooks.gcp_api_base_hook import GoogleCloudBaseHook
 from airflow.exceptions import AirflowException
@@ -554,12 +555,6 @@ def _parse_gcs_url(gsurl):
     Given a Google Cloud Storage URL (gs://<bucket>/<blob>), returns a
     tuple containing the corresponding bucket and blob.
     """
-    # Python 3
-    try:
-        from urllib.parse import urlparse
-    # Python 2
-    except ImportError:
-        from urlparse import urlparse
 
     parsed_url = urlparse(gsurl)
     if not parsed_url.netloc:

airflow/contrib/utils/mlengine_operator_utils.py

@@ -26,7 +26,7 @@
 from airflow.contrib.operators.dataflow_operator import DataFlowPythonOperator
 from airflow.exceptions import AirflowException
 from airflow.operators.python_operator import PythonOperator
-from six.moves.urllib.parse import urlsplit
+from urllib.parse import urlsplit
 
 
 def create_evaluate_ops(task_prefix,

airflow/utils/log/gcs_task_handler.py

@@ -19,11 +19,12 @@
 import os
 
 from cached_property import cached_property
+from urllib.parse import urlparse
 
 from airflow import configuration
 from airflow.exceptions import AirflowException
-from airflow.utils.log.logging_mixin import LoggingMixin
 from airflow.utils.log.file_task_handler import FileTaskHandler
+from airflow.utils.log.logging_mixin import LoggingMixin
 
 
 class GCSTaskHandler(FileTaskHandler, LoggingMixin):
@@ -167,13 +168,6 @@ def parse_gcs_url(gsurl):
         Given a Google Cloud Storage URL (gs://<bucket>/<blob>), returns a
         tuple containing the corresponding bucket and blob.
         """
-        # Python 3
-        try:
-            from urllib.parse import urlparse
-        # Python 2
-        except ImportError:
-            from urlparse import urlparse
-
         parsed_url = urlparse(gsurl)
         if not parsed_url.netloc:
             raise AirflowException('Please provide a bucket name')

airflow/www/app.py

@@ -19,21 +19,21 @@
 #
 import logging
 import socket
-from typing import Any
 
 from flask import Flask
 from flask_appbuilder import AppBuilder, SQLA
 from flask_caching import Cache
 from flask_wtf.csrf import CSRFProtect
-from six.moves.urllib.parse import urlparse
-from werkzeug.wsgi import DispatcherMiddleware
+from typing import Any
+from urllib.parse import urlparse
 from werkzeug.contrib.fixers import ProxyFix
+from werkzeug.wsgi import DispatcherMiddleware
 
-from airflow import settings
 from airflow import configuration as conf
+from airflow import settings
 from airflow.logging_config import configure_logging
-from airflow.www.static_config import configure_manifest_files
 from airflow.utils.json import AirflowJsonEncoder
+from airflow.www.static_config import configure_manifest_files
 
 app = None  # type: Any
 appbuilder = None

airflow/www/utils.py

@@ -38,7 +38,7 @@
 from flask_appbuilder.models.sqla.interface import SQLAInterface
 import flask_appbuilder.models.sqla.filters as fab_sqlafilters
 import sqlalchemy as sqla
-from six.moves.urllib.parse import urlencode
+from urllib.parse import urlencode
 
 from airflow import configuration
 from airflow.models import BaseOperator

tests/contrib/hooks/test_gcp_mlengine_hook.py

@@ -16,20 +16,17 @@
 # under the License.
 
 import json
-import mock
 import unittest
 
-try:  # python 2
-    from urlparse import urlparse, parse_qsl
-except ImportError:  # python 3
-    from urllib.parse import urlparse, parse_qsl
-
-from airflow.contrib.hooks import gcp_mlengine_hook as hook
-from googleapiclient.errors import HttpError
+import mock
+import requests
+from google.auth.exceptions import GoogleAuthError
 from googleapiclient.discovery import build_from_document
+from googleapiclient.errors import HttpError
 from googleapiclient.http import HttpMockSequence
-from google.auth.exceptions import GoogleAuthError
-import requests
+from urllib.parse import urlparse, parse_qsl
+
+from airflow.contrib.hooks import gcp_mlengine_hook as hook
 
 cml_available = True
 try:

tests/contrib/operators/test_gcp_sql_operator_system_helper.py

@@ -27,7 +27,7 @@
 
 import time
 
-from six.moves.urllib.parse import urlsplit
+from urllib.parse import urlsplit
 
 from tests.contrib.utils.base_gcp_system_test_case import RetrieveVariables
 from tests.contrib.utils.gcp_authenticator import GcpAuthenticator, GCP_CLOUDSQL_KEY

tests/www/test_utils.py

@@ -17,16 +17,15 @@
 # specific language governing permissions and limitations
 # under the License.
 
+import unittest
 from datetime import datetime
+from urllib.parse import parse_qs
 
-from bs4 import BeautifulSoup
 import mock
-from six.moves.urllib.parse import parse_qs
+from bs4 import BeautifulSoup
 
 from airflow.www import utils
 
-import unittest
-
 
 class UtilsTest(unittest.TestCase):
     def test_empty_variable_should_not_be_hidden(self):