Improve error for contrib when the dependency packages are not installed

luigi/contrib/batch.py

@@ -71,8 +71,10 @@
 
 try:
     import boto3
+
+    _boto3_enabled = True
 except ImportError:
-    logger.warning("boto3 is not installed. BatchTasks require boto3")
+    _boto3_enabled = False
 
 
 class BatchJobException(Exception):
@@ -88,6 +90,8 @@ def _random_id():
 
 class BatchClient:
     def __init__(self, poll_time=POLL_TIME):
+        if not _boto3_enabled:
+            raise ImportError("boto3 is required for Batch functionality. Install it with: pip install boto3")
         self.poll_time = poll_time
         self._client = boto3.client("batch")
         self._log_client = boto3.client("logs")
@@ -193,6 +197,11 @@ class BatchTask(luigi.Task):
     job_queue = luigi.OptionalParameter(default=None)
     poll_time = luigi.IntParameter(default=POLL_TIME)
 
+    def __init__(self, *args, **kwargs):
+        if not _boto3_enabled:
+            raise ImportError("boto3 is required for Batch functionality. Install it with: pip install boto3")
+        super().__init__(*args, **kwargs)
+
     def run(self):
         bc = BatchClient(self.poll_time)
         job_id = bc.submit_job(self.job_definition, self.parameters, job_name=self.job_name, queue=self.job_queue)

luigi/contrib/bigquery.py

@@ -32,8 +32,10 @@
 try:
     import httplib2
     from googleapiclient import discovery, errors, http
+
+    _bigquery_enabled = True
 except ImportError:
-    logger.warning("BigQuery module imported, but google-api-python-client is not installed. Any BigQuery task will fail")
+    _bigquery_enabled = False
 else:
     RETRYABLE_ERRORS = (httplib2.HttpLib2Error, IOError, TimeoutError, BrokenPipeError)
 
@@ -142,6 +144,8 @@ class BigQueryClient:
     """
 
     def __init__(self, oauth_credentials=None, descriptor="", http_=None):
+        if not _bigquery_enabled:
+            raise ImportError("google-api-python-client is required for BigQuery functionality. Install it with: pip install google-api-python-client")
         # Save initialisation arguments in case we need to re-create client
         # due to connection timeout
         self.oauth_credentials = oauth_credentials
@@ -398,6 +402,8 @@ def copy(self, source_table, dest_table, create_disposition=CreateDisposition.CR
 
 class BigQueryTarget(luigi.target.Target):
     def __init__(self, project_id, dataset_id, table_id, client=None, location=None):
+        if not _bigquery_enabled:
+            raise ImportError("google-api-python-client is required for BigQuery functionality. Install it with: pip install google-api-python-client")
         self.table = BQTable(project_id=project_id, dataset_id=dataset_id, table_id=table_id, location=location)
         self.client = client or BigQueryClient()
 

luigi/contrib/bigquery_avro.py

@@ -11,8 +11,10 @@
 try:
     import avro
     import avro.datafile
+
+    _avro_enabled = True
 except ImportError:
-    logger.warning("bigquery_avro module imported, but avro is not installed. Any BigQueryLoadAvro task will fail to propagate schema documentation")
+    _avro_enabled = False
 
 
 class BigQueryLoadAvro(BigQueryLoadTask):
@@ -30,6 +32,11 @@ class BigQueryLoadAvro(BigQueryLoadTask):
 
     source_format = SourceFormat.AVRO
 
+    def __init__(self, *args, **kwargs):
+        if not _avro_enabled:
+            raise ImportError("avro is required for BigQueryLoadAvro. Install it with: pip install avro-python3")
+        super().__init__(*args, **kwargs)
+
     def _avro_uri(self, target):
         path_or_uri = target.uri if hasattr(target, "uri") else target.path
         return path_or_uri if path_or_uri.endswith(".avro") else path_or_uri.rstrip("/") + "/*.avro"

luigi/contrib/datadog_metric.py

@@ -8,8 +8,10 @@
 
 try:
     from datadog import api, initialize, statsd
+
+    _datadog_enabled = True
 except ImportError:
-    logger.warning("Loading datadog module without datadog installed. Will crash at runtime if datadog functionality is used.")
+    _datadog_enabled = False
 
 
 class datadog(Config):
@@ -24,6 +26,8 @@ class datadog(Config):
 
 class DatadogMetricsCollector(MetricsCollector):
     def __init__(self, *args, **kwargs):
+        if not _datadog_enabled:
+            raise ImportError("datadog is required for DatadogMetricsCollector. Install it with: pip install datadog")
         self._config = datadog(**kwargs)
 
         initialize(api_key=self._config.api_key, app_key=self._config.app_key, statsd_host=self._config.statsd_host, statsd_port=self._config.statsd_port)

luigi/contrib/dataproc.py

@@ -19,11 +19,9 @@
     DEFAULT_CREDENTIALS, _ = google.auth.default()
     authenticate_kwargs = gcp.get_authenticate_kwargs(DEFAULT_CREDENTIALS)
     _dataproc_client = discovery.build("dataproc", "v1", cache_discovery=False, **authenticate_kwargs)
+    _dataproc_enabled = True
 except ImportError:
-    logger.warning(
-        "Loading Dataproc module without the python packages googleapiclient & google-auth. \
-        This will crash at runtime if Dataproc functionality is used."
-    )
+    _dataproc_enabled = False
 
 
 def get_dataproc_client():
@@ -42,6 +40,11 @@ class _DataprocBaseTask(luigi.Task):
 
     dataproc_client = get_dataproc_client()
 
+    def __init__(self, *args, **kwargs):
+        if not _dataproc_enabled:
+            raise ImportError("google-api-python-client is required for Dataproc functionality. Install it with: pip install google-api-python-client")
+        super().__init__(*args, **kwargs)
+
 
 class DataprocBaseTask(_DataprocBaseTask):
     """

