works up until upload metadata

problem - the method in the extractor is not getting the secret key
possible fix - use token for secret key

Author: tcnichol

pyclowder/connectors.py

@@ -58,6 +58,8 @@
 from dotenv import load_dotenv
 load_dotenv()
 
+clowder_version = float(os.getenv('clowder_version'))
+
 
 class Connector(object):
     """ Class that will listen for messages.
@@ -415,17 +417,15 @@ def _process_message(self, body):
             return
 
         # register extractor
-        # TODO make work for clowder2.0
-        if float(os.getenv('clowder_version')) == 2.0:
-            print('do differently')
-            registration_url  = "%sapi/v2/extractors" % source_host
-            if registration_url not in Connector.registered_clowder:
-                Connector.registered_clowder.append(registration_url)
-                self.register_extractor_v2(registration_url, token)
+        if clowder_version >= 2.0:
+            url = "%sapi/v2/extractors" % source_host
         else:
             url = "%sapi/extractors" % source_host
-            if url not in Connector.registered_clowder:
-                Connector.registered_clowder.append(url)
+        if url not in Connector.registered_clowder:
+            Connector.registered_clowder.append(url)
+            if clowder_version >= 2.0:
+                self.register_extractor("%s" % (url), token=token)
+            else:
                 self.register_extractor("%s?key=%s" % (url, secret_key))
 
         # tell everybody we are starting to process the file
@@ -446,16 +446,16 @@ def _process_message(self, body):
                         found_local = False
                         try:
                             if check_result != pyclowder.utils.CheckMessage.bypass:
-                                if float(os.getenv('clowder_version')) == 2.0:
-                                    file_metadata = pyclowder.files.download_info_v2(self, host, token, resource["id"])
+                                if clowder_version >= 2.0:
+                                    file_metadata = pyclowder.files.download_info(self, host, secret_key, resource["id"], token=token)
                                 else:
                                     file_metadata = pyclowder.files.download_info(self, host, secret_key, resource["id"])
                                 file_path = self._check_for_local_file(file_metadata)
                                 if not file_path:
-                                    if float(os.getenv('clowder_version')) == 2.0:
-                                        file_path = pyclowder.files.download_v2(self, host, token, resource["id"],
-                                                                             resource["intermediate_id"],
-                                                                             resource["file_ext"])
+                                    if clowder_version >= 2.0:
+                                        file_path = pyclowder.files.download(self, host, secret_key, resource["id"],
+                                                                         resource["intermediate_id"],
+                                                                         resource["file_ext"], token=token)
                                     else:
                                         file_path = pyclowder.files.download(self, host, secret_key, resource["id"],
                                                                              resource["intermediate_id"],
@@ -537,61 +537,56 @@ def _process_message(self, body):
             else:
                 self.message_error(resource, message)
 
-    def register_extractor(self, endpoints):
+    def register_extractor(self, endpoints, token=None):
         """Register extractor info with Clowder.
 
         This assumes a file called extractor_info.json to be located in either the
         current working directory, or the folder where the main program is started.
         """
+        if clowder_version >= 2.0:
+            if not endpoints or endpoints == "":
+                return
+
+            logger = logging.getLogger(__name__)
+
+            headers = {'Content-Type': 'application/json',
+                       'Authorization': 'Bearer ' + token}
+            data = self.extractor_info
+
+            for url in endpoints.split(','):
+                if url not in Connector.registered_clowder:
+                    Connector.registered_clowder.append(url)
+                    try:
+                        result = requests.post(url.strip(), headers=headers,
+                                               data=json.dumps(data),
+                                               verify=self.ssl_verify)
+                        result.raise_for_status()
+                        logger.debug("Registering extractor with %s : %s", url, result.text)
+                    except Exception as exc:  # pylint: disable=broad-except
+                        logger.exception('Error in registering extractor: ' + str(exc))
+        else:
+            # don't do any work if we wont register the endpoint
+            if not endpoints or endpoints == "":
+                return
 
-        # don't do any work if we wont register the endpoint
-        if not endpoints or endpoints == "":
-            return
-
-        logger = logging.getLogger(__name__)
-
-        headers = {'Content-Type': 'application/json'}
-        data = self.extractor_info
-
-        for url in endpoints.split(','):
-            if url not in Connector.registered_clowder:
-                Connector.registered_clowder.append(url)
-                try:
-                    result = requests.post(url.strip(), headers=headers,
-                                           data=json.dumps(data),
-                                           verify=self.ssl_verify)
-                    result.raise_for_status()
-                    logger.debug("Registering extractor with %s : %s", url, result.text)
-                except Exception as exc:  # pylint: disable=broad-except
-                    logger.exception('Error in registering extractor: ' + str(exc))
-
-    def register_extractor_v2(self, endpoint, token):
-        """Register extractor info with Clowder.
-
-        This assumes a file called extractor_info.json to be located in either the
-        current working directory, or the folder where the main program is started.
-        """
+            logger = logging.getLogger(__name__)
 
-        # don't do any work if we wont register the endpoint
-        if not endpoint or endpoint == "":
-            return
+            headers = {'Content-Type': 'application/json'}
+            data = self.extractor_info
 
-        logger = logging.getLogger(__name__)
 
-        headers = {'Content-Type': 'application/json',
-                   'Authorization': 'Bearer ' + token}
-        data = self.extractor_info
 
-        if endpoint not in Connector.registered_clowder:
-            Connector.registered_clowder.append(endpoint)
-            try:
-                result = requests.post(endpoint.strip(), headers=headers,
-                                       data=json.dumps(data),
-                                       verify=self.ssl_verify)
-                result.raise_for_status()
-                logger.debug("Registering extractor with %s : %s", url, result.text)
-            except Exception as exc:  # pylint: disable=broad-except
-                logger.exception('Error in registering extractor: ' + str(exc))
+            for url in endpoints.split(','):
+                if url not in Connector.registered_clowder:
+                    Connector.registered_clowder.append(url)
+                    try:
+                        result = requests.post(url.strip(), headers=headers,
+                                               data=json.dumps(data),
+                                               verify=self.ssl_verify)
+                        result.raise_for_status()
+                        logger.debug("Registering extractor with %s : %s", url, result.text)
+                    except Exception as exc:  # pylint: disable=broad-except
+                        logger.exception('Error in registering extractor: ' + str(exc))
 
     # pylint: disable=no-self-use
     def status_update(self, status, resource, message):

pyclowder/files.py

@@ -15,6 +15,10 @@
 from pyclowder.datasets import get_file_list
 from pyclowder.collections import get_datasets, get_child_collections
 
+from dotenv import load_dotenv
+load_dotenv()
+clowder_version = float(os.getenv('clowder_version'))
+
 # Some sources of urllib3 support warning suppression, but not all
 try:
     from urllib3 import disable_warnings
@@ -25,7 +29,7 @@
 
 
 # pylint: disable=too-many-arguments
-def download(connector, host, key, fileid, intermediatefileid=None, ext=""):
+def download(connector, host, key, fileid, intermediatefileid=None, ext="", token=None):
     """Download file to be processed from Clowder.
 
     Keyword arguments:
