Change WorkspaceEdit to use documentChanges

pylsp_rope/plugin.py

@@ -140,7 +140,11 @@ def pylsp_execute_command(config, workspace, command, arguments):
     commands = {cmd.name: cmd for cmd in refactoring.Command.__subclasses__()}
 
     try:
-        return commands[command](workspace, **arguments[0])()
+        return commands[command](workspace, **arguments[0])(
+            # FIXME: Hardcode executeCommand to use WorkspaceEditWithChanges
+            #        We need to upgrade this at some point.
+            workspace_edit_format=["changes"],
+        )
     except Exception as exc:
         logger.exception(
             "Exception when doing workspace/executeCommand: %s",

pylsp_rope/project.py

@@ -1,14 +1,22 @@
 import logging
 from functools import lru_cache
-from typing import List, Dict, Tuple
+from typing import List, Dict, Tuple, Optional, Literal, cast
 
 from pylsp import uris, workspace
 from rope.base import libutils
 from rope.base.fscommands import FileSystemCommands
 
 from pylsp_rope import rope
 from pylsp_rope.lsp_diff import lsp_diff
-from pylsp_rope.typing import WorkspaceEdit, DocumentUri, TextEdit, Line
+from pylsp_rope.typing import (
+    WorkspaceEditWithChanges,
+    WorkspaceEditWithDocumentChanges,
+    WorkspaceEdit,
+    DocumentUri,
+    TextEdit,
+    Line,
+    TextDocumentEdit,
+)
 
 
 logger = logging.getLogger(__name__)
@@ -38,36 +46,92 @@ def get_document(workspace, resource: rope.Resource) -> workspace.Document:
     return workspace.get_document(uris.from_fs_path(resource.real_path))
 
 
-def rope_changeset_to_workspace_edit(
-    workspace, rope_changeset: rope.ChangeSet
-) -> WorkspaceEdit:
-    def _get_contents(change: rope.Change) -> Tuple[List[Line], List[Line]]:
-        old = change.old_contents
-        new = change.new_contents
-        if old is None:
-            if change.resource.exists():
-                old = change.resource.read()
-            else:
-                old = ""
-        return old.splitlines(keepends=True), new.splitlines(keepends=True)
+def _get_contents(change: rope.Change) -> Tuple[List[Line], List[Line]]:
+    old = change.old_contents
+    new = change.new_contents
+    if old is None:
+        if change.resource.exists():
+            old = change.resource.read()
+        else:
+            old = ""
+    return old.splitlines(keepends=True), new.splitlines(keepends=True)
+
 
+def convert_workspace_edit_document_changes_to_changes(
+    workspace_edit: WorkspaceEditWithDocumentChanges,
+) -> WorkspaceEditWithChanges:
     workspace_changeset: Dict[DocumentUri, List[TextEdit]] = {}
+    for change in workspace_edit["documentChanges"] or []:
+        document_changes = workspace_changeset.setdefault(
+            change["textDocument"]["uri"],
+            [],
+        )
+        document_changes.extend(change["edits"])
+
+    return {
+        "changes": workspace_changeset,
+    }
+
+
+def _rope_changeset_to_workspace_edit(
+    workspace, rope_changeset: rope.ChangeSet
+) -> WorkspaceEditWithDocumentChanges:
+    workspace_changeset: List[TextDocumentEdit] = []
     for change in rope_changeset.changes:
         lines_old, lines_new = _get_contents(change)
 
         document = get_document(workspace, change.resource)
-        document_changes = workspace_changeset.setdefault(document.uri, [])
-        document_changes.extend(lsp_diff(lines_old, lines_new))
+        workspace_changeset.append(
+            {
+                "textDocument": {
+                    "uri": document.uri,
+                    "version": document.version,
+                },
+                "edits": list(lsp_diff(lines_old, lines_new)),
+            }
+        )
 
     return {
-        "changes": workspace_changeset,
+        "documentChanges": workspace_changeset,
     }
 
 
-def apply_rope_changeset(workspace, rope_changeset: rope.ChangeSet) -> None:
+WorkspaceEditFormat = Literal["changes", "documentChanges"]
+DEFAULT_WORKSPACE_EDIT_FORMAT: List[WorkspaceEditFormat] = ["changes"]
+
+
+def rope_changeset_to_workspace_edit(
+    workspace,
+    rope_changeset: rope.ChangeSet,
+    workspace_edit_format: List[WorkspaceEditFormat] = DEFAULT_WORKSPACE_EDIT_FORMAT,
+) -> WorkspaceEdit:
+    assert len(workspace_edit_format) > 0
+    documentChanges: WorkspaceEditWithDocumentChanges = (
+        _rope_changeset_to_workspace_edit(
+            workspace,
+            rope_changeset,
+        )
+    )
+    workspace_edit: dict = {}
+    if "changes" in workspace_edit_format:
+        changes: WorkspaceEditWithChanges = (
+            convert_workspace_edit_document_changes_to_changes(documentChanges)
+        )
+        workspace_edit.update(changes)
+    if "documentChanges" in workspace_edit_format:
+        workspace_edit.update(documentChanges)
+    return cast(WorkspaceEdit, workspace_edit)
+
+
+def apply_rope_changeset(
+    workspace,
+    rope_changeset: rope.ChangeSet,
+    workspace_edit_format: List[WorkspaceEditFormat] = DEFAULT_WORKSPACE_EDIT_FORMAT,
+) -> None:
     workspace_edit = rope_changeset_to_workspace_edit(
         workspace,
         rope_changeset,
+        workspace_edit_format=workspace_edit_format,
     )
 
     logger.info("applying workspace edit: %s", workspace_edit)

pylsp_rope/refactoring.py

@@ -14,10 +14,12 @@
 
 from pylsp_rope import typing, commands
 from pylsp_rope.project import (
+    WorkspaceEditFormat,
     get_project,
     get_resource,
     get_resources,
     apply_rope_changeset,
+    DEFAULT_WORKSPACE_EDIT_FORMAT,
 )
 from pylsp_rope.typing import DocumentUri, CodeActionKind
 
@@ -32,10 +34,20 @@ def __init__(self, workspace, **arguments):
         self.arguments = arguments
         self.__dict__.update(**arguments)
 
-    def __call__(self):
+    def __call__(
+        self,
+        *,
+        workspace_edit_format: List[
+            WorkspaceEditFormat
+        ] = DEFAULT_WORKSPACE_EDIT_FORMAT,
+    ):
         rope_changeset = self.get_changes()
         if rope_changeset is not None:
-            apply_rope_changeset(self.workspace, rope_changeset)
+            apply_rope_changeset(
+                self.workspace,
+                rope_changeset,
+                workspace_edit_format,
+            )
 
     def get_changes(self):
         """
@@ -250,7 +262,11 @@ def get_changes(self):
             resource=resource,
             offset=current_document.offset_at_position(self.position),
         )
-        resources = get_resources(self.workspace, self.documents) if self.documents is not None else None
+        resources = (
+            get_resources(self.workspace, self.documents)
+            if self.documents is not None
+            else None
+        )
         rope_changeset = refactoring.get_changes(
             resources=resources,
         )

pylsp_rope/text.py

@@ -23,33 +23,29 @@ def Position(
     line: Tuple[AutoLineNumber, Optional[_CharNumberOrMarker]],
     *,
     _default_character: _CharNumberOrMarker = CharNumber(0),
-) -> typing.Position:
-    ...
+) -> typing.Position: ...
 
 
 @overload
 def Position(
     line: AutoLineNumber,
     *,
     _default_character: _CharNumberOrMarker = CharNumber(0),
-) -> typing.Position:
-    ...
+) -> typing.Position: ...
 
 
 @overload
 def Position(
     line: AutoLineNumber,
     character: AutoCharNumber,
-) -> typing.Position:
-    ...
+) -> typing.Position: ...
 
 
 @overload
 def Position(
     line: AutoLineNumber,
     character: Literal["^", "$"],
-) -> typing.Position:
-    ...
+) -> typing.Position: ...
 
 
 def Position(

pylsp_rope/typing.py

@@ -1,5 +1,9 @@
 import sys
-from typing import List, Dict, Optional, NewType, Any
+from typing import List, Dict, Optional, NewType, Any, Union
+try:
+    from typing import TypeGuard
+except ImportError:
+    from typing_extensions import TypeGuard
 
 
 if sys.version_info >= (3, 8):
@@ -25,17 +29,54 @@ class Range(TypedDict):
     end: Position
 
 
+class TextDocumentIdentifier(TypedDict):
+    uri: DocumentUri
+
+
+class OptionalVersionedTextDocumentIdentifier(TextDocumentIdentifier):
+    version: Optional[int]
+
+
 class TextEdit(TypedDict):
     range: Range
     newText: str
 
 
-class WorkspaceEdit(TypedDict):
-    changes: Optional[Dict[DocumentUri, List[TextEdit]]]
-    # documentChanges: ...
+class TextDocumentEdit(TypedDict):
+    textDocument: OptionalVersionedTextDocumentIdentifier
+
+    edits: List[TextEdit]  # FIXME: should be: list[TextEdit| AnnotatedTextEdit]
+
+
+class WorkspaceEditWithChanges(TypedDict):
+    changes: Dict[DocumentUri, List[TextEdit]]
+    # documentChanges: Optional[list[TextDocumentEdit]]  # FIXME: should be: (TextDocumentEdit | CreateFile | RenameFile | DeleteFile)[]
+    # changeAnnotations: ...
+
+
+class WorkspaceEditWithDocumentChanges(TypedDict):
+    # changes: Optional[Dict[DocumentUri, List[TextEdit]]]
+    documentChanges: List[
+        TextDocumentEdit
+    ]  # FIXME: should be: (TextDocumentEdit | CreateFile | RenameFile | DeleteFile)[]
     # changeAnnotations: ...
 
 
+WorkspaceEdit = Union[WorkspaceEditWithChanges, WorkspaceEditWithDocumentChanges]
+
+
+def is_workspace_edit_with_changes(
+    workspace_edit: WorkspaceEdit,
+) -> TypeGuard[WorkspaceEditWithChanges]:
+    return "changes" in workspace_edit
+
+
+def is_workspace_edit_with_document_changes(
+    workspace_edit: WorkspaceEdit,
+) -> TypeGuard[WorkspaceEditWithDocumentChanges]:
+    return "documentChanges" in workspace_edit
+
+
 class ApplyWorkspaceEditParams(TypedDict):
     label: Optional[str]
     edit: WorkspaceEdit
@@ -79,9 +120,3 @@ class CodeAction(TypedDict):
 Line = NewType("Line", str)
 LineNumber = NewType("LineNumber", int)
 CharNumber = NewType("CharNumber", int)
-
-
-class SimpleWorkspaceEdit(TypedDict):
-    """This is identical to WorkspaceEdit, but `changes` field is not optional."""
-
-    changes: Dict[DocumentUri, List[TextEdit]]

setup.cfg

@@ -25,7 +25,7 @@ packages = find:
 install_requires = 
     python-lsp-server
     rope>=0.21.0
-    typing-extensions; python_version < '3.8'
+    typing-extensions; python_version < '3.10'
 
 python_requires = >= 3.6
 

test/helpers.py

@@ -11,7 +11,7 @@
     CodeAction,
     DocumentContent,
     DocumentUri,
-    SimpleWorkspaceEdit,
+    WorkspaceEditWithChanges,
     TextEdit,
 )
 from test.conftest import read_fixture_file
@@ -49,7 +49,7 @@ def assert_single_document_edit(
     return document_edits
 
 
-def assert_is_apply_edit_request(edit_request: Any) -> SimpleWorkspaceEdit:
+def assert_is_apply_edit_request(edit_request: Any) -> WorkspaceEditWithChanges:
     assert edit_request == call(
         "workspace/applyEdit",
         {
@@ -59,7 +59,7 @@ def assert_is_apply_edit_request(edit_request: Any) -> SimpleWorkspaceEdit:
         },
     )
 
-    workspace_edit: SimpleWorkspaceEdit = edit_request[0][1]["edit"]
+    workspace_edit: WorkspaceEditWithChanges = edit_request[0][1]["edit"]
     for document_uri, document_edits in workspace_edit["changes"].items():
         assert is_document_uri(document_uri)
         for change in document_edits:
@@ -79,14 +79,14 @@ def is_document_uri(uri: DocumentUri) -> bool:
 
 
 def assert_modified_documents(
-    workspace_edit: SimpleWorkspaceEdit,
+    workspace_edit: WorkspaceEditWithChanges,
     document_uris: Collection[DocumentUri],
 ) -> None:
     assert workspace_edit["changes"].keys() == set(document_uris)
 
 
 def assert_unmodified_document(
-    workspace_edit: SimpleWorkspaceEdit,
+    workspace_edit: WorkspaceEditWithChanges,
     document_uri: DocumentUri,
 ) -> None:
     assert is_document_uri(document_uri)