luigi/contrib/docker_runner.py

@@ -48,9 +48,9 @@
     import docker
     from docker.errors import APIError, ContainerError, ImageNotFound
 
+    _docker_enabled = True
 except ImportError:
-    logger.warning("docker is not installed. DockerTask requires docker.")
-    docker = None
+    _docker_enabled = False
 
 # TODO: may need to implement this logic for remote hosts
 # class dockerconfig(luigi.Config):
@@ -143,6 +143,8 @@ def __init__(self, *args, **kwargs):
         - create a tmp dir
         - add the temp dir to the volume binds specified in the task
         """
+        if not _docker_enabled:
+            raise ImportError("docker is required for DockerTask. Install it with: pip install docker")
         super(DockerTask, self).__init__(*args, **kwargs)
         self.__logger = logger
 

luigi/contrib/dropbox.py

@@ -33,10 +33,10 @@
     import dropbox.dropbox_client
     import dropbox.exceptions
     import dropbox.files
+
+    _dropbox_enabled = True
 except ImportError:
-    logger.warning(
-        "Loading Dropbox module without the python package dropbox (https://pypi.org/project/dropbox/). Will crash at runtime if Dropbox functionality is used."
-    )
+    _dropbox_enabled = False
 
 
 def accept_trailing_slash_in_existing_dirpaths(func):
@@ -74,6 +74,8 @@ def __init__(self, token, user_agent="Luigi", root_namespace_id=None):
         :param str token: Dropbox Oauth2 Token. See :class:`DropboxTarget` for more information about generating a token
         :param str root_namespace_id: Root namespace ID for interacting with Team Spaces
         """
+        if not _dropbox_enabled:
+            raise ImportError("dropbox is required for Dropbox functionality. Install it with: pip install dropbox")
         if not token:
             raise ValueError("The token parameter must contain a valid Dropbox Oauth2 Token")
 
