fixes for missing schema

- fixing update chunks for files with the same checksum
- added minimal version of sync command

Author: MarcelGeo

mergin/cli.py

@@ -468,6 +468,38 @@ def pull(ctx):
     except Exception as e:
         _print_unhandled_exception()
 
+@cli.command()
+@click.pass_context
+def sync(ctx):
+    """Synchronize the project. Pull latest project version from the server and push split changes."""
+    mc = ctx.obj["client"]
+    if mc is None:
+        return
+    directory = os.getcwd()
+    upload_job = None
+    length = 1
+    try:
+        def on_progress(increment, push_job):
+            nonlocal upload_job
+            upload_job = push_job
+
+        # run pull & push cycles until there are no local changes
+        mc.sync_project_with_callback(directory, progress_callback=on_progress)
+
+        click.secho("Sync complete.", fg="green")
+
+    except InvalidProject as e:
+        click.secho("Invalid project directory ({})".format(str(e)), fg="red")
+    except ClientError as e:
+        click.secho("Error: " + str(e), fg="red")
+        return
+    except KeyboardInterrupt:
+        click.secho("Cancelling...")
+        if upload_job:
+            push_project_cancel(upload_job)
+    except Exception as e:
+        _print_unhandled_exception()
+
 
 @cli.command()
 @click.argument("version")

mergin/client.py

@@ -3,7 +3,6 @@
 import os
 import json
 import shutil
-import time
 import zlib
 import base64
 import urllib.parse
@@ -17,6 +16,7 @@
 import re
 import typing
 import warnings
+from time import sleep
 
 from typing import List
 
@@ -41,7 +41,7 @@
     download_diffs_finalize,
 )
 from .client_pull import pull_project_async, pull_project_wait, pull_project_finalize
-from .client_push import push_project_async, push_project_wait, push_project_finalize, UploadChunksCache
+from .client_push import get_push_changes_batch, push_project_async, push_project_is_running, push_project_wait, push_project_finalize, UploadChunksCache
 from .utils import DateTimeEncoder, get_versions_with_file_changes, int_version, is_version_acceptable
 from .version import __version__
 
