Merge remote-tracking branch 'origin/push-v2-integration' into changes_limits

MarcelGeo · MarcelGeo · commit cf232a652fc7 · 2025-09-18T10:58:19.000+02:00
diff --git a/mergin/cli.py b/mergin/cli.py
@@ -477,17 +477,36 @@ def sync(ctx):
     if mc is None:
         return
     directory = os.getcwd()
-    upload_job = None
+    current_job = None
+    current_bar = None
     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(directory, progress_callback=on_progress)
-
-        click.secho("Sync complete.", fg="green")
+        # Iterate over the generator to get updates
+        for size_change, job in mc.sync_project(directory, upload_progress=True):
+            # 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"Starting 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")
@@ -496,8 +515,8 @@ def on_progress(increment, push_job):
         return
     except KeyboardInterrupt:
         click.secho("Cancelling...")
-        if upload_job:
-            push_project_cancel(upload_job)
+        if current_job:
+            push_project_cancel(current_job)
     except Exception as e:
         _print_unhandled_exception()
 
diff --git a/mergin/client.py b/mergin/client.py
@@ -1512,13 +1512,16 @@ 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_directory, progress_callback=None):
+    def sync_project(self, project_directory, upload_progress=False):
         """
         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
+        :param upload_progress: If True, the method will be a generator yielding upload progress as (size_change, job) tuples.
         """
         mp = MerginProject(project_directory)
         has_changes = True
@@ -1533,19 +1536,18 @@ def sync_project(self, project_directory, progress_callback=None):
                 job = push_project_async(self, project_directory)
                 if not job:
                     break
-                if not progress_callback:
+                if not upload_progress:
                     push_project_wait(job)
                 else:
                     last_size = 0
                     while push_project_is_running(job):
                         sleep(SYNC_CALLBACK_WAIT)
                         current_size = job.transferred_size
-                        progress_callback(
-                            current_size - last_size, job
-                        )  # call callback with transferred size increment
+                        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, mp)
+                server_conflict_attempts = 0
             except ClientError as e:
                 if e.is_retryable_sync() and server_conflict_attempts < PUSH_ATTEMPTS - 1:
                     # retry on conflict, e.g. when server has changes that we do not have yet
@@ -1556,5 +1558,3 @@ def sync_project(self, project_directory, progress_callback=None):
                     sleep(PUSH_ATTEMPT_WAIT)
                     continue
                 raise e
-            else:
-                server_conflict_attempts = 0
diff --git a/mergin/client_push.py b/mergin/client_push.py
@@ -22,7 +22,7 @@
 import time
 from typing import List, Tuple, Optional, ByteString
 
