cleanup

- bumped some requirements (solves github complaining about security issues)
- fixes to be python 3.x compatible
- fix some functions not using key to post
- general cleanup

Author: robkooper

contrib/monitor/monitor.py

@@ -15,6 +15,8 @@
 rabbitmq_uri = os.getenv('RABBITMQ_URI', 'amqp://guest:guest@localhost/%2F')
 rabbitmq_mgmt_port = os.getenv('RABBITMQ_MGMT_PORT', '15672')
 rabbitmq_mgmt_url = ''
+rabbitmq_username = None
+rabbitmq_password = None
 
 extractors = {}
 
@@ -32,7 +34,7 @@ def do_GET(self):
         self.send_response(200)
         self.send_header('Content-type', 'application/json')
         self.end_headers()
-        self.wfile.write(bytes(json.dumps(extractors), 'utf-8'))
+        self.wfile.write(json.dumps(extractors).encode())
 
 
 def http_server():
@@ -52,7 +54,7 @@ def get_mgmt_queue_messages(queue):
         response = requests.get(rabbitmq_mgmt_url + queue, auth=(rabbitmq_username, rabbitmq_password), timeout=5)
         response.raise_for_status()
         return response.json()['messages']
-    except:
+    except Exception:
         logging.exception("Error getting list of messages in %s" % queue)
         return 0
 
@@ -121,9 +123,8 @@ def extractors_monitor():
     connection = pika.BlockingConnection(params)
 
     # create management url
-    rabbitmq_url = ''
     if rabbitmq_mgmt_port != '':
-        if params.ssl:
+        if params.ssl_options:
             rabbitmq_mgmt_url = 'https://'
         else:
             rabbitmq_mgmt_url = 'http://'

pyclowder/client.py

@@ -99,10 +99,10 @@ def get(self, path, params=None, headers=None):
             except requests.HTTPError as e:
                 attempt += 1
                 if attempt > self.retries:
-                    self.logger.exception("Error calling GET url %s: %s" % (url, e.message))
+                    self.logger.exception("Error calling GET url %s: %s" % (url, str(e)))
                     raise e
                 else:
-                    self.logger.debug("Error calling GET url %s: %s" % (url, e.message))
+                    self.logger.debug("Error calling GET url %s: %s" % (url, str(e)))
 
     def post(self, path, content, params=None, headers=None):
         """
@@ -138,10 +138,10 @@ def post(self, path, content, params=None, headers=None):
             except requests.HTTPError as e:
                 attempt += 1
                 if attempt > self.retries:
-                    self.logger.exception("Error calling POST url %s: %s" % (url, e.message))
+                    self.logger.exception("Error calling POST url %s: %s" % (url, str(e)))
                     raise e
                 else:
-                    self.logger.debug("Error calling POST url %s: %s" % (url, e.message))
+                    self.logger.debug("Error calling POST url %s: %s" % (url, str(e)))
 
     def delete(self, path, params=None, headers=None):
         """
@@ -175,10 +175,10 @@ def delete(self, path, params=None, headers=None):
             except requests.HTTPError as e:
                 attempt += 1
                 if attempt > self.retries:
-                    self.logger.exception("Error calling DELETE url %s: %s" % (url, e.message))
+                    self.logger.exception("Error calling DELETE url %s: %s" % (url, str(e)))
                     raise e
                 else:
-                    self.logger.debug("Error calling DELETE url %s: %s" % (url, e.message))
+                    self.logger.debug("Error calling DELETE url %s: %s" % (url, str(e)))
 
     def get_file(self, path, filename=None, params=None, headers=None):
         """
@@ -262,7 +262,7 @@ def post_file(self, path, filename, params=None, headers=None):
             except requests.HTTPError as e:
                 attempt += 1
                 if attempt > self.retries:
-                    self.logger.exception("Error calling POST url %s: %s" % (url, e.message))
+                    self.logger.exception("Error calling POST url %s: %s" % (url, str(e)))
                     raise e
                 else:
-                    self.logger.debug("Error calling POST url %s: %s" % (url, e.message))
+                    self.logger.debug("Error calling POST url %s: %s" % (url, str(e)))

pyclowder/collections.py

@@ -60,7 +60,7 @@ def create_empty(connector, host, key, collectionname, description, parentid=Non
 
 
 def delete(connector, host, key, collectionid):
-    url = "%sapi/collections/%s" % (host, collectionid)
+    url = "%sapi/collections/%s?key=%s" % (host, collectionid, key)
 
     result = requests.delete(url, verify=connector.ssl_verify if connector else True)
     result.raise_for_status()

pyclowder/connectors.py

@@ -254,7 +254,7 @@ def _build_resource(self, body, host, secret_key):
                 "metadata": body['metadata']
             }
 
-    def _check_for_local_file(self, host, secret_key, file_metadata):
+    def _check_for_local_file(self, file_metadata):
         """ Try to get pointer to locally accessible copy of file for extractor."""
 
         # first check if file is accessible locally
@@ -285,7 +285,7 @@ def _download_file_metadata(self, host, secret_key, fileid, filepath):
         md_dir = tempfile.mkdtemp(suffix=fileid)
         (fd, md_file) = tempfile.mkstemp(suffix=md_name, dir=md_dir)
 
-        with os.fdopen(fd, "w") as tmp_file:
+        with os.fdopen(fd, "wb") as tmp_file:
             tmp_file.write(json.dumps(file_md))
 
         return (md_dir, md_file)
@@ -303,13 +303,13 @@ def _prepare_dataset(self, host, secret_key, resource):
         # first check if any files in dataset accessible locally
         ds_file_list = pyclowder.datasets.get_file_list(self, host, secret_key, resource["id"])
         for ds_file in ds_file_list:
-            file_path = self._check_for_local_file(host, secret_key, ds_file)
+            file_path = self._check_for_local_file(ds_file)
             if not file_path:
                 missing_files.append(ds_file)
             else:
                 # Create a link to the original file if the "true" name of the file doesn't match what's on disk
                 if not file_path.lower().endswith(ds_file['filename'].lower()):
-                    ln_name = io.path.join(temp_link_dir, ds_file['filename'])
+                    ln_name = os.path.join(temp_link_dir, ds_file['filename'])
                     os.symlink(file_path, ln_name)
                     tmp_files_created.append(ln_name)
                     file_path = ln_name
@@ -342,7 +342,7 @@ def _prepare_dataset(self, host, secret_key, resource):
             md_name = "%s_dataset_metadata.json" % resource["id"]
             md_dir = tempfile.mkdtemp(suffix=resource["id"])
             (fd, md_file) = tempfile.mkstemp(suffix=md_name, dir=md_dir)
-            with os.fdopen(fd, "w") as tmp_file:
+            with os.fdopen(fd, "wb") as tmp_file:
                 tmp_file.write(json.dumps(ds_md))
             located_files.append(md_file)
             tmp_files_created.append(md_file)
@@ -409,7 +409,7 @@ def _process_message(self, body):
                         try:
                             if check_result != pyclowder.utils.CheckMessage.bypass:
                                 file_metadata = pyclowder.files.download_info(self, host, secret_key, resource["id"])
-                                file_path = self._check_for_local_file(host, secret_key, file_metadata)
+                                file_path = self._check_for_local_file(file_metadata)
                                 if not file_path:
                                     file_path = pyclowder.files.download(self, host, secret_key, resource["id"],
                                                                          resource["intermediate_id"],
@@ -460,7 +460,7 @@ def _process_message(self, body):
             self.message_ok(resource)
 
         except SystemExit as exc:
-            status = "sys.exit : " + exc.message
+            status = "sys.exit : " + str(exc)
             logger.exception("[%s] %s", resource['id'], status)
             self.status_update(pyclowder.utils.StatusMessage.error, resource, status)
             self.message_resubmit(resource, retry_count)
@@ -483,7 +483,7 @@ def _process_message(self, body):
             self.status_update(pyclowder.utils.StatusMessage.error, resource, status)
             self.message_error(resource)
         except Exception as exc:  # pylint: disable=broad-except
-            status = "Error processing : " + exc.message
+            status = "Error processing : " + str(exc)
             logger.exception("[%s] %s", resource['id'], status)
             self.status_update(pyclowder.utils.StatusMessage.error, resource, status)
             if retry_count < 10:
@@ -635,6 +635,7 @@ def __init__(self, extractor_name, extractor_info,
         self.connection = None
         self.consumer_tag = None
         self.worker = None
+        self.announcer = None
 
     def connect(self):
         """connect to rabbitmq using URL parameters"""
@@ -688,13 +689,15 @@ def listen(self):
             self.connect()
 
         # create listener
-        self.consumer_tag = self.channel.basic_consume(self.on_message, queue=self.rabbitmq_queue, no_ack=False)
+        self.consumer_tag = self.channel.basic_consume(queue=self.rabbitmq_queue,
+                                                       on_message_callback=self.on_message,
+                                                       auto_ack=False)
 
         # start listening
         logging.getLogger(__name__).info("Starting to listen for messages.")
         try:
             # pylint: disable=protected-access
-            while self.channel and self.channel._consumer_infos:
+            while self.channel and self.channel.is_open and self.channel._consumer_infos:
                 self.channel.connection.process_data_events(time_limit=1)  # 1 second
                 if self.worker:
                     self.worker.process_messages(self.channel, self.rabbitmq_queue)
@@ -710,20 +713,18 @@ def listen(self):
             logging.getLogger(__name__).exception("Error while consuming messages.")
         finally:
             logging.getLogger(__name__).info("Stopped listening for messages.")
-            if self.channel:
+            if self.channel and self.channel.is_open:
                 try:
                     self.channel.close()
                 except Exception:
                     logging.getLogger(__name__).exception("Error while closing channel.")
-                finally:
-                    self.channel = None
-            if self.connection:
+            self.channel = None
+            if self.connection and self.connection.is_open:
                 try:
                     self.connection.close()
                 except Exception:
                     logging.getLogger(__name__).exception("Error while closing connection.")
-                finally:
-                    self.connection = None
+            self.connection = None
 
     def stop(self):
         """Tell the connector to stop listening for messages."""
@@ -734,7 +735,9 @@ def alive(self):
         return self.connection is not None
 
     @staticmethod
-    def _decode_body(body, codecs=['utf8', 'iso-8859-1']):
+    def _decode_body(body, codecs=None):
+        if not codecs:
+            codecs = ['utf8', 'iso-8859-1']
         # see https://stackoverflow.com/a/15918519
         for i in codecs:
             try:
@@ -778,6 +781,9 @@ def __init__(self, rabbitmq_uri, extractor_info, rabbitmq_queue, heartbeat):
         self.rabbitmq_queue = rabbitmq_queue
         self.heartbeat = heartbeat
         self.id = str(uuid.uuid4())
+        self.connection = None
+        self.channel = None
+        self.thread = None
 
     def start_thread(self):
         parameters = pika.URLParameters(self.rabbitmq_uri)

pyclowder/datasets.py

@@ -71,7 +71,7 @@ def delete(connector, host, key, datasetid):
     key -- the secret key to login to clowder
     datasetid -- the dataset to delete
     """
