sync methods

Author: harminius

mergin/cli.py

@@ -33,8 +33,7 @@
     download_project_is_running,
 )
 from mergin.client_pull import pull_project_async, pull_project_is_running, pull_project_finalize, pull_project_cancel
-from mergin.client_push import push_next_change, push_project_is_running, push_project_finalize, push_project_cancel, push_project_async
-from mergin.client_sync import Syncer, calculate_uploads_size
+from mergin.client_push import push_project_is_running, push_project_finalize, push_project_cancel, push_project_async, total_upload_size
 
 from pygeodiff import GeoDiff
 
@@ -411,8 +410,8 @@ def sync(ctx):
     if mc is None:
         return
     directory = os.getcwd()
-    syncer = Syncer(mc, directory)
-    size = syncer.estimate_total_upload()
+
+    size = total_upload_size(directory)
     if size == 0:
         click.secho("Already up to date.", fg="green")
         return
@@ -428,8 +427,8 @@ def on_progress(last, now):
                 uploaded_so_far += delta
                 bar.update(delta)
 
-            # run pull → push cycles
-            syncer.sync_loop(progress_callback=on_progress)
+            # run pull & push cycles
+            mc.sync_project_with_callback(directory, progress_callback=on_progress)
 
         click.secho("Sync complete.", fg="green")
 
@@ -440,7 +439,8 @@ def on_progress(last, now):
         return
     except KeyboardInterrupt:
         click.secho("Cancelling...")
-        syncer.cancel()
+        push_project_cancel(mc.upload_job)
+        pull_project_cancel(mc.download_job)
     except Exception as e:
         _print_unhandled_exception()
 

mergin/client.py

@@ -1,4 +1,6 @@
 import logging
+from time import sleep
+
 import math
 import os
 import json
@@ -19,7 +21,6 @@
 
 from typing import List
 
-from .client_sync import Syncer
 from .common import ClientError, LoginError, WorkspaceRole, ProjectRole, LOG_FILE_SIZE_TO_SEND, MERGIN_DEFAULT_LOGS_URL
 from .merginproject import MerginProject
 from .client_pull import (
@@ -32,9 +33,13 @@
     download_project_finalize,
     download_project_wait,
     download_diffs_finalize,
+    pull_project_is_running,
+    pull_project_async,
+    pull_project_wait,
+    pull_project_finalize
 )
-from .client_pull import pull_project_async, pull_project_wait, pull_project_finalize
-from .client_push import push_next_change, push_project_wait, push_project_finalize, push_project_async
+from .client_push import push_project_wait, push_project_finalize, push_project_async, push_project_is_running, \
+    ChangesHandler
 from .utils import DateTimeEncoder, get_versions_with_file_changes, int_version, is_version_acceptable
 from .version import __version__
 