@@ -39,60 +43,44 @@ def download(connector, host, key, fileid, intermediatefileid=None, ext=""):
 
     connector.message_process({"type": "file", "id": fileid}, "Downloading file.")
 
-    # TODO: intermediateid doesn't really seem to be used here, can we remove entirely?
-    if not intermediatefileid:
-        intermediatefileid = fileid
-
-    url = '%sapi/files/%s?key=%s' % (host, intermediatefileid, key)
-    result = connector.get(url, stream=True, verify=connector.ssl_verify if connector else True)
 
-    (inputfile, inputfilename) = tempfile.mkstemp(suffix=ext)
-
-    try:
-        with os.fdopen(inputfile, "wb") as outputfile:
-            for chunk in result.iter_content(chunk_size=10*1024):
-                outputfile.write(chunk)
-        return inputfilename
-    except Exception:
-        os.remove(inputfilename)
-        raise
-
-
-# pylint: disable=too-many-arguments
-def download_v2(connector, host, token, fileid, intermediatefileid=None, ext=""):
-    """Download file to be processed from Clowder.
-
-    Keyword arguments:
-    connector -- connector information, used to get missing parameters and send status updates
-    host -- the clowder host, including http and port, should end with a /
-    key -- the secret key to login to clowder
-    fileid -- the file that is currently being processed
-    intermediatefileid -- either same as fileid, or the intermediate file to be used
-    ext -- the file extension, the downloaded file will end with this extension
-    """
-
-    connector.message_process({"type": "file", "id": fileid}, "Downloading file.")
 
     # TODO: intermediateid doesn't really seem to be used here, can we remove entirely?
     if not intermediatefileid:
         intermediatefileid = fileid
 