-    url = "%sapi/datasets/%s" % (host, datasetid)
+    url = "%sapi/datasets/%s?key=%s" % (host, datasetid, key)
 
     result = requests.delete(url, verify=connector.ssl_verify if connector else True)
     result.raise_for_status()
@@ -122,7 +122,7 @@ def download(connector, host, key, datasetid):
     result.raise_for_status()
 
     (filedescriptor, zipfile) = tempfile.mkstemp(suffix=".zip")
-    with os.fdopen(filedescriptor, "w") as outfile:
+    with os.fdopen(filedescriptor, "wb") as outfile:
         for chunk in result.iter_content(chunk_size=10 * 1024):
             outfile.write(chunk)
 
@@ -303,7 +303,7 @@ def datasets_get(self):
         try:
             return self.client.get("/datasets")
         except Exception as e:
-            logging.error("Error retrieving dataset list: %s", e.message)
+            logging.error("Error retrieving dataset list: %s", str(e))
 
     def dataset_get(self, dataset_id):
         """
@@ -316,7 +316,7 @@ def dataset_get(self, dataset_id):
         try:
             return self.client.get("/datasets/%s" % dataset_id)
         except Exception as e:
-            logging.error("Error retrieving dataset %s: %s" % (dataset_id, e.message))
+            logging.error("Error retrieving dataset %s: %s" % (dataset_id, str(e)))
 
     def create_empty(self, dataset_id):
         """
