Apply black & ruff to modules copied from multilspy

This commit is contained in:
Dominik Jain committed 2025-06-23 16:34:56 +02:00
1 parent 27826d89c6
commit c2593a1171
32 files changed
+816 -802

No files matched your search

+4
View File
@@ -241,6 +241,10 @@ ignore = [
"SIM105", # forbids empty/general except clause
"SIM113", # wants to enforce use of enumerate
"E712", # forbids equality comparison with True/False
"UP007", # forbids some uses of Union
"TID252", # forbids relative imports
"B904", # forces use of raise from other_exception
"RUF012", # forbids mutable attributes as ClassVar
]
unfixable = [
"F841",
+3 -3
View File
@@ -34,9 +34,6 @@ from sensai.util.logging import LOG_DEFAULT_FORMAT, FallbackHandler
from sensai.util.string import ToStringMixin, dict_string
from multilspy import SyncLanguageServer
from solidlsp.multilspy_config import Language, MultilspyConfig
from solidlsp.multilspy_logger import MultilspyLogger
from solidlsp.multilspy_types import SymbolKind
from serena import serena_version
from serena.config import SerenaAgentContext, SerenaAgentMode
from serena.constants import (
@@ -57,6 +54,9 @@ from serena.util.inspection import determine_programming_language_composition, i
from serena.util.shell import execute_shell_command
from serena.util.thread import ExecutionResult, execute_with_timeout
from solidlsp.ls import SolidLanguageServer
from solidlsp.multilspy_config import Language, MultilspyConfig
from solidlsp.multilspy_logger import MultilspyLogger
from solidlsp.multilspy_types import SymbolKind
if TYPE_CHECKING:
from serena.gui_log_viewer import GuiLogViewerHandler
+1 -1
View File
@@ -3,8 +3,8 @@ import os
from collections.abc import Generator
from typing import TypeVar
from solidlsp.multilspy_config import Language
from serena.util.file_system import find_all_non_ignored_files
from solidlsp.multilspy_config import Language
T = TypeVar("T")
@@ -9,12 +9,12 @@ import pathlib
import stat
import threading
from solidlsp.ls import SolidLanguageServer
from solidlsp.lsp_protocol_handler.lsp_types import InitializeParams
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.multilspy_config import MultilspyConfig
from solidlsp.multilspy_logger import MultilspyLogger
from solidlsp.multilspy_utils import FileUtils, PlatformUtils
from solidlsp.ls import SolidLanguageServer
class ClangdLanguageServer(SolidLanguageServer):
@@ -4,10 +4,10 @@ import os
import pathlib
import stat
from solidlsp.ls import SolidLanguageServer
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.multilspy_logger import MultilspyLogger
from solidlsp.multilspy_utils import FileUtils, PlatformUtils
from solidlsp.ls import SolidLanguageServer
class DartLanguageServer(SolidLanguageServer):
@@ -15,13 +15,13 @@ from pathlib import PurePath
from overrides import override
from solidlsp.ls import SolidLanguageServer
from solidlsp.lsp_protocol_handler.lsp_types import InitializeParams
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.multilspy_config import MultilspyConfig
from solidlsp.multilspy_logger import MultilspyLogger
from solidlsp.multilspy_settings import MultilspySettings
from solidlsp.multilspy_utils import FileUtils, PlatformUtils
from solidlsp.ls import SolidLanguageServer
@dataclasses.dataclass
+1 -1
View File
@@ -7,11 +7,11 @@ import threading
from overrides import override
from solidlsp.ls import SolidLanguageServer
from solidlsp.lsp_protocol_handler.lsp_types import InitializeParams
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.multilspy_config import MultilspyConfig
from solidlsp.multilspy_logger import MultilspyLogger
from solidlsp.ls import SolidLanguageServer
class Gopls(SolidLanguageServer):
@@ -12,12 +12,12 @@ from time import sleep
from overrides import override
from solidlsp.ls import SolidLanguageServer
from solidlsp.lsp_protocol_handler.lsp_types import DefinitionParams, InitializeParams
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.multilspy_config import MultilspyConfig
from solidlsp.multilspy_logger import MultilspyLogger
from solidlsp.multilspy_utils import PlatformId, PlatformUtils
from solidlsp.ls import SolidLanguageServer
class Intelephense(SolidLanguageServer):
@@ -9,11 +9,11 @@ import pathlib
from overrides import override
from solidlsp.ls import SolidLanguageServer
from solidlsp.lsp_protocol_handler.lsp_types import InitializeParams
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.multilspy_config import MultilspyConfig
from solidlsp.multilspy_logger import MultilspyLogger
from solidlsp.ls import SolidLanguageServer
class JediServer(SolidLanguageServer):
@@ -9,12 +9,12 @@ import os
import pathlib
import stat
from solidlsp.ls import SolidLanguageServer
from solidlsp.lsp_protocol_handler.lsp_types import InitializeParams
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.multilspy_config import MultilspyConfig
from solidlsp.multilspy_logger import MultilspyLogger
from solidlsp.multilspy_utils import FileUtils, PlatformUtils
from solidlsp.ls import SolidLanguageServer
@dataclasses.dataclass
@@ -12,13 +12,13 @@ from collections.abc import Iterable
from overrides import override
from solidlsp.ls import SolidLanguageServer
from solidlsp.lsp_protocol_handler.lsp_types import InitializeParams
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.multilspy_config import MultilspyConfig
from solidlsp.multilspy_exceptions import MultilspyException
from solidlsp.multilspy_logger import MultilspyLogger
from solidlsp.multilspy_utils import DotnetVersion, FileUtils, PlatformId, PlatformUtils
from solidlsp.ls import SolidLanguageServer
def breadth_first_file_scan(root) -> Iterable[str]:
@@ -10,11 +10,11 @@ import threading
from overrides import override
from solidlsp.ls import SolidLanguageServer
from solidlsp.lsp_protocol_handler.lsp_types import InitializeParams
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.multilspy_config import MultilspyConfig
from solidlsp.multilspy_logger import MultilspyLogger
from solidlsp.ls import SolidLanguageServer
class PyrightServer(SolidLanguageServer):
@@ -11,12 +11,12 @@ import threading
from overrides import override
from solidlsp.ls import SolidLanguageServer
from solidlsp.lsp_protocol_handler.lsp_types import InitializeParams
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.multilspy_config import MultilspyConfig
from solidlsp.multilspy_logger import MultilspyLogger
from solidlsp.multilspy_utils import FileUtils, PlatformUtils
from solidlsp.ls import SolidLanguageServer
class RustAnalyzer(SolidLanguageServer):
@@ -12,11 +12,11 @@ import subprocess
import threading
from typing import override
from solidlsp.ls import SolidLanguageServer
from solidlsp.lsp_protocol_handler.lsp_types import InitializeParams
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.multilspy_config import MultilspyConfig
from solidlsp.multilspy_logger import MultilspyLogger
from solidlsp.ls import SolidLanguageServer
class Solargraph(SolidLanguageServer):
@@ -13,12 +13,12 @@ from time import sleep
from overrides import override
from solidlsp.ls import SolidLanguageServer
from solidlsp.lsp_protocol_handler.lsp_types import InitializeParams
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.multilspy_config import MultilspyConfig
from solidlsp.multilspy_logger import MultilspyLogger
from solidlsp.multilspy_utils import PlatformId, PlatformUtils
from solidlsp.ls import SolidLanguageServer
# Platform-specific imports
if os.name != "nt": # Unix-like systems
+8 -10
View File
@@ -16,9 +16,12 @@ from typing import Self, cast
import pathspec
import tqdm
from solidlsp import multilspy_types
from multilspy.language_server import GenericDocumentSymbol, LSPFileBuffer, ReferenceInSymbol
from solidlsp.lsp_protocol_handler import lsp_types, lsp_types as LSPTypes
from serena.text_utils import MatchedConsecutiveLines, search_files
from solidlsp import multilspy_types
from solidlsp.ls_handler import SolidLanguageServerHandler
from solidlsp.lsp_protocol_handler import lsp_types
from solidlsp.lsp_protocol_handler import lsp_types as LSPTypes
from solidlsp.lsp_protocol_handler.lsp_constants import LSPConstants
from solidlsp.lsp_protocol_handler.lsp_types import Definition, DefinitionParams, LocationLink, SymbolKind
from solidlsp.lsp_protocol_handler.server import (
@@ -30,8 +33,6 @@ from solidlsp.multilspy_config import Language, MultilspyConfig
from solidlsp.multilspy_exceptions import MultilspyException
from solidlsp.multilspy_logger import MultilspyLogger
from solidlsp.multilspy_utils import FileUtils, PathUtils, TextUtils
from serena.text_utils import MatchedConsecutiveLines, search_files
from solidlsp.ls_handler import SolidLanguageServerHandler
class SolidLanguageServer(ABC):
@@ -174,8 +175,7 @@ class SolidLanguageServer(ABC):
# load cache first to prevent any racing conditions due to asyncio stuff
self._document_symbols_cache: dict[
str, tuple[str, tuple[list[multilspy_types.UnifiedSymbolInformation], list[
multilspy_types.UnifiedSymbolInformation]]]
str, tuple[str, tuple[list[multilspy_types.UnifiedSymbolInformation], list[multilspy_types.UnifiedSymbolInformation]]]
] = {}
"""Maps file paths to a tuple of (file_content_hash, result_of_request_document_symbols)"""
self._cache_lock = threading.Lock()
@@ -543,8 +543,7 @@ class SolidLanguageServer(ABC):
return ret
# Some LS cause problems with this, so the call is isolated from the rest to allow overriding in subclasses
def _send_references_request(self, relative_file_path: str, line: int, column: int) -> list[
lsp_types.Location] | None:
def _send_references_request(self, relative_file_path: str, line: int, column: int) -> list[lsp_types.Location] | None:
return self.server.send.references(
{
"textDocument": {"uri": PathUtils.path_to_uri(os.path.join(self.repository_root_path, relative_file_path))},
@@ -840,8 +839,7 @@ class SolidLanguageServer(ABC):
if LSPConstants.CHILDREN in root_symbol:
# TODO: l_tree should be a list of TreeRepr. Define the following function to return TreeRepr as well
def visit_tree_nodes_and_build_tree_repr(node: GenericDocumentSymbol) -> list[
multilspy_types.UnifiedSymbolInformation]:
def visit_tree_nodes_and_build_tree_repr(node: GenericDocumentSymbol) -> list[multilspy_types.UnifiedSymbolInformation]:
node = cast(multilspy_types.UnifiedSymbolInformation, node)
l: list[multilspy_types.UnifiedSymbolInformation] = []
turn_item_into_symbol_with_children(node)
+1 -1
View File
@@ -28,8 +28,8 @@ from solidlsp.lsp_protocol_handler.server import (
make_request,
make_response,
)
from solidlsp.multilspy_exceptions import MultilspyException
from solidlsp.lsp_request import SolidLspRequest
from solidlsp.multilspy_exceptions import MultilspyException
log = logging.getLogger(__name__)
@@ -2,6 +2,7 @@
This module contains constants used in the LSP protocol.
"""
class LSPConstants:
"""
This class contains constants used in the LSP protocol.
+172 -227
View File
@@ -29,7 +29,8 @@ OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
"""
from typing import List, Union
from typing import Union
from solidlsp.lsp_protocol_handler import lsp_types
@@ -39,96 +40,94 @@ class LspRequest:
async def implementation(
self, params: lsp_types.ImplementationParams
) -> Union["lsp_types.Definition", List["lsp_types.LocationLink"], None]:
) -> Union["lsp_types.Definition", list["lsp_types.LocationLink"], None]:
"""A request to resolve the implementation locations of a symbol at a given text
document position. The request's parameter is of type [TextDocumentPositionParams]
(#TextDocumentPositionParams) the response is of type {@link Definition} or a
Thenable that resolves to such."""
Thenable that resolves to such.
"""
return await self.send_request("textDocument/implementation", params)
async def type_definition(
self, params: lsp_types.TypeDefinitionParams
) -> Union["lsp_types.Definition", List["lsp_types.LocationLink"], None]:
) -> Union["lsp_types.Definition", list["lsp_types.LocationLink"], None]:
"""A request to resolve the type definition locations of a symbol at a given text
document position. The request's parameter is of type [TextDocumentPositionParams]
(#TextDocumentPositionParams) the response is of type {@link Definition} or a
Thenable that resolves to such."""
Thenable that resolves to such.
"""
return await self.send_request("textDocument/typeDefinition", params)
async def document_color(
self, params: lsp_types.DocumentColorParams
) -> List["lsp_types.ColorInformation"]:
async def document_color(self, params: lsp_types.DocumentColorParams) -> list["lsp_types.ColorInformation"]:
"""A request to list all color symbols found in a given text document. The request's
parameter is of type {@link DocumentColorParams} the
response is of type {@link ColorInformation ColorInformation[]} or a Thenable
that resolves to such."""
that resolves to such.
"""
return await self.send_request("textDocument/documentColor", params)
async def color_presentation(
self, params: lsp_types.ColorPresentationParams
) -> List["lsp_types.ColorPresentation"]:
async def color_presentation(self, params: lsp_types.ColorPresentationParams) -> list["lsp_types.ColorPresentation"]:
"""A request to list all presentation for a color. The request's
parameter is of type {@link ColorPresentationParams} the
response is of type {@link ColorInformation ColorInformation[]} or a Thenable
that resolves to such."""
that resolves to such.
"""
return await self.send_request("textDocument/colorPresentation", params)
async def folding_range(
self, params: lsp_types.FoldingRangeParams
) -> Union[List["lsp_types.FoldingRange"], None]:
async def folding_range(self, params: lsp_types.FoldingRangeParams) -> list["lsp_types.FoldingRange"] | None:
"""A request to provide folding ranges in a document. The request's
parameter is of type {@link FoldingRangeParams}, the
response is of type {@link FoldingRangeList} or a Thenable
that resolves to such."""
that resolves to such.
"""
return await self.send_request("textDocument/foldingRange", params)
async def declaration(
self, params: lsp_types.DeclarationParams
) -> Union["lsp_types.Declaration", List["lsp_types.LocationLink"], None]:
) -> Union["lsp_types.Declaration", list["lsp_types.LocationLink"], None]:
"""A request to resolve the type definition locations of a symbol at a given text
document position. The request's parameter is of type [TextDocumentPositionParams]
(#TextDocumentPositionParams) the response is of type {@link Declaration}
or a typed array of {@link DeclarationLink} or a Thenable that resolves
to such."""
to such.
"""
return await self.send_request("textDocument/declaration", params)
async def selection_range(
self, params: lsp_types.SelectionRangeParams
) -> Union[List["lsp_types.SelectionRange"], None]:
async def selection_range(self, params: lsp_types.SelectionRangeParams) -> list["lsp_types.SelectionRange"] | None:
"""A request to provide selection ranges in a document. The request's
parameter is of type {@link SelectionRangeParams}, the
response is of type {@link SelectionRange SelectionRange[]} or a Thenable
that resolves to such."""
that resolves to such.
"""
return await self.send_request("textDocument/selectionRange", params)
async def prepare_call_hierarchy(
self, params: lsp_types.CallHierarchyPrepareParams
) -> Union[List["lsp_types.CallHierarchyItem"], None]:
async def prepare_call_hierarchy(self, params: lsp_types.CallHierarchyPrepareParams) -> list["lsp_types.CallHierarchyItem"] | None:
"""A request to result a `CallHierarchyItem` in a document at a given position.
Can be used as an input to an incoming or outgoing call hierarchy.
@since 3.16.0"""
@since 3.16.0
"""
return await self.send_request("textDocument/prepareCallHierarchy", params)
async def incoming_calls(
self, params: lsp_types.CallHierarchyIncomingCallsParams
) -> Union[List["lsp_types.CallHierarchyIncomingCall"], None]:
) -> list["lsp_types.CallHierarchyIncomingCall"] | None:
"""A request to resolve the incoming calls for a given `CallHierarchyItem`.
@since 3.16.0"""
@since 3.16.0
"""
return await self.send_request("callHierarchy/incomingCalls", params)
async def outgoing_calls(
self, params: lsp_types.CallHierarchyOutgoingCallsParams
) -> Union[List["lsp_types.CallHierarchyOutgoingCall"], None]:
) -> list["lsp_types.CallHierarchyOutgoingCall"] | None:
"""A request to resolve the outgoing calls for a given `CallHierarchyItem`.
@since 3.16.0"""
@since 3.16.0
"""
return await self.send_request("callHierarchy/outgoingCalls", params)
async def semantic_tokens_full(
self, params: lsp_types.SemanticTokensParams
) -> Union["lsp_types.SemanticTokens", None]:
async def semantic_tokens_full(self, params: lsp_types.SemanticTokensParams) -> Union["lsp_types.SemanticTokens", None]:
"""@since 3.16.0"""
return await self.send_request("textDocument/semanticTokens/full", params)
@@ -138,157 +137,143 @@ class LspRequest:
"""@since 3.16.0"""
return await self.send_request("textDocument/semanticTokens/full/delta", params)
async def semantic_tokens_range(
self, params: lsp_types.SemanticTokensRangeParams
) -> Union["lsp_types.SemanticTokens", None]:
async def semantic_tokens_range(self, params: lsp_types.SemanticTokensRangeParams) -> Union["lsp_types.SemanticTokens", None]:
"""@since 3.16.0"""
return await self.send_request("textDocument/semanticTokens/range", params)
async def linked_editing_range(
self, params: lsp_types.LinkedEditingRangeParams
) -> Union["lsp_types.LinkedEditingRanges", None]:
async def linked_editing_range(self, params: lsp_types.LinkedEditingRangeParams) -> Union["lsp_types.LinkedEditingRanges", None]:
"""A request to provide ranges that can be edited together.
@since 3.16.0"""
@since 3.16.0
"""
return await self.send_request("textDocument/linkedEditingRange", params)
async def will_create_files(
self, params: lsp_types.CreateFilesParams
) -> Union["lsp_types.WorkspaceEdit", None]:
async def will_create_files(self, params: lsp_types.CreateFilesParams) -> Union["lsp_types.WorkspaceEdit", None]:
"""The will create files request is sent from the client to the server before files are actually
created as long as the creation is triggered from within the client.
@since 3.16.0"""
@since 3.16.0
"""
return await self.send_request("workspace/willCreateFiles", params)
async def will_rename_files(
self, params: lsp_types.RenameFilesParams
) -> Union["lsp_types.WorkspaceEdit", None]:
async def will_rename_files(self, params: lsp_types.RenameFilesParams) -> Union["lsp_types.WorkspaceEdit", None]:
"""The will rename files request is sent from the client to the server before files are actually
renamed as long as the rename is triggered from within the client.
@since 3.16.0"""
@since 3.16.0
"""
return await self.send_request("workspace/willRenameFiles", params)
async def will_delete_files(
self, params: lsp_types.DeleteFilesParams
) -> Union["lsp_types.WorkspaceEdit", None]:
async def will_delete_files(self, params: lsp_types.DeleteFilesParams) -> Union["lsp_types.WorkspaceEdit", None]:
"""The did delete files notification is sent from the client to the server when
files were deleted from within the client.
@since 3.16.0"""
@since 3.16.0
"""
return await self.send_request("workspace/willDeleteFiles", params)
async def moniker(
self, params: lsp_types.MonikerParams
) -> Union[List["lsp_types.Moniker"], None]:
async def moniker(self, params: lsp_types.MonikerParams) -> list["lsp_types.Moniker"] | None:
"""A request to get the moniker of a symbol at a given text document position.
The request parameter is of type {@link TextDocumentPositionParams}.
The response is of type {@link Moniker Moniker[]} or `null`."""
The response is of type {@link Moniker Moniker[]} or `null`.
"""
return await self.send_request("textDocument/moniker", params)
async def prepare_type_hierarchy(
self, params: lsp_types.TypeHierarchyPrepareParams
) -> Union[List["lsp_types.TypeHierarchyItem"], None]:
async def prepare_type_hierarchy(self, params: lsp_types.TypeHierarchyPrepareParams) -> list["lsp_types.TypeHierarchyItem"] | None:
"""A request to result a `TypeHierarchyItem` in a document at a given position.
Can be used as an input to a subtypes or supertypes type hierarchy.
@since 3.17.0"""
@since 3.17.0
"""
return await self.send_request("textDocument/prepareTypeHierarchy", params)
async def type_hierarchy_supertypes(
self, params: lsp_types.TypeHierarchySupertypesParams
) -> Union[List["lsp_types.TypeHierarchyItem"], None]:
) -> list["lsp_types.TypeHierarchyItem"] | None:
"""A request to resolve the supertypes for a given `TypeHierarchyItem`.
@since 3.17.0"""
@since 3.17.0
"""
return await self.send_request("typeHierarchy/supertypes", params)
async def type_hierarchy_subtypes(
self, params: lsp_types.TypeHierarchySubtypesParams
) -> Union[List["lsp_types.TypeHierarchyItem"], None]:
async def type_hierarchy_subtypes(self, params: lsp_types.TypeHierarchySubtypesParams) -> list["lsp_types.TypeHierarchyItem"] | None:
"""A request to resolve the subtypes for a given `TypeHierarchyItem`.
@since 3.17.0"""
@since 3.17.0
"""
return await self.send_request("typeHierarchy/subtypes", params)
async def inline_value(
self, params: lsp_types.InlineValueParams
) -> Union[List["lsp_types.InlineValue"], None]:
async def inline_value(self, params: lsp_types.InlineValueParams) -> list["lsp_types.InlineValue"] | None:
"""A request to provide inline values in a document. The request's parameter is of
type {@link InlineValueParams}, the response is of type
{@link InlineValue InlineValue[]} or a Thenable that resolves to such.
@since 3.17.0"""
@since 3.17.0
"""
return await self.send_request("textDocument/inlineValue", params)
async def inlay_hint(
self, params: lsp_types.InlayHintParams
) -> Union[List["lsp_types.InlayHint"], None]:
async def inlay_hint(self, params: lsp_types.InlayHintParams) -> list["lsp_types.InlayHint"] | None:
"""A request to provide inlay hints in a document. The request's parameter is of
type {@link InlayHintsParams}, the response is of type
{@link InlayHint InlayHint[]} or a Thenable that resolves to such.
@since 3.17.0"""
@since 3.17.0
"""
return await self.send_request("textDocument/inlayHint", params)
async def resolve_inlay_hint(
self, params: lsp_types.InlayHint
) -> "lsp_types.InlayHint":
async def resolve_inlay_hint(self, params: lsp_types.InlayHint) -> "lsp_types.InlayHint":
"""A request to resolve additional properties for an inlay hint.
The request's parameter is of type {@link InlayHint}, the response is
of type {@link InlayHint} or a Thenable that resolves to such.
@since 3.17.0"""
@since 3.17.0
"""
return await self.send_request("inlayHint/resolve", params)
async def text_document_diagnostic(
self, params: lsp_types.DocumentDiagnosticParams
) -> "lsp_types.DocumentDiagnosticReport":
async def text_document_diagnostic(self, params: lsp_types.DocumentDiagnosticParams) -> "lsp_types.DocumentDiagnosticReport":
"""The document diagnostic request definition.
@since 3.17.0"""
@since 3.17.0
"""
return await self.send_request("textDocument/diagnostic", params)
async def workspace_diagnostic(
self, params: lsp_types.WorkspaceDiagnosticParams
) -> "lsp_types.WorkspaceDiagnosticReport":
async def workspace_diagnostic(self, params: lsp_types.WorkspaceDiagnosticParams) -> "lsp_types.WorkspaceDiagnosticReport":
"""The workspace diagnostic request definition.
@since 3.17.0"""
@since 3.17.0
"""
return await self.send_request("workspace/diagnostic", params)
async def initialize(
self, params: lsp_types.InitializeParams
) -> "lsp_types.InitializeResult":
async def initialize(self, params: lsp_types.InitializeParams) -> "lsp_types.InitializeResult":
"""The initialize request is sent from the client to the server.
It is sent once as the request after starting up the server.
The requests parameter is of type {@link InitializeParams}
the response if of type {@link InitializeResult} of a Thenable that
resolves to such."""
resolves to such.
"""
return await self.send_request("initialize", params)
async def shutdown(self) -> None:
"""A shutdown request is sent from the client to the server.
It is sent once when the client decides to shutdown the
server. The only notification that is sent after a shutdown request
is the exit event."""
is the exit event.
"""
return await self.send_request("shutdown")
async def will_save_wait_until(
self, params: lsp_types.WillSaveTextDocumentParams
) -> Union[List["lsp_types.TextEdit"], None]:
async def will_save_wait_until(self, params: lsp_types.WillSaveTextDocumentParams) -> list["lsp_types.TextEdit"] | None:
"""A document will save request is sent from the client to the server before
the document is actually saved. The request can return an array of TextEdits
which will be applied to the text document before it is saved. Please note that
clients might drop results if computing the text edits took too long or if a
server constantly fails on this request. This is done to keep the save fast and
reliable."""
reliable.
"""
return await self.send_request("textDocument/willSaveWaitUntil", params)
async def completion(
self, params: lsp_types.CompletionParams
) -> Union[List["lsp_types.CompletionItem"], "lsp_types.CompletionList", None]:
) -> Union[list["lsp_types.CompletionItem"], "lsp_types.CompletionList", None]:
"""Request to request completion at a given text document position. The request's
parameter is of type {@link TextDocumentPosition} the response
is of type {@link CompletionItem CompletionItem[]} or {@link CompletionList}
@@ -301,85 +286,72 @@ class LspRequest:
"""
return await self.send_request("textDocument/completion", params)
async def resolve_completion_item(
self, params: lsp_types.CompletionItem
) -> "lsp_types.CompletionItem":
async def resolve_completion_item(self, params: lsp_types.CompletionItem) -> "lsp_types.CompletionItem":
"""Request to resolve additional information for a given completion item.The request's
parameter is of type {@link CompletionItem} the response
is of type {@link CompletionItem} or a Thenable that resolves to such."""
is of type {@link CompletionItem} or a Thenable that resolves to such.
"""
return await self.send_request("completionItem/resolve", params)
async def hover(
self, params: lsp_types.HoverParams
) -> Union["lsp_types.Hover", None]:
async def hover(self, params: lsp_types.HoverParams) -> Union["lsp_types.Hover", None]:
"""Request to request hover information at a given text document position. The request's
parameter is of type {@link TextDocumentPosition} the response is of
type {@link Hover} or a Thenable that resolves to such."""
type {@link Hover} or a Thenable that resolves to such.
"""
return await self.send_request("textDocument/hover", params)
async def signature_help(
self, params: lsp_types.SignatureHelpParams
) -> Union["lsp_types.SignatureHelp", None]:
async def signature_help(self, params: lsp_types.SignatureHelpParams) -> Union["lsp_types.SignatureHelp", None]:
return await self.send_request("textDocument/signatureHelp", params)
async def definition(
self, params: lsp_types.DefinitionParams
) -> Union["lsp_types.Definition", List["lsp_types.LocationLink"], None]:
async def definition(self, params: lsp_types.DefinitionParams) -> Union["lsp_types.Definition", list["lsp_types.LocationLink"], None]:
"""A request to resolve the definition location of a symbol at a given text
document position. The request's parameter is of type [TextDocumentPosition]
(#TextDocumentPosition) the response is of either type {@link Definition}
or a typed array of {@link DefinitionLink} or a Thenable that resolves
to such."""
to such.
"""
return await self.send_request("textDocument/definition", params)
async def references(
self, params: lsp_types.ReferenceParams
) -> Union[List["lsp_types.Location"], None]:
async def references(self, params: lsp_types.ReferenceParams) -> list["lsp_types.Location"] | None:
"""A request to resolve project-wide references for the symbol denoted
by the given text document position. The request's parameter is of
type {@link ReferenceParams} the response is of type
{@link Location Location[]} or a Thenable that resolves to such."""
{@link Location Location[]} or a Thenable that resolves to such.
"""
return await self.send_request("textDocument/references", params)
async def document_highlight(
self, params: lsp_types.DocumentHighlightParams
) -> Union[List["lsp_types.DocumentHighlight"], None]:
async def document_highlight(self, params: lsp_types.DocumentHighlightParams) -> list["lsp_types.DocumentHighlight"] | None:
"""Request to resolve a {@link DocumentHighlight} for a given
text document position. The request's parameter is of type [TextDocumentPosition]
(#TextDocumentPosition) the request response is of type [DocumentHighlight[]]
(#DocumentHighlight) or a Thenable that resolves to such."""
(#DocumentHighlight) or a Thenable that resolves to such.
"""
return await self.send_request("textDocument/documentHighlight", params)
async def document_symbol(
self, params: lsp_types.DocumentSymbolParams
) -> Union[
List["lsp_types.SymbolInformation"], List["lsp_types.DocumentSymbol"], None
]:
) -> list["lsp_types.SymbolInformation"] | list["lsp_types.DocumentSymbol"] | None:
"""A request to list all symbols found in a given text document. The request's
parameter is of type {@link TextDocumentIdentifier} the
response is of type {@link SymbolInformation SymbolInformation[]} or a Thenable
that resolves to such."""
that resolves to such.
"""
return await self.send_request("textDocument/documentSymbol", params)
async def code_action(
self, params: lsp_types.CodeActionParams
) -> Union[List[Union["lsp_types.Command", "lsp_types.CodeAction"]], None]:
async def code_action(self, params: lsp_types.CodeActionParams) -> list[Union["lsp_types.Command", "lsp_types.CodeAction"]] | None:
"""A request to provide commands for the given text document and range."""
return await self.send_request("textDocument/codeAction", params)
async def resolve_code_action(
self, params: lsp_types.CodeAction
) -> "lsp_types.CodeAction":
async def resolve_code_action(self, params: lsp_types.CodeAction) -> "lsp_types.CodeAction":
"""Request to resolve additional information for a given code action.The request's
parameter is of type {@link CodeAction} the response
is of type {@link CodeAction} or a Thenable that resolves to such."""
is of type {@link CodeAction} or a Thenable that resolves to such.
"""
return await self.send_request("codeAction/resolve", params)
async def workspace_symbol(
self, params: lsp_types.WorkspaceSymbolParams
) -> Union[
List["lsp_types.SymbolInformation"], List["lsp_types.WorkspaceSymbol"], None
]:
) -> list["lsp_types.SymbolInformation"] | list["lsp_types.WorkspaceSymbol"] | None:
"""A request to list project-wide symbols matching the query string given
by the {@link WorkspaceSymbolParams}. The response is
of type {@link SymbolInformation SymbolInformation[]} or a Thenable that
@@ -391,78 +363,60 @@ class LspRequest:
"""
return await self.send_request("workspace/symbol", params)
async def resolve_workspace_symbol(
self, params: lsp_types.WorkspaceSymbol
) -> "lsp_types.WorkspaceSymbol":
async def resolve_workspace_symbol(self, params: lsp_types.WorkspaceSymbol) -> "lsp_types.WorkspaceSymbol":
"""A request to resolve the range inside the workspace
symbol's location.
@since 3.17.0"""
@since 3.17.0
"""
return await self.send_request("workspaceSymbol/resolve", params)
async def code_lens(
self, params: lsp_types.CodeLensParams
) -> Union[List["lsp_types.CodeLens"], None]:
async def code_lens(self, params: lsp_types.CodeLensParams) -> list["lsp_types.CodeLens"] | None:
"""A request to provide code lens for the given text document."""
return await self.send_request("textDocument/codeLens", params)
async def resolve_code_lens(
self, params: lsp_types.CodeLens
) -> "lsp_types.CodeLens":
async def resolve_code_lens(self, params: lsp_types.CodeLens) -> "lsp_types.CodeLens":
"""A request to resolve a command for a given code lens."""
return await self.send_request("codeLens/resolve", params)
async def document_link(
self, params: lsp_types.DocumentLinkParams
) -> Union[List["lsp_types.DocumentLink"], None]:
async def document_link(self, params: lsp_types.DocumentLinkParams) -> list["lsp_types.DocumentLink"] | None:
"""A request to provide document links"""
return await self.send_request("textDocument/documentLink", params)
async def resolve_document_link(
self, params: lsp_types.DocumentLink
) -> "lsp_types.DocumentLink":
async def resolve_document_link(self, params: lsp_types.DocumentLink) -> "lsp_types.DocumentLink":
"""Request to resolve additional information for a given document link. The request's
parameter is of type {@link DocumentLink} the response
is of type {@link DocumentLink} or a Thenable that resolves to such."""
is of type {@link DocumentLink} or a Thenable that resolves to such.
"""
return await self.send_request("documentLink/resolve", params)
async def formatting(
self, params: lsp_types.DocumentFormattingParams
) -> Union[List["lsp_types.TextEdit"], None]:
async def formatting(self, params: lsp_types.DocumentFormattingParams) -> list["lsp_types.TextEdit"] | None:
"""A request to to format a whole document."""
return await self.send_request("textDocument/formatting", params)
async def range_formatting(
self, params: lsp_types.DocumentRangeFormattingParams
) -> Union[List["lsp_types.TextEdit"], None]:
async def range_formatting(self, params: lsp_types.DocumentRangeFormattingParams) -> list["lsp_types.TextEdit"] | None:
"""A request to to format a range in a document."""
return await self.send_request("textDocument/rangeFormatting", params)
async def on_type_formatting(
self, params: lsp_types.DocumentOnTypeFormattingParams
) -> Union[List["lsp_types.TextEdit"], None]:
async def on_type_formatting(self, params: lsp_types.DocumentOnTypeFormattingParams) -> list["lsp_types.TextEdit"] | None:
"""A request to format a document on type."""
return await self.send_request("textDocument/onTypeFormatting", params)
async def rename(
self, params: lsp_types.RenameParams
) -> Union["lsp_types.WorkspaceEdit", None]:
async def rename(self, params: lsp_types.RenameParams) -> Union["lsp_types.WorkspaceEdit", None]:
"""A request to rename a symbol."""
return await self.send_request("textDocument/rename", params)
async def prepare_rename(
self, params: lsp_types.PrepareRenameParams
) -> Union["lsp_types.PrepareRenameResult", None]:
async def prepare_rename(self, params: lsp_types.PrepareRenameParams) -> Union["lsp_types.PrepareRenameResult", None]:
"""A request to test and perform the setup necessary for a rename.
@since 3.16 - support for default behavior"""
@since 3.16 - support for default behavior
"""
return await self.send_request("textDocument/prepareRename", params)
async def execute_command(
self, params: lsp_types.ExecuteCommandParams
) -> Union["lsp_types.LSPAny", None]:
async def execute_command(self, params: lsp_types.ExecuteCommandParams) -> Union["lsp_types.LSPAny", None]:
"""A request send from the client to the server to execute a command. The request might return
a workspace edit which the client will apply to the workspace."""
a workspace edit which the client will apply to the workspace.
"""
return await self.send_request("workspace/executeCommand", params)
@@ -470,92 +424,87 @@ class LspNotification:
def __init__(self, send_notification):
self.send_notification = send_notification
def did_change_workspace_folders(
self, params: lsp_types.DidChangeWorkspaceFoldersParams
) -> None:
def did_change_workspace_folders(self, params: lsp_types.DidChangeWorkspaceFoldersParams) -> None:
"""The `workspace/didChangeWorkspaceFolders` notification is sent from the client to the server when the workspace
folder configuration changes."""
folder configuration changes.
"""
return self.send_notification("workspace/didChangeWorkspaceFolders", params)
def cancel_work_done_progress(
self, params: lsp_types.WorkDoneProgressCancelParams
) -> None:
def cancel_work_done_progress(self, params: lsp_types.WorkDoneProgressCancelParams) -> None:
"""The `window/workDoneProgress/cancel` notification is sent from the client to the server to cancel a progress
initiated on the server side."""
initiated on the server side.
"""
return self.send_notification("window/workDoneProgress/cancel", params)
def did_create_files(self, params: lsp_types.CreateFilesParams) -> None:
"""The did create files notification is sent from the client to the server when
files were created from within the client.
@since 3.16.0"""
@since 3.16.0
"""
return self.send_notification("workspace/didCreateFiles", params)
def did_rename_files(self, params: lsp_types.RenameFilesParams) -> None:
"""The did rename files notification is sent from the client to the server when
files were renamed from within the client.
@since 3.16.0"""
@since 3.16.0
"""
return self.send_notification("workspace/didRenameFiles", params)
def did_delete_files(self, params: lsp_types.DeleteFilesParams) -> None:
"""The will delete files request is sent from the client to the server before files are actually
deleted as long as the deletion is triggered from within the client.
@since 3.16.0"""
@since 3.16.0
"""
return self.send_notification("workspace/didDeleteFiles", params)
def did_open_notebook_document(
self, params: lsp_types.DidOpenNotebookDocumentParams
) -> None:
def did_open_notebook_document(self, params: lsp_types.DidOpenNotebookDocumentParams) -> None:
"""A notification sent when a notebook opens.
@since 3.17.0"""
@since 3.17.0
"""
return self.send_notification("notebookDocument/didOpen", params)
def did_change_notebook_document(
self, params: lsp_types.DidChangeNotebookDocumentParams
) -> None:
def did_change_notebook_document(self, params: lsp_types.DidChangeNotebookDocumentParams) -> None:
return self.send_notification("notebookDocument/didChange", params)
def did_save_notebook_document(
self, params: lsp_types.DidSaveNotebookDocumentParams
) -> None:
def did_save_notebook_document(self, params: lsp_types.DidSaveNotebookDocumentParams) -> None:
"""A notification sent when a notebook document is saved.
@since 3.17.0"""
@since 3.17.0
"""
return self.send_notification("notebookDocument/didSave", params)
def did_close_notebook_document(
self, params: lsp_types.DidCloseNotebookDocumentParams
) -> None:
def did_close_notebook_document(self, params: lsp_types.DidCloseNotebookDocumentParams) -> None:
"""A notification sent when a notebook closes.
@since 3.17.0"""
@since 3.17.0
"""
return self.send_notification("notebookDocument/didClose", params)
def initialized(self, params: lsp_types.InitializedParams) -> None:
"""The initialized notification is sent from the client to the
server after the client is fully initialized and the server
is allowed to send requests from the server to the client."""
is allowed to send requests from the server to the client.
"""
return self.send_notification("initialized", params)
def exit(self) -> None:
"""The exit event is sent from the client to the server to
ask the server to exit its process."""
ask the server to exit its process.
"""
return self.send_notification("exit")
def workspace_did_change_configuration(
self, params: lsp_types.DidChangeConfigurationParams
) -> None:
def workspace_did_change_configuration(self, params: lsp_types.DidChangeConfigurationParams) -> None:
"""The configuration change notification is sent from the client to the server
when the client's configuration has changed. The notification contains
the changed configuration as defined by the language client."""
the changed configuration as defined by the language client.
"""
return self.send_notification("workspace/didChangeConfiguration", params)
def did_open_text_document(
self, params: lsp_types.DidOpenTextDocumentParams
) -> None:
def did_open_text_document(self, params: lsp_types.DidOpenTextDocumentParams) -> None:
"""The document open notification is sent from the client to the server to signal
newly opened text documents. The document's truth is now managed by the client
and the server must not try to read the document's truth using the document's
@@ -563,47 +512,43 @@ class LspNotification:
mean that its content is presented in an editor. An open notification must not
be sent more than once without a corresponding close notification send before.
This means open and close notification must be balanced and the max open count
is one."""
is one.
"""
return self.send_notification("textDocument/didOpen", params)
def did_change_text_document(
self, params: lsp_types.DidChangeTextDocumentParams
) -> None:
def did_change_text_document(self, params: lsp_types.DidChangeTextDocumentParams) -> None:
"""The document change notification is sent from the client to the server to signal
changes to a text document."""
changes to a text document.
"""
return self.send_notification("textDocument/didChange", params)
def did_close_text_document(
self, params: lsp_types.DidCloseTextDocumentParams
) -> None:
def did_close_text_document(self, params: lsp_types.DidCloseTextDocumentParams) -> None:
"""The document close notification is sent from the client to the server when
the document got closed in the client. The document's truth now exists where
the document's uri points to (e.g. if the document's uri is a file uri the
truth now exists on disk). As with the open notification the close notification
is about managing the document's content. Receiving a close notification
doesn't mean that the document was open in an editor before. A close
notification requires a previous open notification to be sent."""
notification requires a previous open notification to be sent.
"""
return self.send_notification("textDocument/didClose", params)
def did_save_text_document(
self, params: lsp_types.DidSaveTextDocumentParams
) -> None:
def did_save_text_document(self, params: lsp_types.DidSaveTextDocumentParams) -> None:
"""The document save notification is sent from the client to the server when
the document got saved in the client."""
the document got saved in the client.
"""
return self.send_notification("textDocument/didSave", params)
def will_save_text_document(
self, params: lsp_types.WillSaveTextDocumentParams
) -> None:
def will_save_text_document(self, params: lsp_types.WillSaveTextDocumentParams) -> None:
"""A document will save notification is sent from the client to the server before
the document is actually saved."""
the document is actually saved.
"""
return self.send_notification("textDocument/willSave", params)
def did_change_watched_files(
self, params: lsp_types.DidChangeWatchedFilesParams
) -> None:
def did_change_watched_files(self, params: lsp_types.DidChangeWatchedFilesParams) -> None:
"""The watched files notification is sent from the client to the server when
the client detects changes to file watched by the language client."""
the client detects changes to file watched by the language client.
"""
return self.send_notification("workspace/didChangeWatchedFiles", params)
def set_trace(self, params: lsp_types.SetTraceParams) -> None:
File diff suppressed because it is too large. Load diff
+39 -51
View File
@@ -1,8 +1,8 @@
"""
This file provides the implementation of the JSON-RPC client, that launches and
This file provides the implementation of the JSON-RPC client, that launches and
communicates with the language server.
The initial implementation of this file was obtained from
The initial implementation of this file was obtained from
https://github.com/predragnikolic/OLSP under the MIT License with the following terms:
MIT License
@@ -34,16 +34,17 @@ import json
import logging
import os
import threading
from collections.abc import Callable
from typing import Any, Union
import psutil
from typing import Any, Callable, Dict, List, Optional, Union
from ..multilspy_exceptions import MultilspyException
from .lsp_requests import LspNotification, LspRequest
from .lsp_types import ErrorCodes
from ..multilspy_exceptions import MultilspyException
StringDict = Dict[str, Any]
PayloadLike = Union[List[StringDict], StringDict, None]
StringDict = dict[str, Any]
PayloadLike = Union[list[StringDict], StringDict, None]
CONTENT_LENGTH = "Content-Length: "
ENCODING = "utf-8"
log = logging.getLogger(__name__)
@@ -59,7 +60,7 @@ class ProcessLaunchInfo:
cmd: str
# The environment variables to set for the process
env: Dict[str, str] = dataclasses.field(default_factory=dict)
env: dict[str, str] = dataclasses.field(default_factory=dict)
# The working directory for the process
cwd: str = os.getcwd()
@@ -120,8 +121,8 @@ class MessageType:
class Request:
def __init__(self) -> None:
self.cv = asyncio.Condition()
self.result: Optional[PayloadLike] = None
self.error: Optional[Error] = None
self.result: PayloadLike | None = None
self.error: Error | None = None
async def on_result(self, params: PayloadLike) -> None:
self.result = params
@@ -134,14 +135,14 @@ class Request:
self.cv.notify()
def content_length(line: bytes) -> Optional[int]:
def content_length(line: bytes) -> int | None:
if line.startswith(b"Content-Length: "):
_, value = line.split(b"Content-Length: ")
value = value.strip()
try:
return int(value)
except ValueError:
raise ValueError("Invalid Content-Length header: {}".format(value))
raise ValueError(f"Invalid Content-Length header: {value}")
return None
@@ -181,12 +182,13 @@ class LanguageServerHandler:
language server process in an independent process group. Default is `True`. Setting it to
`False` means that the language server process will be in the same process group as the
the current process, and any SIGINT and SIGTERM signals will be sent to both processes.
"""
def __init__(
self,
process_launch_info: ProcessLaunchInfo,
logger: Optional[Callable[[str, str, StringDict | str], None]] = None,
logger: Callable[[str, str, StringDict | str], None] | None = None,
start_independent_lsp_process=True,
) -> None:
"""
@@ -203,7 +205,7 @@ class LanguageServerHandler:
self._received_shutdown = False
self.request_id = 1
self._response_handlers: Dict[Any, Request] = {}
self._response_handlers: dict[Any, Request] = {}
self.on_request_handlers = {}
self.on_notification_handlers = {}
self.logger = logger
@@ -218,8 +220,6 @@ class LanguageServerHandler:
self._response_handlers_lock = threading.Lock()
self._tasks_lock = threading.Lock()
def is_running(self) -> bool:
"""
Checks if the language server process is currently running.
@@ -250,7 +250,7 @@ class LanguageServerHandler:
log.error("Language server has already terminated/could not be started")
# Process has already terminated
stderr_data = await self.process.stderr.read()
error_message = stderr_data.decode('utf-8', errors='replace')
error_message = stderr_data.decode("utf-8", errors="replace")
raise RuntimeError(f"Process terminated immediately with code {self.process.returncode}. Error: {error_message}")
self.loop = asyncio.get_event_loop()
@@ -262,21 +262,19 @@ class LanguageServerHandler:
self.tasks[self.task_counter] = self.loop.create_task(self.run_forever_stderr())
self.task_counter += 1
async def stop(self) -> None:
"""
Sends the terminate signal to the language server process and waits for it to exit, with a timeout, killing it if necessary
"""
# First cancel all tasks
await self._cancel_pending_tasks()
process = self.process
self.process = None
if not process:
return
# Clean up the process
await self._cleanup_process(process)
@@ -294,7 +292,7 @@ class LanguageServerHandler:
if pending_tasks:
try:
await asyncio.wait_for(asyncio.gather(*pending_tasks, return_exceptions=True), timeout=5.0)
except (asyncio.TimeoutError, Exception):
except (TimeoutError, Exception):
pass
# Clear tasks dictionary under lock
@@ -306,18 +304,18 @@ class LanguageServerHandler:
# Close stdin first to prevent deadlocks
# See: https://bugs.python.org/issue35539
self._safely_close_pipe(process.stdin)
# Terminate/kill the process if it's still running
if process.returncode is None:
await self._terminate_or_kill_process(process)
# Close stdout and stderr pipes after process has exited
# This is essential to prevent "I/O operation on closed pipe" errors and
# "Event loop is closed" errors during garbage collection
# See: https://bugs.python.org/issue41320 and https://github.com/python/cpython/issues/88050
self._safely_close_pipe(process.stdout)
self._safely_close_pipe(process.stderr)
# Small delay to ensure OS has released file handles
await asyncio.sleep(0.5)
@@ -333,11 +331,11 @@ class LanguageServerHandler:
"""Try to terminate the process gracefully, then forcefully if necessary."""
# First try to terminate the process tree gracefully
self._signal_process_tree(process, terminate=True)
# Wait for the process to exit (with timeout)
try:
await asyncio.wait_for(process.wait(), timeout=10)
except (asyncio.TimeoutError, Exception):
except (TimeoutError, Exception):
# If termination failed, forcefully kill the process tree
self._signal_process_tree(process, terminate=False)
try:
@@ -349,14 +347,14 @@ class LanguageServerHandler:
def _signal_process_tree(self, process, terminate=True):
"""Send signal (terminate or kill) to the process and all its children."""
signal_method = "terminate" if terminate else "kill"
# Try to get the parent process
parent = None
try:
parent = psutil.Process(process.pid)
except (psutil.NoSuchProcess, psutil.AccessDenied, Exception):
pass
# If we have the parent process and it's running, signal the entire tree
if parent and parent.is_running():
# Signal children first
@@ -365,7 +363,7 @@ class LanguageServerHandler:
getattr(child, signal_method)()
except (psutil.NoSuchProcess, psutil.AccessDenied, Exception):
pass
# Then signal the parent
try:
getattr(parent, signal_method)()
@@ -378,7 +376,6 @@ class LanguageServerHandler:
except Exception:
pass
async def shutdown(self) -> None:
"""
Perform the shutdown sequence for the client, including sending the shutdown request to the server and notifying it of exit
@@ -433,7 +430,6 @@ class LanguageServerHandler:
pass
return self._received_shutdown
async def run_forever_stderr(self) -> None:
"""
Continuously read from the language server process stderr and log the messages
@@ -443,7 +439,7 @@ class LanguageServerHandler:
line = await self.process.stderr.readline()
if not line:
continue
self._log("LSP stderr: " + line.decode(ENCODING, errors='replace'))
self._log("LSP stderr: " + line.decode(ENCODING, errors="replace"))
except (BrokenPipeError, ConnectionResetError, StopLoopException):
pass
@@ -453,7 +449,7 @@ class LanguageServerHandler:
"""
try:
await self._receive_payload(json.loads(body))
except IOError as ex:
except OSError as ex:
self._log(f"malformed {ENCODING}: {ex}")
except UnicodeDecodeError as ex:
self._log(f"malformed {ENCODING}: {ex}")
@@ -479,7 +475,7 @@ class LanguageServerHandler:
except Exception as err:
self._log(f"Error handling server payload: {err}")
def send_notification(self, method: str, params: Optional[dict] = None) -> None:
def send_notification(self, method: str, params: dict | None = None) -> None:
"""
Send notification pertaining to the given method to the server with the given parameters
"""
@@ -491,25 +487,19 @@ class LanguageServerHandler:
"""
# Use lock to prevent race conditions on tasks and task_counter
with self._tasks_lock:
self.tasks[self.task_counter] = asyncio.get_event_loop().create_task(
self._send_payload(make_response(request_id, params))
)
self.tasks[self.task_counter] = asyncio.get_event_loop().create_task(self._send_payload(make_response(request_id, params)))
self.task_counter += 1
def send_error_response(self, request_id: Any, err: Error) -> None:
"""
Send error response to the given request id to the server with the given error
"""
# Use lock to prevent race conditions on tasks and task_counter
with self._tasks_lock:
self.tasks[self.task_counter] = asyncio.get_event_loop().create_task(
self._send_payload(make_error_response(request_id, err))
)
self.tasks[self.task_counter] = asyncio.get_event_loop().create_task(self._send_payload(make_error_response(request_id, err)))
self.task_counter += 1
async def send_request(self, method: str, params: Optional[dict] = None) -> PayloadLike:
async def send_request(self, method: str, params: dict | None = None) -> PayloadLike:
"""
Send request to the server, register the request id, and wait for the response
"""
@@ -527,13 +517,14 @@ class LanguageServerHandler:
await self._send_payload(make_request(method, request_id, params))
self._log(f"Waiting for asyncio condition for request {method} with params:\n{params}")
await request.cv.wait()
self._log(f"Finished waiting, processing result")
self._log("Finished waiting, processing result")
if isinstance(request.error, Error):
raise MultilspyException(f"Could not process request {method} with params:\n{params}.\n Language server error: {request.error}") from request.error
raise MultilspyException(
f"Could not process request {method} with params:\n{params}.\n Language server error: {request.error}"
) from request.error
self._log(f"Returning non-error result, which is:\n{request.result}")
return request.result
def _send_payload_sync(self, payload: StringDict) -> None:
"""
Send the payload to the server by writing to its stdin synchronously
@@ -554,7 +545,6 @@ class LanguageServerHandler:
self.logger("client", "logger", f"Failed to write to stdin: {e}")
return
async def _send_payload(self, payload: StringDict) -> None:
"""
Send the payload to the server by writing to its stdin asynchronously.
@@ -575,7 +565,6 @@ class LanguageServerHandler:
self.logger("client", "logger", f"Failed to write to stdin: {e}")
return
def on_request(self, method: str, cb) -> None:
"""
Register the callback function to handle requests from the server to the client for the given method
@@ -602,7 +591,6 @@ class LanguageServerHandler:
else:
await request.on_error(Error(ErrorCodes.InvalidRequest, ""))
async def _request_handler(self, response: StringDict) -> None:
"""
Handle the request received from the server: call the appropriate callback function and return the result
@@ -616,7 +604,7 @@ class LanguageServerHandler:
request_id,
Error(
ErrorCodes.MethodNotFound,
"method '{}' not handled on client.".format(method),
f"method '{method}' not handled on client.",
),
)
return
+5 -6
View File
@@ -1,10 +1,10 @@
"""
Configuration parameters for Multilspy.
"""
import fnmatch
from enum import Enum
from typing import List
from dataclasses import dataclass, field
from enum import Enum
class FilenameMatcher:
@@ -77,6 +77,7 @@ class MultilspyConfig:
"""
Configuration parameters
"""
code_language: Language
trace_lsp_communication: bool = False
start_independent_lsp_process: bool = True
@@ -89,7 +90,5 @@ class MultilspyConfig:
Create a MultilspyConfig instance from a dictionary
"""
import inspect
return cls(**{
k: v for k, v in env.items()
if k in inspect.signature(cls).parameters
})
return cls(**{k: v for k, v in env.items() if k in inspect.signature(cls).parameters})
+2 -1
View File
@@ -2,6 +2,7 @@
This module contains the exceptions raised by the Multilspy framework.
"""
class MultilspyException(Exception):
"""
Exceptions raised by the Multilspy framework.
@@ -11,4 +12,4 @@ class MultilspyException(Exception):
"""
Initializes the exception with the given message.
"""
super().__init__(message)
super().__init__(message)
+2 -1
View File
@@ -1,9 +1,11 @@
"""
Multilspy logger module.
"""
import inspect
import logging
from datetime import datetime
from pydantic import BaseModel
@@ -34,7 +36,6 @@ class MultilspyLogger:
"""
Log the debug and sanitized messages using the logger
"""
debug_message = debug_message.replace("'", '"').replace("\n", " ")
sanitized_error_message = sanitized_error_message.replace("'", '"').replace("\n", " ")
+2
View File
@@ -5,10 +5,12 @@ Defines the settings for multilspy.
import os
import pathlib
class MultilspySettings:
"""
Provides the various settings for multilspy.
"""
@staticmethod
def get_language_server_directory() -> str:
"""Returns the directory for language servers"""
+44 -24
View File
@@ -4,16 +4,19 @@ Defines wrapper objects around the types returned by LSP to ensure decoupling be
from __future__ import annotations
from enum import IntEnum, Enum
from typing_extensions import NotRequired, TypedDict, List, Dict, Union
from enum import Enum, IntEnum
from typing import NotRequired, Union
from typing_extensions import TypedDict
URI = str
DocumentUri = str
Uint = int
RegExp = str
class Position(TypedDict):
"""Position in a text document expressed as zero-based line and character
r"""Position in a text document expressed as zero-based line and character
offset. Prior to 3.17 the offsets were always based on a UTF-16 string
representation. So a string of the form `a𐐀b` the character offset of the
character `a` is 0, the character offset of `𐐀` is 1 and the character
@@ -39,7 +42,8 @@ class Position(TypedDict):
Positions are line end character agnostic. So you can not specify a position
that denotes `\r|\n` or `\n|` where `|` represents the character offset.
@since 3.17.0 - support for negotiated position encoding."""
@since 3.17.0 - support for negotiated position encoding.
"""
line: Uint
""" Line position in a document (zero-based).
@@ -67,7 +71,8 @@ class Range(TypedDict):
start: { line: 5, character: 23 }
end : { line 6, character : 0 }
}
```"""
```
"""
start: Position
""" The range's start position. """
@@ -77,12 +82,14 @@ class Range(TypedDict):
class Location(TypedDict):
"""Represents a location inside a resource, such as a line
inside a text file."""
inside a text file.
"""
uri: DocumentUri
range: Range
absolutePath: str
relativePath: Union[str, None]
relativePath: str | None
class CompletionItemKind(IntEnum):
"""The kind of a completion entry."""
@@ -113,9 +120,11 @@ class CompletionItemKind(IntEnum):
Operator = 24
TypeParameter = 25
class CompletionItem(TypedDict):
"""A completion item represents a text snippet that is
proposed to complete text that is being typed."""
proposed to complete text that is being typed.
"""
completionText: str
""" The completionText of this completion item.
@@ -131,6 +140,7 @@ class CompletionItem(TypedDict):
""" A human-readable string with additional information
about this item, like type or symbol information. """
class SymbolKind(IntEnum):
"""A symbol kind."""
@@ -161,17 +171,21 @@ class SymbolKind(IntEnum):
Operator = 25
TypeParameter = 26
class SymbolTag(IntEnum):
"""Symbol tags are extra annotations that tweak the rendering of a symbol.
@since 3.16"""
@since 3.16
"""
Deprecated = 1
""" Render a symbol as obsolete, usually using a strike-out. """
class UnifiedSymbolInformation(TypedDict):
"""Represents information about programming constructs like variables, classes,
interfaces etc."""
interfaces etc.
"""
deprecated: NotRequired[bool]
""" Indicates if this symbol is deprecated.
@@ -191,7 +205,7 @@ class UnifiedSymbolInformation(TypedDict):
""" The name of this symbol. """
kind: SymbolKind
""" The kind of this symbol. """
tags: NotRequired[List[SymbolTag]]
tags: NotRequired[list[SymbolTag]]
""" Tags for this symbol.
@since 3.16.0 """
@@ -207,7 +221,7 @@ class UnifiedSymbolInformation(TypedDict):
detail: NotRequired[str]
""" More detail for this symbol, e.g the signature of a function. """
range: NotRequired[Range]
""" The range enclosing this symbol not including leading/trailing whitespace but everything else
like comments. This information is typically used to determine if the clients cursor is
@@ -215,37 +229,40 @@ class UnifiedSymbolInformation(TypedDict):
selectionRange: NotRequired[Range]
""" The range that should be selected and revealed when this symbol is being picked, e.g the name of a function.
Must be contained by the `range`. """
body: NotRequired[str]
""" The body of the symbol. """
children: List[UnifiedSymbolInformation]
children: list[UnifiedSymbolInformation]
""" The children of the symbol.
Added to be compatible with `lsp_types.DocumentSymbol`,
since it is sometimes useful to have the children of the symbol as a user-facing feature."""
parent: NotRequired[UnifiedSymbolInformation | None]
"""The parent of the symbol, if there is any. Added with Serena, not part of the LSP.
All symbols except the root packages will have a parent.
"""
class MarkupKind(Enum):
"""Describes the content type that a client supports in various
result literals like `Hover`, `ParameterInfo` or `CompletionItem`.
Please note that `MarkupKinds` must not start with a `$`. This kinds
are reserved for internal usage."""
are reserved for internal usage.
"""
PlainText = "plaintext"
""" Plain text is supported as a content format """
Markdown = "markdown"
""" Markdown is supported as a content format """
class __MarkedString_Type_1(TypedDict):
language: str
value: str
MarkedString = Union[str, "__MarkedString_Type_1"]
""" MarkedString can be used to render human readable text. It is either a markdown string
or a code-block that provides a language and a code snippet. The language identifier
@@ -260,8 +277,9 @@ ${value}
Note that markdown strings will be sanitized - that means html will be escaped.
@deprecated use MarkupContent instead. """
class MarkupContent(TypedDict):
"""A `MarkupContent` literal represents a string value which content is interpreted base on its
r"""A `MarkupContent` literal represents a string value which content is interpreted base on its
kind flag. Currently the protocol supports `plaintext` and `markdown` as markup kinds.
If the kind is `markdown` then the value can contain fenced code blocks like in GitHub issues.
@@ -282,18 +300,20 @@ class MarkupContent(TypedDict):
```
*Please Note* that clients might sanitize the return markdown. A client could decide to
remove HTML from the markdown to avoid script execution."""
remove HTML from the markdown to avoid script execution.
"""
kind: "MarkupKind"
kind: MarkupKind
""" The type of the Markup """
value: str
""" The content itself """
class Hover(TypedDict):
"""The result of a hover request."""
contents: Union["MarkupContent", "MarkedString", List["MarkedString"]]
contents: MarkupContent | MarkedString | list[MarkedString]
""" The hover's content """
range: NotRequired["Range"]
range: NotRequired[Range]
""" An optional range inside the text document that is used to
visualize the hover, e.g. by changing the background color. """
visualize the hover, e.g. by changing the background color. """
+52 -47
View File
@@ -5,17 +5,16 @@ This file contains various utility functions like I/O operations, handling paths
import gzip
import logging
import os
from typing import Tuple, Union
import requests
import shutil
import uuid
import platform
import shutil
import subprocess
import uuid
from enum import Enum
from pathlib import Path, PurePath
import requests
from solidlsp.multilspy_exceptions import MultilspyException
from pathlib import PurePath, Path
from solidlsp.multilspy_logger import MultilspyLogger
from solidlsp.multilspy_types import UnifiedSymbolInformation
@@ -24,8 +23,9 @@ class TextUtils:
"""
Utilities for text operations.
"""
@staticmethod
def get_line_col_from_index(text: str, index: int) -> Tuple[int, int]:
def get_line_col_from_index(text: str, index: int) -> tuple[int, int]:
"""
Returns the zero-indexed line and column number of the given index in the given text
"""
@@ -33,7 +33,7 @@ class TextUtils:
c = 0
idx = 0
while idx < index:
if text[idx] == '\n':
if text[idx] == "\n":
l += 1
c = 0
else:
@@ -41,7 +41,7 @@ class TextUtils:
idx += 1
return l, c
@staticmethod
def get_index_from_line_col(text: str, line: int, col: int) -> int:
"""
@@ -55,35 +55,35 @@ class TextUtils:
idx += 1
idx += col
return idx
@staticmethod
def _get_updated_position_from_line_and_column_and_edit(l: int, c: int, text_to_be_inserted: str) -> Tuple[int, int]:
def _get_updated_position_from_line_and_column_and_edit(l: int, c: int, text_to_be_inserted: str) -> tuple[int, int]:
"""
Utility function to get the position of the cursor after inserting text at a given line and column.
"""
num_newlines_in_gen_text = text_to_be_inserted.count('\n')
num_newlines_in_gen_text = text_to_be_inserted.count("\n")
if num_newlines_in_gen_text > 0:
l += num_newlines_in_gen_text
c = len(text_to_be_inserted.split('\n')[-1])
c = len(text_to_be_inserted.split("\n")[-1])
else:
c += len(text_to_be_inserted)
return (l, c)
@staticmethod
def delete_text_between_positions(text: str, start_line: int, start_col: int, end_line: int, end_col: int) -> Tuple[str, str]:
def delete_text_between_positions(text: str, start_line: int, start_col: int, end_line: int, end_col: int) -> tuple[str, str]:
"""
Deletes the text between the given start and end positions.
Returns the modified text and the deleted text.
"""
del_start_idx = TextUtils.get_index_from_line_col(text, start_line, start_col)
del_end_idx = TextUtils.get_index_from_line_col(text, end_line, end_col)
deleted_text = text[del_start_idx:del_end_idx]
new_text = text[:del_start_idx] + text[del_end_idx:]
return new_text, deleted_text
@staticmethod
def insert_text_at_position(text: str, line: int, col: int, text_to_be_inserted: str) -> Tuple[str, int, int]:
def insert_text_at_position(text: str, line: int, col: int, text_to_be_inserted: str) -> tuple[str, int, int]:
"""
Inserts the given text at the given line and column.
Returns the modified text and the new line and column.
@@ -98,6 +98,7 @@ class PathUtils:
"""
Utilities for platform-agnostic path operations.
"""
@staticmethod
def uri_to_path(uri: str) -> str:
"""
@@ -106,14 +107,15 @@ class PathUtils:
This method was obtained from https://stackoverflow.com/a/61922504
"""
try:
from urllib.parse import urlparse, unquote
from urllib.parse import unquote, urlparse
from urllib.request import url2pathname
except ImportError:
# backwards compatibility
from urlparse import urlparse
from urllib import unquote, url2pathname
from urlparse import urlparse
parsed = urlparse(uri)
host = "{0}{0}{mnt}{0}".format(os.path.sep, mnt=parsed.netloc)
host = f"{os.path.sep}{os.path.sep}{parsed.netloc}{os.path.sep}"
return os.path.normpath(os.path.join(host, url2pathname(unquote(parsed.path))))
@staticmethod
@@ -126,10 +128,10 @@ class PathUtils:
@staticmethod
def is_glob_pattern(pattern: str) -> bool:
"""Check if a pattern contains glob-specific characters."""
return any(c in pattern for c in '*?[]!')
return any(c in pattern for c in "*?[]!")
@staticmethod
def get_relative_path(path: str, base_path: str) -> Union[str, None]:
def get_relative_path(path: str, base_path: str) -> str | None:
"""
Gets relative path if it's possible (paths should be on the same drive),
returns `None` otherwise.
@@ -153,12 +155,12 @@ class FileUtils:
logger.log(f"File read '{file_path}' failed: File does not exist.", logging.ERROR)
raise MultilspyException(f"File read '{file_path}' failed: File does not exist.")
try:
with open(file_path, "r", encoding="utf-8") as inp_file:
with open(file_path, encoding="utf-8") as inp_file:
return inp_file.read()
except Exception as exc:
logger.log(f"File read '{file_path}' failed to read with encoding 'utf-8': {exc}", logging.ERROR)
raise MultilspyException("File read failed.") from None
@staticmethod
def download_file(logger: MultilspyLogger, url: str, target_path: str) -> None:
"""
@@ -215,6 +217,7 @@ class PlatformId(str, Enum):
"""
multilspy supported platforms
"""
WIN_x86 = "win-x86"
WIN_x64 = "win-x64"
WIN_arm64 = "win-arm64"
@@ -232,6 +235,7 @@ class DotnetVersion(str, Enum):
"""
multilspy supported dotnet versions
"""
V4 = "4"
V6 = "6"
V7 = "7"
@@ -260,7 +264,7 @@ class PlatformUtils:
platform_id = system_map[system] + "-" + machine_map[machine]
if system == "Linux" and bitness == "64bit":
libc = platform.libc_ver()[0]
if libc != 'glibc':
if libc != "glibc":
platform_id += "-" + libc
return PlatformId(platform_id)
else:
@@ -274,13 +278,13 @@ class PlatformUtils:
class SYSTEM_INFO(ctypes.Structure):
class _U(ctypes.Union):
class _S(ctypes.Structure):
_fields_ = [("wProcessorArchitecture", wintypes.WORD),
("wReserved", wintypes.WORD)]
_fields_ = [("dwOemId", wintypes.DWORD),
("s", _S)]
_fields_ = [("wProcessorArchitecture", wintypes.WORD), ("wReserved", wintypes.WORD)]
_fields_ = [("dwOemId", wintypes.DWORD), ("s", _S)]
_anonymous_ = ("s",)
_fields_ = [("u", _U),
_fields_ = [
("u", _U),
("dwPageSize", wintypes.DWORD),
("lpMinimumApplicationAddress", wintypes.LPVOID),
("lpMaximumApplicationAddress", wintypes.LPVOID),
@@ -289,22 +293,22 @@ class PlatformUtils:
("dwProcessorType", wintypes.DWORD),
("dwAllocationGranularity", wintypes.DWORD),
("wProcessorLevel", wintypes.WORD),
("wProcessorRevision", wintypes.WORD)]
("wProcessorRevision", wintypes.WORD),
]
_anonymous_ = ("u",)
sys_info = SYSTEM_INFO()
ctypes.windll.kernel32.GetNativeSystemInfo(ctypes.byref(sys_info))
arch_map = {
9: 'AMD64',
5: 'ARM',
12: 'arm64',
6: 'Intel Itanium-based',
0: 'i386',
9: "AMD64",
5: "ARM",
12: "arm64",
6: "Intel Itanium-based",
0: "i386",
}
return arch_map.get(sys_info.wProcessorArchitecture, f'Unknown ({sys_info.wProcessorArchitecture})')
return arch_map.get(sys_info.wProcessorArchitecture, f"Unknown ({sys_info.wProcessorArchitecture})")
@staticmethod
def get_dotnet_version() -> DotnetVersion:
@@ -314,14 +318,14 @@ class PlatformUtils:
try:
result = subprocess.run(["dotnet", "--list-runtimes"], capture_output=True, check=True)
available_version_cmd_output = []
for line in result.stdout.decode('utf-8').split('\n'):
if line.startswith('Microsoft.NETCore.App'):
version_cmd_output = line.split(' ')[1]
for line in result.stdout.decode("utf-8").split("\n"):
if line.startswith("Microsoft.NETCore.App"):
version_cmd_output = line.split(" ")[1]
available_version_cmd_output.append(version_cmd_output)
if not available_version_cmd_output:
raise MultilspyException("dotnet not found on the system")
# Check for supported versions in order of preference (latest first)
for version_cmd_output in available_version_cmd_output:
if version_cmd_output.startswith("8"):
@@ -335,9 +339,11 @@ class PlatformUtils:
for version_cmd_output in available_version_cmd_output:
if version_cmd_output.startswith("4"):
return DotnetVersion.V4
# If no supported version found, raise exception with all available versions
raise MultilspyException(f"No supported dotnet version found. Available versions: {', '.join(available_version_cmd_output)}. Supported versions: 4, 6, 7, 8")
raise MultilspyException(
f"No supported dotnet version found. Available versions: {', '.join(available_version_cmd_output)}. Supported versions: 4, 6, 7, 8"
)
except (FileNotFoundError, subprocess.CalledProcessError):
try:
result = subprocess.run(["mono", "--version"], capture_output=True, check=True)
@@ -346,7 +352,6 @@ class PlatformUtils:
raise MultilspyException("dotnet or mono not found on the system")
class SymbolUtils:
@staticmethod
def symbol_tree_contains_name(roots: list[UnifiedSymbolInformation], name: str) -> bool:
+6 -5
View File
@@ -3,14 +3,15 @@ This module provides type-helpers used across multilspy implementation
"""
import inspect
from typing import Callable, TypeVar, Type
from collections.abc import Callable
from typing import TypeVar
R = TypeVar("R", bound=object)
def ensure_all_methods_implemented(
source_cls: Type[object],
) -> Callable[[Type[R]], Type[R]]:
source_cls: type[object],
) -> Callable[[type[R]], type[R]]:
"""
A decorator to ensure that all methods of source_cls class are implemented in the decorated class.
"""
@@ -24,4 +25,4 @@ def ensure_all_methods_implemented(
return target_cls
return check_all_methods_implemented
return check_all_methods_implemented
+2 -2
View File
@@ -6,11 +6,11 @@ import pytest
from sensai.util.logging import configure
from multilspy.language_server import SyncLanguageServer
from solidlsp.multilspy_config import Language, MultilspyConfig
from solidlsp.multilspy_logger import MultilspyLogger
from serena.constants import USE_SOLID_LSP
from serena.util.file_system import GitignoreParser
from solidlsp.ls import SolidLanguageServer
from solidlsp.multilspy_config import Language, MultilspyConfig
from solidlsp.multilspy_logger import MultilspyLogger
configure(level=logging.DEBUG)
+1 -1
View File
@@ -10,8 +10,8 @@ import os
import pytest
from multilspy.language_server import SyncLanguageServer
from solidlsp.multilspy_config import Language
from serena.text_utils import LineType
from solidlsp.multilspy_config import Language
@pytest.mark.python
+1 -1
View File
@@ -5,9 +5,9 @@ from dataclasses import dataclass
import pytest
from solidlsp.multilspy_config import Language
from serena.agent import FindReferencingSymbolsTool, FindSymbolTool, Project, ProjectConfig, SerenaAgent, SerenaConfigBase
from serena.process_isolated_agent import ProcessIsolatedSerenaAgent
from solidlsp.multilspy_config import Language
from test.conftest import get_repo_path
+1 -1
View File
@@ -11,8 +11,8 @@ from typing import Literal
import pytest
from solidlsp.multilspy_config import Language
from serena.symbol import CodeDiff
from solidlsp.multilspy_config import Language
from src.serena.symbol import SymbolManager
from test.conftest import create_ls, get_repo_path