-from .local_changes import ChangesValidationError, LocalChange, LocalChanges
+from .local_changes import ChangesValidationError, FileChange, LocalPojectChanges
 
 from .common import (
     MAX_UPLOAD_VERSIONED_SIZE,
@@ -73,7 +73,7 @@ def __init__(
 
         self._request_headers = {"Content-Type": "application/octet-stream"}
 
-    def upload_chunk(self, data: ByteString, checksum: str):
+    def upload_chunk_v1_api(self, data: ByteString, checksum: str):
         """
         Uploads the chunk to the server.
         """
@@ -125,7 +125,7 @@ def upload_blocking(self):
                         self.upload_chunk_v2_api(data, checksum_str)
                     else:
                         # use v1 API for uploading chunks
-                        self.upload_chunk(data, checksum_str)
+                        self.upload_chunk_v1_api(data, checksum_str)
                     break  # exit loop if upload was successful
                 except ClientError as e:
                     if attempt < UPLOAD_CHUNK_ATTEMPTS - 1:
@@ -140,10 +140,10 @@ class UploadJob:
     """Keeps all the important data about a pending upload job"""
 
     def __init__(
-        self, version: str, changes: LocalChanges, transaction_id: Optional[str], mp: MerginProject, mc, tmp_dir
+        self, version: str, changes: LocalPojectChanges, transaction_id: Optional[str], mp: MerginProject, mc, tmp_dir
     ):
         self.version = version
-        self.changes: LocalChanges = changes  # dictionary of local changes to the project
+        self.changes: LocalPojectChanges = changes  # dictionary of local changes to the project
         self.transaction_id = transaction_id  # ID of the transaction assigned by the server
         self.total_size = 0  # size of data to upload (in bytes)
         self.transferred_size = 0  # size of data already uploaded (in bytes)
@@ -167,8 +167,8 @@ def _submit_item_to_thread(self, item: UploadQueueItem):
         future = self.executor.submit(_do_upload, item, self)
         self.futures.append(future)
 
-    def add_items(self, items: List[UploadQueueItem]):
-        """Add multiple chunks to the upload queue"""
+    def start(self, items: List[UploadQueueItem]):
+        """Starting upload in background with multiple upload items (UploadQueueItem)"""
         self.total_size = sum(item.size for item in items)
         self.upload_queue_items = items
 
@@ -182,7 +182,10 @@ def add_items(self, items: List[UploadQueueItem]):
             self._submit_item_to_thread(item)
 
     def update_chunks_from_items(self):
-        """Update chunks in LocalChanges from the upload queue items."""
+        """
+        Update chunks in LocalProjectChanges from the upload queue items.
+        Used just before finalizing the transaction to set the server_chunk_id in v2 API.
+        """
         self.changes.update_chunks([(item.chunk_id, item.server_chunk_id) for item in self.upload_queue_items])
 
 
@@ -195,7 +198,7 @@ def _do_upload(item: UploadQueueItem, job: UploadJob):
     job.transferred_size += item.size
 
 
-def create_upload_chunks(mc, mp: MerginProject, local_changes: List[LocalChange]) -> List[UploadQueueItem]:
+def create_upload_chunks(mc, mp: MerginProject, local_changes: List[FileChange]) -> List[UploadQueueItem]:
     """
     Create a list of UploadQueueItem objects from the changes dictionary and calculate total size of files.
     This is used to prepare the upload queue for the upload job.
@@ -228,7 +231,7 @@ def create_upload_chunks(mc, mp: MerginProject, local_changes: List[LocalChange]
 
 
 def create_upload_job(
-    mc, mp: MerginProject, changes: LocalChanges, tmp_dir: tempfile.TemporaryDirectory
+    mc, mp: MerginProject, changes: LocalPojectChanges, tmp_dir: tempfile.TemporaryDirectory
 ) -> Optional[UploadJob]:
     """
     Prepare transaction and create an upload job for the project using the v1 API.
@@ -268,18 +271,20 @@ def create_upload_job(
     if not upload_changes:
         mp.log.info("not uploading any files")
         if push_start_resp:
+            # This is related just to v1 API
             job.server_resp = push_start_resp
         push_project_finalize(job)
         return  # all done - no pending job
 
     if transaction_id:
+        # This is related just to v1 API
         mp.log.info(f"got transaction ID {transaction_id}")
 
     # prepare file chunks for upload
     upload_queue_items = create_upload_chunks(mc, mp, upload_changes)
 
     mp.log.info(f"Starting upload chunks for project {project_id}")
-    job.add_items(upload_queue_items)
+    job.start(upload_queue_items)
     return job
 
 
@@ -451,6 +456,7 @@ def push_project_cancel(job: UploadJob):
 
     job.executor.shutdown(wait=True)
     if not job.transaction_id:
+        # If not v2 api endpoint with transaction, nothing to cancel on server
         job.mp.log.info("--- push cancelled")
         return
     try:
@@ -473,7 +479,7 @@ def remove_diff_files(job: UploadJob) -> None:
                 os.remove(diff_file)
 
 
-def get_push_changes_batch(mc, mp: MerginProject) -> Tuple[LocalChanges, int]:
+def get_push_changes_batch(mc, mp: MerginProject) -> Tuple[LocalPojectChanges, int]:
     """
     Get changes that need to be pushed to the server.
     """
@@ -482,10 +488,10 @@ def get_push_changes_batch(mc, mp: MerginProject) -> Tuple[LocalChanges, int]:
     changes = filter_changes(mc, project_role, changes)
 
     try:
-        local_changes = LocalChanges(
-            added=[LocalChange(**change) for change in changes["added"]],
-            updated=[LocalChange(**change) for change in changes["updated"]],
-            removed=[LocalChange(**change) for change in changes["removed"]],
+        local_changes = LocalPojectChanges(
+            added=[FileChange(**change) for change in changes["added"]],
+            updated=[FileChange(**change) for change in changes["updated"]],
+            removed=[FileChange(**change) for change in changes["removed"]],
         )
     except ChangesValidationError as e:
         raise ClientError(
diff --git a/mergin/editor.py b/mergin/editor.py
@@ -64,7 +64,7 @@ def prevent_conflicted_copy(path: str, mc, project_role: str) -> bool:
     Args:
         path (str): The file path to check.
         mc: The Mergin client object.
-        project_info (dict): Information about the Mergin project from server.
+        project_role: Current project role.
 
     Returns:
         bool: True if the file path should be prevented from ceating conflicted copy, False otherwise.
diff --git a/mergin/local_changes.py b/mergin/local_changes.py
@@ -18,17 +18,28 @@ def __init__(self, message, invalid_changes=[], max_media_upload_size=None, max_
 
 
 @dataclass
-class BaseLocalChange:
+class FileDiffChange:
     path: str
     checksum: str
     size: int
     mtime: datetime
 
 
 @dataclass
-class LocalChange(BaseLocalChange):
+class FileChange:
+    # path to the file relative to the project root
+    path: str
+    # file checksum
+    checksum: str
+    # file size in bytes
+    size: int
+    # file modification time
+    mtime: datetime
+    # original file checksum used for compairison
     origin_checksum: Optional[str] = None
+    # list of chunk ids that make up this file
     chunks: List[str] = field(default_factory=list)
+    # optional diff information for geopackage files with geodiff metadata
     diff: Optional[dict] = None
     upload_file: Optional[str] = None
     # some functions (MerginProject.compare_file_sets) are adding version to the change from project info
@@ -38,9 +49,9 @@ class LocalChange(BaseLocalChange):
     # 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]:
+    def get_diff(self) -> Optional[FileDiffChange]:
         if self.diff:
-            return BaseLocalChange(
+            return FileDiffChange(
                 path=self.diff.get("path", ""),
                 checksum=self.diff.get("checksum", ""),
                 size=self.diff.get("size", 0),
@@ -64,10 +75,10 @@ def to_server_data(self) -> dict:
 
 
 @dataclass
-class LocalChanges:
-    added: List[LocalChange] = field(default_factory=list)
-    updated: List[LocalChange] = field(default_factory=list)
-    removed: List[LocalChange] = field(default_factory=list)
+class LocalPojectChanges:
+    added: List[FileChange] = field(default_factory=list)
+    updated: List[FileChange] = field(default_factory=list)
+    removed: List[FileChange] = field(default_factory=list)
 
     def __post_init__(self):
         """
@@ -101,7 +112,7 @@ def to_server_payload(self) -> dict:
             "removed": [change.to_server_data() for change in self.removed],
         }
 
-    def get_upload_changes(self) -> List[LocalChange]:
+    def get_upload_changes(self) -> List[FileChange]:
         """
         Get all changes that need to be uploaded.
         This includes added and updated files.
@@ -115,9 +126,7 @@ def _map_unique_chunks(self, change_chunks: List[str], server_chunks: List[Tuple
         mapped = []
         seen = set()
         for chunk in change_chunks:
-            for server_chunk in server_chunks:
-                chunk_id = server_chunk[0]
-                server_chunk_id = server_chunk[1]
+            for chunk_id, server_chunk_id in server_chunks:
                 if chunk_id == chunk and server_chunk_id not in seen:
                     mapped.append(server_chunk_id)
                     seen.add(server_chunk_id)
diff --git a/mergin/test/test_local_changes.py b/mergin/test/test_local_changes.py
