Merge pull request #65 from opengisch/adding_push_delta_method_on_cli_sdk

Adding CLI and SDK for push delta files

Author: suricactus

qfieldcloud_sdk/cli.py

@@ -460,6 +460,22 @@ def job_status(ctx: Context, job_id):
         log(f'Job status for {job_id}: {status["status"]}')
 
 
+@cli.command(short_help="Push a delta file to a project.")
+@click.argument("project_id")
+@click.argument("delta_filename", type=click.Path(exists=True))
+@click.pass_context
+def delta_push(ctx: Context, project_id: str, delta_filename: str) -> None:
+    """Push a delta file to a project with PROJECT_ID."""
+    log(f'Pushing delta file "{delta_filename}" to project "{project_id}"…')
+
+    response = ctx.obj["client"].push_delta(project_id, delta_filename)
+
+    if ctx.obj["format_json"]:
+        print_json(response)
+    else:
+        log(f'Delta file "{delta_filename}" pushed to project "{project_id}".')
+
+
 @cli.command()
 @click.argument("project_id")
 @click.pass_context

qfieldcloud_sdk/sdk.py

@@ -183,6 +183,20 @@ def is_empty(self) -> bool:
         return self.limit is None and self.offset is None
 
 
+class DeltaPushResponse(TypedDict):
+    """Represents the structure of the response for pushing a delta file.
+
+    Attributes:
+        status: The status of the response.
+        message: A message providing additional information about the response.
+        details: Additional details about the delta push operation.
+    """
+
+    status: str
+    message: Optional[str]
+    details: Optional[Dict[str, Any]]
+
+
 class Client:
     """The core component of the QFieldCloud SDK, providing methods for interacting with the QFieldCloud platform.
 
@@ -617,6 +631,26 @@ def job_status(self, job_id: str) -> Dict[str, Any]:
 
         return resp.json()
 
+    def push_delta(self, project_id: str, delta_filename: str) -> DeltaPushResponse:
+        """Push a delta file to a project.
+
+        Args:
+            project_id: Project ID.
+            delta_filename: Path to the delta JSON file.
+
+        Returns:
+            A DeltaPushResponse containing the response from the server.
+        """
+        with open(delta_filename, "r") as delta_file:
+            files = {"file": delta_file}
+            response = self._request(
+                "POST",
+                f"deltas/{project_id}/",
+                files=files,
+            )
+
+        return cast(DeltaPushResponse, response)
+
     def delete_files(
         self,
         project_id: str,