ZO-4267: postgresql connector copy move lock

core/docs/changelog/ZO-4267.change

@@ -0,0 +1 @@
+ZO-4267: Implement copy, move and lock in zeit.connector

core/src/zeit/connector/connector.py

@@ -15,7 +15,7 @@
 import zope.cachedescriptors.property
 import zope.interface
 
-from zeit.connector.interfaces import ID_NAMESPACE
+from zeit.connector.interfaces import ID_NAMESPACE, IPersistentCache
 import zeit.connector.cache
 import zeit.connector.dav.davconnection
 import zeit.connector.dav.davresource
@@ -131,8 +131,11 @@ def listCollection(self, id):
         """List the filenames of a collection identified by <id> (see[8])."""
         __traceback_info__ = (id,)
         id = self._get_cannonical_id(id)
-        for child_id in self._get_resource_child_ids(id):
-            yield (self._id_splitlast(child_id)[1].rstrip('/'), child_id)
+        try:
+            for child_id in self._get_resource_child_ids(id):
+                yield (self._id_splitlast(child_id)[1].rstrip('/'), child_id)
+        except zeit.connector.dav.interfaces.DAVNotFoundError as err:
+            raise KeyError(f'The resource {id} does not exist.') from err
 
     def _get_resource_type(self, id):
         __traceback_info__ = (id,)
@@ -260,6 +263,8 @@ def __delitem__(self, id):
             self.get_connection().delete(self._id2loc(id), token)
         except zeit.connector.dav.interfaces.DAVNotFoundError:
             raise KeyError(id)
+        except zeit.connector.dav.interfaces.FailedDependencyError as e:
+            raise zeit.connector.interfaces.LockedByOtherSystemError(id, e) from e
         self._invalidate_cache(id)
 
     def __contains__(self, id):
@@ -281,30 +286,20 @@ def copy(self, old_id, new_id):
 
     def move(self, old_id, new_id):
         """Move the resource with id `old_id` to `new_id`."""
-        self._copy_or_move(
-            'move', zeit.connector.interfaces.MoveError, old_id, new_id, resolve_conflicts=True
-        )
+        self._copy_or_move('move', zeit.connector.interfaces.MoveError, old_id, new_id)
 
-    def _copy_or_move(self, method_name, exception, old_id, new_id, resolve_conflicts=False):
+    def _copy_or_move(self, method_name, exception, old_id, new_id):
         source = self[old_id]  # Makes sure source exists.
         if self._is_descendant(new_id, old_id):
             raise exception(old_id, 'Could not copy or move %s to a decendant of itself.' % old_id)
 
         logger.debug('copy: %s to %s' % (old_id, new_id))
         if self._get_cannonical_id(new_id) in self:
-            target = self[new_id]
-            # The target already exists. It's possible that there was a
-            # conflict. For non-directories verify body.
-            if not (
-                resolve_conflicts
-                and 'httpd/unix-directory' not in (source.contentType, target.contentType)
-                and source.data.read() == self[new_id].data.read()
-            ):
-                raise exception(
-                    old_id,
-                    'Could not copy or move %s to %s, '
-                    'because target alread exists.' % (old_id, new_id),
-                )
+            raise exception(
+                old_id,
+                'Could not copy or move %s to %s, '
+                'because target already exists.' % (old_id, new_id),
+            )
         # Make old_id and new_id canonical. Use the canonical old_id to deduct
         # the canonical new_id:
         old_id = self._get_cannonical_id(old_id)
