Merge pull request #480 from MerginMaps/revert-464-add_non_blocking_push

Revert "Allow concurrent non-blocking uploads"

Author: MarcelGeo

server/mergin/sync/commands.py

@@ -9,6 +9,7 @@
 from datetime import datetime
 from flask import Flask, current_app
 
+from .files import UploadChanges
 from ..app import db
 from .models import Project, ProjectVersion
 from .utils import split_project_path
@@ -51,7 +52,8 @@ def create(name, namespace, username):  # pylint: disable=W0612
         p = Project(**project_params)
         p.updated = datetime.utcnow()
         db.session.add(p)
-        pv = ProjectVersion(p, 0, user.id, [], "127.0.0.1")
+        changes = UploadChanges(added=[], updated=[], removed=[])
+        pv = ProjectVersion(p, 0, user.id, changes, "127.0.0.1")
         pv.project = p
         db.session.commit()
         os.makedirs(p.storage.project_dir, exist_ok=True)

server/mergin/sync/db_events.py

@@ -7,26 +7,16 @@
 from sqlalchemy import event
 
 from ..app import db
-from .models import ProjectVersion
-from .public_api_controller import push_finished
-from .tasks import remove_stale_project_uploads
 
 
 def check(session):
     if os.path.isfile(current_app.config["MAINTENANCE_FILE"]):
         abort(503, "Service unavailable due to maintenance, please try later")
 
 
-def cleanup_on_push_finished(project_version: ProjectVersion) -> None:
-    """On finished push trigger celery job cleanup"""
-    remove_stale_project_uploads.delay(project_version.project_id)
-
-
 def register_events():
     event.listen(db.session, "before_commit", check)
-    push_finished.connect(cleanup_on_push_finished)
 
 
 def remove_events():
     event.remove(db.session, "before_commit", check)
-    push_finished.connect(cleanup_on_push_finished)

server/mergin/sync/files.py

@@ -3,29 +3,14 @@
 # SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-MerginMaps-Commercial
 import datetime
 import os
-import uuid
 from dataclasses import dataclass
-from enum import Enum
-from flask import current_app
-from marshmallow import ValidationError, fields, EXCLUDE, post_dump, validates_schema
-from pathvalidate import sanitize_filename
 from typing import Optional, List
+from marshmallow import fields, EXCLUDE, pre_load, post_load, post_dump
+from pathvalidate import sanitize_filename
 
-from .utils import is_file_name_blacklisted, is_qgis, is_versioned_file
 from ..app import DateTimeWithZ, ma
 
 
-class PushChangeType(Enum):
-    CREATE = "create"
-    UPDATE = "update"
-    DELETE = "delete"
-    UPDATE_DIFF = "update_diff"
-
-    @classmethod
-    def values(cls):
-        return [member.value for member in cls.__members__.values()]
-
-
 def mergin_secure_filename(filename: str) -> str:
     """Generate secure filename for given file"""
     filename = os.path.normpath(filename)
@@ -39,181 +24,94 @@ def mergin_secure_filename(filename: str) -> str:
 
 @dataclass
 class File:
-    """Base class for every file object, either intended to upload or already existing in project"""
+    """Base class for every file object"""
 
     path: str
     checksum: str
     size: int
+    location: str
 
     def is_valid_gpkg(self):
         """Check if diff file is valid"""
         return self.size != 0
 
 
-@dataclass
-class ProjectDiffFile(File):
-    """Metadata for geodiff diff file (aka. changeset) associated with geopackage"""
-
-    # location where file is actually stored
-    location: str
-
-
 @dataclass
 class ProjectFile(File):
-    """Project file metadata including metadata for diff file and location where it is stored"""
+    """Project file metadata including metadata for diff file"""
 
     # metadata for gpkg diff file
-    diff: Optional[ProjectDiffFile]
+    diff: Optional[File]
     # deprecated attribute kept for public API compatibility
     mtime: Optional[datetime.datetime]
-    # location where file is actually stored
-    location: str
 
 
 @dataclass
