Push v2 integration

.github/workflows/autotests.yml

@@ -19,7 +19,7 @@ jobs:
 
       - uses: actions/setup-python@v2
         with:
-          python-version: '3.x'
+          python-version: '3.8'
 
       - name: Install python package dependencies
         run: |

mergin/cli.py

@@ -469,6 +469,58 @@ def pull(ctx):
         _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()
+    current_job = None
+    current_bar = None
+    try:
+        # Iterate over the generator to get updates
+        for size_change, job in mc.sync_project_generator(directory):
+            # Check if this is a new job (a new push operation)
+            if job and job != current_job:
+                # If a previous bar exists, close it
+                if current_bar:
+                    current_bar.finish()
+
+                # A new push job has started. Initialize a new progress bar.
+                click.echo(f"\nStarting upload")
+                current_job = job
+
+                # The length of the progress bar should be the total size of the job
+                # You'll need to get this from your job object (e.g., job.total_size)
+                total_size = job.total_size
+                current_bar = click.progressbar(
+                    length=total_size,
+                    label=f"Uploading project",
+                )
+
+            # Update the current progress bar with the size increment
+            current_bar.update(size_change)
+
+        # After the loop finishes, make sure to close the final progress bar
+        if current_bar:
+            current_bar.finish()
+        click.secho("\nProject synced successfully", 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 current_job:
+            push_project_cancel(current_job)
+    except Exception as e:
+        _print_unhandled_exception()
+
+
 @cli.command()
 @click.argument("version")
 @click.pass_context

mergin/client.py

@@ -16,8 +16,12 @@
 import re
 import typing
 import warnings
+from time import sleep
 
 from .common import (
+    SYNC_ATTEMPT_WAIT,
+    SYNC_ATTEMPTS,
+    SYNC_CALLBACK_WAIT,
     ClientError,
     LoginError,
     WorkspaceRole,
@@ -38,7 +42,14 @@
     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
+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__
 
@@ -119,6 +130,8 @@ def __init__(
         self._user_info = None
         self._server_type = None
         self._server_version = None
+        self._server_features = {}
+        self.upload_chunks_cache = UploadChunksCache()
         self.client_version = "Python-client/" + __version__
         if plugin_version is not None:  # this could be e.g. "Plugin/2020.1 QGIS/3.14"
             self.client_version += " " + plugin_version
@@ -378,8 +391,7 @@ def server_type(self):
         """
         if not self._server_type:
             try:
-                resp = self.get("/config", validate_auth=False)
-                config = json.load(resp)
+                config = self.server_config()
                 if config["server_type"] == "ce":
                     self._server_type = ServerType.CE
                 elif config["server_type"] == "ee":
@@ -400,14 +412,26 @@ def server_version(self):
         """
         if self._server_version is None:
             try:
-                resp = self.get("/config", validate_auth=False)
-                config = json.load(resp)
+                config = self.server_config()
                 self._server_version = config["version"]
             except (ClientError, KeyError):
                 self._server_version = ""
 
         return self._server_version
 
+    def server_features(self):
+        """
+        Returns feature flags of the server.
+        """
+        if self._server_features:
+            return self._server_features
+        config = self.server_config()
+        self._server_features = {
+            "v2_push_enabled": config.get("v2_push_enabled", False),
+            "v2_pull_enabled": config.get("v2_pull_enabled", False),
+        }
+        return self._server_features
+
     def workspaces_list(self):
         """
         Find all available workspaces
@@ -798,7 +822,7 @@ def download_project(self, project_path, directory, version=None):
     def user_info(self):
         server_type = self.server_type()
         if server_type == ServerType.OLD:
-            resp = self.get("/v1/user/" + self.username())
+            resp = self.get(f"/v1/user/{self.username()}")
         else:
             resp = self.get("/v1/user/profile")
         return json.load(resp)
@@ -1487,3 +1511,56 @@ def create_invitation(self, workspace_id: int, email: str, workspace_role: Works
         params = {"email": email, "role": workspace_role.value}
         ws_inv = self.post(f"v2/workspaces/{workspace_id}/invitations", params, json_headers)
         return json.load(ws_inv)
+
+    def sync_project_generator(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.
+
+        :param project_directory: Project's directory
+        """
+        mp = MerginProject(project_directory)
+        has_changes = True
+        server_conflict_attempts = 0
+        while has_changes:
+            self.pull_project(project_directory)
+            try:
+                job = push_project_async(self, project_directory)
+                if not job:
+                    break
+                # waiting for progress
+                last_size = 0
+                while push_project_is_running(job):
+                    sleep(SYNC_CALLBACK_WAIT)
+                    current_size = job.transferred_size
+                    yield (current_size - last_size, job)  # Yields the size change and the job object
+                    last_size = current_size
+                push_project_finalize(job)
+                _, has_changes = get_push_changes_batch(self, project_directory)
+                server_conflict_attempts = 0
+            except ClientError as e:
+                if e.is_retryable_sync() and server_conflict_attempts < SYNC_ATTEMPTS - 1:
+                    # retry on conflict, e.g. when server has changes that we do not have yet
+                    mp.log.info(
+                        f"Restarting sync process (conflict on server) - {server_conflict_attempts + 1}/{SYNC_ATTEMPTS}"
+                    )
+                    server_conflict_attempts += 1
+                    sleep(SYNC_ATTEMPT_WAIT)
+                    continue
+                raise e
+
+    def sync_project(self, project_directory):
+        """
+        Syncs project by pulling server changes and pushing local changes. There is intorduced retry mechanism
+        for handling server conflicts (when server has changes that we do not have yet or somebody else is syncing).
+        See description of _sync_project_generator().
+
+        :param project_directory: Project's directory
+        """
+        # walk through the generator to perform the sync
+        # in this method we do not yield anything to the caller
+        for _ in self.sync_project_generator(project_directory):
+            pass

mergin/client_pull.py

@@ -145,7 +145,7 @@ def download_project_async(mc, project_path, directory, project_version=None):
     mp.log.info("--- version: " + mc.user_agent_info())
     mp.log.info(f"--- start download {project_path}")
 
-    tmp_dir = tempfile.TemporaryDirectory(prefix="python-api-client-", ignore_cleanup_errors=True, delete=True)
+    tmp_dir = tempfile.TemporaryDirectory(prefix="python-api-client-")
 
     try:
         # check whether we download the latest version or not
@@ -384,7 +384,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.
@@ -424,7 +424,7 @@ def pull_project_async(mc, directory):
     # then we just download the whole file
     _pulling_file_with_diffs = lambda f: "diffs" in f and len(f["diffs"]) != 0
 
-    tmp_dir = tempfile.TemporaryDirectory(prefix="mm-pull-", ignore_cleanup_errors=True, delete=True)
+    tmp_dir = tempfile.TemporaryDirectory(prefix="mm-pull-")
     pull_changes = mp.get_pull_changes(server_info["files"])
     mp.log.debug("pull changes:\n" + pprint.pformat(pull_changes))
     fetch_files = []