@@ -333,7 +328,10 @@ def _copy_or_move(self, method_name, exception, old_id, new_id, resolve_conflict
                 del self[old_id]
         else:
             token = self._get_my_locktoken(old_id)
-            method(old_loc, new_loc, locktoken=token)
+            try:
+                method(old_loc, new_loc, locktoken=token)
+            except zeit.connector.dav.interfaces.DAVNotFoundError as err:
+                raise KeyError('The resource %s does not exist.', old_id) from err
 
         self._invalidate_cache(old_id)
         self._invalidate_cache(new_id)
@@ -685,14 +683,13 @@ def invalidate_cache(self, id):
             self._update_child_id_cache(davres)
 
             # Remove no longer existing child entries from property_cache
-            IPersistentCache = zeit.connector.interfaces.IPersistentCache
-            if id.endswith('/') and IPersistentCache.providedBy(self.property_cache):
-                end = id[:-1] + chr(ord('/') + 1)
-                for key in self.property_cache.keys(min=id, max=end):
-                    if key not in self.child_name_cache:
-                        del self.property_cache[key]
+            for key in self._cached_ids_below_parent(self.property_cache, id):
+                if key not in self.child_name_cache:
+                    self._remove_from_caches(key, [self.property_cache])
         else:
             self._remove_from_caches(id, [self.property_cache, self.child_name_cache])
+            for key in self._cached_ids_below_parent(self.property_cache, id):
+                self._remove_from_caches(key, [self.property_cache, self.child_name_cache])
 
         # Update our parent's child_name_cache
         parent, name = self._id_splitlast(id)
@@ -717,6 +714,13 @@ def invalidate_cache(self, id):
             else:
                 self._update_property_cache(davres)
 
+    def _cached_ids_below_parent(self, cache, id):
+        if not (id.endswith('/') or IPersistentCache.providedBy(cache)):
+            return iter(())
+        end = id[:-1] + chr(ord('/') + 1)
+        for key in cache.keys(min=id, max=end):
+            yield key
+
     def _get_cannonical_id(self, id):
         """Add / for collections if not appended yet."""
         if isinstance(id, CannonicalId):

core/src/zeit/connector/dav/davconnection.py

@@ -19,6 +19,7 @@ class DAVConnection(zeit.connector.dav.davbase.DAVConnection):
         http.client.MOVED_PERMANENTLY: zeit.connector.dav.interfaces.DAVRedirectError,
         http.client.NOT_FOUND: zeit.connector.dav.interfaces.DAVNotFoundError,
         http.client.BAD_REQUEST: zeit.connector.dav.interfaces.DAVBadRequestError,
+        http.client.FAILED_DEPENDENCY: zeit.connector.dav.interfaces.FailedDependencyError,
     }
 
     def lock(self, url, owner=None, depth=0, timeout=None, headers=None):

core/src/zeit/connector/dav/interfaces.py

@@ -40,6 +40,10 @@ class DAVRedirectError(DAVError):
     """Raised when a resource was moved."""
 
 
+class FailedDependencyError(DAVError):
+    """Raised when dependency is e.g. locked"""
+
+
 class PreconditionFailedError(http.client.HTTPException):
     """Raised when a precondition (if header) fails."""
 

core/src/zeit/connector/filesystem.py

@@ -12,7 +12,6 @@
 import zope.interface
 
 from zeit.connector.connector import CannonicalId
-from zeit.connector.dav.interfaces import DAVNotFoundError
 from zeit.connector.interfaces import ID_NAMESPACE
 import zeit.connector.dav.interfaces
 import zeit.connector.interfaces
@@ -68,12 +67,7 @@ def listCollection(self, id):
         path = self._path(id)
         names = self._get_collection_names(path)
         if not names:
-            try:
-                self[id]
-            except KeyError:
-                # XXX mimic the real behaviour -- real *should* probably raise
-                # KeyError, but doesn't at the moment.
-                raise DAVNotFoundError(404, 'Not Found', id, '')
+            self[id]
 
         result = []
         for name in sorted(names):

core/src/zeit/connector/interfaces.py

@@ -81,8 +81,6 @@ def listCollection(id):
         raises ValueError if the id is not valid or does not match the
          repository.
         raises KeyError if the resource does not exist.
-
-        XXX define more error cases
         """
 
     def __getitem__(id):
@@ -97,6 +95,7 @@ def __delitem__(id):
         """Remove the resource from the repository.
 
         raises KeyError if the resource could not be found.
+        raises LockedByOtherSystemError if the resource is locked by another user
         """
 
     def __setitem__(id, object):
@@ -125,15 +124,21 @@ def add(object, verify_etag=True):
     def copy(old_id, new_id):
         """Copy the resource old_id to new_id.
 
-        raises KeyError if ther is no resource `old_id`
+        Copy does not duplicate any write locks active on the source.
+
         raises CopyError if there was a problem with copying itself.