@@ -292,6 +294,8 @@ def __init__(self, path, token, format=None, user_agent="Luigi", root_namespace_
 
 
         """
+        if not _dropbox_enabled:
+            raise ImportError("dropbox is required for Dropbox functionality. Install it with: pip install dropbox")
         super(DropboxTarget, self).__init__(path)
 
         if not token:

luigi/contrib/ecs.py

@@ -62,8 +62,9 @@
     import boto3
 
     client = boto3.client("ecs")
+    _boto3_enabled = True
 except ImportError:
-    logger.warning("boto3 is not installed. ECSTasks require boto3")
+    _boto3_enabled = False
 
 POLL_TIME = 2
 
@@ -138,6 +139,11 @@ class ECSTask(luigi.Task):
     task_def = luigi.OptionalParameter(default=None)
     cluster = luigi.Parameter(default="default")
 
+    def __init__(self, *args, **kwargs):
+        if not _boto3_enabled:
+            raise ImportError("boto3 is required for ECSTask. Install it with: pip install boto3")
+        super().__init__(*args, **kwargs)
+
     @property
     def ecs_task_ids(self):
         """Expose the ECS task ID"""

luigi/contrib/esindex.py

@@ -100,8 +100,9 @@ def docs(self):
     from elasticsearch.connection import Urllib3HttpConnection
     from elasticsearch.helpers import bulk
 
+    _elasticsearch_enabled = True
 except ImportError:
-    logger.warning("Loading esindex module without elasticsearch installed. Will crash at runtime if esindex functionality is used.")
+    _elasticsearch_enabled = False
 
 
 class ElasticsearchTarget(luigi.Target):
@@ -129,6 +130,8 @@ def __init__(self, host, port, index, doc_type, update_id, marker_index_hist_siz
         :param extra_elasticsearch_args: extra args for Elasticsearch
         :type Extra: dict
         """
+        if not _elasticsearch_enabled:
+            raise ImportError("elasticsearch is required for ElasticsearchTarget. Install it with: pip install elasticsearch")
         if extra_elasticsearch_args is None:
             extra_elasticsearch_args = {}
 

luigi/contrib/ftp.py

@@ -73,7 +73,7 @@ def _sftp_connect(self):
         try:
             import pysftp
         except ImportError:
-            logger.warning("Please install pysftp to use SFTP.")
+            raise ImportError("pysftp is required for SFTP functionality. Install it with: pip install pysftp")
 
         self.conn = pysftp.Connection(self.host, username=self.username, password=self.password, port=self.port, **self.pysftp_conn_kwargs)
 

luigi/contrib/gcp.py

@@ -9,11 +9,10 @@
 try:
     import google.auth
     import httplib2
+
+    _gcp_enabled = True
 except ImportError:
-    logger.warning(
-        "Loading GCP module without the python packages httplib2, google-auth. \
-        This *could* crash at runtime if no other credentials are provided."
-    )
+    _gcp_enabled = False
 
 
 def get_authenticate_kwargs(oauth_credentials=None, http_=None):
@@ -25,6 +24,8 @@ def get_authenticate_kwargs(oauth_credentials=None, http_=None):
 
     Used by `gcs.GCSClient` and `bigquery.BigQueryClient` to initiate the API Client
     """
+    if not _gcp_enabled:
+        raise ImportError("google-auth and httplib2 are required for GCP functionality. Install them with: pip install google-auth httplib2")
     if oauth_credentials:
         authenticate_kwargs = {"credentials": oauth_credentials}
     elif http_:

luigi/contrib/gcs.py

@@ -40,11 +40,10 @@
 try:
     import httplib2
     from googleapiclient import discovery, errors, http
+
+    _gcs_enabled = True
 except ImportError:
-    logger.warning(
-        "Loading GCS module without the python packages googleapiclient & google-auth. \
-        This will crash at runtime if GCS functionality is used."
-    )
+    _gcs_enabled = False
 else:
     RETRYABLE_ERRORS = (httplib2.HttpLib2Error, IOError)
 
@@ -122,6 +121,8 @@ class GCSClient(luigi.target.FileSystem):
     """
 
     def __init__(self, oauth_credentials=None, descriptor="", http_=None, chunksize=CHUNKSIZE, **discovery_build_kwargs):
+        if not _gcs_enabled:
+            raise ImportError("googleapiclient is required for GCS functionality. Install it with: pip install google-api-python-client")
         self.chunksize = chunksize
         authenticate_kwargs = gcp.get_authenticate_kwargs(oauth_credentials, http_)
 
@@ -435,6 +436,8 @@ class GCSTarget(luigi.target.FileSystemTarget):
     fs = None
 
     def __init__(self, path, format=None, client=None):
+        if not _gcs_enabled:
+            raise ImportError("googleapiclient is required for GCS functionality. Install it with: pip install google-api-python-client")
         super(GCSTarget, self).__init__(path)
         if format is None:
             format = luigi.format.get_default_format()
@@ -484,6 +487,8 @@ def __init__(self, path, format=None, client=None, flag="_SUCCESS"):
         :param flag:
         :type flag: str
         """
+        if not _gcs_enabled:
+            raise ImportError("googleapiclient is required for GCS functionality. Install it with: pip install google-api-python-client")
         if format is None:
             format = luigi.format.get_default_format()
 

luigi/contrib/kubernetes.py

@@ -46,8 +46,10 @@
     from pykube.config import KubeConfig
     from pykube.http import HTTPClient
     from pykube.objects import Job, Pod
+
+    _pykube_enabled = True
 except ImportError:
-    logger.warning("pykube is not installed. KubernetesJobTask requires pykube.")
+    _pykube_enabled = False
 
 
 class kubernetes(luigi.Config):
@@ -62,6 +64,11 @@ class KubernetesJobTask(luigi.Task):
     __DEFAULT_POD_CREATION_INTERVAL = 5
     _kubernetes_config = None  # Needs to be loaded at runtime
 
+    def __init__(self, *args, **kwargs):
+        if not _pykube_enabled:
+            raise ImportError("pykube is required for KubernetesJobTask. Install it with: pip install pykube-ng")
+        super().__init__(*args, **kwargs)
+
     def _init_kubernetes(self):
         self.__logger = logger
         self.__logger.debug("Kubernetes auth method: " + self.auth_method)

luigi/contrib/mssqldb.py

@@ -23,11 +23,10 @@
 
 try:
     from pymssql import _mssql
+
+    _pymssql_enabled = True
 except ImportError:
-    logger.warning(
-        "Loading MSSQL module without the python package pymssql. \
-        This will crash at runtime if SQL Server functionality is used."
-    )
+    _pymssql_enabled = False
 
 
 class MSSqlTarget(luigi.Target):
@@ -56,6 +55,8 @@ def __init__(self, host, database, user, password, table, update_id):
         :param update_id: an identifier for this data set.
         :type update_id: str
         """
+        if not _pymssql_enabled:
+            raise ImportError("pymssql is required for SQL Server functionality. Install it with: pip install pymssql")
         if ":" in host:
             self.host, self.port = host.split(":")
             self.port = int(self.port)

luigi/contrib/mysqldb.py

@@ -25,11 +25,10 @@
 try:
     import mysql.connector
     from mysql.connector import Error, errorcode
+
+    _mysql_enabled = True
 except ImportError:
-    logger.warning(
-        "Loading MySQL module without the python package mysql-connector-python. \
-       This will crash at runtime if MySQL functionality is used."
-    )
+    _mysql_enabled = False
 
 
 class MySqlTarget(luigi.Target):
@@ -56,6 +55,8 @@ def __init__(self, host, database, user, password, table, update_id, **cnx_kwarg
         :param cnx_kwargs: optional params for mysql connector constructor.
             See https://dev.mysql.com/doc/connector-python/en/connector-python-connectargs.html.
         """
+        if not _mysql_enabled:
+            raise ImportError("mysql-connector-python is required for MySQL functionality. Install it with: pip install mysql-connector-python")
         if ":" in host:
             self.host, self.port = host.split(":")
             self.port = int(self.port)

luigi/contrib/pai.py

@@ -44,8 +44,9 @@
     import requests as rs
     from requests.exceptions import HTTPError
 
+    _requests_enabled = True
 except ImportError:
-    logger.warning("requests is not installed. PaiTask requires requests.")
+    _requests_enabled = False
 
 
 def slot_to_dict(o):
@@ -240,6 +241,8 @@ def __init__(self, *args, **kwargs):
         :param pai_url: The rest server url of PAI clusters, default is 'http://127.0.0.1:9186'.
         :param token: The token used to auth the rest server of PAI.
         """
+        if not _requests_enabled:
+            raise ImportError("requests is required for PaiTask. Install it with: pip install requests")
         super(PaiTask, self).__init__(*args, **kwargs)
         self.__init_token()