Merge pull request #150 from pedrxd/task/mr-move-webdav

feat: Add WebDAV resource move/rename/copy functionality
2025-09-11 00:12:23 +02:00
parent b1eb4d2497 5d4902a73e
commit 18973e061a
4 changed files with 259 additions and 0 deletions
@@ -83,6 +83,8 @@ included? Feel free to open an issue, or contribute via a pull-request.
 | `nc_webdav_write_file` | Create or update files in NextCloud |
 | `nc_webdav_create_directory` | Create new directories |
 | `nc_webdav_delete_resource` | Delete files or directories |
+| `nc_webdav_move_resource` | Move or rename files and directories |
+| `nc_webdav_copy_resource` | Copy files and directories |

 ## Available Resources

@@ -121,6 +123,24 @@ await nc_webdav_write_file("NewProject/docs/notes.md", "# My Notes\n\nContent he

 # Delete a file or directory
 await nc_webdav_delete_resource("old_file.txt")
+
+# Move or rename a file
+await nc_webdav_move_resource("document.txt", "new_name.txt")
+
+# Move a file to another directory
+await nc_webdav_move_resource("document.txt", "Archive/document.txt")
+
+# Move a directory
+await nc_webdav_move_resource("Projects/OldProject", "Projects/NewProject")
+
+# Copy a file
+await nc_webdav_copy_resource("document.txt", "document_copy.txt")
+
+# Copy a file to another directory
+await nc_webdav_copy_resource("document.txt", "Backup/document.txt")
+
+# Copy a directory
+await nc_webdav_copy_resource("Projects/ProjectA", "Projects/ProjectA_Backup")
 ```

 ### Calendar Integration
@@ -415,3 +415,158 @@ class WebDAVClient(BaseNextcloudClient):
        except Exception as e:
            logger.error(f"Unexpected error creating directory '{path}': {e}")
            raise e
+
+    async def move_resource(
+        self, source_path: str, destination_path: str, overwrite: bool = False
+    ) -> Dict[str, Any]:
+        """Move or rename a resource (file or directory) via WebDAV MOVE.
+
+        Args:
+            source_path: The path of the file or directory to move
+            destination_path: The new path for the file or directory
+            overwrite: Whether to overwrite the destination if it exists
+
+        Returns:
+            Dict with status_code and optional message
+        """
+        source_webdav_path = f"{self._get_webdav_base_path()}/{source_path.lstrip('/')}"
+        destination_webdav_path = (
+            f"{self._get_webdav_base_path()}/{destination_path.lstrip('/')}"
+        )
+
+        # Ensure paths have consistent trailing slashes for directories
+        if source_path.endswith("/") and not destination_path.endswith("/"):
+            destination_webdav_path += "/"
+        elif not source_path.endswith("/") and destination_path.endswith("/"):
+            source_webdav_path += "/"
+
+        logger.debug(f"Moving resource from '{source_path}' to '{destination_path}'")
+
+        headers = {
+            "OCS-APIRequest": "true",
+            "Destination": destination_webdav_path,
+            "Overwrite": "T" if overwrite else "F",
+        }
+
+        try:
+            response = await self._make_request(
+                "MOVE", source_webdav_path, headers=headers
+            )
+            response.raise_for_status()
+
+            logger.debug(
+                f"Successfully moved resource from '{source_path}' to '{destination_path}'"
+            )
+            return {"status_code": response.status_code}
+
+        except HTTPStatusError as e:
+            if e.response.status_code == 404:
+                logger.debug(f"Source resource '{source_path}' not found")
+                return {"status_code": 404, "message": "Source resource not found"}
+            elif e.response.status_code == 412:
+                logger.debug(
+                    f"Destination '{destination_path}' already exists and overwrite is false"
+                )
+                return {
+                    "status_code": 412,
+                    "message": "Destination already exists and overwrite is false",
+                }
+            elif e.response.status_code == 409:
+                logger.debug(
+                    f"Parent directory of destination '{destination_path}' doesn't exist"
+                )
+                return {
+                    "status_code": 409,
+                    "message": "Parent directory of destination doesn't exist",
+                }
+                logger.debug(
+                    f"Parent directory of destination '{destination_path}' doesn't exist"
+                )
+                return {
+                    "status_code": 409,
+                    "message": "Parent directory of destination doesn't exist",
+                }
+            else:
+                logger.error(
+                    f"HTTP error moving resource from '{source_path}' to '{destination_path}': {e}"
+                )
+                raise e
+        except Exception as e:
+            logger.error(
+                f"Unexpected error moving resource from '{source_path}' to '{destination_path}': {e}"
+            )
+            raise e
+
+    async def copy_resource(
+        self, source_path: str, destination_path: str, overwrite: bool = False
+    ) -> Dict[str, Any]:
+        """Copy a resource (file or directory) via WebDAV COPY.
+
+        Args:
+            source_path: The path of the file or directory to copy
+            destination_path: The destination path for the copy
+            overwrite: Whether to overwrite the destination if it exists
+
+        Returns:
+            Dict with status_code and optional message
+        """
+        source_webdav_path = f"{self._get_webdav_base_path()}/{source_path.lstrip('/')}"
+        destination_webdav_path = (
+            f"{self._get_webdav_base_path()}/{destination_path.lstrip('/')}"
+        )
+
+        # Ensure paths have consistent trailing slashes for directories
+        if source_path.endswith("/") and not destination_path.endswith("/"):
+            destination_webdav_path += "/"
+        elif not source_path.endswith("/") and destination_path.endswith("/"):
+            source_webdav_path += "/"
+
+        logger.debug(f"Copying resource from '{source_path}' to '{destination_path}'")
+
+        headers = {
+            "OCS-APIRequest": "true",
+            "Destination": destination_webdav_path,
+            "Overwrite": "T" if overwrite else "F",
+        }
+
+        try:
+            response = await self._make_request(
+                "COPY", source_webdav_path, headers=headers
+            )
+            response.raise_for_status()
+
+            logger.debug(
+                f"Successfully copied resource from '{source_path}' to '{destination_path}'"
+            )
+            return {"status_code": response.status_code}
+
+        except HTTPStatusError as e:
+            if e.response.status_code == 404:
+                logger.debug(f"Source resource '{source_path}' not found")
+                return {"status_code": 404, "message": "Source resource not found"}
+            elif e.response.status_code == 412:
+                logger.debug(
+                    f"Destination '{destination_path}' already exists and overwrite is false"
+                )
+                return {
+                    "status_code": 412,
+                    "message": "Destination already exists and overwrite is false",
+                }
+            elif e.response.status_code == 409:
+                logger.debug(
+                    f"Parent directory of destination '{destination_path}' doesn't exist"
+                )
+                return {
+                    "status_code": 409,
+                    "message": "Parent directory of destination doesn't exist",
+                }
+            else:
+                logger.error(
+                    f"HTTP error copying resource from '{source_path}' to '{destination_path}': {e}"
+                )
+                raise e
+        except Exception as e:
+            logger.error(
+                f"Unexpected error copying resource from '{source_path}' to '{destination_path}': {e}"
+            )
+            raise e
@@ -86,3 +86,23 @@ class DeleteResourceResponse(StatusResponse):
    items_deleted: Optional[int] = Field(
        None, description="Number of items deleted (for directories)"
    )
+
+
+class MoveResourceResponse(StatusResponse):
+    """Response model for resource move/rename operations."""
+
+    source_path: str = Field(description="Original path of the resource")
+    destination_path: str = Field(description="New path of the resource")
+    overwrite: bool = Field(
+        description="Whether the destination was overwritten if it existed"
+    )
+
+
+class CopyResourceResponse(StatusResponse):
+    """Response model for resource copy operations."""
+
+    source_path: str = Field(description="Original path of the resource")
+    destination_path: str = Field(description="Destination path for the copy")
+    overwrite: bool = Field(
+        description="Whether the destination was overwritten if it existed"
+    )
@@ -149,3 +149,67 @@ def configure_webdav_tools(mcp: FastMCP):
        """
        client: NextcloudClient = ctx.request_context.lifespan_context.client
        return await client.webdav.delete_resource(path)
+
+    @mcp.tool()
+    async def nc_webdav_move_resource(
+        source_path: str, destination_path: str, ctx: Context, overwrite: bool = False
+    ):
+        """Move or rename a file or directory in NextCloud.
+
+        Args:
+            source_path: Full path of the file or directory to move
+            destination_path: New path for the file or directory
+            overwrite: Whether to overwrite the destination if it exists (default: False)
+
+        Returns:
+            Dict with status_code indicating result (404 if source not found, 412 if destination exists and overwrite is False)
+
+        Examples:
+            # Rename a file
+            await nc_webdav_move_resource("document.txt", "new_name.txt")
+
+            # Move a file to another directory
+            await nc_webdav_move_resource("document.txt", "Archive/document.txt")
+
+            # Move a directory
+            await nc_webdav_move_resource("Projects/OldProject", "Projects/NewProject")
+
+            # Move and overwrite if destination exists
+            await nc_webdav_move_resource("document.txt", "Archive/document.txt", overwrite=True)
+        """
+        client: NextcloudClient = ctx.request_context.lifespan_context.client
+        return await client.webdav.move_resource(
+            source_path, destination_path, overwrite
+        )
+
+    @mcp.tool()
+    async def nc_webdav_copy_resource(
+        source_path: str, destination_path: str, ctx: Context, overwrite: bool = False
+    ):
+        """Copy a file or directory in NextCloud.
+
+        Args:
+            source_path: Full path of the file or directory to copy
+            destination_path: Destination path for the copy
+            overwrite: Whether to overwrite the destination if it exists (default: False)
+
+        Returns:
+            Dict with status_code indicating result (404 if source not found, 412 if destination exists and overwrite is False)
+
+        Examples:
+            # Copy a file
+            await nc_webdav_copy_resource("document.txt", "document_copy.txt")
+
+            # Copy a file to another directory
+            await nc_webdav_copy_resource("document.txt", "Backup/document.txt")
+
+            # Copy a directory
+            await nc_webdav_copy_resource("Projects/ProjectA", "Projects/ProjectA_Backup")
+
+            # Copy and overwrite if destination exists
+            await nc_webdav_copy_resource("document.txt", "Backup/document.txt", overwrite=True)
+        """
+        client: NextcloudClient = ctx.request_context.lifespan_context.client
+        return await client.webdav.copy_resource(
+            source_path, destination_path, overwrite
+        )