tiennm99/serena
1
0
Fork 0
mirror of https://github.com/tiennm99/serena.git synced 2026-09-05 08:17:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

Initial implementation of fully synchronous language server (SolidLanguageServer)

Browse Source 
with working PyRight implementation
This commit is contained in:
Dominik Jain
2025-06-21 14:13:20 +02:00
parent 2f607ff2b0
commit de466ef6a9
4 changed files with 374 additions and 385 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/serena/agent.py
+16 -6
View File
@@ -49,6 +49,7 @@ from serena.util.general import load_yaml, save_yaml
from serena.util.inspection import determine_programming_language_composition, iter_subclasses
from serena.util.shell import execute_shell_command
from serena.util.thread import ExecutionResult, execute_with_timeout
from solidlsp.ls import SolidLanguageServer
if TYPE_CHECKING:
from serena.gui_log_viewer import GuiLogViewerHandler
@@ -666,12 +667,21 @@ def create_ls_for_project(
)
ls_logger = MultilspyLogger(log_level=log_level)
log.info(f"Creating language server instance for {project_instance.project_root}.")
return SyncLanguageServer.create(
multilspy_config,
ls_logger,
project_instance.project_root,
timeout=ls_timeout,
)
use_solid_ls = True
if use_solid_ls:
return SolidLanguageServer.create(
multilspy_config,
ls_logger,
project_instance.project_root,
timeout=ls_timeout,
)
else:
return SyncLanguageServer.create(
multilspy_config,
ls_logger,
project_instance.project_root,
timeout=ls_timeout,
)
@click.command()
src/solidlsp/ls.py
+130 -133
View File
@@ -24,7 +24,6 @@ from multilspy.lsp_protocol_handler.lsp_constants import LSPConstants
from multilspy.lsp_protocol_handler.lsp_types import Definition, DefinitionParams, LocationLink, SymbolKind, InitializeParams
from multilspy.lsp_protocol_handler.server import (
Error,
LanguageServerHandler,
ProcessLaunchInfo,
StringDict,
)
@@ -33,9 +32,10 @@ from multilspy.multilspy_exceptions import MultilspyException
from multilspy.multilspy_logger import MultilspyLogger
from multilspy.multilspy_utils import FileUtils, PathUtils, TextUtils
from serena.text_utils import MatchedConsecutiveLines, search_files
from solidlsp.ls_handler import SolidLanguageServerHandler
class LanguageServer:
class SolidLanguageServer:
"""
The LanguageServer class provides a language agnostic interface to the Language Server Protocol.
It is used to communicate with Language Servers of different programming languages.
@@ -50,7 +50,7 @@ class LanguageServer:
return dirname.startswith('.')
@classmethod
def create(cls, config: MultilspyConfig, logger: MultilspyLogger, repository_root_path: str) -> "LanguageServer":
def create(cls, config: MultilspyConfig, logger: MultilspyLogger, repository_root_path: str, timeout) -> "SolidLanguageServer":
"""
Creates a language specific LanguageServer instance based on the given configuration, and appropriate settings for the programming language.
@@ -67,7 +67,7 @@ class LanguageServer:
PyrightServer,
)
return PyrightServer(config, logger, repository_root_path)
return SolidPyrightServer(config, logger, repository_root_path)
# It used to be jedi, but pyright is a bit faster, and also more actively maintained
# Keeping the previous code for reference
# from multilspy.language_servers.jedi_language_server.jedi_server import (
@@ -147,11 +147,6 @@ class LanguageServer:
The command must pass appropriate flags to the binary, so that it runs in the stdio mode,
as opposed to HTTP, TCP modes supported by some language servers.
"""
if type(self) == LanguageServer:
raise MultilspyException(
"LanguageServer is an abstract class and cannot be instantiated directly. Use LanguageServer.create method instead."
)
self.logger = logger
self.repository_root_path: str = repository_root_path
self.logger.log(f"Creating language server instance for {repository_root_path=} with {language_id=} and process launch info: {process_launch_info}", logging.DEBUG)
@@ -168,7 +163,7 @@ class LanguageServer:
self.load_cache()
self.server_started = False
self.completions_available = asyncio.Event()
self.completions_available = threading.Event()
if config.trace_lsp_communication:
def logging_fn(source: str, target: str, msg: StringDict | str):
self.logger.log(f"LSP: {source} -> {target}: {str(msg)[:90]}...", self.logger.logger.level)
@@ -179,7 +174,7 @@ class LanguageServer:
# cmd is obtained from the child classes, which provide the language specific command to start the language server
# LanguageServerHandler provides the functionality to start the language server and communicate with it
self.logger.log(f"Creating language server instance with {language_id=} and process launch info: {process_launch_info}", logging.DEBUG)
self.server = LanguageServerHandler(
self.server = SolidLanguageServerHandler(
process_launch_info,
logger=logging_fn,
start_independent_lsp_process=config.start_independent_lsp_process,
@@ -201,6 +196,8 @@ class LanguageServer:
processed_patterns
)
self._server_context = None
def get_ignore_spec(self) -> pathspec.PathSpec:
"""Returns the pathspec matcher for the paths that were configured to be ignored through
the multilspy config.
@@ -262,7 +259,7 @@ class LanguageServer:
return False
async def _shutdown(self, timeout: float = 5.0):
def _shutdown(self, timeout: float = 5.0):
"""
A robust shutdown process designed to terminate cleanly on all platforms, including Windows,
by explicitly closing all I/O pipes.
@@ -276,27 +273,19 @@ class LanguageServer:
reader_tasks = list(self.server.tasks.values())
# --- Main Shutdown Logic ---
# Stage 1: Graceful Termination Request
# Send LSP shutdown and close stdin to signal no more input.
try:
# Stage 1: Graceful Termination Request
# Send LSP shutdown and close stdin to signal no more input.
try:
await asyncio.wait_for(self.server.shutdown(), timeout=2.0)
if process.stdin and not process.stdin.is_closing():
process.stdin.close()
except Exception:
pass # Ignore errors here, we are proceeding to terminate anyway.
self.server.shutdown()
if process.stdin and not process.stdin.is_closing():
process.stdin.close()
except Exception:
pass # Ignore errors here, we are proceeding to terminate anyway.
# Stage 2: Terminate and Concurrently Drain stdout/stderr
process.terminate()
# Wait for the process to exit AND for the output pipes to be drained.
# The reader tasks will exit when they hit EOF.
await asyncio.wait_for(
asyncio.gather(process.wait(), *reader_tasks),
timeout=timeout - 2.0
)
self.logger.log("Process terminated and output pipes drained.", logging.INFO)
# Stage 2: Terminate and Concurrently Drain stdout/stderr
process.terminate()
"""
except asyncio.TimeoutError:
# Stage 3: Forceful Kill
self.logger.log("Graceful termination failed. Forcefully killing process...", logging.WARNING)
@@ -307,7 +296,6 @@ class LanguageServer:
await process.wait()
except Exception as e:
self.logger.log(f"Error during forceful kill: {e}", logging.ERROR)
except Exception as e:
self.logger.log(f"An unexpected error occurred during shutdown logic: {e}", logging.ERROR)
@@ -334,18 +322,16 @@ class LanguageServer:
# 3. Null out the process object in the handler.
self.server.process = None
self.logger.log("Shutdown sequence fully finished.", logging.DEBUG)
"""
@asynccontextmanager
async def start_server(self) -> AsyncIterator["LanguageServer"]:
"""
Starts the Language Server and yields the LanguageServer instance.
"""
@contextmanager
def start_server(self) -> Iterator["SolidLanguageServer"]:
self.start()
yield self
self._shutdown()
def _start_server(self) -> None:
self.server_started = True
try:
yield self
finally:
self.server_started = False
await self._shutdown()
@contextmanager
def open_file(self, relative_file_path: str) -> Iterator[LSPFileBuffer]:
@@ -404,7 +390,7 @@ class LanguageServer:
self, relative_file_path: str, line: int, column: int, text_to_be_inserted: str
) -> multilspy_types.Position:
"""
Insert text at the given line and column in the given file and return
Insert text at the given line and column in the given file and return
the updated cursor position after inserting the text.
:param relative_file_path: The relative path of the file to open.
@@ -486,10 +472,10 @@ class LanguageServer:
)
return deleted_text
async def _send_definition_request(self, definition_params: DefinitionParams) -> Union[Definition, List[LocationLink], None]:
return await self.server.send.definition(definition_params)
def _send_definition_request(self, definition_params: DefinitionParams) -> Union[Definition, List[LocationLink], None]:
return self.server.send.definition(definition_params)
async def request_definition(
def request_definition(
self, relative_file_path: str, line: int, column: int
) -> List[multilspy_types.Location]:
"""
@@ -523,7 +509,7 @@ class LanguageServer:
LSPConstants.CHARACTER: column,
},
})
response = await self._send_definition_request(definition_params)
response = self._send_definition_request(definition_params)
ret: List[multilspy_types.Location] = []
if isinstance(response, list):
@@ -573,8 +559,8 @@ class LanguageServer:
return ret
# Some LS cause problems with this, so the call is isolated from the rest to allow overriding in subclasses
async def _send_references_request(self, relative_file_path: str, line: int, column: int) -> List[lsp_types.Location] | None:
return await self.server.send.references(
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))},
"position": {"line": line, "character": column},
@@ -582,7 +568,7 @@ class LanguageServer:
}
)
async def request_references(
def request_references(
self, relative_file_path: str, line: int, column: int
) -> List[multilspy_types.Location]:
"""
@@ -607,7 +593,7 @@ class LanguageServer:
with self.open_file(relative_file_path):
try:
response = await self._send_references_request(relative_file_path, line=line, column=column)
response = self._send_references_request(relative_file_path, line=line, column=column)
except Exception as e:
# Catch LSP internal error (-32603) and raise a more informative exception
if isinstance(e, Error) and getattr(e, 'code', None) == -32603:
@@ -640,7 +626,7 @@ class LanguageServer:
return ret
async def request_references_with_content(
def request_references_with_content(
self, relative_file_path: str, line: int, column: int, context_lines_before: int = 0, context_lines_after: int = 0
) -> List[MatchedConsecutiveLines]:
"""
@@ -654,7 +640,7 @@ class LanguageServer:
:return: A list of MatchedConsecutiveLines objects, one for each reference.
"""
references = await self.request_references(relative_file_path, line, column)
references = self.request_references(relative_file_path, line, column)
return [self.retrieve_content_around_line(ref["relativePath"], ref["range"]["start"]["line"], context_lines_before, context_lines_after) for ref in references]
def retrieve_full_file_content(self, relative_file_path: str) -> str:
@@ -679,8 +665,7 @@ class LanguageServer:
file_contents = file_data.contents
return MatchedConsecutiveLines.from_file_contents(file_contents, line=line, context_lines_before=context_lines_before, context_lines_after=context_lines_after, source_file_path=relative_file_path)
async def request_completions(
def request_completions(
self, relative_file_path: str, line: int, column: int, allow_incomplete: bool = False
) -> List[multilspy_types.CompletionItem]:
"""
@@ -706,10 +691,10 @@ class LanguageServer:
num_retries = 0
while response is None or (response["isIncomplete"] and num_retries < 30):
await self.completions_available.wait()
self.completions_available.wait()
response: Union[
List[LSPTypes.CompletionItem], LSPTypes.CompletionList, None
] = await self.server.send.completion(completion_params)
] = self.server.send.completion(completion_params)
if isinstance(response, list):
response = {"items": response, "isIncomplete": False}
num_retries += 1
@@ -774,7 +759,7 @@ class LanguageServer:
for json_repr in set([json.dumps(item, sort_keys=True) for item in completions_list])
]
async def request_document_symbols(self, relative_file_path: str, include_body: bool = False) -> Tuple[List[multilspy_types.UnifiedSymbolInformation], List[multilspy_types.UnifiedSymbolInformation]]:
def request_document_symbols(self, relative_file_path: str, include_body: bool = False) -> Tuple[List[multilspy_types.UnifiedSymbolInformation], List[multilspy_types.UnifiedSymbolInformation]]:
"""
Raise a [textDocument/documentSymbol](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#textDocument_documentSymbol) request to the Language Server
to find symbols in the given file. Wait for the response and return the result.
@@ -805,7 +790,7 @@ class LanguageServer:
self.logger.log(f"No cache hit for symbols with {include_body=} in {relative_file_path}", logging.DEBUG)
self.logger.log(f"Requesting document symbols for {relative_file_path} from the Language Server", logging.DEBUG)
response = await self.server.send.document_symbol(
response = self.server.send.document_symbol(
{
"textDocument": {
"uri": pathlib.Path(os.path.join(self.repository_root_path, relative_file_path)).as_uri()
@@ -892,9 +877,9 @@ class LanguageServer:
self._cache_has_changed = True
return result
async def request_full_symbol_tree(self, within_relative_path: str | None = None, include_body: bool = False) -> List[multilspy_types.UnifiedSymbolInformation]:
def request_full_symbol_tree(self, within_relative_path: str | None = None, include_body: bool = False) -> List[multilspy_types.UnifiedSymbolInformation]:
"""
Will go through all files in the project or within a relative path and build a tree of symbols.
Will go through all files in the project or within a relative path and build a tree of symbols.
Note: this may be slow the first time it is called, especially if `within_relative_path` is not used to restrict the search.
For each file, a symbol of kind File (2) will be created. For directories, a symbol of kind Package (4) will be created.
@@ -921,11 +906,11 @@ class LanguageServer:
self.logger.log(f"You passed a file explicitly, but it is ignored. This is probably an error. File: {within_relative_path}", logging.ERROR)
return []
else:
_, root_nodes = await self.request_document_symbols(within_relative_path, include_body=include_body)
_, root_nodes = self.request_document_symbols(within_relative_path, include_body=include_body)
return root_nodes
# Helper function to recursively process directories
async def process_directory(rel_dir_path: str) -> List[multilspy_types.UnifiedSymbolInformation]:
def process_directory(rel_dir_path: str) -> List[multilspy_types.UnifiedSymbolInformation]:
abs_dir_path = self.repository_root_path if rel_dir_path == "." else os.path.join(self.repository_root_path, rel_dir_path)
abs_dir_path = os.path.realpath(abs_dir_path)
@@ -961,13 +946,13 @@ class LanguageServer:
continue
if os.path.isdir(contained_dir_or_file_abs_path):
child_symbols = await process_directory(contained_dir_or_file_rel_path)
child_symbols = process_directory(contained_dir_or_file_rel_path)
package_symbol["children"].extend(child_symbols)
for child in child_symbols:
child["parent"] = package_symbol
elif os.path.isfile(contained_dir_or_file_abs_path):
_, file_root_nodes = await self.request_document_symbols(contained_dir_or_file_rel_path, include_body=include_body)
_, file_root_nodes = self.request_document_symbols(contained_dir_or_file_rel_path, include_body=include_body)
# Create file symbol, link with children
file_rel_path = str(Path(contained_dir_or_file_abs_path).resolve().relative_to(self.repository_root_path))
@@ -1013,7 +998,7 @@ class LanguageServer:
# Start from the root or the specified directory
start_rel_path = within_relative_path or "."
return await process_directory(start_rel_path)
return process_directory(start_rel_path)
@staticmethod
def _get_range_from_file_content(file_content: str) -> multilspy_types.Range:
@@ -1028,14 +1013,14 @@ class LanguageServer:
end=multilspy_types.Position(line=end_line, character=end_column)
)
async def request_dir_overview(self, relative_dir_path: str) -> dict[str, list[tuple[str, multilspy_types.SymbolKind, int, int]]]:
def request_dir_overview(self, relative_dir_path: str) -> dict[str, list[tuple[str, multilspy_types.SymbolKind, int, int]]]:
"""
An overview of the given directory.
Maps relative paths of all contained files to info about top-level symbols in the file
(name, kind, line, column).
"""
symbol_tree = await self.request_full_symbol_tree(relative_dir_path)
symbol_tree = self.request_full_symbol_tree(relative_dir_path)
# Initialize result dictionary
result: dict[str, list[tuple[str, multilspy_types.SymbolKind, int, int]]] = defaultdict(list)
@@ -1062,12 +1047,12 @@ class LanguageServer:
process_symbol(root)
return result
async def request_document_overview(self, relative_file_path: str) -> list[tuple[str, multilspy_types.SymbolKind, int, int]]:
def request_document_overview(self, relative_file_path: str) -> list[tuple[str, multilspy_types.SymbolKind, int, int]]:
"""
An overview of the given file.
Returns the list of tuples (name, kind, line, column) of all top-level symbols in the file.
"""
_, document_roots = await self.request_document_symbols(relative_file_path)
_, document_roots = self.request_document_symbols(relative_file_path)
result = []
for root in document_roots:
try:
@@ -1083,7 +1068,7 @@ class LanguageServer:
) from e
return result
async def request_overview(self, within_relative_path: str) -> dict[str, list[tuple[str, multilspy_types.SymbolKind, int, int]]]:
def request_overview(self, within_relative_path: str) -> dict[str, list[tuple[str, multilspy_types.SymbolKind, int, int]]]:
"""
An overview of all symbols in the given file or directory.
@@ -1095,12 +1080,12 @@ class LanguageServer:
raise FileNotFoundError(f"File or directory not found: {abs_path}")
if abs_path.is_file():
symbols_overview = await self.request_document_overview(within_relative_path)
symbols_overview = self.request_document_overview(within_relative_path)
return {within_relative_path: symbols_overview}
else:
return await self.request_dir_overview(within_relative_path)
return self.request_dir_overview(within_relative_path)
async def request_hover(self, relative_file_path: str, line: int, column: int) -> Union[multilspy_types.Hover, None]:
def request_hover(self, relative_file_path: str, line: int, column: int) -> Union[multilspy_types.Hover, None]:
"""
Raise a [textDocument/hover](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#textDocument_hover) request to the Language Server
to find the hover information at the given line and column in the given file. Wait for the response and return the result.
@@ -1112,7 +1097,7 @@ class LanguageServer:
:return None
"""
with self.open_file(relative_file_path):
response = await self.server.send.hover(
response = self.server.send.hover(
{
"textDocument": {
"uri": pathlib.Path(os.path.join(self.repository_root_path, relative_file_path)).as_uri()
@@ -1154,7 +1139,7 @@ class LanguageServer:
symbol_body = symbol_body[symbol_start_column:]
return symbol_body
async def request_parsed_files(self) -> list[str]:
def request_parsed_files(self) -> list[str]:
"""Retrieves relative paths of all files analyzed by the Language Server."""
if not self.server_started:
self.logger.log(
@@ -1171,7 +1156,7 @@ class LanguageServer:
rel_file_paths.append(rel_file_path)
return rel_file_paths
async def search_files_for_pattern(
def search_files_for_pattern(
self,
pattern: re.Pattern | str,
context_lines_before: int = 0,
@@ -1192,7 +1177,7 @@ class LanguageServer:
if isinstance(pattern, str):
pattern = re.compile(pattern)
relative_file_paths = await self.request_parsed_files()
relative_file_paths = self.request_parsed_files()
return search_files(
relative_file_paths,
pattern,
@@ -1203,7 +1188,7 @@ class LanguageServer:
paths_exclude_glob=paths_exclude_glob
)
async def request_referencing_symbols(
def request_referencing_symbols(
self,
relative_file_path: str,
line: int,
@@ -1239,7 +1224,7 @@ class LanguageServer:
raise MultilspyException("Language Server not started")
# First, get all references to the symbol
references = await self.request_references(relative_file_path, line, column)
references = self.request_references(relative_file_path, line, column)
if not references:
return []
@@ -1253,7 +1238,7 @@ class LanguageServer:
with self.open_file(ref_path) as file_data:
# Get the containing symbol for this reference
containing_symbol = await self.request_containing_symbol(
containing_symbol = self.request_containing_symbol(
ref_path, ref_line, ref_col, include_body=include_body
)
if containing_symbol is None:
@@ -1273,7 +1258,7 @@ class LanguageServer:
ref_text = file_data.contents.split("\n")[ref_line]
if "." in ref_text:
containing_symbol_name = ref_text.split(".")[0]
all_symbols, _ = await self.request_document_symbols(ref_path)
all_symbols, _ = self.request_document_symbols(ref_path)
for symbol in all_symbols:
if symbol["name"] == containing_symbol_name and symbol["kind"] == multilspy_types.SymbolKind.Variable:
containing_symbol = copy(symbol)
@@ -1350,7 +1335,7 @@ class LanguageServer:
return result
async def request_containing_symbol(
def request_containing_symbol(
self,
relative_file_path: str,
line: int,
@@ -1397,7 +1382,7 @@ class LanguageServer:
)
return None
symbols, _ = await self.request_document_symbols(relative_file_path)
symbols, _ = self.request_document_symbols(relative_file_path)
# make jedi and pyright api compatible
# the former has no location, the later has no range
@@ -1470,7 +1455,7 @@ class LanguageServer:
else:
return None
async def request_container_of_symbol(self, symbol: multilspy_types.UnifiedSymbolInformation, include_body: bool = False) -> multilspy_types.UnifiedSymbolInformation | None:
def request_container_of_symbol(self, symbol: multilspy_types.UnifiedSymbolInformation, include_body: bool = False) -> multilspy_types.UnifiedSymbolInformation | None:
"""
Finds the container of the given symbol if there is one. If the parent attribute is present, the parent is returned
without further searching.
@@ -1482,7 +1467,7 @@ class LanguageServer:
if "parent" in symbol:
return symbol["parent"]
assert "location" in symbol, f"Symbol {symbol} has no location and no parent attribute"
return await self.request_containing_symbol(
return self.request_containing_symbol(
symbol["location"]["relativePath"],
symbol["location"]["range"]["start"]["line"],
symbol["location"]["range"]["start"]["character"],
@@ -1490,7 +1475,7 @@ class LanguageServer:
include_body=include_body,
)
async def request_defining_symbol(
def request_defining_symbol(
self,
relative_file_path: str,
line: int,
@@ -1517,7 +1502,7 @@ class LanguageServer:
raise MultilspyException("Language Server not started")
# Get the definition location(s)
definitions = await self.request_definition(relative_file_path, line, column)
definitions = self.request_definition(relative_file_path, line, column)
if not definitions:
return None
@@ -1528,7 +1513,7 @@ class LanguageServer:
def_col = definition["range"]["start"]["character"]
# Find the symbol at or containing this location
defining_symbol = await self.request_containing_symbol(
defining_symbol = self.request_containing_symbol(
def_path, def_line, def_col, strict=False, include_body=include_body
)
@@ -1541,19 +1526,19 @@ class LanguageServer:
"""
return Path(self.repository_root_path) / ".serena" / "cache" / self.language_id / "document_symbols_cache_v20-05-25.pkl"
async def index_repository(self, progress_bar: bool = True, save_after_n_files: int = 10) -> None:
def index_repository(self, progress_bar: bool = True, save_after_n_files: int = 10) -> None:
"""Will go through the entire repository and "index" all files, meaning save their symbols to the cache.
:param progress_bar: Whether to show a progress bar while indexing the repository.
:param save_after_n_files: How many files to process before saving a checkpoint of the cache.
"""
parsed_files = await self.request_parsed_files()
parsed_files = self.request_parsed_files()
files_processed = 0
pbar = tqdm.tqdm(parsed_files, disable=not progress_bar)
for relative_file_path in pbar:
pbar.set_description(f"Indexing ({os.path.basename(relative_file_path)})")
await self.request_document_symbols(relative_file_path, include_body=False)
await self.request_document_symbols(relative_file_path, include_body=True)
self.request_document_symbols(relative_file_path, include_body=False)
self.request_document_symbols(relative_file_path, include_body=True)
files_processed += 1
if files_processed % save_after_n_files == 0:
self.save_cache()
@@ -1596,8 +1581,7 @@ class LanguageServer:
logging.ERROR,
)
async def request_workspace_symbol(self, query: str) -> Union[List[multilspy_types.UnifiedSymbolInformation], None]:
def request_workspace_symbol(self, query: str) -> Union[List[multilspy_types.UnifiedSymbolInformation], None]:
"""
Raise a [workspace/symbol](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#workspace_symbol) request to the Language Server
to find symbols across the whole workspace. Wait for the response and return the result.
@@ -1606,7 +1590,7 @@ class LanguageServer:
:return: A list of matching symbols
"""
response = await self.server.send.workspace_symbol({"query": query})
response = self.server.send.workspace_symbol({"query": query})
if response is None:
return None
@@ -1624,8 +1608,25 @@ class LanguageServer:
return ret
def start(self) -> "SolidLanguageServer":
"""
Starts the language server process and connects to it. Call shutdown when ready.
class PyrightServer(LanguageServer):
:return: self for method chaining
"""
self.logger.log(f"Starting language server with language {self.language_server.language} for {self.language_server.repository_root_path}", logging.INFO)
self._server_context = self._start_server()
return self
@property
def language_server(self) -> Self:
return self
def is_running(self) -> bool:
return self.server.is_running()
class SolidPyrightServer(SolidLanguageServer):
"""
Provides Python specific instantiation of the LanguageServer class using Pyright.
Contains various configurations and settings specific to Python.
@@ -1646,7 +1647,7 @@ class PyrightServer(LanguageServer):
)
# Event to signal when initial workspace analysis is complete
self.analysis_complete = asyncio.Event()
self.analysis_complete = threading.Event()
self.found_source_files = False
@override
@@ -1753,8 +1754,7 @@ class PyrightServer(LanguageServer):
return initialize_params
@asynccontextmanager
async def start_server(self) -> AsyncIterator["PyrightServer"]:
def _start_server(self):
"""
Starts the Pyright Language Server and waits for initial workspace analysis to complete.
@@ -1772,13 +1772,13 @@ class PyrightServer(LanguageServer):
```
"""
async def execute_client_command_handler(params):
def execute_client_command_handler(params):
return []
async def do_nothing(params):
def do_nothing(params):
return
async def window_log_message(msg):
def window_log_message(msg):
"""
Monitor Pyright's log messages to detect when initial analysis is complete.
Pyright logs "Found X source files" when it finishes scanning the workspace.
@@ -1794,7 +1794,7 @@ class PyrightServer(LanguageServer):
self.analysis_complete.set()
self.completions_available.set()
async def check_experimental_status(params):
def check_experimental_status(params):
"""
Also listen for experimental/serverStatus as a backup signal
"""
@@ -1814,38 +1814,35 @@ class PyrightServer(LanguageServer):
self.server.on_notification("language/actionableNotification", do_nothing)
self.server.on_notification("experimental/serverStatus", check_experimental_status)
async with super().start_server():
self.logger.log("Starting pyright-langserver server process", logging.INFO)
await self.server.start()
super()._start_server()
self.logger.log("Starting pyright-langserver server process", logging.INFO)
self.server.start()
# Send proper initialization parameters
initialize_params = self._get_initialize_params(self.repository_root_path)
# Send proper initialization parameters
initialize_params = self._get_initialize_params(self.repository_root_path)
self.logger.log(
"Sending initialize request from LSP client to pyright server and awaiting response",
logging.INFO,
)
init_response = await self.server.send.initialize(initialize_params)
self.logger.log(f"Received initialize response from pyright server: {init_response}", logging.INFO)
self.logger.log(
"Sending initialize request from LSP client to pyright server and awaiting response",
logging.INFO,
)
init_response = self.server.send.initialize(initialize_params)
self.logger.log(f"Received initialize response from pyright server: {init_response}", logging.INFO)
# Verify that the server supports our required features
assert "textDocumentSync" in init_response["capabilities"]
assert "completionProvider" in init_response["capabilities"]
assert "definitionProvider" in init_response["capabilities"]
# Verify that the server supports our required features
assert "textDocumentSync" in init_response["capabilities"]
assert "completionProvider" in init_response["capabilities"]
assert "definitionProvider" in init_response["capabilities"]
# Complete the initialization handshake
self.server.notify.initialized({})
# Complete the initialization handshake
self.server.notify.initialized({})
# Wait for Pyright to complete its initial workspace analysis
# This prevents zombie processes by ensuring background tasks finish
self.logger.log("Waiting for Pyright to complete initial workspace analysis...", logging.INFO)
try:
await asyncio.wait_for(self.analysis_complete.wait(), timeout=1.0)
self.logger.log("Pyright initial analysis complete, server ready", logging.INFO)
except asyncio.TimeoutError:
self.logger.log("Timeout waiting for Pyright analysis completion, proceeding anyway", logging.WARNING)
# Fallback: assume analysis is complete after timeout
self.analysis_complete.set()
self.completions_available.set()
yield self
# Wait for Pyright to complete its initial workspace analysis
# This prevents zombie processes by ensuring background tasks finish
self.logger.log("Waiting for Pyright to complete initial workspace analysis...", logging.INFO)
if self.analysis_complete.wait(timeout=5.0):
self.logger.log("Pyright initial analysis complete, server ready", logging.INFO)
else:
self.logger.log("Timeout waiting for Pyright analysis completion, proceeding anyway", logging.WARNING)
# Fallback: assume analysis is complete after timeout
self.analysis_complete.set()
self.completions_available.set()
src/solidlsp/ls_handler.py
+125 -143
View File
@@ -1,25 +1,51 @@
import asyncio
import dataclasses
import json
import logging
import os
import subprocess
import threading
import time
from dataclasses import dataclass
from queue import Queue
from typing import Any, Callable, Dict, Optional
import psutil
from typing import Any, Callable, Dict, List, Optional, Union
from multilspy.lsp_protocol_handler.lsp_requests import LspNotification, LspRequest
from multilspy.lsp_protocol_handler.lsp_requests import LspNotification
from multilspy.lsp_protocol_handler.lsp_types import ErrorCodes
from multilspy.lsp_protocol_handler.server import ProcessLaunchInfo, StringDict, Request, content_length, ENCODING, Error, MessageType, \
from multilspy.lsp_protocol_handler.server import ProcessLaunchInfo, StringDict, content_length, ENCODING, Error, MessageType, \
make_notification, PayloadLike, make_response, make_error_response, make_request, create_message
from multilspy.multilspy_exceptions import MultilspyException
from solidlsp.lsp_request import SolidLspRequest
log = logging.getLogger(__name__)
class LanguageServerHandler:
class Request:
@dataclass
class Result:
payload: Optional[PayloadLike] = None
error: Optional[Error] = None
def is_error(self) -> bool:
return self.error is not None
def __init__(self) -> None:
self._result_queue = Queue()
def on_result(self, params: PayloadLike) -> None:
self._result_queue.put(Request.Result(payload=params))
def on_error(self, err: Error) -> None:
self._result_queue.put(Request.Result(error=err))
def get_result(self) -> Result:
# TODO could add timeout
return self._result_queue.get()
class SolidLanguageServerHandler:
"""
This class provides the implementation of Python client for the Language Server Protocol.
A class that launches the language server and communicates with it
@@ -69,7 +95,7 @@ class LanguageServerHandler:
logger: An optional function that takes two strings (source and destination) and
a payload dictionary, and logs the communication between the client and the server.
"""
self.send = LspRequest(self.send_request)
self.send = SolidLspRequest(self.send_request)
self.notify = LspNotification(self.send_notification)
self.process_launch_info = process_launch_info
@@ -92,15 +118,13 @@ 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.
"""
return self.process is not None and self.process.returncode is None
async def start(self) -> None:
def start(self) -> None:
"""
Starts the language server process and creates a task to continuously read from its stdout to handle communications
from the server to the client
@@ -109,73 +133,47 @@ class LanguageServerHandler:
child_proc_env.update(self.process_launch_info.env)
log.info("Starting language server process via command: %s", self.process_launch_info.cmd)
self.process = await asyncio.create_subprocess_shell(
self.process = subprocess.Popen(
self.process_launch_info.cmd,
stdout=asyncio.subprocess.PIPE,
stdin=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
stdout=subprocess.PIPE,
stdin=subprocess.PIPE,
stderr=subprocess.PIPE,
env=child_proc_env,
cwd=self.process_launch_info.cwd,
start_new_session=self.start_independent_lsp_process,
shell=True
)
# Check if process terminated immediately
if self.process.returncode is not None:
log.error("Language server has already terminated/could not be started")
# Process has already terminated
stderr_data = await self.process.stderr.read()
stderr_data = self.process.stderr.read()
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()
# start threads to read stdout and stderr of the process
threading.Thread(
target=self.run_forever,
name="LSP stdout reader",
daemon=True,
).start()
threading.Thread(
target=self.run_forever_stderr,
name="LSP stderr reader",
daemon=True,
).start()
# Use lock to prevent race conditions on tasks and task_counter during startup
with self._tasks_lock:
self.tasks[self.task_counter] = self.loop.create_task(self.run_forever())
self.task_counter += 1
self.tasks[self.task_counter] = self.loop.create_task(self.run_forever_stderr())
self.task_counter += 1
async def stop(self) -> None:
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 process:
self._cleanup_process(process)
if not process:
return
# Clean up the process
await self._cleanup_process(process)
async def _cancel_pending_tasks(self):
"""Cancel all pending tasks and wait for them to complete or timeout."""
pending_tasks = []
# Use lock to safely access tasks dictionary
with self._tasks_lock:
for task in self.tasks.values():
if not task.done():
task.cancel()
pending_tasks.append(task)
if pending_tasks:
try:
await asyncio.wait_for(asyncio.gather(*pending_tasks, return_exceptions=True), timeout=5.0)
except (asyncio.TimeoutError, Exception):
pass
# Clear tasks dictionary under lock
with self._tasks_lock:
self.tasks = {}
async def _cleanup_process(self, process):
def _cleanup_process(self, process):
"""Clean up a process: close stdin, terminate/kill process, close stdout/stderr."""
# Close stdin first to prevent deadlocks
# See: https://bugs.python.org/issue35539
@@ -183,7 +181,7 @@ class LanguageServerHandler:
# Terminate/kill the process if it's still running
if process.returncode is None:
await self._terminate_or_kill_process(process)
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
@@ -192,9 +190,6 @@ class LanguageServerHandler:
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)
def _safely_close_pipe(self, pipe):
"""Safely close a pipe, ignoring any exceptions."""
if pipe:
@@ -203,14 +198,16 @@ class LanguageServerHandler:
except Exception:
pass
async def _terminate_or_kill_process(self, process):
def _terminate_or_kill_process(self, process):
"""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)
# TODO
"""
# Wait for the process to exit (with timeout)
try:
await asyncio.wait_for(process.wait(), timeout=10)
asyncio.wait_for(process.wait(), timeout=10)
except (asyncio.TimeoutError, Exception):
# If termination failed, forcefully kill the process tree
self._signal_process_tree(process, terminate=False)
@@ -219,6 +216,7 @@ class LanguageServerHandler:
await asyncio.wait_for(process.wait(), timeout=2)
except Exception:
pass
"""
def _signal_process_tree(self, process, terminate=True):
"""Send signal (terminate or kill) to the process and all its children."""
@@ -252,23 +250,25 @@ class LanguageServerHandler:
except Exception:
pass
async def shutdown(self) -> None:
def shutdown(self) -> None:
"""
Perform the shutdown sequence for the client, including sending the shutdown request to the server and notifying it of exit
"""
self._log("Sending shutdown request to server")
await self.send.shutdown()
self.send.shutdown()
self._log("Received shutdown response from server")
self._received_shutdown = True
self._log("Sending exit notification to server")
self.notify.exit()
self._log("Sent exit notification to server")
# TODO
"""
if self.process and self.process.stdout:
self.process.stdout.set_exception(StopLoopException())
# This yields the control to the event loop to allow the exception to be handled
# in the run_forever and run_forever_stderr methods
await asyncio.sleep(0)
"""
def _log(self, message: str | StringDict) -> None:
"""
@@ -277,14 +277,33 @@ class LanguageServerHandler:
if self.logger is not None:
self.logger("client", "logger", message)
async def run_forever(self) -> bool:
@staticmethod
def _read_bytes_from_process(process, stream, num_bytes):
"""Read exactly num_bytes from process stdout"""
if process.poll() is not None:
# Process has terminated, check if we can still read
pass
data = b''
while len(data) < num_bytes:
chunk = stream.read(num_bytes - len(data))
if not chunk:
if process.poll() is not None:
raise EOFError(f"Process terminated. Expected {num_bytes} bytes, got {len(data)}")
# Process still running but no data available yet
time.sleep(0.01) # Small delay
continue
data += chunk
return data
def run_forever(self) -> bool:
"""
Continuously read from the language server process stdout and handle the messages
invoking the registered response and notification handlers
"""
try:
while self.process and self.process.stdout and not self.process.stdout.at_eof():
line = await self.process.stdout.readline()
while self.process and self.process.stdout and self.process.stdout.readable():
line = self.process.stdout.readline()
if not line:
continue
try:
@@ -294,39 +313,35 @@ class LanguageServerHandler:
if num_bytes is None:
continue
while line and line.strip():
line = await self.process.stdout.readline()
line = self.process.stdout.readline()
if not line:
continue
body = await self.process.stdout.readexactly(num_bytes)
body = self._read_bytes_from_process(self.process, self.process.stdout, num_bytes)
# 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._handle_body(body))
self.task_counter += 1
except (BrokenPipeError, ConnectionResetError, StopLoopException):
self._handle_body(body)
except (BrokenPipeError, ConnectionResetError):
pass
return self._received_shutdown
async def run_forever_stderr(self) -> None:
def run_forever_stderr(self) -> None:
"""
Continuously read from the language server process stderr and log the messages
"""
try:
while self.process and self.process.stderr and not self.process.stderr.at_eof():
line = await self.process.stderr.readline()
while self.process and self.process.stderr and self.process.stderr.readable():
line = self.process.stderr.readline()
if not line:
continue
self._log("LSP stderr: " + line.decode(ENCODING, errors='replace'))
except (BrokenPipeError, ConnectionResetError, StopLoopException):
except (BrokenPipeError, ConnectionResetError):
pass
async def _handle_body(self, body: bytes) -> None:
def _handle_body(self, body: bytes) -> None:
"""
Parse the body text received from the language server process and invoke the appropriate handler
"""
try:
await self._receive_payload(json.loads(body))
self._receive_payload(json.loads(body))
except IOError as ex:
self._log(f"malformed {ENCODING}: {ex}")
except UnicodeDecodeError as ex:
@@ -334,7 +349,7 @@ class LanguageServerHandler:
except json.JSONDecodeError as ex:
self._log(f"malformed JSON: {ex}")
async def _receive_payload(self, payload: StringDict) -> None:
def _receive_payload(self, payload: StringDict) -> None:
"""
Determine if the payload received from server is for a request, response, or notification and invoke the appropriate handler
"""
@@ -343,11 +358,11 @@ class LanguageServerHandler:
try:
if "method" in payload:
if "id" in payload:
await self._request_handler(payload)
self._request_handler(payload)
else:
await self._notification_handler(payload)
self._notification_handler(payload)
elif "id" in payload:
await self._response_handler(payload)
self._response_handler(payload)
else:
self._log(f"Unknown payload type: {payload}")
except Exception as err:
@@ -357,33 +372,22 @@ class LanguageServerHandler:
"""
Send notification pertaining to the given method to the server with the given parameters
"""
self._send_payload_sync(make_notification(method, params))
self._send_payload(make_notification(method, params))
def send_response(self, request_id: Any, params: PayloadLike) -> None:
"""
Send response to the given request id to the server with the given parameters
"""
# 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.task_counter += 1
self._send_payload(make_response(request_id, params))
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.task_counter += 1
self._send_payload(make_error_response(request_id, err))
async def send_request(self, method: str, params: Optional[dict] = None) -> PayloadLike:
def send_request(self, method: str, params: Optional[dict] = None) -> PayloadLike:
"""
Send request to the server, register the request id, and wait for the response
"""
@@ -397,39 +401,19 @@ class LanguageServerHandler:
with self._response_handlers_lock:
self._response_handlers[request_id] = request
async with request.cv:
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")
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
self._log(f"Returning non-error result, which is:\n{request.result}")
return request.result
self._send_payload(make_request(method, request_id, params))
self._log(f"Waiting for response to request {method} with params:\n{params}")
result = request.get_result()
def _send_payload_sync(self, payload: StringDict) -> None:
"""
Send the payload to the server by writing to its stdin synchronously
"""
if not self.process or not self.process.stdin:
return
msg = create_message(payload)
if self.logger:
self.logger("client", "server", payload)
self._log(f"Processing result")
if result.is_error():
raise MultilspyException(f"Could not process request {method} with params:\n{params}.\n Language server error: {result.error}") from result.error
self._log(f"Returning non-error result, which is:\n{result.payload}")
return result.payload
# Use lock to prevent concurrent writes to stdin that cause buffer corruption
with self._stdin_lock:
try:
self.process.stdin.writelines(msg)
except (BrokenPipeError, ConnectionResetError, OSError) as e:
# Log the error but don't raise to prevent cascading failures
if self.logger:
self.logger("client", "logger", f"Failed to write to stdin: {e}")
return
async def _send_payload(self, payload: StringDict) -> None:
def _send_payload(self, payload: StringDict) -> None:
"""
Send the payload to the server by writing to its stdin asynchronously.
"""
@@ -442,14 +426,13 @@ class LanguageServerHandler:
with self._stdin_lock:
try:
self.process.stdin.writelines(msg)
await self.process.stdin.drain()
self.process.stdin.flush()
except (BrokenPipeError, ConnectionResetError, OSError) as e:
# Log the error but don't raise to prevent cascading failures
if self.logger:
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
@@ -462,7 +445,7 @@ class LanguageServerHandler:
"""
self.on_notification_handlers[method] = cb
async def _response_handler(self, response: StringDict) -> None:
def _response_handler(self, response: StringDict) -> None:
"""
Handle the response received from the server for a request, using the id to determine the request
"""
@@ -470,14 +453,13 @@ class LanguageServerHandler:
request = self._response_handlers.pop(response["id"])
if "result" in response and "error" not in response:
await request.on_result(response["result"])
request.on_result(response["result"])
elif "result" not in response and "error" in response:
await request.on_error(Error.from_lsp(response["error"]))
request.on_error(Error.from_lsp(response["error"]))
else:
await request.on_error(Error(ErrorCodes.InvalidRequest, ""))
request.on_error(Error(ErrorCodes.InvalidRequest, ""))
async def _request_handler(self, response: StringDict) -> None:
def _request_handler(self, response: StringDict) -> None:
"""
Handle the request received from the server: call the appropriate callback function and return the result
"""
@@ -495,13 +477,13 @@ class LanguageServerHandler:
)
return
try:
self.send_response(request_id, await handler(params))
self.send_response(request_id, handler(params))
except Error as ex:
self.send_error_response(request_id, ex)
except Exception as ex:
self.send_error_response(request_id, Error(ErrorCodes.InternalError, str(ex)))
async def _notification_handler(self, response: StringDict) -> None:
def _notification_handler(self, response: StringDict) -> None:
"""
Handle the notification received from the server: call the appropriate callback function
"""
@@ -512,7 +494,7 @@ class LanguageServerHandler:
self._log(f"unhandled {method}")
return
try:
await handler(params)
handler(params)
except asyncio.CancelledError:
return
except Exception as ex:
src/solidlsp/lsp_request.py
+103 -103
View File
@@ -2,56 +2,56 @@ from typing import List, Union
from multilspy.lsp_protocol_handler import lsp_types
class LspRequest:
class SolidLspRequest:
def __init__(self, send_request):
self.send_request = send_request
async def implementation(
def implementation(
self, params: lsp_types.ImplementationParams
) -> 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."""
return await self.send_request("textDocument/implementation", params)
return self.send_request("textDocument/implementation", params)
async def type_definition(
def type_definition(
self, params: lsp_types.TypeDefinitionParams
) -> 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."""
return await self.send_request("textDocument/typeDefinition", params)
return self.send_request("textDocument/typeDefinition", params)
async def document_color(
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."""
return await self.send_request("textDocument/documentColor", params)
return self.send_request("textDocument/documentColor", params)
async def color_presentation(
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."""
return await self.send_request("textDocument/colorPresentation", params)
return self.send_request("textDocument/colorPresentation", params)
async def folding_range(
def folding_range(
self, params: lsp_types.FoldingRangeParams
) -> Union[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."""
return await self.send_request("textDocument/foldingRange", params)
return self.send_request("textDocument/foldingRange", params)
async def declaration(
def declaration(
self, params: lsp_types.DeclarationParams
) -> Union["lsp_types.Declaration", List["lsp_types.LocationLink"], None]:
"""A request to resolve the type definition locations of a symbol at a given text
@@ -59,129 +59,129 @@ class LspRequest:
(#TextDocumentPositionParams) the response is of type {@link Declaration}
or a typed array of {@link DeclarationLink} or a Thenable that resolves
to such."""
return await self.send_request("textDocument/declaration", params)
return self.send_request("textDocument/declaration", params)
async def selection_range(
def selection_range(
self, params: lsp_types.SelectionRangeParams
) -> Union[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."""
return await self.send_request("textDocument/selectionRange", params)
return self.send_request("textDocument/selectionRange", params)
async def prepare_call_hierarchy(
def prepare_call_hierarchy(
self, params: lsp_types.CallHierarchyPrepareParams
) -> Union[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"""
return await self.send_request("textDocument/prepareCallHierarchy", params)
return self.send_request("textDocument/prepareCallHierarchy", params)
async def incoming_calls(
def incoming_calls(
self, params: lsp_types.CallHierarchyIncomingCallsParams
) -> Union[List["lsp_types.CallHierarchyIncomingCall"], None]:
"""A request to resolve the incoming calls for a given `CallHierarchyItem`.
@since 3.16.0"""
return await self.send_request("callHierarchy/incomingCalls", params)
return self.send_request("callHierarchy/incomingCalls", params)
async def outgoing_calls(
def outgoing_calls(
self, params: lsp_types.CallHierarchyOutgoingCallsParams
) -> Union[List["lsp_types.CallHierarchyOutgoingCall"], None]:
"""A request to resolve the outgoing calls for a given `CallHierarchyItem`.
@since 3.16.0"""
return await self.send_request("callHierarchy/outgoingCalls", params)
return self.send_request("callHierarchy/outgoingCalls", params)
async def semantic_tokens_full(
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)
return self.send_request("textDocument/semanticTokens/full", params)
async def semantic_tokens_delta(
def semantic_tokens_delta(
self, params: lsp_types.SemanticTokensDeltaParams
) -> Union["lsp_types.SemanticTokens", "lsp_types.SemanticTokensDelta", None]:
"""@since 3.16.0"""
return await self.send_request("textDocument/semanticTokens/full/delta", params)
return self.send_request("textDocument/semanticTokens/full/delta", params)
async def semantic_tokens_range(
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)
return self.send_request("textDocument/semanticTokens/range", params)
async def linked_editing_range(
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"""
return await self.send_request("textDocument/linkedEditingRange", params)
return self.send_request("textDocument/linkedEditingRange", params)
async def will_create_files(
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"""
return await self.send_request("workspace/willCreateFiles", params)
return self.send_request("workspace/willCreateFiles", params)
async def will_rename_files(
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"""
return await self.send_request("workspace/willRenameFiles", params)
return self.send_request("workspace/willRenameFiles", params)
async def will_delete_files(
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"""
return await self.send_request("workspace/willDeleteFiles", params)
return self.send_request("workspace/willDeleteFiles", params)
async def moniker(
def moniker(
self, params: lsp_types.MonikerParams
) -> Union[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`."""
return await self.send_request("textDocument/moniker", params)
return self.send_request("textDocument/moniker", params)
async def prepare_type_hierarchy(
def prepare_type_hierarchy(
self, params: lsp_types.TypeHierarchyPrepareParams
) -> Union[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"""
return await self.send_request("textDocument/prepareTypeHierarchy", params)
return self.send_request("textDocument/prepareTypeHierarchy", params)
async def type_hierarchy_supertypes(
def type_hierarchy_supertypes(
self, params: lsp_types.TypeHierarchySupertypesParams
) -> Union[List["lsp_types.TypeHierarchyItem"], None]:
"""A request to resolve the supertypes for a given `TypeHierarchyItem`.
@since 3.17.0"""
return await self.send_request("typeHierarchy/supertypes", params)
return self.send_request("typeHierarchy/supertypes", params)
async def type_hierarchy_subtypes(
def type_hierarchy_subtypes(
self, params: lsp_types.TypeHierarchySubtypesParams
) -> Union[List["lsp_types.TypeHierarchyItem"], None]:
"""A request to resolve the subtypes for a given `TypeHierarchyItem`.
@since 3.17.0"""
return await self.send_request("typeHierarchy/subtypes", params)
return self.send_request("typeHierarchy/subtypes", params)
async def inline_value(
def inline_value(
self, params: lsp_types.InlineValueParams
) -> Union[List["lsp_types.InlineValue"], None]:
"""A request to provide inline values in a document. The request's parameter is of
@@ -189,9 +189,9 @@ class LspRequest:
{@link InlineValue InlineValue[]} or a Thenable that resolves to such.
@since 3.17.0"""
return await self.send_request("textDocument/inlineValue", params)
return self.send_request("textDocument/inlineValue", params)
async def inlay_hint(
def inlay_hint(
self, params: lsp_types.InlayHintParams
) -> Union[List["lsp_types.InlayHint"], None]:
"""A request to provide inlay hints in a document. The request's parameter is of
@@ -199,9 +199,9 @@ class LspRequest:
{@link InlayHint InlayHint[]} or a Thenable that resolves to such.
@since 3.17.0"""
return await self.send_request("textDocument/inlayHint", params)
return self.send_request("textDocument/inlayHint", params)
async def resolve_inlay_hint(
def resolve_inlay_hint(
self, params: lsp_types.InlayHint
) -> "lsp_types.InlayHint":
"""A request to resolve additional properties for an inlay hint.
@@ -209,25 +209,25 @@ class LspRequest:
of type {@link InlayHint} or a Thenable that resolves to such.
@since 3.17.0"""
return await self.send_request("inlayHint/resolve", params)
return self.send_request("inlayHint/resolve", params)
async def text_document_diagnostic(
def text_document_diagnostic(
self, params: lsp_types.DocumentDiagnosticParams
) -> "lsp_types.DocumentDiagnosticReport":
"""The document diagnostic request definition.
@since 3.17.0"""
return await self.send_request("textDocument/diagnostic", params)
return self.send_request("textDocument/diagnostic", params)
async def workspace_diagnostic(
def workspace_diagnostic(
self, params: lsp_types.WorkspaceDiagnosticParams
) -> "lsp_types.WorkspaceDiagnosticReport":
"""The workspace diagnostic request definition.
@since 3.17.0"""
return await self.send_request("workspace/diagnostic", params)
return self.send_request("workspace/diagnostic", params)
async def initialize(
def initialize(
self, params: lsp_types.InitializeParams
) -> "lsp_types.InitializeResult":
"""The initialize request is sent from the client to the server.
@@ -235,16 +235,16 @@ class LspRequest:
The requests parameter is of type {@link InitializeParams}
the response if of type {@link InitializeResult} of a Thenable that
resolves to such."""
return await self.send_request("initialize", params)
return self.send_request("initialize", params)
async def shutdown(self) -> None:
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."""
return await self.send_request("shutdown")
return self.send_request("shutdown")
async def will_save_wait_until(
def will_save_wait_until(
self, params: lsp_types.WillSaveTextDocumentParams
) -> Union[List["lsp_types.TextEdit"], None]:
"""A document will save request is sent from the client to the server before
@@ -253,9 +253,9 @@ class LspRequest:
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."""
return await self.send_request("textDocument/willSaveWaitUntil", params)
return self.send_request("textDocument/willSaveWaitUntil", params)
async def completion(
def completion(
self, params: lsp_types.CompletionParams
) -> Union[List["lsp_types.CompletionItem"], "lsp_types.CompletionList", None]:
"""Request to request completion at a given text document position. The request's
@@ -268,30 +268,30 @@ class LspRequest:
request. However, properties that are needed for the initial sorting and filtering, like `sortText`,
`filterText`, `insertText`, and `textEdit`, must not be changed during resolve.
"""
return await self.send_request("textDocument/completion", params)
return self.send_request("textDocument/completion", params)
async def resolve_completion_item(
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."""
return await self.send_request("completionItem/resolve", params)
return self.send_request("completionItem/resolve", params)
async def hover(
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."""
return await self.send_request("textDocument/hover", params)
return self.send_request("textDocument/hover", params)
async def signature_help(
def signature_help(
self, params: lsp_types.SignatureHelpParams
) -> Union["lsp_types.SignatureHelp", None]:
return await self.send_request("textDocument/signatureHelp", params)
return self.send_request("textDocument/signatureHelp", params)
async def definition(
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
@@ -299,27 +299,27 @@ class LspRequest:
(#TextDocumentPosition) the response is of either type {@link Definition}
or a typed array of {@link DefinitionLink} or a Thenable that resolves
to such."""
return await self.send_request("textDocument/definition", params)
return self.send_request("textDocument/definition", params)
async def references(
def references(
self, params: lsp_types.ReferenceParams
) -> Union[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."""
return await self.send_request("textDocument/references", params)
return self.send_request("textDocument/references", params)
async def document_highlight(
def document_highlight(
self, params: lsp_types.DocumentHighlightParams
) -> Union[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."""
return await self.send_request("textDocument/documentHighlight", params)
return self.send_request("textDocument/documentHighlight", params)
async def document_symbol(
def document_symbol(
self, params: lsp_types.DocumentSymbolParams
) -> Union[
List["lsp_types.SymbolInformation"], List["lsp_types.DocumentSymbol"], None
@@ -328,23 +328,23 @@ class LspRequest:
parameter is of type {@link TextDocumentIdentifier} the
response is of type {@link SymbolInformation SymbolInformation[]} or a Thenable
that resolves to such."""
return await self.send_request("textDocument/documentSymbol", params)
return self.send_request("textDocument/documentSymbol", params)
async def code_action(
def code_action(
self, params: lsp_types.CodeActionParams
) -> Union[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)
return self.send_request("textDocument/codeAction", params)
async def resolve_code_action(
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."""
return await self.send_request("codeAction/resolve", params)
return self.send_request("codeAction/resolve", params)
async def workspace_symbol(
def workspace_symbol(
self, params: lsp_types.WorkspaceSymbolParams
) -> Union[
List["lsp_types.SymbolInformation"], List["lsp_types.WorkspaceSymbol"], None
@@ -358,78 +358,78 @@ class LspRequest:
need to advertise support for WorkspaceSymbols via the client capability
`workspace.symbol.resolveSupport`.
"""
return await self.send_request("workspace/symbol", params)
return self.send_request("workspace/symbol", params)
async def resolve_workspace_symbol(
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"""
return await self.send_request("workspaceSymbol/resolve", params)
return self.send_request("workspaceSymbol/resolve", params)
async def code_lens(
def code_lens(
self, params: lsp_types.CodeLensParams
) -> Union[List["lsp_types.CodeLens"], None]:
"""A request to provide code lens for the given text document."""
return await self.send_request("textDocument/codeLens", params)
return self.send_request("textDocument/codeLens", params)
async def resolve_code_lens(
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)
return self.send_request("codeLens/resolve", params)
async def document_link(
def document_link(
self, params: lsp_types.DocumentLinkParams
) -> Union[List["lsp_types.DocumentLink"], None]:
"""A request to provide document links"""
return await self.send_request("textDocument/documentLink", params)
return self.send_request("textDocument/documentLink", params)
async def resolve_document_link(
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."""
return await self.send_request("documentLink/resolve", params)
return self.send_request("documentLink/resolve", params)
async def formatting(
def formatting(
self, params: lsp_types.DocumentFormattingParams
) -> Union[List["lsp_types.TextEdit"], None]:
"""A request to to format a whole document."""
return await self.send_request("textDocument/formatting", params)
return self.send_request("textDocument/formatting", params)
async def range_formatting(
def range_formatting(
self, params: lsp_types.DocumentRangeFormattingParams
) -> Union[List["lsp_types.TextEdit"], None]:
"""A request to to format a range in a document."""
return await self.send_request("textDocument/rangeFormatting", params)
return self.send_request("textDocument/rangeFormatting", params)
async def on_type_formatting(
def on_type_formatting(
self, params: lsp_types.DocumentOnTypeFormattingParams
) -> Union[List["lsp_types.TextEdit"], None]:
"""A request to format a document on type."""
return await self.send_request("textDocument/onTypeFormatting", params)
return self.send_request("textDocument/onTypeFormatting", params)
async def rename(
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)
return self.send_request("textDocument/rename", params)
async def prepare_rename(
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"""
return await self.send_request("textDocument/prepareRename", params)
return self.send_request("textDocument/prepareRename", params)
async def execute_command(
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."""
return await self.send_request("workspace/executeCommand", params)
return self.send_request("workspace/executeCommand", params)