Ignore current environment site-packages & republish diagnostics after changing configuration (#32)

vrslev · web-flow · commit 902f83633ac4 · 2024-07-08T19:45:29.000+03:00
diff --git a/.vscode/tasks.json b/.vscode/tasks.json
@@ -4,7 +4,10 @@
 		{
 			"type": "npm",
 			"script": "watch",
-			"problemMatcher": "$tsc-watch",
+			"problemMatcher": [
+				"$ts-webpack-watch",
+				"$tslint-webpack-watch"
+			],
 			"isBackground": true,
 			"presentation": {
 				"reveal": "never"
diff --git a/auto_typing_final/lsp.py b/auto_typing_final/lsp.py
@@ -1,14 +1,17 @@
+import os
+import sys
 import uuid
-from collections.abc import Iterable
+from dataclasses import dataclass
 from importlib.metadata import version
-from typing import Final, cast, get_args
+from pathlib import Path
+from typing import Any, Final, TypedDict, cast
+from urllib.parse import unquote_to_bytes
 
 import attr
 import cattrs
 import lsprotocol.types as lsp
 from ast_grep_py import SgRoot
-from pygls import server
-from pygls.workspace import TextDocument
+from pygls.server import LanguageServer
 
 from auto_typing_final.transform import (
     IMPORT_STYLES_TO_IMPORT_CONFIGS,
@@ -19,19 +22,39 @@
     make_replacements,
 )
 
-LSP_SERVER = server.LanguageServer(name="auto-typing-final", version=version("auto-typing-final"), max_workers=5)
-IMPORT_CONFIG: ImportConfig | None = None
 
+# From Python 3.13: https://github.com/python/cpython/blob/0790418a0406cc5419bfd9d718522a749542bbc8/Lib/pathlib/_local.py#L815
+def path_from_uri(uri: str) -> Path | None:
+    if not uri.startswith("file:"):
+        return None
+    path = uri[5:]
+    if path[:3] == "///":
+        # Remove empty authority
+        path = path[2:]
+    elif path[:12] == "//localhost/":
+        # Remove 'localhost' authority
+        path = path[11:]
+    if path[:3] == "///" or (path[:1] == "/" and path[2:3] in ":|"):
+        # Remove slash before DOS device/UNC path
+        path = path[1:]
+    if path[1:2] == "|":
+        # Replace bar with colon in DOS drive
+        path = path[:1] + ":" + path[2:]
+
+    path_: Final = Path(os.fsdecode(unquote_to_bytes(path)))
+    if not path_.is_absolute():
+        return None
+    return path_
 
-@attr.define
-class Fix:
-    message: str
-    text_edits: list[lsp.TextEdit]
+
+ClientSettings = TypedDict("ClientSettings", {"import-style": ImportStyle})
+FullClientSettings = TypedDict("FullClientSettings", {"auto-typing-final": ClientSettings})
 
 
 @attr.define
-class DiagnosticData:
-    fix: Fix
+class Fix:
+    message: str
+    text_edits: list[lsp.TextEdit | lsp.AnnotatedTextEdit]
 
 
 def make_import_text_edit(import_text: str) -> lsp.TextEdit:
@@ -52,108 +75,127 @@ def make_text_edit(edit: Edit) -> lsp.TextEdit:
     )
 
 
-def make_diagnostics(source: str) -> Iterable[lsp.Diagnostic]:
-    if not IMPORT_CONFIG:
-        return
-    result: Final = make_replacements(root=SgRoot(source, "python").root(), import_config=IMPORT_CONFIG)
-
-    for replacement in result.replacements:
-        if replacement.operation_type == AddFinal:
-            fix_message = f"{LSP_SERVER.name}: Add {IMPORT_CONFIG.value}"
-            diagnostic_message = f"Missing {IMPORT_CONFIG.value}"
-        else:
-            fix_message = f"{LSP_SERVER.name}: Remove {IMPORT_CONFIG.value}"
-            diagnostic_message = f"Unexpected {IMPORT_CONFIG.value}"
-
-        fix = Fix(message=fix_message, text_edits=[make_text_edit(edit) for edit in replacement.edits])
-        if result.import_text:
-            fix.text_edits.append(make_import_text_edit(result.import_text))
-
-        for applied_edit in replacement.edits:
-            node_range = applied_edit.node.range()
-            yield lsp.Diagnostic(
-                range=lsp.Range(
-                    start=lsp.Position(line=node_range.start.line, character=node_range.start.column),
-                    end=lsp.Position(line=node_range.end.line, character=node_range.end.column),
-                ),
-                message=diagnostic_message,
-                severity=lsp.DiagnosticSeverity.Warning,
-                source=LSP_SERVER.name,
-                data=cattrs.unstructure(DiagnosticData(fix=fix)),
-            )
-
+@dataclass(init=False)
+class Service:
+    ls_name: str
+    ignored_paths: list[Path]
+    import_config: ImportConfig
 
-def make_fixall_text_edits(source: str) -> Iterable[lsp.TextEdit]:
-    if not IMPORT_CONFIG:
-        return
-    result: Final = make_replacements(root=SgRoot(source, "python").root(), import_config=IMPORT_CONFIG)
+    def __init__(self, ls_name: str, settings: Any) -> None:  # noqa: ANN401
+        self.ls_name = ls_name
 
-    for replacement in result.replacements:
-        for edit in replacement.edits:
-            yield make_text_edit(edit)
+        executable_path: Final = Path(sys.executable)
+        if executable_path.parent.name == "bin":
+            self.ignored_paths = [executable_path.parent.parent]
 
-    if result.import_text:
-        yield make_import_text_edit(result.import_text)
+        try:
+            validated_settings: Final = cattrs.structure(settings, FullClientSettings)
+        except cattrs.BaseValidationError:
+            return
+        self.import_config = IMPORT_STYLES_TO_IMPORT_CONFIGS[validated_settings["auto-typing-final"]["import-style"]]
 
+    def make_diagnostics(self, source: str) -> list[lsp.Diagnostic]:
+        replacement_result: Final = make_replacements(
+            root=SgRoot(source, "python").root(), import_config=self.import_config
+        )
+        result: Final = []
+
+        for replacement in replacement_result.replacements:
+            if replacement.operation_type == AddFinal:
+                fix_message = f"{self.ls_name}: Add {self.import_config.value}"
+                diagnostic_message = f"Missing {self.import_config.value}"
+            else:
+                fix_message = f"{self.ls_name}: Remove {self.import_config.value}"
+                diagnostic_message = f"Unexpected {self.import_config.value}"
+
+            fix = Fix(message=fix_message, text_edits=[make_text_edit(edit) for edit in replacement.edits])
+            if replacement_result.import_text:
+                fix.text_edits.append(make_import_text_edit(replacement_result.import_text))
+
+            for applied_edit in replacement.edits:
+                node_range = applied_edit.node.range()
+                result.append(
+                    lsp.Diagnostic(
+                        range=lsp.Range(
+                            start=lsp.Position(line=node_range.start.line, character=node_range.start.column),
+                            end=lsp.Position(line=node_range.end.line, character=node_range.end.column),
+                        ),
+                        message=diagnostic_message,
+                        severity=lsp.DiagnosticSeverity.Warning,
+                        source=self.ls_name,
+                        data=cattrs.unstructure(fix),
+                    )
+                )
+        return result
 
-def make_workspace_edit(text_document: TextDocument, text_edits: list[lsp.TextEdit]) -> lsp.WorkspaceEdit:
-    return lsp.WorkspaceEdit(
-        document_changes=[
-            lsp.TextDocumentEdit(
-                text_document=lsp.OptionalVersionedTextDocumentIdentifier(
-                    uri=text_document.uri, version=text_document.version
-                ),
-                edits=text_edits,  # type: ignore[arg-type]
-            )
+    def make_fix_all_text_edits(self, source: str) -> list[lsp.TextEdit | lsp.AnnotatedTextEdit]:
+        replacement_result: Final = make_replacements(
+            root=SgRoot(source, "python").root(), import_config=self.import_config
+        )
+        result: Final[list[lsp.TextEdit | lsp.AnnotatedTextEdit]] = [
+            make_text_edit(edit) for replacement in replacement_result.replacements for edit in replacement.edits
         ]
-    )
+        if replacement_result.import_text:
+            result.append(make_import_text_edit(replacement_result.import_text))
+        return result
+
+    def path_is_ignored(self, uri: str) -> bool:
+        if path := path_from_uri(uri):
+            return any(path.is_relative_to(ignored_path) for ignored_path in self.ignored_paths)
+        return False
+
+
+class CustomLanguageServer(LanguageServer):
+    service: Service | None = None
+
+
+LSP_SERVER = CustomLanguageServer(name="auto-typing-final", version=version("auto-typing-final"), max_workers=5)
 
 
 @LSP_SERVER.feature(lsp.INITIALIZE)
-async def initialize(_: lsp.InitializeParams) -> None: ...
+def initialize(_: lsp.InitializeParams) -> None: ...
 
 
 @LSP_SERVER.feature(lsp.INITIALIZED)
-async def initialized(_: lsp.InitializedParams) -> None:
-    await LSP_SERVER.register_capability_async(
+async def initialized(ls: CustomLanguageServer, _: lsp.InitializedParams) -> None:
+    await ls.register_capability_async(
         params=lsp.RegistrationParams(
             registrations=[
                 lsp.Registration(
                     id=str(uuid.uuid4()),
                     method=lsp.WORKSPACE_DID_CHANGE_CONFIGURATION,
-                    register_options=lsp.DidChangeConfigurationRegistrationOptions(section=LSP_SERVER.name),
-                )
+                    register_options=lsp.DidChangeConfigurationRegistrationOptions(section=ls.name),
+                ),
             ]
         )
     )
 
 
 @LSP_SERVER.feature(lsp.WORKSPACE_DID_CHANGE_CONFIGURATION)
-def workspace_did_change_configuration(params: lsp.DidChangeConfigurationParams) -> None:
-    if (
-        isinstance(params.settings, dict)
-        and (settings := params.settings.get(LSP_SERVER.name))
-        and isinstance(settings, dict)
-        and (import_style := settings.get("import-style"))
-        and (import_style in get_args(ImportStyle))
-    ):
-        global IMPORT_CONFIG  # noqa: PLW0603
-        IMPORT_CONFIG = IMPORT_STYLES_TO_IMPORT_CONFIGS[import_style]
+def workspace_did_change_configuration(ls: CustomLanguageServer, params: lsp.DidChangeConfigurationParams) -> None:
+    ls.service = Service(ls_name=ls.name, settings=params.settings)
+    for text_document in ls.workspace.text_documents.values():
+        ls.publish_diagnostics(text_document.uri, diagnostics=ls.service.make_diagnostics(text_document.source))
 
 
 @LSP_SERVER.feature(lsp.TEXT_DOCUMENT_DID_OPEN)
 @LSP_SERVER.feature(lsp.TEXT_DOCUMENT_DID_SAVE)
 @LSP_SERVER.feature(lsp.TEXT_DOCUMENT_DID_CHANGE)
 def did_open_did_save_did_change(
+    ls: CustomLanguageServer,
     params: lsp.DidOpenTextDocumentParams | lsp.DidSaveTextDocumentParams | lsp.DidChangeTextDocumentParams,
 ) -> None:
-    text_document: Final = LSP_SERVER.workspace.get_text_document(params.text_document.uri)
-    LSP_SERVER.publish_diagnostics(text_document.uri, diagnostics=list(make_diagnostics(text_document.source)))
+    if not ls.service:
+        return
+    if ls.service.path_is_ignored(params.text_document.uri):
+        return
+    text_document: Final = ls.workspace.get_text_document(params.text_document.uri)
+    ls.publish_diagnostics(text_document.uri, diagnostics=ls.service.make_diagnostics(text_document.source))
 
 
 @LSP_SERVER.feature(lsp.TEXT_DOCUMENT_DID_CLOSE)
-def did_close(params: lsp.DidCloseTextDocumentParams) -> None:
-    LSP_SERVER.publish_diagnostics(params.text_document.uri, [])
+def did_close(ls: CustomLanguageServer, params: lsp.DidCloseTextDocumentParams) -> None:
+    ls.publish_diagnostics(params.text_document.uri, [])
 
 
 @LSP_SERVER.feature(
@@ -162,31 +204,40 @@ def did_close(params: lsp.DidCloseTextDocumentParams) -> None:
         code_action_kinds=[lsp.CodeActionKind.QuickFix, lsp.CodeActionKind.SourceFixAll], resolve_provider=True
     ),
 )
-def code_action(params: lsp.CodeActionParams) -> list[lsp.CodeAction] | None:
+def code_action(ls: CustomLanguageServer, params: lsp.CodeActionParams) -> list[lsp.CodeAction] | None:
     requested_kinds: Final = params.context.only or {lsp.CodeActionKind.QuickFix, lsp.CodeActionKind.SourceFixAll}
     actions: Final[list[lsp.CodeAction]] = []
 
     if lsp.CodeActionKind.QuickFix in requested_kinds:
-        text_document: Final = LSP_SERVER.workspace.get_text_document(params.text_document.uri)
+        text_document: Final = ls.workspace.get_text_document(params.text_document.uri)
         our_diagnostics: Final = [
-            diagnostic for diagnostic in params.context.diagnostics if diagnostic.source == LSP_SERVER.name
+            diagnostic for diagnostic in params.context.diagnostics if diagnostic.source == ls.name
         ]
 
         for diagnostic in our_diagnostics:
-            data = cattrs.structure(diagnostic.data, DiagnosticData)
+            fix = cattrs.structure(diagnostic.data, Fix)
             actions.append(
                 lsp.CodeAction(
-                    title=data.fix.message,
+                    title=fix.message,
                     kind=lsp.CodeActionKind.QuickFix,
-                    edit=make_workspace_edit(text_document=text_document, text_edits=data.fix.text_edits),
+                    edit=lsp.WorkspaceEdit(
+                        document_changes=[
+                            lsp.TextDocumentEdit(
+                                text_document=lsp.OptionalVersionedTextDocumentIdentifier(
+                                    uri=text_document.uri, version=text_document.version
+                                ),
+                                edits=fix.text_edits,
+                            )
+                        ]
+                    ),
                     diagnostics=[diagnostic],
                 )
             )
 
         if our_diagnostics:
             actions.append(
                 lsp.CodeAction(
-                    title=f"{LSP_SERVER.name}: Fix All",
+                    title=f"{ls.name}: Fix All",
                     kind=lsp.CodeActionKind.QuickFix,
                     data=params.text_document.uri,
                     edit=None,
@@ -197,10 +248,9 @@ def code_action(params: lsp.CodeActionParams) -> list[lsp.CodeAction] | None:
     if lsp.CodeActionKind.SourceFixAll in requested_kinds:
         actions.append(
             lsp.CodeAction(
-                title=f"{LSP_SERVER.name}: Fix All",
+                title=f"{ls.name}: Fix All",
                 kind=lsp.CodeActionKind.SourceFixAll,
                 data=params.text_document.uri,
-                edit=None,
                 diagnostics=params.context.diagnostics,
             ),
         )
@@ -209,18 +259,19 @@ def code_action(params: lsp.CodeActionParams) -> list[lsp.CodeAction] | None:
 
 
 @LSP_SERVER.feature(lsp.CODE_ACTION_RESOLVE)
-def resolve_code_action(params: lsp.CodeAction) -> lsp.CodeAction:
-    text_document: Final = LSP_SERVER.workspace.get_text_document(cast(str, params.data))
-    params.edit = lsp.WorkspaceEdit(
-        document_changes=[
-            lsp.TextDocumentEdit(
-                text_document=lsp.OptionalVersionedTextDocumentIdentifier(
-                    uri=text_document.uri, version=text_document.version
-                ),
-                edits=list(make_fixall_text_edits(text_document.source)),
-            )
-        ],
-    )
+def resolve_code_action(ls: CustomLanguageServer, params: lsp.CodeAction) -> lsp.CodeAction:
+    if ls.service:
+        text_document: Final = ls.workspace.get_text_document(cast(str, params.data))
+        params.edit = lsp.WorkspaceEdit(
+            document_changes=[
+                lsp.TextDocumentEdit(
+                    text_document=lsp.OptionalVersionedTextDocumentIdentifier(
+                        uri=text_document.uri, version=text_document.version
+                    ),
+                    edits=ls.service.make_fix_all_text_edits(text_document.source),
+                )
+            ],
+        )
     return params
 
 
