Syncer

Author: harminius

README.md

@@ -85,6 +85,7 @@ Commands:
   show-file-history    Display information about a single version of a...
   show-version         Display information about a single version of a...
   status               Show all changes in project files - upstream and...
+  sync                 Syncronize the project.
 ```
 
 ### Examples

mergin/cli.py

@@ -33,8 +33,8 @@
     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
-
+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 pygeodiff import GeoDiff
 
@@ -403,6 +403,47 @@ def status(ctx):
     pretty_summary(push_changes_summary)
 
 
+@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()
+    syncer = Syncer(mc, directory)
+    size = syncer.estimate_total_upload()
+    if size == 0:
+        click.secho("Already up to date.", fg="green")
+        return
+
+    try:
+        uploaded_so_far = 0
+
+        with click.progressbar(length=size, label="Uploading changes") as bar:
+            # updates the progress bar
+            def on_progress(last, now):
+                nonlocal uploaded_so_far
+                delta = now - last
+                uploaded_so_far += delta
+                bar.update(delta)
+
+            # run pull → push cycles
+            syncer.sync_loop(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...")
+        syncer.cancel()
+    except Exception as e:
+        _print_unhandled_exception()
+
 @cli.command()
 @click.pass_context
 def push(ctx):
@@ -412,24 +453,17 @@ def push(ctx):
         return
     directory = os.getcwd()
     try:
-        # keep going until there are no more changes
-        while True:
-            job = push_next_change(mc, directory)
-            if job is None:
-                click.echo("All changes uploaded.")
-                break
-
-            # show progress for this single change upload
-            with click.progressbar(length=job.total_size, label="Uploading change") as bar:
+        job = push_project_async(mc, directory)
+        if job is not None:  # if job is none, we don't upload any files, and the transaction is finished already
+            with click.progressbar(length=job.total_size) as bar:
                 last_transferred_size = 0
                 while push_project_is_running(job):
                     time.sleep(1 / 10)  # 100ms
                     new_transferred_size = job.transferred_size
                     bar.update(new_transferred_size - last_transferred_size)  # the update() needs increment only
                     last_transferred_size = new_transferred_size
-            # finalize this change upload (bump versions on server & locally)
             push_project_finalize(job)
-            click.echo("Change pushed, checking for more…")
+        click.secho("Done", fg="green")
     except InvalidProject as e:
         click.secho("Invalid project directory ({})".format(str(e)), fg="red")
     except ClientError as e:
@@ -463,7 +497,7 @@ def pull(ctx):
                 bar.update(new_transferred_size - last_transferred_size)  # the update() needs increment only
                 last_transferred_size = new_transferred_size
         pull_project_finalize(job)
-        click.echo("Done")
+        click.secho("Done", fg="green")
     except InvalidProject as e:
         click.secho("Invalid project directory ({})".format(str(e)), fg="red")
     except ClientError as e:

mergin/client.py

@@ -19,6 +19,7 @@
 
 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 (
@@ -33,7 +34,7 @@
     download_diffs_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
+from .client_push import push_next_change, push_project_wait, push_project_finalize, push_project_async
 from .utils import DateTimeEncoder, get_versions_with_file_changes, int_version, is_version_acceptable
 from .version import __version__
 
@@ -890,19 +891,24 @@ 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.
 
         :param directory: Project's directory
         :type directory: String
         """
-        while True:
-            job = push_next_change(self, directory)
-            if not job:
-                return  # there is nothing to push (or we only deleted some files)
-            push_project_wait(job)
-            push_project_finalize(job)
+        # while True:
+        job = push_project_async(self, directory)
+        if not job:
+            return  # there is nothing to push (or we only deleted some files)
+        push_project_wait(job)
+        push_project_finalize(job)
 
     def pull_project(self, directory):
         """
@@ -917,6 +923,25 @@ def pull_project(self, directory):
         pull_project_wait(job)
         return pull_project_finalize(job)
 
+
+    def sync_project(self, directory):
+
+        syncer = Syncer(self, directory)
+        syncer.sync_loop()
+
+        # 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)
+
+
     def clone_project(self, source_project_path, cloned_project_name, cloned_project_namespace=None):
         """
         Clone project on server.

mergin/client_push.py

@@ -15,8 +15,7 @@
 import tempfile
 import concurrent.futures
 import os
-from typing import Dict, List, Optional
-import click
+from typing import Dict, List, Optional, Tuple
 
 from .common import UPLOAD_CHUNK_SIZE, ClientError
 from .merginproject import MerginProject
@@ -27,7 +26,7 @@
 class UploadJob:
     """Keeps all the important data about a pending upload job"""
 
-    def __init__(self, project_path, changes, transaction_id, mp, mc, tmp_dir):
+    def __init__(self, project_path, changes, transaction_id, mp, mc, tmp_dir, exclusive: bool):
         self.project_path = project_path  # full project name ("username/projectname")
         self.changes = changes  # dictionary of local changes to the project
         self.transaction_id = transaction_id  # ID of the transaction assigned by the server
@@ -37,6 +36,7 @@ def __init__(self, project_path, changes, transaction_id, mp, mc, tmp_dir):
         self.mp = mp  # MerginProject instance
         self.mc = mc  # MerginClient instance
         self.tmp_dir = tmp_dir  # TemporaryDirectory instance for any temp file we need
+        self.exclusive = exclusive  # flag whether this upload blocks other uploads
         self.is_cancelled = False  # whether upload has been cancelled
         self.executor = None  # ThreadPoolExecutor that manages background upload tasks
         self.futures = []  # list of futures submitted to the executor
@@ -160,7 +160,7 @@ def split(self) -> List[Dict[str, List[dict]]]:
         return changes_list
 
 
-def push_next_change(mc, directory) -> Optional[UploadJob]:
+def push_project_async(mc, directory) -> Optional[UploadJob]:
     """Starts push of a change of a project and returns pending upload job"""
 
     mp = MerginProject(directory)
@@ -198,14 +198,8 @@ def push_next_change(mc, directory) -> Optional[UploadJob]:
             + f"\n\nLocal version: {local_version}\nServer version: {server_version}"
         )
 
-    all_changes = mp.get_push_changes()
-    changes_list = ChangesHandler(mc, project_info, all_changes).split()
-    if not changes_list:
-        return None
-
-    # take only the first change
-    change = changes_list[0]
-    mp.log.debug("push change:\n" + pprint.pformat(change))
+    changes = change_batch or mp.get_push_changes()
+    mp.log.debug("push change:\n" + pprint.pformat(changes))
 
     tmp_dir = tempfile.TemporaryDirectory(prefix="python-api-client-")
 
@@ -214,22 +208,22 @@ def push_next_change(mc, directory) -> Optional[UploadJob]:
     # That's because if there are pending transactions, checkpointing or switching from WAL mode
     # won't work, and we would end up with some changes left in -wal file which do not get
     # uploaded. The temporary copy using geodiff uses sqlite backup API and should copy everything.
-    for f in change["updated"]:
+    for f in changes["updated"]:
         if mp.is_versioned_file(f["path"]) and "diff" not in f:
             mp.copy_versioned_file_for_upload(f, tmp_dir.name)
 
-    for f in change["added"]:
+    for f in changes["added"]:
         if mp.is_versioned_file(f["path"]):
             mp.copy_versioned_file_for_upload(f, tmp_dir.name)
 
-    if not any(len(v) for v in change.values()):
+    if not any(len(v) for v in changes.values()):
         mp.log.info(f"--- push {project_path} - nothing to do")
         return
 
     # drop internal info from being sent to server
-    for item in change["updated"]:
+    for item in changes["updated"]:
         item.pop("origin_checksum", None)
-    data = {"version": local_version, "changes": change}
+    data = {"version": local_version, "changes": changes}
 
     try:
         resp = mc.post(
@@ -246,15 +240,16 @@ def push_next_change(mc, directory) -> Optional[UploadJob]:
     upload_files = data["changes"]["added"] + data["changes"]["updated"]
 
     transaction_id = server_resp["transaction"] if upload_files else None
-    job = UploadJob(project_path, change, transaction_id, mp, mc, tmp_dir)
+    exclusive = server_resp.get("exclusive", True)
+    job = UploadJob(project_path, changes, transaction_id, mp, mc, tmp_dir, exclusive)
 
     if not upload_files:
         mp.log.info("not uploading any files")
         job.server_resp = server_resp
         push_project_finalize(job)
         return None  # all done - no pending job
 
-    mp.log.info(f"got transaction ID {transaction_id}")
+    mp.log.info(f"got transaction ID {transaction_id}, {'exclusive' if exclusive else 'non-exclusive'} upload")
 
     upload_queue_items = []
     total_size = 0
@@ -347,7 +342,7 @@ def push_project_finalize(job):
 
     if with_upload_of_files:
         try:
-            job.mp.log.info(f"Finishing transaction {job.transaction_id}")
+            job.mp.log.info(f"Finishing {'exclusive' if job.exclusive else 'non-exclusive'} transaction {job.transaction_id}")
             resp = job.mc.post("/v1/project/push/finish/%s" % job.transaction_id)
             job.server_resp = json.load(resp)
         except ClientError as err:
@@ -417,3 +412,10 @@ def remove_diff_files(job) -> None:
             diff_file = job.mp.fpath_meta(change["diff"]["path"])
             if os.path.exists(diff_file):
                 os.remove(diff_file)
+
+def get_next_batch(project_dir) -> Tuple[Dict[str], bool]:
+    """
+    Return the next dictionary with changes, similar to changes[0] in push_project_async.
+    """
+    # TODO
+    return {"added": [], "updated": [], "removed": []}, True

mergin/client_sync.py

@@ -0,0 +1,81 @@
+import time
+from typing import Dict, Any
+
+from .merginproject import MerginProject
+from .client_pull import DownloadJob, pull_project_async, pull_project_wait, pull_project_finalize, pull_project_cancel
+from .client_push import UploadJob, push_project_async, push_project_wait, push_project_finalize, push_next_change, \
+    push_project_cancel, push_project_is_running
+
+
+class Syncer:
+    def __init__(self, client, directory, download_job: DownloadJob = None, upload_job: UploadJob = None, total_upload_size: int = 0, uploaded_size: int = 0):
+        self.client = client
+        self.directory = directory
+        self.mp = MerginProject(directory)
+        self.download_job = download_job
+        self.upload_job = upload_job
+        self.total_upload_size = total_upload_size
+        self.uploaded_size = uploaded_size
+
+    def _one_cycle(self):
+        """Pull remote, then kick off an upload (if there are changes).
+        Returns an UploadJob (with .transferred_size) or None if up‑to‑date."""
+        download_job = pull_project_async(self.client, self.directory)
+        if download_job:
+            pull_project_wait(download_job)
+            pull_project_finalize(download_job)
+
+        changes = self.mp.get_push_changes()
+        if not changes:
+            return None
+
+        upload_job = push_next_change(self.client, self.mp, changes)
+        return upload_job
+
+    def sync_loop(self, progress_callback=None):
+        """
+        Repeatedly do pull → push cycles until no more changes.
+        If progress_callback is provided, it’s called as:
+            progress_callback(upload_job, last_transferred, new_transferred)
+        """
+        while True:
+            upload_job = self._one_cycle()
+            if upload_job is None:
+                break
+
+            # no UI: just wait & finalize
+            if progress_callback is None:
+                push_project_wait(upload_job)
+                push_project_finalize(upload_job)
+                continue
+
+            # drive the callback
+            last = 0
+            while push_project_is_running(upload_job):
+                time.sleep(0.1)
+                now = upload_job.transferred_size
+                progress_callback(upload_job, last, now)
+                last = now
+
+            push_project_wait(upload_job)
+            push_project_finalize(upload_job)
+
+    def cancel(self):
+        if self.download_job is not None:
+            pull_project_cancel(self.download_job)
+        if self.upload_job is not None:
+            push_project_cancel(self.upload_job)
+
+    def estimate_total_upload(self):
+        """
+        One‑shot estimate of the _current_ total upload size.
+        """
+        changes = self.mp.get_push_changes()
+        return calculate_uploads_size(changes) if changes else 0
+
+def calculate_uploads_size(changes: Dict[str, Any]) -> int:
+    files = changes.get("added", []) + changes.get("updated", [])
+    return sum(
+        f.get("diff", {}).get("size", f.get("size", 0))
+        for f in files
+    )