Cadc 9508 (#145)

* Added support for If-Match in caom2repo

caom2/caom2/caom_util.py

@@ -79,11 +79,12 @@
 from __future__ import (absolute_import, division, print_function,
                         unicode_literals)
 
-import collections
 import sys
+import collections
 from datetime import datetime
 
 import six
+from six.moves import collections_abc
 from six.moves.urllib.parse import urlsplit
 from builtins import int, str as newstr
 
@@ -203,7 +204,7 @@ def value_check(value, min_value, max_value, variable, override=None):
     return True
 
 
-class TypedList(collections.MutableSequence):
+class TypedList(collections_abc.MutableSequence):
     """
     Class that implements a typed list in Python. Supported types
     are specified when instance is created. Example:
@@ -266,7 +267,7 @@ def key_type(self):
         return self._oktypes
 
 
-class TypedSet(collections.MutableSet):
+class TypedSet(collections_abc.MutableSet):
     """
     Class that implements a typed set in Python. Supported types
     are specified when instance is created. Example:

caom2/setup.cfg

@@ -20,8 +20,6 @@ omit = */tests/*
 
 [tool:pytest]
 minversion = 2.2
-norecursedirs = build docs/_build
-doctest_plus = enabled
 testpaths = caom2
 
 [bdist_wheel]

caom2repo/caom2repo/core.py

@@ -3,7 +3,7 @@
 # ******************  CANADIAN ASTRONOMY DATA CENTRE  *******************
 # *************  CENTRE CANADIEN DE DONNÉES ASTRONOMIQUES  **************
 #
-#  (c) 2016.                            (c) 2016.
+#  (c) 2021.                            (c) 2021.
 #  Government of Canada                 Gouvernement du Canada
 #  National Research Council            Conseil national de recherches
 #  Ottawa, Canada, K1A 0R6              Ottawa, Canada, K1A 0R6
@@ -80,8 +80,7 @@
 import sys
 from datetime import datetime
 
-from cadcutils import net
-from cadcutils import util
+from cadcutils import net, util, exceptions
 from caom2.obs_reader_writer import ObservationReader, ObservationWriter
 from caom2 import obs_reader_writer
 from caom2.version import version as caom2_version
@@ -236,8 +235,8 @@ def visit(self, plugin, collection, start=None, end=None, obs_file=None,
         while len(observations) > 0:
             if nthreads is None:
                 results = [
-                    self._process_observation_id(collection, observationID,
-                                                 halt_on_error)
+                    self.process_observation_id(collection, observationID,
+                                                halt_on_error)
                     for observationID in observations]
                 for v, u, s, f in results:
                     if v:
@@ -284,24 +283,39 @@ def visit(self, plugin, collection, start=None, end=None, obs_file=None,
                 break
         return visited, updated, skipped, failed
 
-    def _process_observation_id(self, collection, observationID,
-                                halt_on_error):
-        visited = None
+    def process_observation_id(self, collection, observation_id,
+                               halt_on_error):
+        """
+        Reads an observation, calls a plugin to update it and, if modified,
+        uploads it to the repo.
+
+        :param collection:
+        :param observation_id:
+        :param halt_on_error: if true, raise and exception when error
+        encountered otherwise log the error.
+        :return: (visited, updated, skipped, failed) tuple with values
+        equalled to None or observation_id depending on the outcome case
+        (visited=observationID always)
+        """
+        visited = observation_id
         updated = None
         skipped = None
         failed = None
-        self.logger.info('Process observation: ' + observationID)
+        self.logger.info('Process observation: ' + observation_id)
         try:
-            observation = self.get_observation(collection, observationID)
+            observation = self.get_observation(collection, observation_id)
+            orig_checksum = observation.acc_meta_checksum
+            if orig_checksum:
+                orig_checksum = orig_checksum.uri
             if self.plugin.update(observation=observation,
                                   subject=self._subject) is False:
                 self.logger.info('SKIP {}'.format(observation.observation_id))
                 skipped = observation.observation_id
             else:
-                self.post_observation(observation)
+                self.post_observation(observation, orig_checksum)
                 self.logger.debug(
                     'UPDATED {}'.format(observation.observation_id))
-                updated = observation.observation_id
+                updated = observation_id
         except TypeError as e:
             if "unexpected keyword argument" in str(e):
                 raise RuntimeError(
@@ -311,17 +325,30 @@ def _process_observation_id(self, collection, observationID,
             else:
                 # other unexpected TypeError
                 raise e
+        except exceptions.UnexpectedException as e:
+            if e.orig_exception.response.status_code == 412:
+                self.logger.info(
+                    'Race condition: observation {} updated on the server '
+                    'while being visited. Re-trying.'.format(observation_id))
+                return self.process_observation_id(collection, observation_id,
+                                                   halt_on_error)
+            else:
+                failed = observation_id
+                self._handle_error(e, observation_id, halt_on_error)
         except Exception as e:
-            failed = observationID
-            self.logger.error(
-                'FAILED {} - Reason: {}'.format(observationID, e))
-            if halt_on_error:
-                raise e
-
-        visited = observationID
-
+            failed = observation_id
+            self._handle_error(e, observation_id, halt_on_error)
+        visited = observation_id
         return visited, updated, skipped, failed
 
+    def _handle_error(self, exception, observation_id, halt_on_error):
+        self.logger.error(
+            'FAILED {} - Reason: {}'.format(observation_id, str(exception)))
+        if halt_on_error:
+            raise exception
+        else:
+            self.logger.debug(exception)
+
     def _get_obs_from_file(self, obs_file, start, end, halt_on_error):
         obs = []
         failed = []
@@ -452,10 +479,13 @@ def get_observation(self, collection, observation_id):
             raise Exception('Got empty response for resource: {}'.format(path))
         return obs_reader.read(BytesIO(content))
 
-    def post_observation(self, observation):
+    def post_observation(self, observation, orig_checksum=None):
         """
         Updates an observation in the CAOM2 repo
         :param observation: observation to update
+        :param orig_checksum: the checksum of the observation to be updated.
+        Posting this value prevents race conditions when observations are
+        updated concurrently
         :return: updated observation
         """
         assert observation.collection is not None
@@ -469,6 +499,8 @@ def post_observation(self, observation):
             observation, ibuffer)
         obs_xml = ibuffer.getvalue()
         headers = {'Content-Type': 'application/xml'}
+        if orig_checksum:
+            headers['If-Match'] = orig_checksum
         self._repo_client.post(
             (self.capability_id, path), headers=headers, data=obs_xml)
 
@@ -535,9 +567,9 @@ def multiprocess_observation_id(collection, observationID, plugin, subject,
                                 log_level, resource_id, host, agent,
                                 halt_on_error):
     """
-    Multi-process version of CAOM2RepoClient._process_observation_id().
+    Multi-process version of CAOM2RepoClient.process_observation_id().
     Each process handles Control-C via KeyboardInterrupt, which is not needed
-    in CAOM2RepoClient._process_observation_id().
+    in CAOM2RepoClient.process_observation_id().
     :param collection: Name of the collection
     :param observationID: Observation identifier
     :param plugin: path to python file that contains the algorithm to be
@@ -551,13 +583,6 @@ def multiprocess_observation_id(collection, observationID, plugin, subject,
     :return: Tuple of observationID representing visited, updated, skipped
         and failed
     """
-    visited = None
-    updated = None
-    skipped = None
-    failed = None
-    observation = None
-    # set up logging for each process
-    subject = subject
     logging.basicConfig(
         format='%(asctime)s %(process)d %(levelname)-8s %(name)-12s ' +
                '%(funcName)s %(message)s',
@@ -566,34 +591,10 @@ def multiprocess_observation_id(collection, observationID, plugin, subject,
         'multiprocess_observation_id(): {}'.format(observationID))
 
     client = CAOM2RepoClient(subject, log_level, resource_id, host, agent)
-    try:
-        observation = client.get_observation(collection, observationID)
-        if plugin.update(observation=observation,
-                         subject=subject) is False:
-            rootLogger.info('SKIP {}'.format(observation.observation_id))
-            skipped = observation.observation_id
-        else:
-            client.post_observation(observation)
-            rootLogger.debug('UPDATED {}'.format(observation.observation_id))
-            updated = observation.observation_id
-    except TypeError as e:
-        if "unexpected keyword argument" in str(e):
-            raise RuntimeError(
-                "{} - To fix the problem, please add the **kwargs "
-                "argument to the list of arguments for the update"
-                " method of your plugin.".format(str(e)))
-        else:
-            # other unexpected TypeError
-            raise e
-    except Exception as e:
-        failed = observationID
-        rootLogger.error('FAILED {} - Reason: {}'.format(observationID, e))
-        if halt_on_error:
-            raise e
-
-    visited = observationID
-
-    return visited, updated, skipped, failed
+    client.plugin = plugin
+    client.logger = rootLogger
+    return \
+        client.process_observation_id(collection, observationID, halt_on_error)
 
 
 def main_app():