+        raises KeyError if there is not resource with old_id
         """
 
     def move(old_id, new_id):
         """Move the resource with id `old_id` to `new_id`.
 
-        raises KeyError if ther is no resource `old_id`
+        A successful MOVE request on a write locked resource will not move the
+        write lock with the resource.
+
+        raises KeyError if there is no resource with old_id
         raises MoveError if there was a problem with moving itself.
+        raises LockedByOtherSystemError if the resource is deleted by another user
         """
 
     def changeProperties(id, properties):
@@ -146,14 +151,31 @@ def changeProperties(id, properties):
     def lock(id, principal, until):
         """Lock resource for principal until a given datetime.
 
+        A client MUST NOT submit the same write lock request twice.
+
+        A successful request for an write lock results in the generation of a unique lock token
+        associated with the requesting principal.
+
+        A write locked null resource, referred to as a lock-null resource is possible.
+
+        A write lock request is issued to a collection containing member URIs
+        identifying resources that are currently locked in a manner which conflicts
+        with the write lock, will fail.
+
         id: unique id
         until: datetime until the lock is valid.
 
         returns locktoken.
+        raises LockingError if lock already exists for another principal
         """
 
     def unlock(id):
-        """Unlock resource using the stored locktoken."""
+        """Unlock resource
+
+        returns used locktoken, if the id of the resource does not exist, return none
+        raises KeyError if the resource with id does not exist
+        raises LockedByOtherSystemError if the resource is locked by another user
+        """
 
     def _unlock(id, locktoken):
         """Unlock resource using the given locktoken. For tests."""
@@ -167,8 +189,7 @@ def locked(id):
             my lock: True if the lock was issued by this system, False
                 otherwise.
 
-        returns None, None, None if the resource is not locked.
-        raises KeyError if the resource does not exist.
+        returns None, None, None if the resource is not locked or is non-existant
         """
 
     def search(attributes, search_expression):

core/src/zeit/connector/mock.py

@@ -14,7 +14,14 @@
 import zope.event
 
 from zeit.connector.connector import CannonicalId
-from zeit.connector.interfaces import ID_NAMESPACE, UUID_PROPERTY, CopyError, MoveError
+from zeit.connector.interfaces import (
+    ID_NAMESPACE,
+    UUID_PROPERTY,
+    CopyError,
+    LockedByOtherSystemError,
+    LockingError,
+    MoveError,
+)
 import zeit.connector.cache
 import zeit.connector.dav.interfaces
 import zeit.connector.filesystem
@@ -167,7 +174,10 @@ def __setitem__(self, id, object):
     def __delitem__(self, id):
         id = self._get_cannonical_id(id)
         self[id]  # may raise KeyError
-        for _name, uid in self.listCollection(id):
+        list_collection = self.listCollection(id)
+        for _name, uid in list_collection:
+            if uid in self._locked:
+                raise LockedByOtherSystemError(uid, '')
             del self[uid]
         self._deleted.add(id)
         self._data.pop(id, None)
@@ -193,16 +203,21 @@ def copy(self, old_id, new_id):
 
     def move(self, old_id, new_id):
         self._prevent_overwrite(old_id, new_id, MoveError)
-        self._ignore_uuid_checks = True
         r = self[old_id]
+
+        if new_id in self:
+            raise MoveError(new_id, f'The resource {new_id} already exists.')
         r.id = new_id
         try:
+            self._ignore_uuid_checks = True
             self.add(r, verify_etag=False)
         finally:
             self._ignore_uuid_checks = False
         if not new_id.endswith('/'):
             new_id = new_id + '/'
         for name, uid in self.listCollection(old_id):
+            if uid in self._locked and self._locked[uid][0] != 'zope.user':
+                raise LockedByOtherSystemError(uid, '')
             self.move(uid, urllib.parse.urljoin(new_id, name))
         del self[old_id]
 
@@ -230,6 +245,8 @@ def changeProperties(self, id, properties):
     def lock(self, id, principal, until):
         """Lock resource for principal until a given datetime."""
         id = self._get_cannonical_id(id)
+        if id in self._locked:
+            raise LockingError('Resource is already locked by another principal')
         self._locked[id] = (principal, until, True)
 
     def unlock(self, id):