Added return types fo new functions

+ descriptions
+ tests for appl_pull_actions
+ restore get_pull_changes for backward compatibility with project status

Author: MarcelGeo

mergin/client.py

@@ -17,7 +17,14 @@
 import typing
 import warnings
 
-from mergin.models import ProjectDelta, ProjectDeltaItemDiff, ProjectDeltaItem
+from mergin.models import (
+    ProjectDelta,
+    ProjectDeltaItemDiff,
+    ProjectDeltaItem,
+    ProjectResponse,
+    ProjectFile,
+    ProjectWorkspace,
+)
 
 from .common import (
     ClientError,
@@ -722,7 +729,7 @@ def project_info(self, project_path_or_id, since=None, version=None):
             resp = self.get("/v1/project/{}".format(project_path_or_id), params)
         return json.load(resp)
 
-    def project_info_v2(self, project_id: str, files_at_version=None):
+    def project_info_v2(self, project_id: str, files_at_version=None) -> ProjectResponse:
         """
         Fetch info about project.
 
@@ -731,11 +738,37 @@ def project_info_v2(self, project_id: str, files_at_version=None):
         :param files_at_version: Version to track files at given version
         :type files_at_version: String
         """
+        self.check_v2_project_info_support()
+
         params = {}
         if files_at_version:
             params = {"files_at_version": files_at_version}
         resp = self.get(f"/v2/projects/{project_id}", params)
-        return json.load(resp)
+        resp_json = json.load(resp)
+        project_workspace = resp_json.get("workspace", {})
+        return ProjectResponse(
+            id=resp_json.get("id"),
+            name=resp_json.get("name"),
+            created_at=resp_json.get("created_at"),
+            updated_at=resp_json.get("updated_at"),
+            version=resp_json.get("version"),
+            public=resp_json.get("public"),
+            role=resp_json.get("role"),
+            size=resp_json.get("size"),
+            workspace=ProjectWorkspace(
+                id=project_workspace.get("id"),
+                name=project_workspace.get("name"),
+            ),
+            files=[
+                ProjectFile(
+                    checksum=f.get("checksum"),
+                    mtime=f.get("mtime"),
+                    path=f.get("path"),
+                    size=f.get("size"),
+                )
+                for f in resp_json.get("files", [])
+            ],
+        )
 
     def get_project_delta(self, project_id: str, since: str, to: typing.Optional[str] = None) -> ProjectDelta:
         """
@@ -749,6 +782,10 @@ def get_project_delta(self, project_id: str, since: str, to: typing.Optional[str
         :type since: String
         :rtype: Dict
         """
+        # If it is not enabled on the server, raise error
+        if not self.server_features().get("v2_pull_enabled", False):
+            raise ClientError("Project delta is not supported by the server")
+
         params = {"since": since}
         if to:
             params["to"] = to
@@ -1114,7 +1151,7 @@ def project_status(self, directory):
         mp = MerginProject(directory)
         server_info = self.project_info(mp.project_full_name(), since=mp.version())
 
-        pull_changes = mp.get_pull_delta(server_info["files"])
+        pull_changes = mp.get_pull_changes(server_info["files"])
 
         push_changes = mp.get_push_changes()
         push_changes_summary = mp.get_list_of_push_changes(push_changes)
@@ -1385,13 +1422,21 @@ def check_collaborators_members_support(self):
         if not is_version_acceptable(self.server_version(), f"{min_version}"):
             raise NotImplementedError(f"This needs server at version {min_version} or later")
 
+    def check_v2_project_info_support(self):
+        """
+        Check if the server is compatible with v2 endpoint for project info
+        """
+        min_version = "2025.8.2"
+        if not is_version_acceptable(self.server_version(), f"{min_version}"):
+            raise NotImplementedError(f"This needs server at version {min_version} or later")
+
     def create_user(
         self,
         email: str,
         password: str,
         workspace_id: int,
         workspace_role: WorkspaceRole,
-        username: str = None,
+        username: typing.Optional[str] = None,
         notify_user: bool = False,
     ) -> dict:
         """

mergin/client_pull.py

@@ -17,10 +17,10 @@
 import tempfile
 import typing
 import traceback
+from dataclasses import asdict
 
 import concurrent.futures
 
-from dataclasses import asdict
 
 from .common import CHUNK_SIZE, ClientError, DeltaChangeType, PullActionType
 from .models import ProjectDelta, ProjectDeltaItem, PullAction
@@ -135,7 +135,7 @@ def merge(self):
             raise ClientError("Download of file {} failed. Please try it again.".format(self.dest_file))
 
 
-def get_download_items(file_path: str, file_size: int, file_version: str, directory, diff_only=False):
+def get_download_items(file_path: str, file_size: int, file_version: str, directory: str, diff_only=False):
     """Returns an array of download queue items"""
 
     file_dir = os.path.dirname(os.path.normpath(os.path.join(directory, file_path)))
@@ -433,6 +433,7 @@ def __init__(
         )
         self.mc = mc
         self.futures = []  # list of concurrent.futures.Future instances
+        self.v2_pull = mc.server_features().get("v2_pull_enabled", False)
 
     def dump(self):
         print("--- JOB ---", self.total_size, "bytes")
@@ -447,7 +448,7 @@ def dump(self):
         print("--- END ---")
 
 
-def prepare_chunks_destination(target_dir: str, path: str) -> str:
+def prepare_file_destination(target_dir: str, path: str) -> str:
     """Prepares destination path for downloaded files chunks"""
 
     # figure out destination path for the file
@@ -458,46 +459,19 @@ def prepare_chunks_destination(target_dir: str, path: str) -> str:
     return dest_file_path
 
 
-_pulling_file_with_diffs = lambda f: "diffs" in f and len(f["diffs"]) != 0
-
-
 def get_diff_merge_files(delta_item: ProjectDeltaItem, target_dir: str) -> List[FileToMerge]:
     """
-    Extracts list of diff files to be downloaded delta item.
+    Extracts list of diff files to be downloaded from delta item using v1 endpoint.
     """
     result = []
 
     for diff in delta_item.diffs:
-        dest_file_path = prepare_chunks_destination(target_dir, diff.id)
+        dest_file_path = prepare_file_destination(target_dir, diff.id)
         download_items = get_download_items(diff.id, diff.size, diff.version, target_dir, True)
         result.append(FileToMerge(dest_file_path, download_items))
     return result
 
 
-def get_delta_download_files(
-    delta_items: List[ProjectDeltaItem], target_dir: str
-) -> Tuple[List[FileToMerge], List[DownloadDiffQueueItem]]:
-    """
-    Extracts list of files to be merged from delta dictionary. If there is any diff files to be downloaded, they
-    are returned as a separate list.
-    """
-    merge_files = []
-    diff_files = []
-    for item in delta_items:
-        change = item.change
-        if change == DeltaChangeType.CREATE or change == DeltaChangeType.UPDATE:
-            path = item.path
-            download_items = get_download_items(asdict(item), target_dir, False)
-            dest_file_path = prepare_chunks_destination(target_dir, path)
-            merge_files.append(FileToMerge(dest_file_path, download_items))
-        elif change == DeltaChangeType.UPDATE_DIFF:
-            diffs = item.diffs
-            for diff in diffs:
-                diff_path = diff.id
-                diff_files.append(DownloadDiffQueueItem(diff_path, os.path.join(target_dir, diff_path)))
-    return merge_files, diff_files
-
-
 def pull_project_async(mc, directory) -> Optional[PullJob]:
     """
     Starts project pull in background and returns handle to the pending job.
@@ -531,7 +505,7 @@ def pull_project_async(mc, directory) -> Optional[PullJob]:
         else:
             server_info = mc.project_info(project_path, since=local_version)
             server_version = server_info.get("version")
-            delta = ProjectDelta(to_version=server_version, items=mp.get_pull_delta(server_info))
+            delta = mp.get_pull_delta(server_info)
     except ClientError as err:
         mp.log.error("Error getting project info: " + str(err))
         mp.log.info("--- pull aborted")
@@ -562,13 +536,7 @@ def pull_project_async(mc, directory) -> Optional[PullJob]:
             continue  # no action needed
 
         pull_actions.append(PullAction(pull_action, item, local_item))
-        if pull_action == PullActionType.COPY:
-            # simply download the server version of the files
-            dest_file_path = prepare_chunks_destination(tmp_dir.name, item.path)
-            download_items = get_download_items(item.path, item.size, server_version, mp.cache_dir)
-            merge_files.append(FileToMerge(dest_file_path, download_items))
-
-        elif pull_action == PullActionType.APPLY_DIFF or (
+        if pull_action == PullActionType.APPLY_DIFF or (
             pull_action == PullActionType.COPY_CONFLICT and item.change == DeltaChangeType.UPDATE_DIFF
         ):
             # if we have diff to apply, let's download the diff files
@@ -594,12 +562,10 @@ def pull_project_async(mc, directory) -> Optional[PullJob]:
                 dest_file_path = mp.fpath(item.path, tmp_dir.name)
                 merge_files.append(FileToMerge(dest_file_path, items))
             basefiles_to_patch.append((item.path, [diff.id for diff in item.diffs]))
-
-        elif pull_action == PullActionType.COPY_CONFLICT:
-            # if we have conflict and create or update action, let's just download the server version of the file
-            # let's download server version of the file
-            dest_file_path = prepare_chunks_destination(tmp_dir.name, item.path)
-            download_items = get_download_items(item.path, item.size, server_version, mp.cache_dir)
+        elif pull_action == PullActionType.COPY or pull_action == PullActionType.COPY_CONFLICT:
+            # simply download the server version of the files
+            dest_file_path = prepare_file_destination(tmp_dir.name, item.path)
+            download_items = get_download_items(item.path, item.size, server_version, tmp_dir.name)
             merge_files.append(FileToMerge(dest_file_path, download_items))
         # Do nothing for DELETE actions
 
@@ -696,8 +662,11 @@ def pull_project_finalize(job: PullJob):
             raise future.exception()
 
     job.mp.log.info("finalizing pull")
-    if not job.project_info:
-        job.project_info = job.mc.project_info_v2(job.mp.project_id(), files_at_version=job.version)
+    if not job.project_info and job.v2_pull:
+        project_info_response = job.mc.project_info(job.project_path, version=job.version)
+        job.project_info = asdict(project_info_response)
+    else:
+        raise ClientError("Missing project info for pull finalization")
 
     # merge downloaded chunks
     try:
@@ -729,7 +698,7 @@ def pull_project_finalize(job: PullJob):
             job.mp.log.info("--- pull aborted")
             os.remove(basefile)
             raise ClientError("Cannot patch basefile {}! Please try syncing again.".format(basefile))
-
+    conflicts = []
     try:
         if job.pull_actions:
             conflicts = job.mp.apply_pull_actions(job.pull_actions, job.tmp_dir.name, job.project_info, job.mc)