@@ -104,6 +109,8 @@ def __init__(
         if plugin_version is not None:  # this could be e.g. "Plugin/2020.1 QGIS/3.14"
             self.client_version += " " + plugin_version
         self.setup_logging()
+        self.pull_job = None
+        self.push_job = None
         if auth_token:
             try:
                 token_data = decode_token_data(auth_token)
@@ -891,11 +898,6 @@ def project_user_permissions(self, project_path):
         result["readers"] = access.get("readersnames", [])
         return result
 
-
-
-
-
-
     def push_project(self, directory):
         """
         Upload local changes to the repository.
@@ -923,23 +925,56 @@ def pull_project(self, directory):
         pull_project_wait(job)
         return pull_project_finalize(job)
 
+    def sync_project(self, project_dir):
+        """
+        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 changes pending.
+        """
+        has_more_changes = True
+        while has_more_changes:
+            self.pull_job = pull_project_async(self, project_dir)
+            pull_project_wait(self.pull_job)
+            pull_project_finalize(self.pull_job)
+
+            changes_handler = ChangesHandler(self, project_dir)
+            changes_batch, has_more_changes = changes_handler.get_change_batch()
+
+            self.push_job = push_project_async(project_dir, changes_batch)
+            push_project_wait(self.push_job)
+            push_project_finalize(self.push_job)
+
+    def sync_project_with_callback(self, project_dir, progress_callback=None, sleep_time=100):
+        """
+        Syncs project while sending push progress info as callback.
+        Sync is done in this loop:
+            Pending changes? -> Pull -> Push change batch -> repeat
+        """
+        has_more_changes = True
+        while has_more_changes:
+            changes_handler = ChangesHandler(self, project_dir)
+
+            self.pull_job = pull_project_async(self, project_dir)
+            if self.pull_job is None:
+                return
+            pull_project_wait(self.pull_job)
+            pull_project_finalize(self.pull_job)
 
-    def sync_project(self, directory):
+            changes_batch, has_more_changes = changes_handler.get_change_batch()
 
-        syncer = Syncer(self, directory)
-        syncer.sync_loop()
+            self.push_job = push_project_async(self, project_dir, changes_batch)
+            if not self.push_job:
+                return
 
-        # self.pull_project(directory)
-        #
-        # mp = MerginProject(directory)
-        # changes = mp.get_push_changes()
-        #
-        # if not changes:
-        #     return
-        #
-        # push_changes(self, mp, changes)
-        #
-        # self.sync_project(directory)
+            last = 0
+            while push_project_is_running(self.push_job):
+                sleep(sleep_time)
+                now = self.push_job.transferred_size
+                progress_callback(last, now)
+                last = now
+            push_project_finalize(self.push_job)
 
 
     def clone_project(self, source_project_path, cloned_project_name, cloned_project_namespace=None):

mergin/client_push.py

@@ -96,10 +96,10 @@ class ChangesHandler:
     - Generating upload-ready change groups for asynchronous job creation.
     """
 
-    def __init__(self, client, project_info, changes):
+    def __init__(self, client, mp):
         self.client = client
-        self.project_info = project_info
-        self._raw_changes = changes
+        self.mp = MerginProject(mp)
+        self._raw_changes = self.mp.get_push_changes()
 
     @staticmethod
     def is_blocking_file(file):
@@ -117,7 +117,12 @@ def _filter_changes(self, changes: Dict[str, List[dict]]) -> Dict[str, List[dict
         Returns:
             dict[str, list[dict]]: The filtered changes dictionary.
         """
-        if not is_editor_enabled(self.client, self.project_info):
+        project_name = self.mp.project_full_name()
+        try:
+            project_info = self.client.project_info(project_name)
+        except Exception as e:
+            self.mp.log.error(f"Failed to get project info for project {project_name}: {e}")
+        if not is_editor_enabled(self.client, project_info):
             return changes
         return _apply_editor_filters(changes)
 
@@ -159,8 +164,17 @@ def split(self) -> List[Dict[str, List[dict]]]:
         # TODO: apply limits; changes = self._limit_by_file_count(changes)
         return changes_list
 
+    def get_change_batch(self) -> Tuple[Optional[Dict[str, List[dict]]], bool]:
+        """
+        Return the next changes dictionary and flag if there are more changes
+        """
+        changes_list = self.split()
+        if not changes_list:
+            return None, False
+        return changes_list[0], len(changes_list) > 1
+
 
-def push_project_async(mc, directory) -> Optional[UploadJob]:
+def push_project_async(mc, directory, change_batch=None) -> Optional[UploadJob]:
     """Starts push of a change of a project and returns pending upload job"""
 
     mp = MerginProject(directory)
@@ -413,9 +427,15 @@ def remove_diff_files(job) -> None:
             if os.path.exists(diff_file):
                 os.remove(diff_file)
 
-def get_next_batch(project_dir) -> Tuple[Dict[str], bool]:
+
+def total_upload_size(directory) -> int:
     """
-    Return the next dictionary with changes, similar to changes[0] in push_project_async.
+    Total up the number of bytes that need to be uploaded.
     """
-    # TODO
-    return {"added": [], "updated": [], "removed": []}, True
+    mp = MerginProject(directory)
+    changes = mp.get_push_changes()
+    files = changes.get("added", []) + changes.get("updated", [])
+    return sum(
+        f.get("diff", {}).get("size", f.get("size", 0))
+        for f in files
+    )

mergin/client_sync.py

@@ -9,7 +9,7 @@
 import uuid
 import tempfile
 from datetime import datetime
-from typing import List, Dict, Tuple
+from typing import List, Dict
 from dateutil.tz import tzlocal
 
 from .editor import prevent_conflicted_copy