Split push for blocking and non-blocking

README.md

@@ -81,10 +81,11 @@ Commands:
   share                Fetch permissions to project
   share-add            Add permissions to [users] to project
   share-remove         Remove [users] permissions from project
-  show-file-changeset  Displays information about project changes.
-  show-file-history    Displays information about a single version of a...
-  show-version         Displays information about a single version of a...
+  show-file-changeset  Display information about project changes.
+  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                 Synchronize the project.
 ```
 
 ### Examples
@@ -99,7 +100,7 @@ To download a specific version of a project:
 $ mergin --username john download --version v42 john/project1 ~/mergin/project1
 ```
 
-To download a sepecific version of a single file:
+To download a specific version of a single file:
 
 1. First you need to download the project:
 ```

examples/README.md

@@ -5,10 +5,10 @@ Here you can find some tailored Jupyter Notebooks examples on how to interact wi
 These examples are split into scenarios in order to highlight some of the core capabilities of the Mergin Maps API.
 
 ## [Scenario 1](01_users.ipynb) - Users Management
-On this scenario you'll create some random users from an existing CSV into Mergin Maps, change user's `ROLE` on a `WORKSPACE` and `PROJECT` perspective as well remove user's from a specific project.
+In this scenario you'll create some random users from an existing CSV into Mergin Maps, change user's `ROLE` on a `WORKSPACE` and `PROJECT` perspective as well remove a user from a specific project.
 
-## [Scenario 2](02_sync.ipynb) - Synchronization
-On this scenario you'll learn on how to do basic synchronisation (`PUSH`, `PULL`) of projects with the Mergin Maps Python API client.
+## [Scenario 2](02_sync.ipynb) - Synchronisation
+In this scenario you'll learn how to do basic synchronisation (`PUSH`, `PULL`) of projects with the Mergin Maps Python API client.
 
 ## [Scenario 3](03_projects.ipynb) - Projects Management
-On this scenario you'll learn how to manage project with the Mergin Maps Python API client, namely how to clone projects and how to separate team members in isolated projects from the cloned template.
+In this scenario you'll learn how to manage projects with Mergin Maps Python API client, namely how to clone projects and how to separate team members in isolated projects from the cloned template.

mergin/cli.py

@@ -33,8 +33,13 @@
     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_project_async, push_project_is_running, push_project_finalize, push_project_cancel
-
+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
 
@@ -403,6 +408,42 @@ 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()
+
+    try:
+        size = total_upload_size(directory)
+        with click.progressbar(length=size, label="Syncing") as bar:
+
+            def on_progress(increment):
+                bar.update(increment)
+
+            # 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 mc.pull_job:
+            pull_project_cancel(mc.pull_job)
+        if mc.push_job:
+            push_project_cancel(mc.push_job)
+    except Exception as e:
+        _print_unhandled_exception()
+
+
 @cli.command()
 @click.pass_context
 def push(ctx):
@@ -422,7 +463,7 @@ def push(ctx):
                     bar.update(new_transferred_size - last_transferred_size)  # the update() needs increment only
                     last_transferred_size = new_transferred_size
             push_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:
@@ -456,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:
@@ -473,7 +514,7 @@ def pull(ctx):
 @click.argument("version")
 @click.pass_context
 def show_version(ctx, version):
-    """Displays information about a single version of a project. `version` is 'v1', 'v2', etc."""
+    """Display information about a single version of a project. `version` is 'v1', 'v2', etc."""
     mc = ctx.obj["client"]
     if mc is None:
         return
@@ -492,7 +533,7 @@ def show_version(ctx, version):
 @click.argument("path")
 @click.pass_context
 def show_file_history(ctx, path):
-    """Displays information about a single version of a project."""
+    """Display information about a single version of a project."""
     mc = ctx.obj["client"]
     if mc is None:
         return
@@ -516,7 +557,7 @@ def show_file_history(ctx, path):
 @click.argument("version")
 @click.pass_context
 def show_file_changeset(ctx, path, version):
-    """Displays information about project changes."""
+    """Display information about project changes."""
     mc = ctx.obj["client"]
     if mc is None:
         return

mergin/client.py

@@ -1,4 +1,6 @@
 import logging
+from time import sleep
+
 import math
 import os
 import json
@@ -31,9 +33,17 @@
     download_project_finalize,
     download_project_wait,
     download_diffs_finalize,
+    pull_project_async,
+    pull_project_wait,
+    pull_project_finalize,
+)
+from .client_push import (
+    push_project_wait,
+    push_project_finalize,
+    push_project_async,
+    push_project_is_running,
+    get_change_batch,
 )
-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 .utils import DateTimeEncoder, get_versions_with_file_changes, int_version, is_version_acceptable
 from .version import __version__
 
@@ -45,6 +55,10 @@ class TokenError(Exception):
     pass
 
 
+class AuthTokenExpiredError(Exception):
+    pass
+
+
 class ServerType(Enum):
     OLD = auto()  # Server is old and does not support workspaces
     CE = auto()  # Server is Community Edition
@@ -80,8 +94,16 @@ class MerginClient:
         Currently, only HTTP proxies are supported.
     """
 
-    def __init__(self, url=None, auth_token=None, login=None, password=None, plugin_version=None, proxy_config=None):
-        self.url = url if url is not None else MerginClient.default_url()
+    def __init__(
+        self,
+        url=None,
+        auth_token=None,
+        login=None,
+        password=None,
+        plugin_version=None,
+        proxy_config=None,
+    ):
+        self.url = (url if url is not None else MerginClient.default_url()).rstrip("/") + "/"
         self._auth_params = None
         self._auth_session = None
         self._user_info = None
@@ -91,6 +113,8 @@ def __init__(self, url=None, auth_token=None, login=None, password=None, plugin_
         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)
@@ -190,11 +214,19 @@ def wrapper(self, *args):
                     delta = self._auth_session["expire"] - datetime.now(timezone.utc)
                     if delta.total_seconds() < 5:
                         self.log.info("Token has expired - refreshing...")
-                        self.login(self._auth_params["login"], self._auth_params["password"])
+                        if self._auth_params.get("login", None) and self._auth_params.get("password", None):
+                            self.log.info("Token has expired - refreshing...")
+                            self.login(self._auth_params["login"], self._auth_params["password"])
+                        else:
+                            raise AuthTokenExpiredError("Token has expired - please re-login")
                 else:
                     # Create a new authorization token
                     self.log.info(f"No token - login user: {self._auth_params['login']}")
-                    self.login(self._auth_params["login"], self._auth_params["password"])
+                    if self._auth_params.get("login", None) and self._auth_params.get("password", None):
+                        self.login(self._auth_params["login"], self._auth_params["password"])
+                    else:
+                        raise ClientError("Missing login or password")
+
             return f(self, *args)
 
         return wrapper
@@ -453,7 +485,8 @@ def create_project(self, project_name, is_public=False, namespace=None):
 
     def create_project_and_push(self, project_name, directory, is_public=False, namespace=None):
         """
-        Convenience method to create project and push the initial version right after that.
+        Convenience method to create project and push the the files right after that.
+        Creates two versions when directory contains blocking and non-blocking changes.
 
         :param project_name: Project's full name (<namespace>/<name>)
         :type project_name: String
@@ -877,7 +910,7 @@ def push_project(self, directory):
         :type directory: String
         """
         job = push_project_async(self, directory)
-        if job is None:
+        if not job:
             return  # there is nothing to push (or we only deleted some files)
         push_project_wait(job)
         push_project_finalize(job)
@@ -895,6 +928,62 @@ 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 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)
+        """
+        has_more_changes = True
+        while has_more_changes:
+            self.pull_job = pull_project_async(self, project_dir)
+            if self.pull_job is not None:
+                pull_project_wait(self.pull_job)
+                pull_project_finalize(self.pull_job)
+
+            changes_batch, has_more_changes = get_change_batch(self, project_dir)
+            if not changes_batch:  # no changes to upload, quit sync
+                return
+
+            self.push_job = push_project_async(self, project_dir, changes_batch)
+            if self.push_job:
+                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=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
+        """
+        has_more_changes = True
+        while has_more_changes:
+            self.pull_job = pull_project_async(self, project_dir)
+            if self.pull_job is not None:
+                pull_project_wait(self.pull_job)
+                pull_project_finalize(self.pull_job)
+
+            changes_batch, has_more_changes = get_change_batch(self, project_dir)
+            if not changes_batch:  # no changes to upload, quit sync
+                return
+
+            self.push_job = push_project_async(self, project_dir, changes_batch)
+            if self.push_job:
+                last = 0
+                while push_project_is_running(self.push_job):
+                    sleep(sleep_time)
+                    now = self.push_job.transferred_size
+                    progress_callback(now - last)  # update progressbar with transferred size increment
+                    last = now
+                push_project_finalize(self.push_job)
+
     def clone_project(self, source_project_path, cloned_project_name, cloned_project_namespace=None):
         """
         Clone project on server.

mergin/client_pull.py

@@ -129,7 +129,7 @@ def download_project_async(mc, project_path, directory, project_version=None):
     """
 
     if "/" not in project_path:
-        raise ClientError("Project name needs to be fully qualified, e.g. <username>/<projectname>")
+        raise ClientError("Project name needs to be fully qualified, e.g. <workspacename>/<projectname>")
     if os.path.exists(directory):
         raise ClientError("Project directory already exists")
     os.makedirs(directory)