@@ -1486,32 +1486,72 @@ def create_invitation(self, workspace_id: int, email: str, workspace_role: Works
         ws_inv = self.post(f"v2/workspaces/{workspace_id}/invitations", params, json_headers)
         return json.load(ws_inv)
 
-    def sync_project(self, project_dir):
+    def sync_project(self, project_directory):
         """
         Syncs project by loop with these steps:
         1. Pull server version
         2. Get local changes
         3. Push first change batch
         Repeat if there are more local changes.
-        The batch pushing makes use of the server ability to handle simultaneously exclusive upload (that blocks
-            other uploads) and non-exclusive upload (for adding assets)
         """
-        attempts = 2
-        for attempt in range(attempts):
+        mp = MerginProject(project_directory)
+        has_changes = True
+        server_conflict_attempts = 0
+        while has_changes:
+            pull_job = pull_project_async(self, project_directory)
+            if pull_job:
+                pull_project_wait(pull_job)
+                pull_project_finalize(pull_job)
+
             try:
-                pull_job = pull_project_async(self, project_dir)
-                if pull_job:
-                    pull_project_wait(pull_job)
-                    pull_project_finalize(pull_job)
-
-                job = push_project_async(self, project_dir)
-                if job:
-                    push_project_wait(job)
-                    push_project_finalize(job)
-                break
+                job = push_project_async(self, project_directory)
+                if not job:
+                    break
+                push_project_wait(job)
+                push_project_finalize(job)
+                _, has_changes = get_push_changes_batch(self, mp, job.server_resp)
+            except ClientError as e:
+                if e.http_error == 409 and server_conflict_attempts < 2:
+                    # retry on conflict, e.g. when server has changes that we do not have yet
+                    mp.log.info("Attempting sync process due to conflicts between server and local directory or another user is syncing.")
+                    server_conflict_attempts += 1
+                    sleep(5)
+                    continue
+                raise e
+
+    def sync_project_with_callback(self, project_directory, progress_callback=None, sleep_time=0.1):
+        """
+        Syncs project while sending push progress info as callback.
+        Sync is done in this loop:
+            Pending changes? -> Pull -> Get changes batch -> Push the changes -> repeat
+        :param progress_callback: updates the progress bar in CLI, on_progress(increment)
+        :param sleep_time: sleep time between calling the callback function
+        """
+        mp = MerginProject(project_directory)
+        has_changes = True
+        server_conflict_attempts = 0
+        while has_changes:
+            pull_job = pull_project_async(self, project_directory)
+            if pull_job:
+                pull_project_wait(pull_job)
+                pull_project_finalize(pull_job)
+            try:
+                job = push_project_async(self, project_directory)
+                if not job:
+                    break
+                last = 0
+                while push_project_is_running(job):
+                    sleep(sleep_time)
+                    now = job.transferred_size
+                    progress_callback(now - last, job)  # update progressbar with transferred size increment
+                    last = now
+                push_project_finalize(job)
+                _, has_changes = get_push_changes_batch(self, mp, job.server_resp)
             except ClientError as e:
-                if e.http_error == 409 and attempt < attempts - 1:
+                if e.http_error == 409 and server_conflict_attempts < 2:
                     # retry on conflict, e.g. when server has changes that we do not have yet
-                    time.sleep(5)
+                    mp.log.info("Attempting sync process due to conflicts between server and local directory or another user is syncing.")
+                    server_conflict_attempts += 1
+                    sleep(5)
                     continue
                 raise e

mergin/client_pull.py

@@ -373,7 +373,7 @@ def dump(self):
         print("--- END ---")
 
 
-def pull_project_async(mc, directory):
+def pull_project_async(mc, directory) -> PullJob:
     """
     Starts project pull in background and returns handle to the pending job.
     Using that object it is possible to watch progress or cancel the ongoing work.

mergin/client_push.py

@@ -109,7 +109,6 @@ def upload_chunk_v2_api(self, data: bytes, checksum: str):
         resp_dict = json.load(resp)
         self.server_chunk_id = resp_dict.get("id")
         self.mc.upload_chunks_cache.add(checksum, self.server_chunk_id)
-        self.mp.log.debug(f"Upload chunk finished: {self.file_path}")
 
     def upload_blocking(self):
         with open(self.file_path, "rb") as file_handle:
@@ -137,7 +136,7 @@ def upload_blocking(self):
                         continue
                     raise
 
-            self.mp.log.debug(f"Upload chunk finished: {self.file_path}")
+            self.mp.log.debug(f"Upload chunk {self.chunk_id} finished: {self.file_path}")
 
 
 class UploadJob:
@@ -188,7 +187,9 @@ def add_items(self, items: List[UploadQueueItem], total_size: int):
 
     def update_chunks_from_items(self):
         """Update chunks in LocalChanges from the upload queue items."""
-        self.changes.update_chunks([(item.file_checksum, item.server_chunk_id) for item in self.upload_queue_items])
+        self.changes.update_chunks(
+            [(item.file_checksum, item.server_chunk_id) for item in self.upload_queue_items]
+        )
 
 
 def create_upload_chunks(mc, mp: MerginProject, local_changes: List[LocalChange]) -> Tuple[List[UploadQueueItem], int]:
@@ -348,11 +349,13 @@ def push_project_async(mc, directory) -> Optional[UploadJob]:
             + f"\n\nLocal version: {local_version}\nServer version: {server_version}"
         )
 
-    changes = mp.get_push_changes()
-    changes = filter_changes(mc, project_info, changes)
-    mp.log.debug("push changes:\n" + pprint.pformat(changes))
-
     tmp_dir = tempfile.TemporaryDirectory(prefix="python-api-client-")
+    changes, changes_len = get_push_changes_batch(mc, mp, project_info)
+    if not changes_len:
+        mp.log.info(f"--- push {project_path} - nothing to do")
+        return
+
+    mp.log.debug("push changes:\n" + pprint.pformat(changes))
 
     # If there are any versioned files (aka .gpkg) that are not updated through a diff,
     # we need to make a temporary copy somewhere to be sure that we are uploading full content.
@@ -367,10 +370,6 @@ def push_project_async(mc, directory) -> Optional[UploadJob]:
         if mp.is_versioned_file(f["path"]):
             mp.copy_versioned_file_for_upload(f, tmp_dir.name)
 
-    if not sum(len(v) for v in changes.values()):
-        mp.log.info(f"--- push {project_path} - nothing to do")
-        return
-
     server_feaure_flags = mc.server_features()
     job = None
     local_changes = LocalChanges(
@@ -527,3 +526,13 @@ def remove_diff_files(job: UploadJob) -> None:
             diff_file = job.mp.fpath_meta(diff.path)
             if os.path.exists(diff_file):
                 os.remove(diff_file)
+
+
+def get_push_changes_batch(mc, mp: MerginProject, project_info: dict) -> Tuple[dict, int]:
+    """
+    Get changes that need to be pushed to the server.
+    """
+    changes = mp.get_push_changes()
+    changes = filter_changes(mc, project_info, changes)
+
+    return changes, sum(len(v) for v in changes.values())

mergin/local_changes.py

@@ -15,6 +15,12 @@ class LocalChange(BaseLocalChange):
     chunks: List[str] = field(default_factory=list)
     diff: Optional[dict] = None 
     upload_file: Optional[str] = None
+    # some functions (MerginProject.compare_file_sets) are adding version to the change from project info
+    version: Optional[str] = None
+    # some functions (MerginProject.compare_file_sets) are adding history dict to the change from project info
+    history: Optional[dict] = None
+    # some functions (MerginProject.compare_file_sets) are adding location dict to the change from project info
+    location: Optional[str] = None
 
     def get_diff(self) -> Optional[BaseLocalChange]:
         if self.diff:
@@ -69,15 +75,15 @@ def update_chunks(self, chunks: List[Tuple[str, str]]) -> None:
         lists based on the provided `chunks` list, which contains tuples of (checksum, chunk_id).
         """
         for change in self.added:
-            change.chunks = [
+            change.chunks = list({
                 chunk[1]
                 for chunk in chunks
                 if chunk[0] == change.checksum
-            ]
+            })
 
         for change in self.updated:
-            change.chunks = [
+            change.chunks = list({
                 chunk[1]
                 for chunk in chunks
                 if chunk[0] == change.checksum
-            ]
+            })