chore(lsprotocol): migrate to 2025.0.0 and cleanup artifacts
The changes align the project with the 2025.0.0 lsprotocol release, removing the old backport and updating type hints in the protocol hooks to use Sequence where appropriate. The dist-info and packaging metadata for older lsprotocol versions are replaced with the new 2025.0.0 artifacts. - Remove exceptiongroup backport used on Python <3.11 - Use Sequence instead of List in LS protocol hooks - Replace old dist-info with 2025.0.0 metadata
This commit is contained in:
Christoph Brandau
@@ -15,88 +15,34 @@
|
||||
# limitations under the License. #
|
||||
############################################################################
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import inspect
|
||||
import json
|
||||
import logging
|
||||
import sys
|
||||
from concurrent.futures import Future
|
||||
import typing
|
||||
from functools import lru_cache
|
||||
from itertools import zip_longest
|
||||
from typing import (
|
||||
Callable,
|
||||
List,
|
||||
Optional,
|
||||
Type,
|
||||
TypeVar,
|
||||
Union,
|
||||
)
|
||||
|
||||
from lsprotocol import types
|
||||
|
||||
from pygls.capabilities import ServerCapabilitiesBuilder
|
||||
from pygls.lsp import ConfigCallbackType, ShowDocumentCallbackType
|
||||
from lsprotocol.types import (
|
||||
CLIENT_REGISTER_CAPABILITY,
|
||||
CLIENT_UNREGISTER_CAPABILITY,
|
||||
EXIT,
|
||||
INITIALIZE,
|
||||
INITIALIZED,
|
||||
METHOD_TO_TYPES,
|
||||
NOTEBOOK_DOCUMENT_DID_CHANGE,
|
||||
NOTEBOOK_DOCUMENT_DID_CLOSE,
|
||||
NOTEBOOK_DOCUMENT_DID_OPEN,
|
||||
LOG_TRACE,
|
||||
SET_TRACE,
|
||||
SHUTDOWN,
|
||||
TEXT_DOCUMENT_DID_CHANGE,
|
||||
TEXT_DOCUMENT_DID_CLOSE,
|
||||
TEXT_DOCUMENT_DID_OPEN,
|
||||
TEXT_DOCUMENT_PUBLISH_DIAGNOSTICS,
|
||||
WINDOW_LOG_MESSAGE,
|
||||
WINDOW_SHOW_DOCUMENT,
|
||||
WINDOW_SHOW_MESSAGE,
|
||||
WINDOW_WORK_DONE_PROGRESS_CANCEL,
|
||||
WORKSPACE_APPLY_EDIT,
|
||||
WORKSPACE_CONFIGURATION,
|
||||
WORKSPACE_DID_CHANGE_WORKSPACE_FOLDERS,
|
||||
WORKSPACE_EXECUTE_COMMAND,
|
||||
WORKSPACE_SEMANTIC_TOKENS_REFRESH,
|
||||
)
|
||||
from lsprotocol.types import (
|
||||
ApplyWorkspaceEditParams,
|
||||
Diagnostic,
|
||||
DidChangeNotebookDocumentParams,
|
||||
DidChangeTextDocumentParams,
|
||||
DidChangeWorkspaceFoldersParams,
|
||||
DidCloseNotebookDocumentParams,
|
||||
DidCloseTextDocumentParams,
|
||||
DidOpenNotebookDocumentParams,
|
||||
DidOpenTextDocumentParams,
|
||||
ExecuteCommandParams,
|
||||
InitializeParams,
|
||||
InitializeResult,
|
||||
LogMessageParams,
|
||||
LogTraceParams,
|
||||
MessageType,
|
||||
PublishDiagnosticsParams,
|
||||
RegistrationParams,
|
||||
SetTraceParams,
|
||||
ShowDocumentParams,
|
||||
ShowMessageParams,
|
||||
TraceValues,
|
||||
UnregistrationParams,
|
||||
WorkspaceApplyEditResponse,
|
||||
WorkspaceEdit,
|
||||
InitializeResultServerInfoType,
|
||||
WorkspaceConfigurationParams,
|
||||
WorkDoneProgressCancelParams,
|
||||
)
|
||||
from pygls.constants import PARAM_LS
|
||||
from pygls.exceptions import JsonRpcInvalidParams
|
||||
from pygls.protocol.json_rpc import JsonRPCProtocol
|
||||
from pygls.protocol.lsp_meta import LSPMeta
|
||||
from pygls.uris import from_fs_path
|
||||
from pygls.workspace import Workspace
|
||||
|
||||
if typing.TYPE_CHECKING:
|
||||
from collections.abc import Generator
|
||||
from typing import Any, Callable, Optional, Type, TypeVar
|
||||
|
||||
F = TypeVar("F", bound=Callable)
|
||||
from cattrs import Converter
|
||||
|
||||
from pygls.lsp.server import LanguageServer
|
||||
|
||||
F = TypeVar("F", bound=Callable)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -109,7 +55,7 @@ def lsp_method(method_name: str) -> Callable[[F], F]:
|
||||
return decorator
|
||||
|
||||
|
||||
class LanguageServerProtocol(JsonRPCProtocol, metaclass=LSPMeta):
|
||||
class LanguageServerProtocol(JsonRPCProtocol):
|
||||
"""A class that represents language server protocol.
|
||||
|
||||
It contains implementations for generic LSP features.
|
||||
@@ -118,17 +64,19 @@ class LanguageServerProtocol(JsonRPCProtocol, metaclass=LSPMeta):
|
||||
workspace(Workspace): In memory workspace
|
||||
"""
|
||||
|
||||
def __init__(self, server, converter):
|
||||
_server: LanguageServer
|
||||
|
||||
def __init__(self, server: LanguageServer, converter: Converter):
|
||||
super().__init__(server, converter)
|
||||
|
||||
self._workspace: Optional[Workspace] = None
|
||||
self.trace = None
|
||||
self.trace = types.TraceValue.Off
|
||||
|
||||
from pygls.progress import Progress
|
||||
|
||||
self.progress = Progress(self)
|
||||
|
||||
self.server_info = InitializeResultServerInfoType(
|
||||
self.server_info = types.ServerInfo(
|
||||
name=server.name,
|
||||
version=server.version,
|
||||
)
|
||||
@@ -155,40 +103,38 @@ class LanguageServerProtocol(JsonRPCProtocol, metaclass=LSPMeta):
|
||||
return self._workspace
|
||||
|
||||
@lru_cache()
|
||||
def get_message_type(self, method: str) -> Optional[Type]:
|
||||
def get_message_type(self, method: str) -> Type[Any] | None:
|
||||
"""Return LSP type definitions, as provided by `lsprotocol`"""
|
||||
return METHOD_TO_TYPES.get(method, (None,))[0]
|
||||
return types.METHOD_TO_TYPES.get(method, (None,))[0]
|
||||
|
||||
@lru_cache()
|
||||
def get_result_type(self, method: str) -> Optional[Type]:
|
||||
return METHOD_TO_TYPES.get(method, (None, None))[1]
|
||||
def get_result_type(self, method: str) -> Type[Any] | None:
|
||||
return types.METHOD_TO_TYPES.get(method, (None, None))[1]
|
||||
|
||||
def apply_edit(
|
||||
self, edit: WorkspaceEdit, label: Optional[str] = None
|
||||
) -> WorkspaceApplyEditResponse:
|
||||
"""Sends apply edit request to the client."""
|
||||
return self.send_request(
|
||||
WORKSPACE_APPLY_EDIT, ApplyWorkspaceEditParams(edit=edit, label=label)
|
||||
)
|
||||
|
||||
def apply_edit_async(
|
||||
self, edit: WorkspaceEdit, label: Optional[str] = None
|
||||
) -> WorkspaceApplyEditResponse:
|
||||
"""Sends apply edit request to the client. Should be called with `await`"""
|
||||
return self.send_request_async(
|
||||
WORKSPACE_APPLY_EDIT, ApplyWorkspaceEditParams(edit=edit, label=label)
|
||||
)
|
||||
|
||||
@lsp_method(EXIT)
|
||||
def lsp_exit(self, *args) -> None:
|
||||
@lsp_method(types.EXIT)
|
||||
def lsp_exit(self, *args) -> Generator[Any, Any, None]:
|
||||
"""Stops the server process."""
|
||||
if self.transport is not None:
|
||||
self.transport.close()
|
||||
|
||||
sys.exit(0 if self._shutdown else 1)
|
||||
# Ensure that the user handler is called first
|
||||
if (user_handler := self.fm.features.get(types.EXIT)) is not None:
|
||||
yield user_handler, args, None
|
||||
|
||||
@lsp_method(INITIALIZE)
|
||||
def lsp_initialize(self, params: InitializeParams) -> InitializeResult:
|
||||
returncode = 0 if self._shutdown else 1
|
||||
if self.writer is None:
|
||||
sys.exit(returncode)
|
||||
|
||||
res = self.writer.close()
|
||||
if inspect.isawaitable(res):
|
||||
# Only call sys.exit once the close task has completed.
|
||||
fut = asyncio.ensure_future(res)
|
||||
fut.add_done_callback(lambda t: sys.exit(returncode))
|
||||
else:
|
||||
sys.exit(returncode)
|
||||
|
||||
@lsp_method(types.INITIALIZE)
|
||||
def lsp_initialize(
|
||||
self, params: types.InitializeParams
|
||||
) -> Generator[Any, Any, types.InitializeResult]:
|
||||
"""Method that initializes language server.
|
||||
It will compute and return server capabilities based on
|
||||
registered features.
|
||||
@@ -200,19 +146,9 @@ class LanguageServerProtocol(JsonRPCProtocol, metaclass=LSPMeta):
|
||||
text_document_sync_kind = self._server._text_document_sync_kind
|
||||
notebook_document_sync = self._server._notebook_document_sync
|
||||
|
||||
# Initialize server capabilities
|
||||
self.client_capabilities = params.capabilities
|
||||
self.server_capabilities = ServerCapabilitiesBuilder(
|
||||
self.client_capabilities,
|
||||
set({**self.fm.features, **self.fm.builtin_features}.keys()),
|
||||
self.fm.feature_options,
|
||||
list(self.fm.commands.keys()),
|
||||
text_document_sync_kind,
|
||||
notebook_document_sync,
|
||||
).build()
|
||||
logger.debug(
|
||||
"Server capabilities: %s",
|
||||
json.dumps(self.server_capabilities, default=self._serialize_message),
|
||||
position_encoding = ServerCapabilitiesBuilder.choose_position_encoding(
|
||||
self.client_capabilities
|
||||
)
|
||||
|
||||
root_path = params.root_path
|
||||
@@ -220,86 +156,144 @@ class LanguageServerProtocol(JsonRPCProtocol, metaclass=LSPMeta):
|
||||
if root_path is not None and root_uri is None:
|
||||
root_uri = from_fs_path(root_path)
|
||||
|
||||
# Initialize the workspace
|
||||
# Initialize the workspace before yielding to the user's initialize handler
|
||||
workspace_folders = params.workspace_folders or []
|
||||
self._workspace = Workspace(
|
||||
root_uri,
|
||||
text_document_sync_kind,
|
||||
workspace_folders,
|
||||
self.server_capabilities.position_encoding,
|
||||
position_encoding,
|
||||
)
|
||||
|
||||
self.trace = TraceValues.Off
|
||||
if (user_handler := self.fm.features.get(types.INITIALIZE)) is not None:
|
||||
yield user_handler, (params,), None
|
||||
|
||||
return InitializeResult(
|
||||
# Now that the user has had the opportunity to setup additional features, calculate
|
||||
# the server's capabilities
|
||||
self.server_capabilities = ServerCapabilitiesBuilder(
|
||||
self.client_capabilities,
|
||||
set({**self.fm.features, **self.fm.builtin_features}.keys()),
|
||||
self.fm.feature_options,
|
||||
list(self.fm.commands.keys()),
|
||||
text_document_sync_kind,
|
||||
notebook_document_sync,
|
||||
position_encoding,
|
||||
).build()
|
||||
logger.debug(
|
||||
"Server capabilities: %s",
|
||||
json.dumps(self.server_capabilities, default=self._serialize_message),
|
||||
)
|
||||
|
||||
return types.InitializeResult(
|
||||
capabilities=self.server_capabilities,
|
||||
server_info=self.server_info,
|
||||
)
|
||||
|
||||
@lsp_method(INITIALIZED)
|
||||
def lsp_initialized(self, *args) -> None:
|
||||
@lsp_method(types.INITIALIZED)
|
||||
def lsp_initialized(self, *args):
|
||||
"""Notification received when client and server are connected."""
|
||||
pass
|
||||
|
||||
@lsp_method(SHUTDOWN)
|
||||
def lsp_shutdown(self, *args) -> None:
|
||||
if (user_handler := self.fm.features.get(types.INITIALIZED)) is not None:
|
||||
yield user_handler, args, None
|
||||
|
||||
@lsp_method(types.SHUTDOWN)
|
||||
def lsp_shutdown(self, *args) -> Generator[Any, Any, None]:
|
||||
"""Request from client which asks server to shutdown."""
|
||||
for future in self._request_futures.values():
|
||||
future.cancel()
|
||||
|
||||
if (user_handler := self.fm.features.get(types.SHUTDOWN)) is not None:
|
||||
yield user_handler, args, None
|
||||
|
||||
# Don't cancel the future for this request!
|
||||
current_id = self.msg_id
|
||||
|
||||
for msg_id, future in self._request_futures.items():
|
||||
if msg_id != current_id and not future.done():
|
||||
future.cancel()
|
||||
|
||||
self._shutdown = True
|
||||
return None
|
||||
|
||||
@lsp_method(TEXT_DOCUMENT_DID_CHANGE)
|
||||
def lsp_text_document__did_change(
|
||||
self, params: DidChangeTextDocumentParams
|
||||
) -> None:
|
||||
@lsp_method(types.TEXT_DOCUMENT_DID_CHANGE)
|
||||
def lsp_text_document__did_change(self, params: types.DidChangeTextDocumentParams):
|
||||
"""Updates document's content.
|
||||
(Incremental(from server capabilities); not configurable for now)
|
||||
"""
|
||||
for change in params.content_changes:
|
||||
self.workspace.update_text_document(params.text_document, change)
|
||||
|
||||
@lsp_method(TEXT_DOCUMENT_DID_CLOSE)
|
||||
def lsp_text_document__did_close(self, params: DidCloseTextDocumentParams) -> None:
|
||||
if (
|
||||
user_handler := self.fm.features.get(types.TEXT_DOCUMENT_DID_CHANGE)
|
||||
) is not None:
|
||||
yield user_handler, (params,), None
|
||||
|
||||
@lsp_method(types.TEXT_DOCUMENT_DID_CLOSE)
|
||||
def lsp_text_document__did_close(self, params: types.DidCloseTextDocumentParams):
|
||||
"""Removes document from workspace."""
|
||||
self.workspace.remove_text_document(params.text_document.uri)
|
||||
|
||||
@lsp_method(TEXT_DOCUMENT_DID_OPEN)
|
||||
def lsp_text_document__did_open(self, params: DidOpenTextDocumentParams) -> None:
|
||||
if (
|
||||
user_handler := self.fm.features.get(types.TEXT_DOCUMENT_DID_CLOSE)
|
||||
) is not None:
|
||||
yield user_handler, (params,), None
|
||||
|
||||
@lsp_method(types.TEXT_DOCUMENT_DID_OPEN)
|
||||
def lsp_text_document__did_open(self, params: types.DidOpenTextDocumentParams):
|
||||
"""Puts document to the workspace."""
|
||||
self.workspace.put_text_document(params.text_document)
|
||||
|
||||
@lsp_method(NOTEBOOK_DOCUMENT_DID_OPEN)
|
||||
if (
|
||||
user_handler := self.fm.features.get(types.TEXT_DOCUMENT_DID_OPEN)
|
||||
) is not None:
|
||||
yield user_handler, (params,), None
|
||||
|
||||
@lsp_method(types.NOTEBOOK_DOCUMENT_DID_OPEN)
|
||||
def lsp_notebook_document__did_open(
|
||||
self, params: DidOpenNotebookDocumentParams
|
||||
) -> None:
|
||||
self, params: types.DidOpenNotebookDocumentParams
|
||||
):
|
||||
"""Put a notebook document into the workspace"""
|
||||
self.workspace.put_notebook_document(params)
|
||||
|
||||
@lsp_method(NOTEBOOK_DOCUMENT_DID_CHANGE)
|
||||
if (
|
||||
user_handler := self.fm.features.get(types.NOTEBOOK_DOCUMENT_DID_OPEN)
|
||||
) is not None:
|
||||
yield user_handler, (params,), None
|
||||
|
||||
@lsp_method(types.NOTEBOOK_DOCUMENT_DID_CHANGE)
|
||||
def lsp_notebook_document__did_change(
|
||||
self, params: DidChangeNotebookDocumentParams
|
||||
) -> None:
|
||||
self, params: types.DidChangeNotebookDocumentParams
|
||||
):
|
||||
"""Update a notebook's contents"""
|
||||
self.workspace.update_notebook_document(params)
|
||||
|
||||
@lsp_method(NOTEBOOK_DOCUMENT_DID_CLOSE)
|
||||
if (
|
||||
user_handler := self.fm.features.get(types.NOTEBOOK_DOCUMENT_DID_CHANGE)
|
||||
) is not None:
|
||||
yield user_handler, (params,), None
|
||||
|
||||
@lsp_method(types.NOTEBOOK_DOCUMENT_DID_CLOSE)
|
||||
def lsp_notebook_document__did_close(
|
||||
self, params: DidCloseNotebookDocumentParams
|
||||
) -> None:
|
||||
self, params: types.DidCloseNotebookDocumentParams
|
||||
):
|
||||
"""Remove a notebook document from the workspace."""
|
||||
self.workspace.remove_notebook_document(params)
|
||||
|
||||
@lsp_method(SET_TRACE)
|
||||
def lsp_set_trace(self, params: SetTraceParams) -> None:
|
||||
if (
|
||||
user_handler := self.fm.features.get(types.NOTEBOOK_DOCUMENT_DID_CLOSE)
|
||||
) is not None:
|
||||
yield user_handler, (params,), None
|
||||
|
||||
@lsp_method(types.SET_TRACE)
|
||||
def lsp_set_trace(self, params: types.SetTraceParams) -> Generator[Any, Any, None]:
|
||||
"""Changes server trace value."""
|
||||
self.trace = params.value
|
||||
|
||||
@lsp_method(WORKSPACE_DID_CHANGE_WORKSPACE_FOLDERS)
|
||||
if (user_handler := self.fm.features.get(types.SET_TRACE)) is not None:
|
||||
yield user_handler, (params,), None
|
||||
|
||||
@lsp_method(types.WORKSPACE_DID_CHANGE_WORKSPACE_FOLDERS)
|
||||
def lsp_workspace__did_change_workspace_folders(
|
||||
self, params: DidChangeWorkspaceFoldersParams
|
||||
) -> None:
|
||||
self, params: types.DidChangeWorkspaceFoldersParams
|
||||
):
|
||||
"""Adds/Removes folders from the workspace."""
|
||||
logger.info("Workspace folders changed: %s", params)
|
||||
|
||||
@@ -312,18 +306,35 @@ class LanguageServerProtocol(JsonRPCProtocol, metaclass=LSPMeta):
|
||||
if f_remove:
|
||||
self.workspace.remove_folder(f_remove.uri)
|
||||
|
||||
@lsp_method(WORKSPACE_EXECUTE_COMMAND)
|
||||
def lsp_workspace__execute_command(
|
||||
self, params: ExecuteCommandParams, msg_id: str
|
||||
) -> None:
|
||||
"""Executes commands with passed arguments and returns a value."""
|
||||
cmd_handler = self.fm.commands[params.command]
|
||||
self._execute_request(msg_id, cmd_handler, params.arguments)
|
||||
if (
|
||||
user_handler := self.fm.features.get(
|
||||
types.WORKSPACE_DID_CHANGE_WORKSPACE_FOLDERS
|
||||
)
|
||||
) is not None:
|
||||
yield user_handler, (params,), None
|
||||
|
||||
@lsp_method(WINDOW_WORK_DONE_PROGRESS_CANCEL)
|
||||
def lsp_work_done_progress_cancel(
|
||||
self, params: WorkDoneProgressCancelParams
|
||||
) -> None:
|
||||
@lsp_method(types.WORKSPACE_EXECUTE_COMMAND)
|
||||
def lsp_workspace__execute_command(
|
||||
self, params: types.ExecuteCommandParams
|
||||
) -> Generator[Any, Any, Any]:
|
||||
"""Executes commands with passed arguments and returns a value."""
|
||||
|
||||
if (handler := self.fm.commands.get(params.command, None)) is None:
|
||||
raise JsonRpcInvalidParams.of(
|
||||
ValueError(f"Command name {params.command!r} is not defined")
|
||||
)
|
||||
|
||||
try:
|
||||
args, kwargs = _prepare_command_arguments(handler, params, self._converter)
|
||||
except Exception as exc:
|
||||
raise JsonRpcInvalidParams.of(exc)
|
||||
|
||||
# Call the user's command handler.
|
||||
result = yield handler, args, kwargs
|
||||
return result
|
||||
|
||||
@lsp_method(types.WINDOW_WORK_DONE_PROGRESS_CANCEL)
|
||||
def lsp_work_done_progress_cancel(self, params: types.WorkDoneProgressCancelParams):
|
||||
"""Received a progress cancellation from client."""
|
||||
future = self.progress.tokens.get(params.token)
|
||||
if future is None:
|
||||
@@ -333,237 +344,85 @@ class LanguageServerProtocol(JsonRPCProtocol, metaclass=LSPMeta):
|
||||
else:
|
||||
future.cancel()
|
||||
|
||||
def get_configuration(
|
||||
self,
|
||||
params: WorkspaceConfigurationParams,
|
||||
callback: Optional[ConfigCallbackType] = None,
|
||||
) -> Future:
|
||||
"""Sends configuration request to the client.
|
||||
if (
|
||||
user_handler := self.fm.features.get(types.WINDOW_WORK_DONE_PROGRESS_CANCEL)
|
||||
) is not None:
|
||||
yield user_handler, (params,), None
|
||||
|
||||
Args:
|
||||
params(WorkspaceConfigurationParams): WorkspaceConfigurationParams from lsp specs
|
||||
callback(callable): Callabe which will be called after
|
||||
response from the client is received
|
||||
Returns:
|
||||
concurrent.futures.Future object that will be resolved once a
|
||||
response has been received
|
||||
"""
|
||||
return self.send_request(WORKSPACE_CONFIGURATION, params, callback)
|
||||
|
||||
def get_configuration_async(
|
||||
self, params: WorkspaceConfigurationParams
|
||||
) -> asyncio.Future:
|
||||
"""Calls `get_configuration` method but designed to use with coroutines
|
||||
def _prepare_command_arguments(
|
||||
handler: Callable[..., Any],
|
||||
params: types.ExecuteCommandParams,
|
||||
converter: Converter,
|
||||
) -> tuple[tuple[Any, ...], dict[str, Any]]:
|
||||
"""Prepare the arguments to pass to the command handler."""
|
||||
|
||||
Args:
|
||||
params(WorkspaceConfigurationParams): WorkspaceConfigurationParams from lsp specs
|
||||
Returns:
|
||||
asyncio.Future that can be awaited
|
||||
"""
|
||||
return asyncio.wrap_future(self.get_configuration(params))
|
||||
if params.arguments is None:
|
||||
return tuple(), {}
|
||||
|
||||
def log_trace(self, message: str, verbose: Optional[str] = None) -> None:
|
||||
"""Sends trace notification to the client."""
|
||||
if self.trace == TraceValues.Off:
|
||||
return
|
||||
# Import this here to not introduce an import cycle at the module level
|
||||
from pygls.lsp.server import LanguageServer
|
||||
|
||||
params = LogTraceParams(message=message)
|
||||
if verbose and self.trace == TraceValues.Verbose:
|
||||
params.verbose = verbose
|
||||
param_vals = iter(params.arguments)
|
||||
param_defs, annotations = _get_handler_params_annotations(handler)
|
||||
|
||||
self.notify(LOG_TRACE, params)
|
||||
args: list[Any] = []
|
||||
kwargs: dict[str, Any] = {}
|
||||
|
||||
def _publish_diagnostics_deprecator(
|
||||
self,
|
||||
params_or_uri: Union[str, PublishDiagnosticsParams],
|
||||
diagnostics: Optional[List[Diagnostic]],
|
||||
version: Optional[int],
|
||||
**kwargs,
|
||||
) -> PublishDiagnosticsParams:
|
||||
if isinstance(params_or_uri, str):
|
||||
message = "DEPRECATION: "
|
||||
"`publish_diagnostics("
|
||||
"self, doc_uri: str, diagnostics: List[Diagnostic], version: Optional[int] = None)`"
|
||||
"will be replaced with `publish_diagnostics(self, params: PublishDiagnosticsParams)`"
|
||||
logging.warning(message)
|
||||
# param_defs is an OrderedDict so *in theory* at least we don't have to
|
||||
# worry about argument order.
|
||||
found_ls = False
|
||||
for idx, (name, param) in enumerate(param_defs.items()):
|
||||
ptype = annotations.get(name, None)
|
||||
|
||||
# We don't need to provide the injected server instance here.
|
||||
# The @server.command decorator will have already handled it.
|
||||
if idx == 0:
|
||||
if name == PARAM_LS:
|
||||
found_ls = True
|
||||
continue
|
||||
|
||||
if (ptype is not None) and issubclass(ptype, LanguageServer):
|
||||
found_ls = True
|
||||
continue
|
||||
|
||||
if param.kind == inspect.Parameter.VAR_POSITIONAL: # i.e. *args
|
||||
# consume the remaining values
|
||||
args.extend(param_vals)
|
||||
|
||||
params = self._construct_publish_diagnostic_type(
|
||||
params_or_uri, diagnostics, version, **kwargs
|
||||
)
|
||||
else:
|
||||
params = params_or_uri
|
||||
return params
|
||||
try:
|
||||
value = converter.structure(next(param_vals), ptype)
|
||||
except StopIteration as exc:
|
||||
raise TypeError(
|
||||
f"Expected {len(param_defs) - found_ls} arguments, "
|
||||
f"got {len(params.arguments)}"
|
||||
) from exc
|
||||
|
||||
def _construct_publish_diagnostic_type(
|
||||
self,
|
||||
uri: str,
|
||||
diagnostics: Optional[List[Diagnostic]],
|
||||
version: Optional[int],
|
||||
**kwargs,
|
||||
) -> PublishDiagnosticsParams:
|
||||
if diagnostics is None:
|
||||
diagnostics = []
|
||||
args.append(value)
|
||||
|
||||
args = {
|
||||
**{"uri": uri, "diagnostics": diagnostics, "version": version},
|
||||
**kwargs,
|
||||
}
|
||||
|
||||
params = PublishDiagnosticsParams(**args) # type:ignore
|
||||
return params
|
||||
|
||||
def publish_diagnostics(
|
||||
self,
|
||||
params_or_uri: Union[str, PublishDiagnosticsParams],
|
||||
diagnostics: Optional[List[Diagnostic]] = None,
|
||||
version: Optional[int] = None,
|
||||
**kwargs,
|
||||
):
|
||||
"""Sends diagnostic notification to the client.
|
||||
|
||||
.. deprecated:: 1.0.1
|
||||
|
||||
Passing ``(uri, diagnostics, version)`` as arguments is deprecated.
|
||||
Pass an instance of :class:`~lsprotocol.types.PublishDiagnosticParams`
|
||||
instead.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
params_or_uri
|
||||
The :class:`~lsprotocol.types.PublishDiagnosticParams` to send to the client.
|
||||
|
||||
diagnostics
|
||||
*Deprecated*. The diagnostics to publish
|
||||
|
||||
version
|
||||
*Deprecated*: The version number
|
||||
"""
|
||||
params = self._publish_diagnostics_deprecator(
|
||||
params_or_uri, diagnostics, version, **kwargs
|
||||
)
|
||||
self.notify(TEXT_DOCUMENT_PUBLISH_DIAGNOSTICS, params)
|
||||
|
||||
def register_capability(
|
||||
self, params: RegistrationParams, callback: Optional[Callable[[], None]] = None
|
||||
) -> Future:
|
||||
"""Register a new capability on the client.
|
||||
|
||||
Args:
|
||||
params(RegistrationParams): RegistrationParams from lsp specs
|
||||
callback(callable): Callabe which will be called after
|
||||
response from the client is received
|
||||
Returns:
|
||||
concurrent.futures.Future object that will be resolved once a
|
||||
response has been received
|
||||
"""
|
||||
return self.send_request(CLIENT_REGISTER_CAPABILITY, params, callback)
|
||||
|
||||
def register_capability_async(self, params: RegistrationParams) -> asyncio.Future:
|
||||
"""Register a new capability on the client.
|
||||
|
||||
Args:
|
||||
params(RegistrationParams): RegistrationParams from lsp specs
|
||||
|
||||
Returns:
|
||||
asyncio.Future object that will be resolved once a
|
||||
response has been received
|
||||
"""
|
||||
return asyncio.wrap_future(self.register_capability(params, None))
|
||||
|
||||
def semantic_tokens_refresh(
|
||||
self, callback: Optional[Callable[[], None]] = None
|
||||
) -> Future:
|
||||
"""Requesting a refresh of all semantic tokens.
|
||||
|
||||
Args:
|
||||
callback(callable): Callabe which will be called after
|
||||
response from the client is received
|
||||
|
||||
Returns:
|
||||
concurrent.futures.Future object that will be resolved once a
|
||||
response has been received
|
||||
"""
|
||||
return self.send_request(WORKSPACE_SEMANTIC_TOKENS_REFRESH, callback=callback)
|
||||
|
||||
def semantic_tokens_refresh_async(self) -> asyncio.Future:
|
||||
"""Requesting a refresh of all semantic tokens.
|
||||
|
||||
Returns:
|
||||
asyncio.Future object that will be resolved once a
|
||||
response has been received
|
||||
"""
|
||||
return asyncio.wrap_future(self.semantic_tokens_refresh(None))
|
||||
|
||||
def show_document(
|
||||
self,
|
||||
params: ShowDocumentParams,
|
||||
callback: Optional[ShowDocumentCallbackType] = None,
|
||||
) -> Future:
|
||||
"""Display a particular document in the user interface.
|
||||
|
||||
Args:
|
||||
params(ShowDocumentParams): ShowDocumentParams from lsp specs
|
||||
callback(callable): Callabe which will be called after
|
||||
response from the client is received
|
||||
|
||||
Returns:
|
||||
concurrent.futures.Future object that will be resolved once a
|
||||
response has been received
|
||||
"""
|
||||
return self.send_request(WINDOW_SHOW_DOCUMENT, params, callback)
|
||||
|
||||
def show_document_async(self, params: ShowDocumentParams) -> asyncio.Future:
|
||||
"""Display a particular document in the user interface.
|
||||
|
||||
Args:
|
||||
params(ShowDocumentParams): ShowDocumentParams from lsp specs
|
||||
|
||||
Returns:
|
||||
asyncio.Future object that will be resolved once a
|
||||
response has been received
|
||||
"""
|
||||
return asyncio.wrap_future(self.show_document(params, None))
|
||||
|
||||
def show_message(self, message, msg_type=MessageType.Info):
|
||||
"""Sends message to the client to display message."""
|
||||
self.notify(
|
||||
WINDOW_SHOW_MESSAGE, ShowMessageParams(type=msg_type, message=message)
|
||||
# did we consume all the values?
|
||||
if len(list(param_vals)) > 0:
|
||||
raise TypeError(
|
||||
f"Expected {len(param_defs) - found_ls} arguments, "
|
||||
f"got {len(params.arguments)}"
|
||||
)
|
||||
|
||||
def show_message_log(self, message, msg_type=MessageType.Log):
|
||||
"""Sends message to the client's output channel."""
|
||||
self.notify(
|
||||
WINDOW_LOG_MESSAGE, LogMessageParams(type=msg_type, message=message)
|
||||
)
|
||||
return tuple(args), kwargs
|
||||
|
||||
def unregister_capability(
|
||||
self,
|
||||
params: UnregistrationParams,
|
||||
callback: Optional[Callable[[], None]] = None,
|
||||
) -> Future:
|
||||
"""Unregister a new capability on the client.
|
||||
|
||||
Args:
|
||||
params(UnregistrationParams): UnregistrationParams from lsp specs
|
||||
callback(callable): Callabe which will be called after
|
||||
response from the client is received
|
||||
Returns:
|
||||
concurrent.futures.Future object that will be resolved once a
|
||||
response has been received
|
||||
"""
|
||||
return self.send_request(CLIENT_UNREGISTER_CAPABILITY, params, callback)
|
||||
def _get_handler_params_annotations(handler: Callable[..., Any]):
|
||||
"""Return the parameters and corresponding type annotations for the given handler
|
||||
function."""
|
||||
|
||||
def unregister_capability_async(
|
||||
self, params: UnregistrationParams
|
||||
) -> asyncio.Future:
|
||||
"""Unregister a new capability on the client.
|
||||
# If the user's handler requests the language server instance, the real function
|
||||
# is wrapped inside whatever `functools.partial()` returns.
|
||||
if hasattr(handler, "func"):
|
||||
annotations = typing.get_type_hints(handler.func)
|
||||
params = inspect.signature(handler.func).parameters
|
||||
|
||||
Args:
|
||||
params(UnregistrationParams): UnregistrationParams from lsp specs
|
||||
callback(callable): Callabe which will be called after
|
||||
response from the client is received
|
||||
Returns:
|
||||
asyncio.Future object that will be resolved once a
|
||||
response has been received
|
||||
"""
|
||||
return asyncio.wrap_future(self.unregister_capability(params, None))
|
||||
else:
|
||||
annotations = typing.get_type_hints(handler)
|
||||
params = inspect.signature(handler).parameters
|
||||
|
||||
return params, annotations
|
||||
|
||||
Reference in New Issue
Block a user