@@ -329,7 +329,7 @@ def create_empty(self, dataset_id):
         try:
             return self.client.post("/datasets/createempty", dataset_id)
         except Exception as e:
-            logging.error("Error adding datapoint %s: %s" % (dataset_id, e.message))
+            logging.error("Error adding datapoint %s: %s" % (dataset_id, str(e)))
 
     def dataset_delete(self, dataset_id):
         """
@@ -342,7 +342,7 @@ def dataset_delete(self, dataset_id):
         try:
             return self.client.delete("/datasets/%s" % dataset_id)
         except Exception as e:
-            logging.error("Error retrieving dataset %s: %s" % (dataset_id, e.message))
+            logging.error("Error retrieving dataset %s: %s" % (dataset_id, str(e)))
 
     def upload_file(self, dataset_id, file):
         """
@@ -355,7 +355,7 @@ def upload_file(self, dataset_id, file):
         try:
             return self.client.post_file("/uploadToDataset/%s" % dataset_id, file)
         except Exception as e:
-            logging.error("Error upload to dataset %s: %s" % (dataset_id, e.message))
+            logging.error("Error upload to dataset %s: %s" % (dataset_id, str(e)))
 
     def add_metadata(self, dataset_id, metadata):
         """
@@ -368,5 +368,5 @@ def add_metadata(self, dataset_id, metadata):
         logging.debug("Update metadata of dataset %s" % dataset_id)
         try:
             return self.client.post("/datasets/%s/metadata" % dataset_id, metadata)
-        except Exception:
-            logging.error("Error upload to dataset %s: %s" % (dataset_id, e.message))
+        except Exception as e:
+            logging.error("Error upload to dataset %s: %s" % (dataset_id, str(e)))

pyclowder/extractors.py

@@ -205,8 +205,8 @@ def start(self):
         except BaseException:
             logger.exception("Error while consuming messages.")
 
-        while connectors:
-            connectors.pop(0).stop()
+        for c in connectors:
+            c.stop()
 
     def get_metadata(self, content, resource_type, resource_id, server=None):
         """Generate a metadata field.
@@ -297,9 +297,9 @@ class SimpleExtractor(Extractor):
     """
 
     def __init__(self):
-        '''
+        """
         Initialize the extractor and setup the logger.
-        '''
+        """
         Extractor.__init__(self)
         self.setup()
 

pyclowder/files.py

@@ -48,11 +48,11 @@ def download(connector, host, key, fileid, intermediatefileid=None, ext=""):
 
     (inputfile, inputfilename) = tempfile.mkstemp(suffix=ext)
     try:
-        with os.fdopen(inputfile, "w") as outputfile:
+        with os.fdopen(inputfile, "wb") as outputfile:
             for chunk in result.iter_content(chunk_size=10*1024):
                 outputfile.write(chunk)
         return inputfilename
-    except:
+    except Exception:
         os.remove(inputfilename)
         raise
 
@@ -188,7 +188,7 @@ def upload_metadata(connector, host, key, fileid, metadata):
 
 
 # pylint: disable=too-many-arguments
-def upload_preview(connector, host, key, fileid, previewfile, previewmetadata, preview_mimetype=None):
+def upload_preview(connector, host, key, fileid, previewfile, previewmetadata=None, preview_mimetype=None):
     """Upload preview to Clowder.
 
     Keyword arguments: