Merge pull request #5 from oarepo/parallel

Parallel

.github/workflows/main.yml

@@ -36,10 +36,11 @@ jobs:
         uses: elastic/elastic-github-actions/elasticsearch@master
         with:
           stack-version: 7.6.0
-
+      - name: Runs RabbitMQ
+        uses: nijel/rabbitmq-action@v1.0.0
       - name: Install dependencies
         run: |
-          python -m pip install --upgrade pip wheel setuptools
+          python -m pip install --upgrade 'pip<20.0.0' wheel setuptools
           pip install -e .[tests]
       - name: Test with pytest
         run: |

MANIFEST.in

@@ -27,6 +27,7 @@ recursive-exclude 3rdparty .gitkeep
 recursive-exclude 3rdparty *.yaml
 recursive-include 3rdparty *.toml
 recursive-include 3rdparty *.txt
+recursive-exclude 3rdparty *
 include .gitmodules
 include docker-compose.yml
 include docker-services.yml

oarepo_s3/version.py

@@ -13,4 +13,4 @@
 
 from __future__ import absolute_import, print_function
 
-__version__ = '1.2.4'
+__version__ = '1.2.5'

oarepo_s3/views.py

@@ -42,16 +42,18 @@
 import json
 from functools import wraps
 
-from flask import abort, jsonify
+from flask import abort, jsonify, request
 from flask.views import MethodView
 from invenio_db import db
 from invenio_files_rest.models import ObjectVersion, ObjectVersionTag
 from invenio_files_rest.proxies import current_permission_factory
 from invenio_files_rest.signals import file_deleted
 from invenio_files_rest.tasks import remove_file_data
 from invenio_files_rest.views import check_permission
+from invenio_records_rest.errors import PIDResolveRESTError
 from invenio_records_rest.views import pass_record
 from invenio_rest import csrf
+from sqlalchemy.exc import SQLAlchemyError
 from webargs import fields
 from webargs.flaskparser import use_kwargs
 
@@ -67,6 +69,24 @@
 }
 
 
+def pass_locked_record(f):
+    """Decorator to retrieve persistent identifier and record.
+
+    This decorator will resolve the ``pid_value`` parameter from the route
+    pattern and resolve it to a PID and a record, which are then available in
+    the decorated function as ``pid`` and ``record`` kwargs respectively.
+    """
+    @wraps(f)
+    def inner(self, pid_value, *args, **kwargs):
+        try:
+            pid, record = request.view_args['pid_value'].data
+            record = lock_record(record)
+            return f(self, pid=pid, record=record, *args, **kwargs)
+        except SQLAlchemyError:
+            raise PIDResolveRESTError(pid_value)
+    return inner
+
+
 def pass_file_rec(f):
     """Decorator to retrieve a FileObject
        given by key from record FilesIterator.
@@ -125,11 +145,20 @@ def delete_file_object_version(bucket, obj):
     file_deleted.send(obj)
 
 
+def lock_record(record):
+    # lock the record in case of multiple uploads to the same record
+    cls = type(record)
+    obj = cls.model_cls.query.filter_by(id=record.id). \
+        filter(cls.model_cls.json != None).with_for_update().one()
+    db.session.expire(obj)
+    return cls.get_record(record.id)
+
+
 class MultipartUploadCompleteResource(MethodView):
     """Complete multipart upload method view."""
     view_name = '{endpoint}_upload_complete'
 
-    @pass_record
+    @pass_locked_record
     @pass_file_rec
     @pass_multipart_config
     @use_kwargs(multipart_complete_args)
@@ -167,7 +196,7 @@ class MultipartUploadAbortResource(MethodView):
     """Cancel a multipart upload method view."""
     view_name = '{endpoint}_upload_abort'
 
-    @pass_record
+    @pass_locked_record
     @pass_file_rec
     @pass_multipart_config
     def post(self, pid, record, files, file_rec, multipart_config, key):

setup.py

@@ -24,7 +24,7 @@
     'tests': {
         'oarepo[tests]>=3.3.46',
         *tests_require,
-        'oarepo-records-draft>=5.5.1',
+        'oarepo-records-draft>=5.5.2',
     }
 }