-class ProjectFileChange(ProjectFile):
-    """Metadata of changed file in project version.
-
-    This item is saved into database into file_history.
-    """
-
-    change: PushChangeType
-
-
-def files_changes_from_upload(changes: dict, version: int) -> List["ProjectFileChange"]:
-    """Create a list of version file changes from upload changes dictionary used by public API.
-
-    It flattens changes dict and adds change type to each item. Also generates location for each file.
-    """
-    secure_filenames = []
-    version_changes = []
-    version = "v" + str(version)
-    for key in ("added", "updated", "removed"):
-        for item in changes.get(key, []):
-            location = os.path.join(version, mergin_secure_filename(item["path"]))
-            diff = None
-
-            # make sure we have unique location for each file
-            if location in secure_filenames:
-                filename, file_extension = os.path.splitext(location)
-                location = filename + f".{str(uuid.uuid4())}" + file_extension
-
-            secure_filenames.append(location)
-
-            if key == "removed":
-                change = PushChangeType.DELETE
-                location = None
-            elif key == "added":
-                change = PushChangeType.CREATE
-            else:
-                change = PushChangeType.UPDATE
-                if item.get("diff"):
-                    change = PushChangeType.UPDATE_DIFF
-                    diff_location = os.path.join(
-                        version, mergin_secure_filename(item["diff"]["path"])
-                    )
-                    if diff_location in secure_filenames:
-                        filename, file_extension = os.path.splitext(diff_location)
-                        diff_location = (
-                            filename + f".{str(uuid.uuid4())}" + file_extension
-                        )
-
-                    secure_filenames.append(diff_location)
-                    diff = ProjectDiffFile(
-                        path=item["diff"]["path"],
-                        checksum=item["diff"]["checksum"],
-                        size=item["diff"]["size"],
-                        location=diff_location,
-                    )
-
-            file_change = ProjectFileChange(
-                path=item["path"],
-                checksum=item["checksum"],
-                size=item["size"],
-                mtime=None,
-                change=change,
-                location=location,
-                diff=diff,
-            )
-            version_changes.append(file_change)
+class UploadFile(File):
+    """File to be uploaded coming from client push process"""
 
-    return version_changes
+    # determined by client
+    chunks: Optional[List[str]]
+    diff: Optional[File]
+
+
+@dataclass
+class UploadChanges:
+    added: List[UploadFile]
+    updated: List[UploadFile]
+    removed: List[UploadFile]
 
 
 class FileSchema(ma.Schema):
     path = fields.String()
     size = fields.Integer()
     checksum = fields.String()
+    location = fields.String(load_default="", load_only=True)
 
     class Meta:
         unknown = EXCLUDE
 
+    @post_load
+    def create_obj(self, data, **kwargs):
+        return File(**data)
+
 
 class UploadFileSchema(FileSchema):
     chunks = fields.List(fields.String(), load_default=[])
     diff = fields.Nested(FileSchema(), many=False, load_default=None)
 
+    @pre_load
+    def pre_load(self, data, **kwargs):
+        # add future location based on context version
+        version = f"v{self.context.get('version')}"
+        if not data.get("location"):
+            data["location"] = os.path.join(
+                version, mergin_secure_filename(data["path"])
+            )
+        if data.get("diff") and not data.get("diff").get("location"):
+            data["diff"]["location"] = os.path.join(
+                version, mergin_secure_filename(data["diff"]["path"])
+            )
+        return data
+
+    @post_load
+    def create_obj(self, data, **kwargs):
+        return UploadFile(**data)
+
 
 class ChangesSchema(ma.Schema):
     """Schema for upload changes"""
 
-    added = fields.List(
-        fields.Nested(UploadFileSchema()), load_default=[], dump_default=[]
-    )
-    updated = fields.List(
-        fields.Nested(UploadFileSchema()), load_default=[], dump_default=[]
-    )
-    removed = fields.List(
-        fields.Nested(UploadFileSchema()), load_default=[], dump_default=[]
-    )
-    is_blocking = fields.Method("_is_blocking")
+    added = fields.List(fields.Nested(UploadFileSchema()), load_default=[])
+    updated = fields.List(fields.Nested(UploadFileSchema()), load_default=[])
+    removed = fields.List(fields.Nested(UploadFileSchema()), load_default=[])
 
     class Meta:
         unknown = EXCLUDE
 
-    def _is_blocking(self, obj) -> bool:
-        """Check if changes would be blocking."""
-        # let's mark upload as non-blocking only if there are new non-spatial data added (e.g. photos)
-        return bool(
-            len(obj.get("updated", []))
-            or len(obj.get("removed", []))
-            or any(
-                is_qgis(f["path"]) or is_versioned_file(f["path"])
-                for f in obj.get("added", [])
-            )
-        )
-
-    @post_dump
-    def remove_blacklisted_files(self, data, **kwargs):
-        """Files which are blacklisted are not allowed to be uploaded and are simple ignored."""
-        for key in ("added", "updated", "removed"):
-            data[key] = [
-                f
-                for f in data[key]
-                if not is_file_name_blacklisted(
-                    f["path"], current_app.config["BLACKLIST"]
-                )
-            ]
-        return data
-
-    @validates_schema
-    def validate(self, data, **kwargs):
-        """Basic consistency validations for upload metadata"""
-        changes_files = [
-            f["path"] for f in data["added"] + data["updated"] + data["removed"]
-        ]
-
-        if len(changes_files) == 0:
-            raise ValidationError("No changes")
-
-        # changes' files must be unique
-        if len(set(changes_files)) != len(changes_files):
-            raise ValidationError("Not unique changes")
-
-        # check if all .gpkg file are valid
-        for file in data["added"] + data["updated"]:
-            if is_versioned_file(file["path"]) and file["size"] == 0:
-                raise ValidationError("File is not valid")
+    @post_load
+    def create_obj(self, data, **kwargs):
+        return UploadChanges(**data)
 
 
 class ProjectFileSchema(FileSchema):