-    url = '%sapi/v2/files/%s' % (host, intermediatefileid)
-    headers = {"Authorization": "Bearer " + token}
-    result = connector.get(url, stream=True, verify=connector.ssl_verify if connector else True, headers=headers)
+    if clowder_version >= 2.0:
+        url = '%sapi/v2/files/%s' % (host, intermediatefileid)
+        headers = {"Authorization": "Bearer " + token}
+        result = connector.get(url, stream=True, verify=connector.ssl_verify if connector else True, headers=headers)
+
+        (inputfile, inputfilename) = tempfile.mkstemp(suffix=ext)
+
+        try:
+            with os.fdopen(inputfile, "wb") as outputfile:
+                for chunk in result.iter_content(chunk_size=10 * 1024):
+                    outputfile.write(chunk)
+            return inputfilename
+        except Exception:
+            os.remove(inputfilename)
+            raise
+    else:
+        url = '%sapi/files/%s?key=%s' % (host, intermediatefileid, key)
+        result = connector.get(url, stream=True, verify=connector.ssl_verify if connector else True)
+
+        (inputfile, inputfilename) = tempfile.mkstemp(suffix=ext)
 
-    (inputfile, inputfilename) = tempfile.mkstemp(suffix=ext)
+        try:
+            with os.fdopen(inputfile, "wb") as outputfile:
+                for chunk in result.iter_content(chunk_size=10*1024):
+                    outputfile.write(chunk)
+            return inputfilename
+        except Exception:
+            os.remove(inputfilename)
+            raise
 
-    try:
-        with os.fdopen(inputfile, "wb") as outputfile:
-            for chunk in result.iter_content(chunk_size=10*1024):
-                outputfile.write(chunk)
-        return inputfilename
-    except Exception:
-        os.remove(inputfilename)
-        raise
 
-def download_info(connector, host, key, fileid):
+def download_info(connector, host, key, fileid, token=None):
     """Download file summary metadata from Clowder.
 
     Keyword arguments:
@@ -102,30 +90,21 @@ def download_info(connector, host, key, fileid):
     fileid -- the file to fetch metadata of
     """
 
-    url = '%sapi/files/%s/metadata?key=%s' % (host, fileid, key)
-    headers = {"Authorization": "Bearer " + token}
-
-    # fetch data
-    result = connector.get(url, stream=True, verify=connector.ssl_verify if connector else True)
-
-    return result.json()
-
-def download_info_v2(connector, host, token, fileid):
-    """Download file summary metadata from Clowder.
+    if clowder_version >= 2.0:
+        url = '%sapi/v2/files/%s/metadata' % (host, fileid)
+        headers = {"Authorization": "Bearer " + token}
+        # fetch data
+        result = connector.get(url, stream=True, verify=connector.ssl_verify if connector else True, headers=headers)
 
-    Keyword arguments:
-    connector -- connector information, used to get missing parameters and send status updates
-    host -- the clowder host, including http and port, should end with a /
-    key -- the secret key to login to clowder
-    fileid -- the file to fetch metadata of
-    """
+        return result.json()
+    else:
+        url = '%sapi/files/%s/metadata?key=%s' % (host, fileid, key)
+        headers = {"Authorization": "Bearer " + token}
 
-    url = '%sapi/v2/files/%s/metadata' % (host, fileid)
-    headers = {"Authorization": "Bearer " + token}
-    # fetch data
-    result = connector.get(url, stream=True, verify=connector.ssl_verify if connector else True, headers=headers)
+        # fetch data
+        result = connector.get(url, stream=True, verify=connector.ssl_verify if connector else True)
 
-    return result.json()
+        return result.json()
 
 
 def download_metadata(connector, host, key, fileid, extractor=None):
@@ -235,6 +214,8 @@ def upload_metadata(connector, host, key, fileid, metadata):
     connector.message_process({"type": "file", "id": fileid}, "Uploading file metadata.")
 
     headers = {'Content-Type': 'application/json'}
+    # TODO if version 2.0
+
     url = '%sapi/files/%s/metadata.jsonld?key=%s' % (host, fileid, key)
     result = connector.post(url, headers=headers, data=json.dumps(metadata),
                             verify=connector.ssl_verify if connector else True)

sample-extractors/csv-precipitation/Dockerfile

@@ -0,0 +1,21 @@
+ARG PYCLOWDER_PYTHON=""
+FROM clowder/pyclowder${PYCLOWDER_PYTHON}:onbuild
+
+ENV MAIN_SCRIPT="binary_extractor.py" \
+    RABBITMQ_QUEUE="" \
+    IMAGE_BINARY="" \
+    IMAGE_TYPE="" \
+    IMAGE_THUMBNAIL_COMMAND="" \
+    IMAGE_PREVIEW_COMMAND="" \
+    PREVIEW_BINARY="" \
+    PREVIEW_TYPE="" \
+    PREVIEW_COMMAND=""
+
+ONBUILD COPY packages.* Dockerfile /home/clowder/
+ONBUILD RUN if [ -e packages.apt ]; then \
+                apt-get -q -q update \
+                && xargs apt-get -y install --no-install-recommends < packages.apt \
+                && rm -rf /var/lib/apt/lists/*; \
+            fi
+
+ONBUILD COPY extractor_info.json /home/clowder/