Merge pull request #370 from oraios/mypy_on_solidlsp

Make mypy run through on solidlsp
This commit is contained in:
Michael Panchenko authored and GitHub committed 2025-11-20 22:47:25 +07:00
commit 4702fbe1ea
49 files changed
+489 -423

No files matched your search

+24 -21
View File
@@ -38,6 +38,27 @@ jobs:
mkdir -p $HOME/.serena/language_servers/static
mkdir -p $HOME/.cache/go-build
mkdir -p $HOME/go/bin
- name: Install uv
shell: bash
run: curl -LsSf https://astral.sh/uv/install.sh | sh
- name: Cache uv virtualenv
id: cache-uv
uses: actions/cache@v3
with:
path: .venv
key: uv-venv-${{ runner.os }}-${{ matrix.python-version }}-${{ hashFiles('uv.lock') }}
- name: Create virtual environment
shell: bash
run: |
if [ ! -d ".venv" ]; then
uv venv
fi
- name: Install dependencies
shell: bash
run: uv pip install -e ".[dev]"
- name: Check formatting
shell: bash
run: uv run poe lint
# Add Go bin directory to PATH for this workflow
# GITHUB_PATH is a special file that GitHub Actions uses to modify PATH
# Writing to this file adds the directory to the PATH for subsequent steps
@@ -330,15 +351,6 @@ jobs:
mv regal.exe "$HOME/bin/"
echo "$HOME/bin" >> $GITHUB_PATH
fi
- name: Install uv
shell: bash
run: curl -LsSf https://astral.sh/uv/install.sh | sh
- name: Cache uv virtualenv
id: cache-uv
uses: actions/cache@v3
with:
path: .venv
key: uv-venv-${{ runner.os }}-${{ matrix.python-version }}-${{ hashFiles('uv.lock') }}
- name: Cache language servers
id: cache-language-servers
uses: actions/cache@v3
@@ -347,18 +359,9 @@ jobs:
key: language-servers-${{ runner.os }}-v1
restore-keys: |
language-servers-${{ runner.os }}-
- name: Create virtual environment
shell: bash
run: |
if [ ! -d ".venv" ]; then
uv venv
fi
- name: Install dependencies
shell: bash
run: uv pip install -e ".[dev]"
- name: Check formatting
shell: bash
run: uv run poe lint
- name: Test with pytest
shell: bash
run: uv run poe test
- name: Type-checking with mypy
shell: bash
run: uv run poe type-check
+1 -1
View File
@@ -128,7 +128,7 @@ _black_format = "black src scripts test"
_ruff_format = "ruff check --fix src scripts test"
lint = ["_black_check", "_ruff_check"]
format = ["_ruff_format", "_black_format"]
_mypy = "mypy src/serena"
_mypy = "mypy src/serena src/solidlsp"
type-check = ["_mypy"]
# docs
_autogen_rst = "python docs/autogen_rst.py"
@@ -16,6 +16,7 @@ from solidlsp.language_servers.common import quote_windows_path
from solidlsp.ls import SolidLanguageServer
from solidlsp.ls_config import LanguageServerConfig
from solidlsp.ls_logger import LanguageServerLogger
from solidlsp.ls_types import SymbolKind, UnifiedSymbolInformation
from solidlsp.lsp_protocol_handler.lsp_types import Definition, DefinitionParams, LocationLink
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.settings import SolidLSPSettings
@@ -485,7 +486,7 @@ class ALLanguageServer(SolidLanguageServer):
return initialize_params
@override
def _start_server(self):
def _start_server(self) -> None:
"""
Starts the AL Language Server process and initializes it.
@@ -495,19 +496,19 @@ class ALLanguageServer(SolidLanguageServer):
"""
# Set up event handlers
def do_nothing(params):
def do_nothing(params: str) -> None:
return
def window_log_message(msg):
def window_log_message(msg: str) -> None:
self.logger.log(f"AL LSP: window/logMessage: {msg}", logging.INFO)
def publish_diagnostics(params):
def publish_diagnostics(params: dict) -> None:
# AL server publishes diagnostics during initialization
uri = params.get("uri", "")
diagnostics = params.get("diagnostics", [])
self.logger.log(f"AL LSP: Diagnostics for {uri}: {len(diagnostics)} issues", logging.DEBUG)
def handle_al_notifications(params):
def handle_al_notifications(params: dict) -> None:
# AL server sends custom notifications during project loading
self.logger.log("AL LSP: Notification received", logging.DEBUG)
@@ -635,8 +636,8 @@ class ALLanguageServer(SolidLanguageServer):
"workspacePath": self.repository_root_path,
"setActiveWorkspace": True,
},
"timeout": 2, # Quick timeout since this is optional
},
timeout=2, # Quick timeout since this is optional
)
self.logger.log(f"Set active workspace result: {result}", logging.DEBUG)
except Exception as e:
@@ -680,7 +681,7 @@ class ALLanguageServer(SolidLanguageServer):
return super().is_ignored_dirname(dirname) or dirname in al_ignore_dirs
@override
def request_full_symbol_tree(self, within_relative_path: str | None = None) -> list[dict]:
def request_full_symbol_tree(self, within_relative_path: str | None = None) -> list[UnifiedSymbolInformation]:
"""
Override to handle AL's requirement of opening files before requesting symbols.
@@ -749,8 +750,9 @@ class ALLanguageServer(SolidLanguageServer):
return []
# Collect all symbols from all files
all_file_symbols = []
all_file_symbols: list[UnifiedSymbolInformation] = []
file_symbol: UnifiedSymbolInformation
for file_path, relative_path in al_files:
try:
# Use our overridden request_document_symbols which handles opening
@@ -761,7 +763,7 @@ class ALLanguageServer(SolidLanguageServer):
# Create a file-level symbol containing the document symbols
file_symbol = {
"name": file_path.stem, # Just the filename without extension
"kind": 1, # File
"kind": SymbolKind.File,
"children": root_syms,
"location": {
"uri": file_path.as_uri(),
@@ -776,7 +778,7 @@ class ALLanguageServer(SolidLanguageServer):
# If we only got all_syms but not root, use all_syms
file_symbol = {
"name": file_path.stem,
"kind": 1, # File
"kind": SymbolKind.File,
"children": all_syms,
"location": {
"uri": file_path.as_uri(),
@@ -795,10 +797,11 @@ class ALLanguageServer(SolidLanguageServer):
self.logger.log(f"AL: Returning symbols from {len(all_file_symbols)} files", logging.DEBUG)
# Group files by directory
directory_structure = {}
directory_structure: dict[str, list] = {}
for file_symbol in all_file_symbols:
rel_path = file_symbol["location"]["relativePath"]
assert rel_path is not None
path_parts = rel_path.split("/")
if len(path_parts) > 1:
@@ -824,7 +827,7 @@ class ALLanguageServer(SolidLanguageServer):
# Create directory symbol
dir_symbol = {
"name": Path(dir_path).name,
"kind": 4, # Package/Directory
"kind": SymbolKind.Package, # Package/Directory
"children": file_symbols,
"location": {
"relativePath": dir_path,
@@ -860,7 +863,7 @@ class ALLanguageServer(SolidLanguageServer):
# Use custom AL command instead of standard LSP
response = self.server.send_request("al/gotodefinition", al_params)
self.logger.log(f"AL gotodefinition response: {response}", logging.DEBUG)
return response
return response # type: ignore[return-value]
except Exception as e:
self.logger.log(f"Failed to use al/gotodefinition, falling back to standard: {e}", logging.WARNING)
# Fallback to standard LSP method if custom command fails
@@ -889,7 +892,7 @@ class ALLanguageServer(SolidLanguageServer):
try:
# Use a very short timeout since this is just a status check
response = self.server.send_request("al/hasProjectClosureLoadedRequest", {}, timeout=1)
response = self.server.send_request("al/hasProjectClosureLoadedRequest", {"timeout": 1})
# Response can be boolean directly, dict with 'loaded' field, or None
if isinstance(response, bool):
return response
@@ -125,27 +125,27 @@ class BashLanguageServer(SolidLanguageServer):
}
],
}
return initialize_params
return initialize_params # type: ignore
def _start_server(self):
def _start_server(self) -> None:
"""
Starts the Bash Language Server, waits for the server to be ready and yields the LanguageServer instance.
"""
def register_capability_handler(params):
def register_capability_handler(params: dict) -> None:
assert "registrations" in params
for registration in params["registrations"]:
if registration["method"] == "workspace/executeCommand":
self.initialize_searcher_command_available.set()
return
def execute_client_command_handler(params):
def execute_client_command_handler(params: dict) -> list:
return []
def do_nothing(params):
def do_nothing(params: dict) -> None:
return
def window_log_message(msg):
def window_log_message(msg: dict) -> None:
self.logger.log(f"LSP: window/logMessage: {msg}", logging.INFO)
# Check for bash-language-server ready signals
message_text = msg.get("message", "")
@@ -6,6 +6,7 @@ import logging
import os
import pathlib
import threading
from typing import Any, cast
from solidlsp.ls import SolidLanguageServer
from solidlsp.ls_config import LanguageServerConfig
@@ -154,9 +155,9 @@ class ClangdLanguageServer(SolidLanguageServer):
],
}
return initialize_params
return cast(InitializeParams, initialize_params)
def _start_server(self):
def _start_server(self) -> None:
"""
Starts the Clangd Language Server, waits for the server to be ready and yields the LanguageServer instance.
@@ -170,7 +171,7 @@ class ClangdLanguageServer(SolidLanguageServer):
# LanguageServer has been shutdown
"""
def register_capability_handler(params):
def register_capability_handler(params: Any) -> None:
assert "registrations" in params
for registration in params["registrations"]:
if registration["method"] == "workspace/executeCommand":
@@ -178,24 +179,24 @@ class ClangdLanguageServer(SolidLanguageServer):
self.resolve_main_method_available.set()
return
def lang_status_handler(params):
def lang_status_handler(params: Any) -> None:
# TODO: Should we wait for
# server -> client: {'jsonrpc': '2.0', 'method': 'language/status', 'params': {'type': 'ProjectStatus', 'message': 'OK'}}
# Before proceeding?
if params["type"] == "ServiceReady" and params["message"] == "ServiceReady":
self.service_ready_event.set()
def execute_client_command_handler(params):
def execute_client_command_handler(params: Any) -> list:
return []
def do_nothing(params):
def do_nothing(params: Any) -> None:
return
def check_experimental_status(params):
def check_experimental_status(params: Any) -> None:
if params["quiescent"] == True:
self.server_ready.set()
def window_log_message(msg):
def window_log_message(msg: dict) -> None:
self.logger.log(f"LSP: window/logMessage: {msg}", logging.INFO)
self.server.on_request("client/registerCapability", register_capability_handler)
@@ -216,7 +217,7 @@ class ClangdLanguageServer(SolidLanguageServer):
logging.INFO,
)
init_response = self.server.send.initialize(initialize_params)
assert init_response["capabilities"]["textDocumentSync"]["change"] == 2
assert init_response["capabilities"]["textDocumentSync"]["change"] == 2 # type: ignore
assert "completionProvider" in init_response["capabilities"]
assert init_response["capabilities"]["completionProvider"] == {
"triggerCharacters": [".", "<", ">", ":", '"', "/", "*"],
+14 -12
View File
@@ -8,6 +8,7 @@ import pathlib
import shutil
import subprocess
import threading
from typing import cast
from solidlsp.ls import SolidLanguageServer
from solidlsp.ls_config import LanguageServerConfig
@@ -25,7 +26,7 @@ def run_command(cmd: list, capture_output: bool = True) -> subprocess.CompletedP
)
def verify_clojure_cli():
def verify_clojure_cli() -> None:
install_msg = "Please install the official Clojure CLI from:\n https://clojure.org/guides/getting_started"
if shutil.which("clojure") is None:
raise FileNotFoundError("`clojure` not found.\n" + install_msg)
@@ -131,7 +132,7 @@ class ClojureLSP(SolidLanguageServer):
def _get_initialize_params(repository_absolute_path: str) -> InitializeParams:
"""Returns the init params for clojure-lsp."""
root_uri = pathlib.Path(repository_absolute_path).as_uri()
return { # type: ignore
result = { # type: ignore
"processId": os.getpid(),
"rootPath": repository_absolute_path,
"rootUri": root_uri,
@@ -159,9 +160,10 @@ class ClojureLSP(SolidLanguageServer):
"trace": "off",
"workspaceFolders": [{"uri": root_uri, "name": os.path.basename(repository_absolute_path)}],
}
return cast(InitializeParams, result)
def _start_server(self):
def register_capability_handler(params):
def _start_server(self) -> None:
def register_capability_handler(params: dict) -> None:
assert "registrations" in params
for registration in params["registrations"]:
if registration["method"] == "workspace/executeCommand":
@@ -169,24 +171,24 @@ class ClojureLSP(SolidLanguageServer):
self.resolve_main_method_available.set()
return
def lang_status_handler(params):
def lang_status_handler(params: dict) -> None:
# TODO: Should we wait for
# server -> client: {'jsonrpc': '2.0', 'method': 'language/status', 'params': {'type': 'ProjectStatus', 'message': 'OK'}}
# Before proceeding?
if params["type"] == "ServiceReady" and params["message"] == "ServiceReady":
self.service_ready_event.set()
def execute_client_command_handler(params):
def execute_client_command_handler(params: dict) -> list:
return []
def do_nothing(params):
def do_nothing(params: dict) -> None:
return
def check_experimental_status(params):
if params["quiescent"] == True:
def check_experimental_status(params: dict) -> None:
if params["quiescent"] is True:
self.server_ready.set()
def window_log_message(msg):
def window_log_message(msg: dict) -> None:
self.logger.log(f"LSP: window/logMessage: {msg}", logging.INFO)
self.server.on_request("client/registerCapability", register_capability_handler)
@@ -208,11 +210,11 @@ class ClojureLSP(SolidLanguageServer):
logging.INFO,
)
init_response = self.server.send.initialize(initialize_params)
assert init_response["capabilities"]["textDocumentSync"]["change"] == 2
assert init_response["capabilities"]["textDocumentSync"]["change"] == 2 # type: ignore
assert "completionProvider" in init_response["capabilities"]
# Clojure-lsp completion provider capabilities are more flexible than other servers'
completion_provider = init_response["capabilities"]["completionProvider"]
assert completion_provider["resolveProvider"] == True
assert completion_provider["resolveProvider"] is True
assert "triggerCharacters" in completion_provider
self.server.notify.initialized({})
# after initialize, Clojure-lsp is ready to serve
+5 -3
View File
@@ -103,7 +103,7 @@ class RuntimeDependencyCollection:
if not PlatformUtils.get_platform_id().is_windows():
import pwd
kwargs["user"] = pwd.getpwuid(os.getuid()).pw_name
kwargs["user"] = pwd.getpwuid(os.getuid()).pw_name # type: ignore
is_windows = platform.system() == "Windows"
if not isinstance(command, str) and not is_windows:
@@ -121,7 +121,7 @@ class RuntimeDependencyCollection:
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
**kwargs,
)
) # type: ignore
if completed_process.returncode != 0:
log.warning("Command '%s' failed with return code %d", command, completed_process.returncode)
log.warning("Command output:\n%s", completed_process.stdout)
@@ -132,7 +132,9 @@ class RuntimeDependencyCollection:
@staticmethod
def _install_from_url(dep: RuntimeDependency, logger: LanguageServerLogger, target_dir: str) -> None:
assert dep.url is not None
if not dep.url:
raise ValueError(f"Dependency {dep.id} has no URL")
if dep.archive_type in ("gz", "binary") and dep.binary_name:
dest = os.path.join(target_dir, dep.binary_name)
FileUtils.download_and_extract_archive(logger, dep.url, dest, dep.archive_type)
@@ -12,6 +12,7 @@ import tarfile
import threading
import urllib.request
import zipfile
from collections.abc import Iterable
from pathlib import Path
from typing import Any, cast
@@ -22,7 +23,7 @@ from solidlsp.ls_config import LanguageServerConfig
from solidlsp.ls_exceptions import SolidLSPException
from solidlsp.ls_logger import LanguageServerLogger
from solidlsp.ls_utils import PathUtils
from solidlsp.lsp_protocol_handler.lsp_types import InitializeParams
from solidlsp.lsp_protocol_handler.lsp_types import InitializeParams, InitializeResult
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.settings import SolidLSPSettings
from solidlsp.util.zip import SafeZipExtractor
@@ -133,7 +134,7 @@ _RUNTIME_DEPENDENCIES = [
]
def breadth_first_file_scan(root_dir):
def breadth_first_file_scan(root_dir: str) -> Iterable[str]:
"""
Perform a breadth-first scan of files in the given directory.
Yields file paths in breadth-first order.
@@ -155,7 +156,7 @@ def breadth_first_file_scan(root_dir):
pass
def find_solution_or_project_file(root_dir) -> str | None:
def find_solution_or_project_file(root_dir: str) -> str | None:
"""
Find the first .sln file in breadth-first order.
If no .sln file is found, look for a .csproj file.
@@ -301,6 +302,7 @@ class CSharpLanguageServer(SolidLanguageServer):
package_version = lang_server_dep.package_version
server_dir = Path(cls.ls_resources_dir(solidlsp_settings)) / f"{package_name}.{package_version}"
assert lang_server_dep.binary_name is not None
server_dll = server_dir / lang_server_dep.binary_name
if server_dll.exists():
@@ -309,6 +311,8 @@ class CSharpLanguageServer(SolidLanguageServer):
# Download and install the language server
logger.log(f"Downloading {package_name} version {package_version}...", logging.INFO)
assert package_version is not None
assert package_name is not None
package_path = cls._download_nuget_package_direct(logger, package_name, package_version, solidlsp_settings)
# Extract and install
@@ -431,6 +435,7 @@ class CSharpLanguageServer(SolidLanguageServer):
# Download .NET 9 runtime using config
dotnet_dir = Path(cls.ls_resources_dir(solidlsp_settings)) / "dotnet-runtime-9.0"
assert dotnet_runtime_dep.binary_name is not None, "Runtime dependency must have a binary_name"
dotnet_exe = dotnet_dir / dotnet_runtime_dep.binary_name
if dotnet_exe.exists():
@@ -534,11 +539,11 @@ class CSharpLanguageServer(SolidLanguageServer):
},
)
def _start_server(self):
def do_nothing(params):
def _start_server(self) -> None:
def do_nothing(params: dict) -> None:
return
def window_log_message(msg):
def window_log_message(msg: dict) -> None:
"""Log messages from the language server."""
message_text = msg.get("message", "")
level = msg.get("type", 4) # Default to Log level
@@ -548,7 +553,7 @@ class CSharpLanguageServer(SolidLanguageServer):
self.logger.log(f"LSP: {message_text}", level_map.get(level, logging.DEBUG))
def handle_progress(params):
def handle_progress(params: dict) -> None:
"""Handle progress notifications from the language server."""
token = params.get("token", "")
value = params.get("value", {})
@@ -582,10 +587,10 @@ class CSharpLanguageServer(SolidLanguageServer):
message = value.get("message", "Operation completed")
self.logger.log(f"Progress [{token}]: {message}", logging.INFO)
def handle_workspace_configuration(params):
def handle_workspace_configuration(params: dict) -> list:
"""Handle workspace/configuration requests from the server."""
items = params.get("items", [])
result = []
result: list[Any] = []
for item in items:
section = item.get("section", "")
@@ -628,17 +633,17 @@ class CSharpLanguageServer(SolidLanguageServer):
return result
def handle_work_done_progress_create(params):
def handle_work_done_progress_create(params: dict) -> None:
"""Handle work done progress create requests."""
# Just acknowledge the request
return
def handle_register_capability(params):
def handle_register_capability(params: dict) -> None:
"""Handle client/registerCapability requests."""
# Just acknowledge the request - we don't need to track these for now
return
def handle_project_needs_restore(params):
def handle_project_needs_restore(params: dict) -> None:
return
# Set up notification handlers
@@ -702,13 +707,13 @@ class CSharpLanguageServer(SolidLanguageServer):
logging.INFO,
)
def _force_pull_diagnostics(self, init_response: dict) -> None:
def _force_pull_diagnostics(self, init_response: dict | InitializeResult) -> None:
"""
Apply the diagnostic capabilities hack.
Forces the server to support pull diagnostics.
"""
capabilities = init_response.get("capabilities", {})
diagnostic_provider = capabilities.get("diagnosticProvider", {})
diagnostic_provider: Any = capabilities.get("diagnosticProvider", {})
# Add the diagnostic capabilities hack
if isinstance(diagnostic_provider, dict):
@@ -1,12 +1,15 @@
import logging
import os
import pathlib
from typing import cast
from solidlsp.ls import SolidLanguageServer
from solidlsp.ls_logger import LanguageServerLogger
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.settings import SolidLSPSettings
from ..ls_config import LanguageServerConfig
from ..lsp_protocol_handler.lsp_types import InitializeParams
from .common import RuntimeDependency, RuntimeDependencyCollection
@@ -15,7 +18,9 @@ class DartLanguageServer(SolidLanguageServer):
Provides Dart specific instantiation of the LanguageServer class. Contains various configurations and settings specific to Dart.
"""
def __init__(self, config, logger, repository_root_path, solidlsp_settings: SolidLSPSettings):
def __init__(
self, config: LanguageServerConfig, logger: LanguageServerLogger, repository_root_path: str, solidlsp_settings: SolidLSPSettings
) -> None:
"""
Creates a DartServer instance. This class is not meant to be instantiated directly. Use LanguageServer.create() instead.
"""
@@ -88,7 +93,7 @@ class DartLanguageServer(SolidLanguageServer):
return f"{dart_executable_path} language-server --client-id multilspy.dart --client-version 1.2"
@staticmethod
def _get_initialize_params(repository_absolute_path: str):
def _get_initialize_params(repository_absolute_path: str) -> InitializeParams:
"""
Returns the initialize params for the Dart Language Server.
"""
@@ -114,23 +119,23 @@ class DartLanguageServer(SolidLanguageServer):
],
}
return initialize_params
return cast(InitializeParams, initialize_params)
def _start_server(self):
def _start_server(self) -> None:
"""
Start the language server and yield when the server is ready.
"""
def execute_client_command_handler(params):
def execute_client_command_handler(params: dict) -> list:
return []
def do_nothing(params):
def do_nothing(params: dict) -> None:
return
def check_experimental_status(params):
def check_experimental_status(params: dict) -> None:
pass
def window_log_message(msg):
def window_log_message(msg: dict) -> None:
self.logger.log(f"LSP: window/logMessage: {msg}", logging.INFO)
self.server.on_request("client/registerCapability", do_nothing)
@@ -149,7 +154,7 @@ class DartLanguageServer(SolidLanguageServer):
"Sending initialize request to dart-language-server",
logging.DEBUG,
)
init_response = self.server.send_request("initialize", initialize_params)
init_response = self.server.send_request("initialize", initialize_params) # type: ignore
self.logger.log(
f"Received initialize response from dart-language-server: {init_response}",
logging.INFO,
+25 -19
View File
@@ -10,14 +10,16 @@ import shutil
import threading
import uuid
from pathlib import PurePath
from typing import cast
from overrides import override
from solidlsp.ls import GenericDocumentSymbol, LSPFileBuffer, SolidLanguageServer
from solidlsp.ls import LSPFileBuffer, SolidLanguageServer
from solidlsp.ls_config import LanguageServerConfig
from solidlsp.ls_logger import LanguageServerLogger
from solidlsp.ls_types import UnifiedSymbolInformation
from solidlsp.ls_utils import FileUtils, PlatformUtils
from solidlsp.lsp_protocol_handler.lsp_types import InitializeParams
from solidlsp.lsp_protocol_handler.lsp_types import DocumentSymbol, InitializeParams, SymbolInformation
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.settings import SolidLSPSettings
@@ -690,29 +692,29 @@ class EclipseJDTLS(SolidLanguageServer):
],
}
initialize_params["initializationOptions"]["workspaceFolders"] = [repo_uri]
initialize_params["initializationOptions"]["workspaceFolders"] = [repo_uri] # type: ignore
bundles = [self.runtime_dependency_paths.intellicode_jar_path]
initialize_params["initializationOptions"]["bundles"] = bundles
initialize_params["initializationOptions"]["settings"]["java"]["configuration"]["runtimes"] = [
initialize_params["initializationOptions"]["bundles"] = bundles # type: ignore
initialize_params["initializationOptions"]["settings"]["java"]["configuration"]["runtimes"] = [ # type: ignore
{"name": "JavaSE-21", "path": self.runtime_dependency_paths.jre_home_path, "default": True}
]
for runtime in initialize_params["initializationOptions"]["settings"]["java"]["configuration"]["runtimes"]:
for runtime in initialize_params["initializationOptions"]["settings"]["java"]["configuration"]["runtimes"]: # type: ignore
assert "name" in runtime
assert "path" in runtime
assert os.path.exists(runtime["path"]), f"Runtime required for eclipse_jdtls at path {runtime['path']} does not exist"
gradle_settings = initialize_params["initializationOptions"]["settings"]["java"]["import"]["gradle"]
gradle_settings = initialize_params["initializationOptions"]["settings"]["java"]["import"]["gradle"] # type: ignore
gradle_settings["home"] = self.runtime_dependency_paths.gradle_path
gradle_settings["java"]["home"] = self.runtime_dependency_paths.jre_path
return initialize_params
return cast(InitializeParams, initialize_params)
def _start_server(self):
def _start_server(self) -> None:
"""
Starts the Eclipse JDTLS Language Server
"""
def register_capability_handler(params):
def register_capability_handler(params: dict) -> None:
assert "registrations" in params
for registration in params["registrations"]:
if registration["method"] == "textDocument/completion":
@@ -730,22 +732,22 @@ class EclipseJDTLS(SolidLanguageServer):
self.intellicode_enable_command_available.set()
return
def lang_status_handler(params):
def lang_status_handler(params: dict) -> None:
# TODO: Should we wait for
# server -> client: {'jsonrpc': '2.0', 'method': 'language/status', 'params': {'type': 'ProjectStatus', 'message': 'OK'}}
# Before proceeding?
if params["type"] == "ServiceReady" and params["message"] == "ServiceReady":
self.service_ready_event.set()
def execute_client_command_handler(params):
def execute_client_command_handler(params: dict) -> list:
assert params["command"] == "_java.reloadBundles.command"
assert params["arguments"] == []
return []
def window_log_message(msg):
def window_log_message(msg: dict) -> None:
self.logger.log(f"LSP: window/logMessage: {msg}", logging.INFO)
def do_nothing(params):
def do_nothing(params: dict) -> None:
return
self.server.on_request("client/registerCapability", register_capability_handler)
@@ -765,13 +767,13 @@ class EclipseJDTLS(SolidLanguageServer):
logging.INFO,
)
init_response = self.server.send.initialize(initialize_params)
assert init_response["capabilities"]["textDocumentSync"]["change"] == 2
assert init_response["capabilities"]["textDocumentSync"]["change"] == 2 # type: ignore
assert "completionProvider" not in init_response["capabilities"]
assert "executeCommandProvider" not in init_response["capabilities"]
self.server.notify.initialized({})
self.server.notify.workspace_did_change_configuration({"settings": initialize_params["initializationOptions"]["settings"]})
self.server.notify.workspace_did_change_configuration({"settings": initialize_params["initializationOptions"]["settings"]}) # type: ignore
self.intellicode_enable_command_available.wait()
@@ -788,19 +790,23 @@ class EclipseJDTLS(SolidLanguageServer):
# TODO: Add comments about why we wait here, and how this can be optimized
self.service_ready_event.wait()
def _request_document_symbols(self, relative_file_path: str, file_data: LSPFileBuffer | None) -> list[GenericDocumentSymbol]:
def _request_document_symbols(
self, relative_file_path: str, file_data: LSPFileBuffer | None
) -> list[SymbolInformation] | list[DocumentSymbol] | None:
result = super()._request_document_symbols(relative_file_path, file_data=file_data)
if result is None:
return None
# JDTLS sometimes returns symbol names with type information to handle overloads,
# e.g. "myMethod(int) <T>", but we want overloads to be handled via overload_idx,
# which requires the name to be just "myMethod".
def fix_name(symbol: GenericDocumentSymbol):
def fix_name(symbol: SymbolInformation | DocumentSymbol | UnifiedSymbolInformation) -> None:
if "(" in symbol["name"]:
symbol["name"] = symbol["name"][: symbol["name"].index("(")]
children = symbol.get("children")
if children:
for child in children:
for child in children: # type: ignore
fix_name(child)
for root_symbol in result:
@@ -4,6 +4,7 @@ import pathlib
import stat
import subprocess
import threading
from typing import Any, cast
from overrides import override
@@ -11,6 +12,7 @@ from solidlsp.ls import SolidLanguageServer
from solidlsp.ls_config import LanguageServerConfig
from solidlsp.ls_logger import LanguageServerLogger
from solidlsp.ls_utils import FileUtils, PlatformId, PlatformUtils
from solidlsp.lsp_protocol_handler import lsp_types
from solidlsp.lsp_protocol_handler.lsp_types import InitializeParams
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.settings import SolidLSPSettings
@@ -23,6 +25,10 @@ class ElixirTools(SolidLanguageServer):
Provides Elixir specific instantiation of the LanguageServer class using Next LS from elixir-tools.
"""
@override
def _get_wait_time_for_cross_file_referencing(self) -> float:
return 10.0 # Elixir projects need a lot of time to compile and index before cross-file references work
@override
def is_ignored_dirname(self, dirname: str) -> bool:
# For Elixir projects, we should ignore:
@@ -42,7 +48,8 @@ class ElixirTools(SolidLanguageServer):
return super().is_ignored_path(relative_path, ignore_unsupported_files)
def _is_next_ls_internal_file(self, abs_path: str) -> bool:
@staticmethod
def _is_next_ls_internal_file(abs_path: str) -> bool:
"""Check if an absolute path is a Next LS internal file that should be ignored."""
return any(
pattern in abs_path
@@ -55,7 +62,7 @@ class ElixirTools(SolidLanguageServer):
)
@override
def _send_references_request(self, relative_file_path: str, line: int, column: int):
def _send_references_request(self, relative_file_path: str, line: int, column: int) -> list[lsp_types.Location] | None:
"""Override to filter out Next LS internal files from references."""
from solidlsp.ls_utils import PathUtils
@@ -78,7 +85,7 @@ class ElixirTools(SolidLanguageServer):
return filtered_response
@classmethod
def _get_elixir_version(cls):
def _get_elixir_version(cls) -> str | None:
"""Get the installed Elixir version or None if not found."""
try:
result = subprocess.run(["elixir", "--version"], capture_output=True, text=True, check=False)
@@ -164,10 +171,12 @@ class ElixirTools(SolidLanguageServer):
dependency = runtime_deps[platform_id]
executable_path = os.path.join(next_ls_dir, "nextls")
assert dependency.binary_name is not None
binary_path = os.path.join(next_ls_dir, dependency.binary_name)
if not os.path.exists(executable_path):
logger.log(f"Downloading Next LS binary from {dependency.url}", logging.INFO)
assert dependency.url is not None
FileUtils.download_file(logger, dependency.url, binary_path)
# Make the binary executable on Unix-like systems
@@ -267,15 +276,15 @@ class ElixirTools(SolidLanguageServer):
"workspaceFolders": [{"uri": root_uri, "name": os.path.basename(repository_absolute_path)}],
}
return initialize_params
return cast(InitializeParams, initialize_params)
def _start_server(self):
def _start_server(self) -> None:
"""Start Next LS server process"""
def register_capability_handler(params):
def register_capability_handler(params: Any) -> None:
return
def window_log_message(msg):
def window_log_message(msg: Any) -> None:
"""Handle window/logMessage notifications from Next LS"""
message_text = msg.get("message", "")
self.logger.log(f"LSP: window/logMessage: {message_text}", logging.INFO)
@@ -286,10 +295,10 @@ class ElixirTools(SolidLanguageServer):
self.logger.log("Next LS runtime is ready based on official log message", logging.INFO)
self.server_ready.set()
def do_nothing(params):
def do_nothing(params: Any) -> None:
return
def check_server_ready(params):
def check_server_ready(params: Any) -> None:
"""
Handle $/progress notifications from Next LS.
Keep as fallback for error detection, but primary readiness detection
@@ -304,7 +313,7 @@ class ElixirTools(SolidLanguageServer):
self.logger.log("Next LS initialization progress completed", logging.INFO)
# Note: We don't set server_ready here - we wait for the log message
def work_done_progress(params):
def work_done_progress(params: Any) -> None:
"""
Handle $/workDoneProgress notifications from Next LS.
Keep for completeness but primary readiness detection is via window/logMessage.
@@ -143,17 +143,17 @@ class ElmLanguageServer(SolidLanguageServer):
}
],
}
return initialize_params
return initialize_params # type: ignore[return-value]
def _start_server(self):
def _start_server(self) -> None:
"""
Starts the Elm Language Server, waits for the server to be ready and yields the LanguageServer instance.
"""
def do_nothing(params):
def do_nothing(params: dict) -> None:
return
def window_log_message(msg):
def window_log_message(msg: dict) -> None:
self.logger.log(f"LSP: window/logMessage: {msg}", logging.INFO)
self.server.on_notification("window/logMessage", window_log_message)
@@ -61,7 +61,7 @@ class ErlangLanguageServer(SolidLanguageServer):
return False
@classmethod
def _get_erlang_version(cls):
def _get_erlang_version(cls) -> str | None:
"""Get the installed Erlang/OTP version or None if not found."""
try:
result = subprocess.run(["erl", "-version"], check=False, capture_output=True, text=True, timeout=10)
@@ -80,13 +80,13 @@ class ErlangLanguageServer(SolidLanguageServer):
except (subprocess.SubprocessError, FileNotFoundError):
return False
def _start_server(self):
def _start_server(self) -> None:
"""Start Erlang LS server process with proper initialization waiting."""
def register_capability_handler(params):
def register_capability_handler(params: dict) -> None:
return
def window_log_message(msg):
def window_log_message(msg: dict) -> None:
"""Handle window/logMessage notifications from Erlang LS"""
message_text = msg.get("message", "")
self.logger.log(f"LSP: window/logMessage: {message_text}", logging.INFO)
@@ -109,10 +109,10 @@ class ErlangLanguageServer(SolidLanguageServer):
self.server_ready.set()
break
def do_nothing(params):
def do_nothing(params: dict) -> None:
return
def check_server_ready(params):
def check_server_ready(params: dict) -> None:
"""Handle $/progress notifications from Erlang LS as fallback."""
value = params.get("value", {})
@@ -154,7 +154,7 @@ class ErlangLanguageServer(SolidLanguageServer):
}
self.logger.log("Sending initialize request to Erlang LS", logging.INFO)
init_response = self.server.send.initialize(initialize_params)
init_response = self.server.send.initialize(initialize_params) # type: ignore[arg-type]
# Verify server capabilities
if "capabilities" in init_response:
@@ -125,7 +125,7 @@ class FortranLanguageServer(SolidLanguageServer):
# Create modified symbol with corrected selectionRange
corrected_symbol = symbol.copy()
corrected_symbol["selectionRange"] = new_sel_range
corrected_symbol["selectionRange"] = new_sel_range # type: ignore[typeddict-item]
self.logger.log(
f"Fixed fortls selectionRange for {identifier_name}: char {start_char} -> {identifier_start}",
@@ -174,7 +174,7 @@ class FortranLanguageServer(SolidLanguageServer):
return DocumentSymbols(fixed_root_symbols)
@staticmethod
def _check_fortls_installation():
def _check_fortls_installation() -> str:
"""Check if fortls is available."""
fortls_path = shutil.which("fortls")
if fortls_path is None:
@@ -255,18 +255,18 @@ class FortranLanguageServer(SolidLanguageServer):
}
],
}
return initialize_params
return initialize_params # type: ignore[return-value]
def _start_server(self):
def _start_server(self) -> None:
"""Start Fortran Language Server process."""
def window_log_message(msg):
def window_log_message(msg: dict) -> None:
self.logger.log(f"Fortran LSP: window/logMessage: {msg}", logging.INFO)
def do_nothing(params):
def do_nothing(params: dict) -> None:
return
def register_capability_handler(params):
def register_capability_handler(params: dict) -> None:
return
# Register LSP message handlers
+9 -8
View File
@@ -3,6 +3,7 @@ import os
import pathlib
import subprocess
import threading
from typing import cast
from overrides import override
@@ -46,7 +47,7 @@ class Gopls(SolidLanguageServer):
return SolidLanguageServer._determine_log_level(line)
@staticmethod
def _get_go_version():
def _get_go_version() -> str | None:
"""Get the installed Go version or None if not found."""
try:
result = subprocess.run(["go", "version"], capture_output=True, text=True, check=False)
@@ -57,7 +58,7 @@ class Gopls(SolidLanguageServer):
return None
@staticmethod
def _get_gopls_version():
def _get_gopls_version() -> str | None:
"""Get the installed gopls version or None if not found."""
try:
result = subprocess.run(["gopls", "version"], capture_output=True, text=True, check=False)
@@ -68,7 +69,7 @@ class Gopls(SolidLanguageServer):
return None
@staticmethod
def _setup_runtime_dependency():
def _setup_runtime_dependency() -> bool:
"""
Check if required Go runtime dependencies are available.
Raises RuntimeError with helpful message if dependencies are missing.
@@ -135,18 +136,18 @@ class Gopls(SolidLanguageServer):
}
],
}
return initialize_params
return cast(InitializeParams, initialize_params)
def _start_server(self):
def _start_server(self) -> None:
"""Start gopls server process"""
def register_capability_handler(params):
def register_capability_handler(params: dict) -> None:
return
def window_log_message(msg):
def window_log_message(msg: dict) -> None:
self.logger.log(f"LSP: window/logMessage: {msg}", logging.INFO)
def do_nothing(params):
def do_nothing(params: dict) -> None:
return
self.server.on_request("client/registerCapability", register_capability_handler)
@@ -7,6 +7,7 @@ import os
import pathlib
import shutil
import time
from typing import Any
from overrides import override
@@ -25,7 +26,7 @@ class HaskellLanguageServer(SolidLanguageServer):
"""
@staticmethod
def _ensure_hls_installed():
def _ensure_hls_installed() -> str:
"""Ensure haskell-language-server-wrapper is available."""
# Try common locations
common_paths = [
@@ -310,20 +311,20 @@ class HaskellLanguageServer(SolidLanguageServer):
}
],
}
return initialize_params
return initialize_params # type: ignore
def _start_server(self):
def _start_server(self) -> None:
"""
Starts the Haskell Language Server
"""
def do_nothing(params):
def do_nothing(params: Any) -> None:
return
def window_log_message(msg):
def window_log_message(msg: dict) -> None:
self.logger.log(f"LSP: window/logMessage: {msg}", logging.INFO)
def register_capability_handler(params):
def register_capability_handler(params: dict) -> None:
"""Handle dynamic capability registration from HLS"""
if "registrations" in params:
for registration in params.get("registrations", []):
@@ -331,7 +332,7 @@ class HaskellLanguageServer(SolidLanguageServer):
self.logger.log(f"HLS registered capability: {method}", logging.INFO)
return
def workspace_configuration_handler(params):
def workspace_configuration_handler(params: dict) -> Any:
"""Handle workspace/configuration requests from HLS"""
self.logger.log(f"HLS requesting configuration: {params}", logging.INFO)
@@ -14,10 +14,11 @@ from solidlsp.ls import SolidLanguageServer
from solidlsp.ls_config import LanguageServerConfig
from solidlsp.ls_logger import LanguageServerLogger
from solidlsp.ls_utils import PlatformId, PlatformUtils
from solidlsp.lsp_protocol_handler.lsp_types import DefinitionParams, InitializeParams
from solidlsp.lsp_protocol_handler.lsp_types import Definition, DefinitionParams, InitializeParams, LocationLink
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.settings import SolidLSPSettings
from ..lsp_protocol_handler import lsp_types
from .common import RuntimeDependency, RuntimeDependencyCollection
@@ -147,18 +148,18 @@ class Intelephense(SolidLanguageServer):
initialization_options["intelephense.files.maxSize"] = max_file_size
initialize_params["initializationOptions"] = initialization_options
return initialize_params
return initialize_params # type: ignore
def _start_server(self):
def _start_server(self) -> None:
"""Start Intelephense server process"""
def register_capability_handler(params):
def register_capability_handler(params: dict) -> None:
return
def window_log_message(msg):
def window_log_message(msg: dict) -> None:
self.logger.log(f"LSP: window/logMessage: {msg}", logging.INFO)
def do_nothing(params):
def do_nothing(params: dict) -> None:
return
self.server.on_request("client/registerCapability", register_capability_handler)
@@ -193,7 +194,7 @@ class Intelephense(SolidLanguageServer):
@override
# For some reason, the LS may need longer to process this, so we just retry
def _send_references_request(self, relative_file_path: str, line: int, column: int):
def _send_references_request(self, relative_file_path: str, line: int, column: int) -> list[lsp_types.Location] | None:
# TODO: The LS doesn't return references contained in other files if it doesn't sleep. This is
# despite the LS having processed requests already. I don't know what causes this, but sleeping
# one second helps. It may be that sleeping only once is enough but that's hard to reliably test.
@@ -203,7 +204,7 @@ class Intelephense(SolidLanguageServer):
return super()._send_references_request(relative_file_path, line, column)
@override
def _send_definition_request(self, definition_params: DefinitionParams):
def _send_definition_request(self, definition_params: DefinitionParams) -> Definition | list[LocationLink] | None:
# TODO: same as above, also only a problem if the definition is in another file
sleep(1)
return super()._send_definition_request(definition_params)
+8 -7
View File
@@ -5,6 +5,7 @@ Provides Python specific instantiation of the LanguageServer class. Contains var
import logging
import os
import pathlib
from typing import cast
from overrides import override
@@ -150,24 +151,24 @@ class JediServer(SolidLanguageServer):
}
],
}
return initialize_params
return cast(InitializeParams, initialize_params)
def _start_server(self):
def _start_server(self) -> None:
"""
Starts the JEDI Language Server
"""
def execute_client_command_handler(params):
def execute_client_command_handler(params: dict) -> list:
return []
def do_nothing(params):
def do_nothing(params: dict) -> None:
return
def check_experimental_status(params):
def check_experimental_status(params: dict) -> None:
if params["quiescent"] == True:
self.completions_available.set()
def window_log_message(msg):
def window_log_message(msg: dict) -> None:
self.logger.log(f"LSP: window/logMessage: {msg}", logging.INFO)
self.server.on_request("client/registerCapability", do_nothing)
@@ -188,7 +189,7 @@ class JediServer(SolidLanguageServer):
logging.INFO,
)
init_response = self.server.send.initialize(initialize_params)
assert init_response["capabilities"]["textDocumentSync"]["change"] == 2
assert init_response["capabilities"]["textDocumentSync"]["change"] == 2 # type: ignore
assert "completionProvider" in init_response["capabilities"]
assert init_response["capabilities"]["completionProvider"] == {
"triggerCharacters": [".", "'", '"'],
@@ -4,6 +4,7 @@ import pathlib
import platform
import shutil
import subprocess
from typing import Any
from overrides import override
@@ -30,6 +31,7 @@ class JuliaLanguageServer(SolidLanguageServer):
julia_executable = self._setup_runtime_dependency(logger) # PASS LOGGER
julia_code = "using LanguageServer; runserver()"
julia_ls_cmd: str | list[str]
if platform.system() == "Windows":
# On Windows, pass as list (Serena handles shell=True differently)
julia_ls_cmd = [julia_executable, "--startup-file=no", "--history-file=no", "-e", julia_code, repository_root_path]
@@ -147,15 +149,15 @@ class JuliaLanguageServer(SolidLanguageServer):
}
],
}
return initialize_params
return initialize_params # type: ignore
def _start_server(self):
def _start_server(self) -> None:
"""Start the LanguageServer.jl server process."""
def do_nothing(params):
def do_nothing(params: Any) -> None:
return
def window_log_message(msg):
def window_log_message(msg: dict) -> None:
self.logger.log(f"LSP: window/logMessage: {msg}", logging.INFO)
self.server.on_notification("window/logMessage", window_log_message)
@@ -7,6 +7,7 @@ import logging
import os
import pathlib
import stat
from typing import cast
from solidlsp.ls import SolidLanguageServer
from solidlsp.ls_config import LanguageServerConfig
@@ -114,7 +115,7 @@ class KotlinLanguageServer(SolidLanguageServer):
}
kotlin_dependency = runtime_dependencies["runtimeDependency"]
java_dependency = runtime_dependencies["java"][platform_id.value]
java_dependency = runtime_dependencies["java"][platform_id.value] # type: ignore
# Setup paths for dependencies
static_dir = os.path.join(cls.ls_resources_dir(solidlsp_settings), "kotlin_language_server")
@@ -149,7 +150,7 @@ class KotlinLanguageServer(SolidLanguageServer):
# Download and extract Kotlin Language Server if script doesn't exist
if not os.path.exists(kotlin_script):
logger.log("Downloading Kotlin Language Server...", logging.INFO)
FileUtils.download_and_extract_archive(logger, kotlin_dependency["url"], static_dir, kotlin_dependency["archiveType"])
FileUtils.download_and_extract_archive(logger, kotlin_dependency["url"], static_dir, kotlin_dependency["archiveType"]) # type: ignore
# Make script executable on Unix platforms
if os.path.exists(kotlin_script) and not platform_id.value.startswith("win-"):
@@ -420,20 +421,20 @@ class KotlinLanguageServer(SolidLanguageServer):
}
],
}
return initialize_params
return cast(InitializeParams, initialize_params)
def _start_server(self):
def _start_server(self) -> None:
"""
Starts the Kotlin Language Server
"""
def execute_client_command_handler(params):
def execute_client_command_handler(params: dict) -> list:
return []
def do_nothing(params):
def do_nothing(params: dict) -> None:
return
def window_log_message(msg):
def window_log_message(msg: dict) -> None:
self.logger.log(f"LSP: window/logMessage: {msg}", logging.INFO)
self.server.on_request("client/registerCapability", do_nothing)
+8 -8
View File
@@ -37,7 +37,7 @@ class LuaLanguageServer(SolidLanguageServer):
return super().is_ignored_dirname(dirname) or dirname in [".luarocks", "lua_modules", "node_modules", "build", "dist", ".cache"]
@staticmethod
def _get_lua_ls_path():
def _get_lua_ls_path() -> str | None:
"""Get the path to lua-language-server executable."""
# First check if it's in PATH
lua_ls = shutil.which("lua-language-server")
@@ -69,7 +69,7 @@ class LuaLanguageServer(SolidLanguageServer):
return None
@staticmethod
def _download_lua_ls():
def _download_lua_ls() -> str:
"""Download and install lua-language-server if not present."""
system = platform.system()
machine = platform.machine().lower()
@@ -140,7 +140,7 @@ class LuaLanguageServer(SolidLanguageServer):
raise RuntimeError("Failed to find lua-language-server executable after extraction")
@staticmethod
def _setup_runtime_dependency():
def _setup_runtime_dependency() -> str:
"""
Check if required Lua runtime dependencies are available.
Downloads lua-language-server if not present.
@@ -254,18 +254,18 @@ class LuaLanguageServer(SolidLanguageServer):
},
},
}
return initialize_params
return initialize_params # type: ignore[return-value]
def _start_server(self):
def _start_server(self) -> None:
"""Start Lua Language Server process"""
def register_capability_handler(params):
def register_capability_handler(params: dict) -> None:
return
def window_log_message(msg):
def window_log_message(msg: dict) -> None:
self.logger.log(f"LSP: window/logMessage: {msg}", logging.INFO)
def do_nothing(params):
def do_nothing(params: dict) -> None:
return
self.server.on_request("client/registerCapability", register_capability_handler)
+6 -6
View File
@@ -130,9 +130,9 @@ class Marksman(SolidLanguageServer):
"documentSymbol": {
"dynamicRegistration": True,
"hierarchicalDocumentSymbolSupport": True,
"symbolKind": {"valueSet": list(range(1, 27))},
"symbolKind": {"valueSet": list(range(1, 27))}, # type: ignore[arg-type]
},
"hover": {"dynamicRegistration": True, "contentFormat": ["markdown", "plaintext"]},
"hover": {"dynamicRegistration": True, "contentFormat": ["markdown", "plaintext"]}, # type: ignore[list-item]
"codeAction": {"dynamicRegistration": True},
},
"workspace": {
@@ -150,18 +150,18 @@ class Marksman(SolidLanguageServer):
}
return initialize_params
def _start_server(self):
def _start_server(self) -> None:
"""
Starts the Marksman Language Server and waits for it to be ready.
"""
def register_capability_handler(_params):
def register_capability_handler(_params: dict) -> None:
return
def window_log_message(msg):
def window_log_message(msg: dict) -> None:
self.logger.log(f"LSP: window/logMessage: {msg}", logging.INFO)
def do_nothing(_params):
def do_nothing(_params: dict) -> None:
return
self.server.on_request("client/registerCapability", register_capability_handler)
+1
View File
@@ -1,3 +1,4 @@
# type: ignore
"""
Provides Nix specific instantiation of the LanguageServer class using nixd (Nix Language Server).
+9 -9
View File
@@ -21,7 +21,7 @@ from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.settings import SolidLSPSettings
def breadth_first_file_scan(root) -> Iterable[str]:
def breadth_first_file_scan(root: str) -> Iterable[str]:
"""
This function was obtained from https://stackoverflow.com/questions/49654234/is-there-a-breadth-first-search-option-available-in-os-walk-or-equivalent-py
It traverses the directory tree in breadth first order.
@@ -47,7 +47,7 @@ def breadth_first_file_scan(root) -> Iterable[str]:
dirs = next_dirs
def find_least_depth_sln_file(root_dir) -> str | None:
def find_least_depth_sln_file(root_dir: str) -> str | None:
for filename in breadth_first_file_scan(root_dir):
if filename.endswith(".sln"):
return filename
@@ -204,12 +204,12 @@ class OmniSharp(SolidLanguageServer):
return omnisharp_executable_path, razor_omnisharp_dll_path
def _start_server(self):
def _start_server(self) -> None:
"""
Starts the Omnisharp Language Server
"""
def register_capability_handler(params):
def register_capability_handler(params: dict) -> None:
assert "registrations" in params
for registration in params["registrations"]:
if registration["method"] == "textDocument/definition":
@@ -219,7 +219,7 @@ class OmniSharp(SolidLanguageServer):
if registration["method"] == "textDocument/completion":
self.completions_available.set()
def lang_status_handler(params):
def lang_status_handler(params: dict) -> None:
# TODO: Should we wait for
# server -> client: {'jsonrpc': '2.0', 'method': 'language/status', 'params': {'type': 'ProjectStatus', 'message': 'OK'}}
# Before proceeding?
@@ -227,17 +227,17 @@ class OmniSharp(SolidLanguageServer):
# self.service_ready_event.set()
pass
def execute_client_command_handler(params):
def execute_client_command_handler(params: dict) -> list:
return []
def do_nothing(params):
def do_nothing(params: dict) -> None:
return
def check_experimental_status(params):
def check_experimental_status(params: dict) -> None:
if params["quiescent"] is True:
self.server_ready.set()
def workspace_configuration_handler(params):
def workspace_configuration_handler(params: dict) -> list[dict]:
# TODO: We do not know the appropriate way to handle this request. Should ideally contact the OmniSharp dev team
return [
{
@@ -9,6 +9,7 @@ import os
import pathlib
import subprocess
import time
from typing import Any
from overrides import override
@@ -16,7 +17,7 @@ from solidlsp.ls import SolidLanguageServer
from solidlsp.ls_config import LanguageServerConfig
from solidlsp.ls_logger import LanguageServerLogger
from solidlsp.ls_utils import PlatformId, PlatformUtils
from solidlsp.lsp_protocol_handler.lsp_types import InitializeParams
from solidlsp.lsp_protocol_handler.lsp_types import DidChangeConfigurationParams, InitializeParams
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.settings import SolidLSPSettings
@@ -27,7 +28,7 @@ class PerlLanguageServer(SolidLanguageServer):
"""
@staticmethod
def _get_perl_version():
def _get_perl_version() -> str | None:
"""Get the installed Perl version or None if not found."""
try:
result = subprocess.run(["perl", "-v"], capture_output=True, text=True, check=False)
@@ -38,7 +39,7 @@ class PerlLanguageServer(SolidLanguageServer):
return None
@staticmethod
def _get_perl_language_server_version():
def _get_perl_language_server_version() -> str | None:
"""Get the installed Perl::LanguageServer version or None if not found."""
try:
result = subprocess.run(
@@ -147,21 +148,21 @@ class PerlLanguageServer(SolidLanguageServer):
],
}
return initialize_params
return initialize_params # type: ignore
def _start_server(self):
def _start_server(self) -> None:
"""Start Perl::LanguageServer process"""
def register_capability_handler(params):
def register_capability_handler(params: Any) -> None:
return
def window_log_message(msg):
def window_log_message(msg: str) -> None:
self.logger.log(f"LSP: window/logMessage: {msg}", logging.INFO)
def do_nothing(params):
def do_nothing(params: Any) -> None:
return
def workspace_configuration_handler(params):
def workspace_configuration_handler(params: Any) -> Any:
"""Handle workspace/configuration request from Perl::LanguageServer."""
self.logger.log(f"Received workspace/configuration request: {params}", logging.INFO)
@@ -203,7 +204,7 @@ class PerlLanguageServer(SolidLanguageServer):
# Send workspace configuration to Perl::LanguageServer
# Perl::LanguageServer requires didChangeConfiguration to set perlInc, fileFilter, and ignoreDirs
# See: Perl::LanguageServer::Methods::workspace::_rpcnot_didChangeConfiguration
perl_config = {
perl_config: DidChangeConfigurationParams = {
"settings": {
"perl": {
"perlInc": [self.repository_root_path, "."],
@@ -7,6 +7,7 @@ import os
import pathlib
import re
import threading
from typing import cast
from overrides import override
@@ -56,7 +57,7 @@ class PyrightServer(SolidLanguageServer):
Returns the initialize params for the Pyright Language Server.
"""
# Create basic initialization parameters
initialize_params: InitializeParams = { # type: ignore
initialize_params = { # type: ignore
"processId": os.getpid(),
"rootPath": repository_absolute_path,
"rootUri": pathlib.Path(repository_absolute_path).as_uri(),
@@ -107,9 +108,9 @@ class PyrightServer(SolidLanguageServer):
],
}
return initialize_params
return cast(InitializeParams, initialize_params)
def _start_server(self):
def _start_server(self) -> None:
"""
Starts the Pyright Language Server and waits for initial workspace analysis to complete.
@@ -127,13 +128,13 @@ class PyrightServer(SolidLanguageServer):
```
"""
def execute_client_command_handler(params):
def execute_client_command_handler(params: dict) -> list:
return []
def do_nothing(params):
def do_nothing(params: dict) -> None:
return
def window_log_message(msg):
def window_log_message(msg: dict) -> None:
"""
Monitor Pyright's log messages to detect when initial analysis is complete.
Pyright logs "Found X source files" when it finishes scanning the workspace.
@@ -149,7 +150,7 @@ class PyrightServer(SolidLanguageServer):
self.analysis_complete.set()
self.completions_available.set()
def check_experimental_status(params):
def check_experimental_status(params: dict) -> None:
"""
Also listen for experimental/serverStatus as a backup signal
"""
@@ -3,6 +3,7 @@ import os
import pathlib
import subprocess
import threading
from typing import Any
from overrides import override
@@ -32,7 +33,7 @@ class RLanguageServer(SolidLanguageServer):
]
@staticmethod
def _check_r_installation():
def _check_r_installation() -> None:
"""Check if R and languageserver are available."""
try:
# Check R installation
@@ -126,18 +127,18 @@ class RLanguageServer(SolidLanguageServer):
}
],
}
return initialize_params
return initialize_params # type: ignore
def _start_server(self):
def _start_server(self) -> None:
"""Start R Language Server process."""
def window_log_message(msg):
def window_log_message(msg: str) -> None:
self.logger.log(f"R LSP: window/logMessage: {msg}", logging.INFO)
def do_nothing(params):
def do_nothing(params: Any) -> None:
return
def register_capability_handler(params):
def register_capability_handler(params: Any) -> None:
return
# Register LSP message handlers
+5 -4
View File
@@ -16,7 +16,7 @@ from overrides import override
from solidlsp.ls import SolidLanguageServer
from solidlsp.ls_config import LanguageServerConfig
from solidlsp.ls_logger import LanguageServerLogger
from solidlsp.lsp_protocol_handler.lsp_types import InitializeParams
from solidlsp.lsp_protocol_handler.lsp_types import InitializeParams, InitializeResult
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.settings import SolidLSPSettings
@@ -180,6 +180,7 @@ class RubyLsp(SolidLanguageServer):
if not bundle_path:
# Try common bundle executables
for bundle_executable in ["bin/bundle", "bundle"]:
bundle_full_path: str | None
if bundle_executable.startswith("bin/"):
bundle_full_path = os.path.join(repository_root_path, bundle_executable)
else:
@@ -340,7 +341,7 @@ class RubyLsp(SolidLanguageServer):
},
}
return initialize_params
return initialize_params # type: ignore
def _start_server(self) -> None:
"""
@@ -394,7 +395,7 @@ class RubyLsp(SolidLanguageServer):
self.analysis_complete.set()
self.completions_available.set()
def window_work_done_progress_create(params: dict) -> None:
def window_work_done_progress_create(params: dict) -> dict:
"""Handle workDoneProgress/create requests from ruby-lsp"""
self.logger.log(f"LSP: window/workDoneProgress/create: {params}", logging.DEBUG)
return {}
@@ -442,7 +443,7 @@ class RubyLsp(SolidLanguageServer):
self.analysis_complete.set()
self.completions_available.set()
def _handle_initialization_response(self, init_response):
def _handle_initialization_response(self, init_response: InitializeResult) -> None:
"""
Handle the initialization response from ruby-lsp and validate capabilities.
"""
+18 -13
View File
@@ -7,6 +7,7 @@ import os
import pathlib
import subprocess
import threading
from typing import cast
from overrides import override
@@ -40,7 +41,7 @@ class RustAnalyzer(SolidLanguageServer):
return SolidLanguageServer._determine_log_level(line)
@staticmethod
def _get_rustup_version():
def _get_rustup_version() -> str | None:
"""Get installed rustup version or None if not found."""
try:
result = subprocess.run(["rustup", "--version"], capture_output=True, text=True, check=False)
@@ -51,7 +52,7 @@ class RustAnalyzer(SolidLanguageServer):
return None
@staticmethod
def _get_rust_analyzer_path():
def _get_rust_analyzer_path() -> str | None:
"""Get rust-analyzer path via rustup. Returns None if not found."""
try:
# Note: we avoid using system PATH to avoid picking up incorrect aliases
@@ -65,8 +66,12 @@ class RustAnalyzer(SolidLanguageServer):
return None
@staticmethod
def _ensure_rust_analyzer_installed():
"""Ensure rust-analyzer is available, install via rustup if needed."""
def _ensure_rust_analyzer_installed() -> str:
"""
Ensure rust-analyzer is available, install via rustup if needed.
:return: path to rust-analyzer executable
"""
path = RustAnalyzer._get_rust_analyzer_path()
if path:
return path
@@ -588,14 +593,14 @@ class RustAnalyzer(SolidLanguageServer):
}
],
}
return initialize_params
return cast(InitializeParams, initialize_params)
def _start_server(self):
def _start_server(self) -> None:
"""
Starts the Rust Analyzer Language Server
"""
def register_capability_handler(params):
def register_capability_handler(params: dict) -> None:
assert "registrations" in params
for registration in params["registrations"]:
if registration["method"] == "workspace/executeCommand":
@@ -603,24 +608,24 @@ class RustAnalyzer(SolidLanguageServer):
self.resolve_main_method_available.set()
return
def lang_status_handler(params):
def lang_status_handler(params: dict) -> None:
# TODO: Should we wait for
# server -> client: {'jsonrpc': '2.0', 'method': 'language/status', 'params': {'type': 'ProjectStatus', 'message': 'OK'}}
# Before proceeding?
if params["type"] == "ServiceReady" and params["message"] == "ServiceReady":
self.service_ready_event.set()
def execute_client_command_handler(params):
def execute_client_command_handler(params: dict) -> list:
return []
def do_nothing(params):
def do_nothing(params: dict) -> None:
return
def check_experimental_status(params):
def check_experimental_status(params: dict) -> None:
if params["quiescent"] == True:
self.server_ready.set()
def window_log_message(msg):
def window_log_message(msg: dict) -> None:
self.logger.log(f"LSP: window/logMessage: {msg}", logging.INFO)
self.server.on_request("client/registerCapability", register_capability_handler)
@@ -641,7 +646,7 @@ class RustAnalyzer(SolidLanguageServer):
logging.INFO,
)
init_response = self.server.send.initialize(initialize_params)
assert init_response["capabilities"]["textDocumentSync"]["change"] == 2
assert init_response["capabilities"]["textDocumentSync"]["change"] == 2 # type: ignore
assert "completionProvider" in init_response["capabilities"]
assert init_response["capabilities"]["completionProvider"] == {
"resolveProvider": True,
@@ -64,6 +64,7 @@ class ScalaLanguageServer(SolidLanguageServer):
os.makedirs(metals_home, exist_ok=True)
metals_executable = os.path.join(metals_home, "metals")
coursier_command_path = shutil.which("coursier")
assert coursier_command_path is not None, "coursier is not installed or not in PATH."
cs_command_path = shutil.which("cs")
if not os.path.exists(metals_executable):
@@ -152,9 +153,9 @@ class ScalaLanguageServer(SolidLanguageServer):
},
"capabilities": {"textDocument": {"documentSymbol": {"hierarchicalDocumentSymbolSupport": True}}},
}
return initialize_params
return initialize_params # type: ignore
def _start_server(self):
def _start_server(self) -> None:
"""
Starts the Scala Language Server
"""
+11 -11
View File
@@ -134,7 +134,7 @@ class Solargraph(SolidLanguageServer):
if bundle_cmd.startswith("bin/"):
bundle_full_path = os.path.join(repository_root_path, bundle_cmd)
else:
bundle_full_path = find_executable_with_extensions(bundle_cmd)
bundle_full_path = find_executable_with_extensions(bundle_cmd) # type: ignore[assignment]
if bundle_full_path and os.path.exists(bundle_full_path):
bundle_path = bundle_full_path if bundle_cmd.startswith("bin/") else bundle_cmd
break
@@ -282,7 +282,7 @@ class Solargraph(SolidLanguageServer):
"rootPath": repository_absolute_path,
"rootUri": root_uri,
"initializationOptions": {
"exclude": exclude_patterns,
"exclude": exclude_patterns, # type: ignore[dict-item]
},
"capabilities": {
"workspace": {
@@ -291,11 +291,11 @@ class Solargraph(SolidLanguageServer):
"textDocument": {
"documentSymbol": {
"hierarchicalDocumentSymbolSupport": True,
"symbolKind": {"valueSet": list(range(1, 27))},
"symbolKind": {"valueSet": list(range(1, 27))}, # type: ignore[arg-type]
},
},
},
"trace": "verbose",
"trace": "verbose", # type: ignore[typeddict-item]
"workspaceFolders": [
{
"uri": root_uri,
@@ -303,14 +303,14 @@ class Solargraph(SolidLanguageServer):
}
],
}
return initialize_params
return initialize_params # type: ignore[return-value]
def _start_server(self):
def _start_server(self) -> None:
"""
Starts the Solargraph Language Server for Ruby
"""
def register_capability_handler(params):
def register_capability_handler(params: dict) -> None:
assert "registrations" in params
for registration in params["registrations"]:
if registration["method"] == "workspace/executeCommand":
@@ -318,20 +318,20 @@ class Solargraph(SolidLanguageServer):
self.resolve_main_method_available.set()
return
def lang_status_handler(params):
def lang_status_handler(params: dict) -> None:
self.logger.log(f"LSP: language/status: {params}", logging.INFO)
if params.get("type") == "ServiceReady" and params.get("message") == "Service is ready.":
self.logger.log("Solargraph service is ready.", logging.INFO)
self.analysis_complete.set()
self.completions_available.set()
def execute_client_command_handler(params):
def execute_client_command_handler(params: dict) -> list:
return []
def do_nothing(params):
def do_nothing(params: dict) -> None:
return
def window_log_message(msg):
def window_log_message(msg: dict) -> None:
self.logger.log(f"LSP: window/logMessage: {msg}", logging.INFO)
self.server.on_request("client/registerCapability", register_capability_handler)
@@ -62,7 +62,7 @@ class SourceKitLSP(SolidLanguageServer):
self.server_ready = threading.Event()
self.request_id = 0
self._did_sleep_before_requesting_references = False
self._initialization_timestamp = None
self._initialization_timestamp: float | None = None
@staticmethod
def _get_initialize_params(repository_absolute_path: str) -> InitializeParams:
@@ -305,18 +305,18 @@ class SourceKitLSP(SolidLanguageServer):
],
}
return initialize_params
return initialize_params # type: ignore[return-value]
def _start_server(self):
def _start_server(self) -> None:
"""Start sourcekit-lsp server process"""
def register_capability_handler(_params):
def register_capability_handler(_params: dict) -> None:
return
def window_log_message(msg):
def window_log_message(msg: dict) -> None:
self.logger.log(f"LSP: window/logMessage: {msg}", logging.INFO)
def do_nothing(_params):
def do_nothing(_params: dict) -> None:
return
self.server.on_request("client/registerCapability", register_capability_handler)
@@ -2,6 +2,7 @@ import logging
import os
import shutil
import threading
from typing import cast
from overrides import override
@@ -54,7 +55,7 @@ class TerraformLS(SolidLanguageServer):
return SolidLanguageServer._determine_log_level(line)
@staticmethod
def _ensure_tf_command_available(logger: LanguageServerLogger):
def _ensure_tf_command_available(logger: LanguageServerLogger) -> None:
logger.log("Starting terraform version detection...", logging.DEBUG)
# 1. Try to find terraform using shutil.which
@@ -176,7 +177,7 @@ class TerraformLS(SolidLanguageServer):
Returns the initialize params for the Terraform Language Server.
"""
root_uri = PathUtils.path_to_uri(repository_absolute_path)
return {
result = {
"processId": os.getpid(),
"locale": "en",
"rootPath": repository_absolute_path,
@@ -201,17 +202,18 @@ class TerraformLS(SolidLanguageServer):
}
],
}
return cast(InitializeParams, result)
def _start_server(self):
def _start_server(self) -> None:
"""Start terraform-ls server process"""
def register_capability_handler(params):
def register_capability_handler(params: dict) -> None:
return
def window_log_message(msg):
def window_log_message(msg: dict) -> None:
self.logger.log(f"LSP: window/logMessage: {msg}", logging.INFO)
def do_nothing(params):
def do_nothing(params: dict) -> None:
return
self.server.on_request("client/registerCapability", register_capability_handler)
@@ -7,6 +7,7 @@ import os
import pathlib
import shutil
import threading
from typing import Any, cast
from overrides import override
from sensai.util.logging import LogTime
@@ -26,9 +27,9 @@ if os.name != "nt": # Unix-like systems
import pwd
else:
# Dummy pwd module for Windows
class pwd:
class pwd: # type: ignore
@staticmethod
def getpwuid(uid):
def getpwuid(uid: Any) -> Any:
return type("obj", (), {"pw_name": os.environ.get("USERNAME", "unknown")})()
@@ -176,9 +177,9 @@ class TypeScriptLanguageServer(SolidLanguageServer):
}
],
}
return initialize_params
return cast(InitializeParams, initialize_params)
def _start_server(self):
def _start_server(self) -> None:
"""
Starts the TypeScript Language Server, waits for the server to be ready and yields the LanguageServer instance.
@@ -192,7 +193,7 @@ class TypeScriptLanguageServer(SolidLanguageServer):
# LanguageServer has been shutdown
"""
def register_capability_handler(params):
def register_capability_handler(params: dict) -> None:
assert "registrations" in params
for registration in params["registrations"]:
if registration["method"] == "workspace/executeCommand":
@@ -202,16 +203,16 @@ class TypeScriptLanguageServer(SolidLanguageServer):
# self.resolve_main_method_available.set()
return
def execute_client_command_handler(params):
def execute_client_command_handler(params: dict) -> list:
return []
def do_nothing(params):
def do_nothing(params: dict) -> None:
return
def window_log_message(msg):
def window_log_message(msg: dict) -> None:
self.logger.log(f"LSP: window/logMessage: {msg}", logging.INFO)
def check_experimental_status(params):
def check_experimental_status(params: dict) -> None:
"""
Also listen for experimental/serverStatus as a backup signal
"""
@@ -9,6 +9,7 @@ import os
import pathlib
import shutil
import threading
from typing import cast
from overrides import override
@@ -144,9 +145,9 @@ class VtsLanguageServer(SolidLanguageServer):
}
],
}
return initialize_params
return cast(InitializeParams, initialize_params)
def _start_server(self):
def _start_server(self) -> None:
"""
Starts the VTS Language Server, waits for the server to be ready and yields the LanguageServer instance.
@@ -160,30 +161,30 @@ class VtsLanguageServer(SolidLanguageServer):
# LanguageServer has been shutdown
"""
def register_capability_handler(params):
def register_capability_handler(params: dict) -> None:
assert "registrations" in params
for registration in params["registrations"]:
if registration["method"] == "workspace/executeCommand":
self.initialize_searcher_command_available.set()
return
def execute_client_command_handler(params):
def execute_client_command_handler(params: dict) -> list:
return []
def workspace_configuration_handler(params):
def workspace_configuration_handler(params: dict) -> list[dict] | dict:
# VTS may request workspace configuration
# Return empty configuration for each requested item
if "items" in params:
return [{}] * len(params["items"])
return {}
def do_nothing(params):
def do_nothing(params: dict) -> None:
return
def window_log_message(msg):
def window_log_message(msg: dict) -> None:
self.logger.log(f"LSP: window/logMessage: {msg}", logging.INFO)
def check_experimental_status(params):
def check_experimental_status(params: dict) -> None:
"""
Also listen for experimental/serverStatus as a backup signal
"""
@@ -8,6 +8,7 @@ import os
import pathlib
import shutil
import threading
from typing import Any
from solidlsp.language_servers.common import RuntimeDependency, RuntimeDependencyCollection
from solidlsp.ls import SolidLanguageServer
@@ -148,20 +149,20 @@ class YamlLanguageServer(SolidLanguageServer):
}
},
}
return initialize_params
return initialize_params # type: ignore
def _start_server(self):
def _start_server(self) -> None:
"""
Starts the YAML Language Server, waits for the server to be ready and yields the LanguageServer instance.
"""
def register_capability_handler(params):
def register_capability_handler(params: Any) -> None:
return
def do_nothing(params):
def do_nothing(params: Any) -> None:
return
def window_log_message(msg):
def window_log_message(msg: str) -> None:
self.logger.log(f"LSP: window/logMessage: {msg}", logging.INFO)
self.server.on_request("client/registerCapability", register_capability_handler)
+9 -9
View File
@@ -35,7 +35,7 @@ class ZigLanguageServer(SolidLanguageServer):
return super().is_ignored_dirname(dirname) or dirname in ["zig-cache", "zig-out", ".zig-cache", "node_modules", "build", "dist"]
@staticmethod
def _get_zig_version():
def _get_zig_version() -> str | None:
"""Get the installed Zig version or None if not found."""
try:
result = subprocess.run(["zig", "version"], capture_output=True, text=True, check=False)
@@ -46,7 +46,7 @@ class ZigLanguageServer(SolidLanguageServer):
return None
@staticmethod
def _get_zls_version():
def _get_zls_version() -> str | None:
"""Get the installed ZLS version or None if not found."""
try:
result = subprocess.run(["zls", "--version"], capture_output=True, text=True, check=False)
@@ -57,12 +57,12 @@ class ZigLanguageServer(SolidLanguageServer):
return None
@staticmethod
def _check_zls_installed():
def _check_zls_installed() -> bool:
"""Check if ZLS is installed in the system."""
return shutil.which("zls") is not None
@staticmethod
def _setup_runtime_dependency():
def _setup_runtime_dependency() -> bool:
"""
Check if required Zig runtime dependencies are available.
Raises RuntimeError with helpful message if dependencies are missing.
@@ -188,18 +188,18 @@ class ZigLanguageServer(SolidLanguageServer):
"inlay_hints_hide_redundant_param_names_last_token": False,
},
}
return initialize_params
return initialize_params # type: ignore[return-value]
def _start_server(self):
def _start_server(self) -> None:
"""Start ZLS server process"""
def register_capability_handler(params):
def register_capability_handler(params: dict) -> None:
return
def window_log_message(msg):
def window_log_message(msg: str) -> None:
self.logger.log(f"LSP: window/logMessage: {msg}", logging.INFO)
def do_nothing(params):
def do_nothing(params: dict) -> None:
return
self.server.on_request("client/registerCapability", register_capability_handler)
+84 -70
View File
@@ -9,7 +9,7 @@ import subprocess
import threading
from abc import ABC, abstractmethod
from collections import defaultdict
from collections.abc import Iterator
from collections.abc import Hashable, Iterator
from contextlib import contextmanager
from copy import copy
from pathlib import Path, PurePath
@@ -31,7 +31,14 @@ from solidlsp.ls_utils import FileUtils, PathUtils, TextUtils
from solidlsp.lsp_protocol_handler import lsp_types
from solidlsp.lsp_protocol_handler import lsp_types as LSPTypes
from solidlsp.lsp_protocol_handler.lsp_constants import LSPConstants
from solidlsp.lsp_protocol_handler.lsp_types import Definition, DefinitionParams, LocationLink
from solidlsp.lsp_protocol_handler.lsp_types import (
Definition,
DefinitionParams,
DocumentSymbol,
LocationLink,
RenameParams,
SymbolInformation,
)
from solidlsp.lsp_protocol_handler.server import (
LSPError,
ProcessLaunchInfo,
@@ -76,7 +83,7 @@ class LSPFileBuffer:
content_hash: str = ""
def __post_init__(self):
def __post_init__(self) -> None:
self.content_hash = hashlib.md5(self.contents.encode("utf-8")).hexdigest()
def split_lines(self) -> list[str]:
@@ -91,7 +98,7 @@ class DocumentSymbols:
self.root_symbols = root_symbols
self._all_symbols: list[ls_types.UnifiedSymbolInformation] | None = None
def __getstate__(self):
def __getstate__(self) -> dict:
return getstate(DocumentSymbols, self, transient_properties=["_all_symbols"])
def iter_symbols(self) -> Iterator[ls_types.UnifiedSymbolInformation]:
@@ -103,7 +110,7 @@ class DocumentSymbols:
yield from self._all_symbols
return
def traverse(s: ls_types.UnifiedSymbolInformation):
def traverse(s: ls_types.UnifiedSymbolInformation) -> Iterator[ls_types.UnifiedSymbolInformation]:
yield s
for child in s.get("children", []):
yield from traverse(child)
@@ -238,7 +245,7 @@ class SolidLanguageServer(ABC):
process_launch_info: ProcessLaunchInfo,
language_id: str,
solidlsp_settings: SolidLSPSettings,
cache_version_raw_document_symbols: int = 1,
cache_version_raw_document_symbols: Hashable = 1,
):
"""
Initializes a LanguageServer instance.
@@ -283,7 +290,7 @@ class SolidLanguageServer(ABC):
self.cache_dir.mkdir(parents=True, exist_ok=True)
# * raw document symbols cache
self._ls_specific_raw_document_symbols_cache_version = cache_version_raw_document_symbols
self._raw_document_symbols_cache: dict[str, tuple[str, list[GenericDocumentSymbol]]] = {}
self._raw_document_symbols_cache: dict[str, tuple[str, list[DocumentSymbol] | list[SymbolInformation] | None]] = {}
"""maps relative file paths to a tuple of (file_content_hash, raw_root_symbols)"""
self._raw_document_symbols_cache_is_modified: bool = False
self._load_raw_document_symbols_cache()
@@ -297,11 +304,11 @@ class SolidLanguageServer(ABC):
self.completions_available = threading.Event()
if config.trace_lsp_communication:
def logging_fn(source: str, target: str, msg: StringDict | str):
def logging_fn(source: str, target: str, msg: StringDict | str) -> None:
self.logger.log(f"LSP: {source} -> {target}: {msg!s}", self.logger.logger.level)
else:
logging_fn = None
logging_fn = None # type: ignore
# 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
@@ -328,7 +335,6 @@ class SolidLanguageServer(ABC):
# Create a pathspec matcher from the processed patterns
self._ignore_spec = pathspec.PathSpec.from_lines(pathspec.patterns.GitWildMatchPattern, processed_patterns)
self._server_context = None
self._request_timeout: float | None = None
self._has_waited_for_cross_file_references = False
@@ -395,7 +401,7 @@ class SolidLanguageServer(ABC):
return match_path(relative_path, self.get_ignore_spec(), root_path=self.repository_root_path)
def _shutdown(self, timeout: float = 5.0):
def _shutdown(self, timeout: float = 5.0) -> None:
"""
A robust shutdown process designed to terminate cleanly on all platforms, including Windows,
by explicitly closing all I/O pipes.
@@ -406,6 +412,9 @@ class SolidLanguageServer(ABC):
self.logger.log(f"Initiating final robust shutdown with a {timeout}s timeout...", logging.INFO)
process = self.server.process
if process is None:
self.logger.log("Server process is None, cannot shutdown.", logging.DEBUG)
return
# --- Main Shutdown Logic ---
# Stage 1: Graceful Termination Request
@@ -423,7 +432,7 @@ class SolidLanguageServer(ABC):
else:
self.logger.log("LSP shutdown request completed.", logging.DEBUG)
if process.stdin and not process.stdin.is_closing():
if process.stdin and not process.stdin.closed:
process.stdin.close()
self.logger.log("Stage 1 shutdown complete.", logging.DEBUG)
except Exception as e:
@@ -462,7 +471,7 @@ class SolidLanguageServer(ABC):
self._start_server()
@abstractmethod
def _start_server(self):
def _start_server(self) -> None:
pass
@contextmanager
@@ -497,7 +506,7 @@ class SolidLanguageServer(ABC):
self.server.notify.did_open_text_document(
{
LSPConstants.TEXT_DOCUMENT: {
LSPConstants.TEXT_DOCUMENT: { # type: ignore
LSPConstants.URI: uri,
LSPConstants.LANGUAGE_ID: self.language_id,
LSPConstants.VERSION: 0,
@@ -511,7 +520,7 @@ class SolidLanguageServer(ABC):
if self.open_file_buffers[uri].ref_count == 0:
self.server.notify.did_close_text_document(
{
LSPConstants.TEXT_DOCUMENT: {
LSPConstants.TEXT_DOCUMENT: { # type: ignore
LSPConstants.URI: uri,
}
}
@@ -562,7 +571,7 @@ class SolidLanguageServer(ABC):
file_buffer.contents = new_contents
self.server.notify.did_change_text_document(
{
LSPConstants.TEXT_DOCUMENT: {
LSPConstants.TEXT_DOCUMENT: { # type: ignore
LSPConstants.VERSION: file_buffer.version,
LSPConstants.URI: file_buffer.uri,
},
@@ -609,7 +618,7 @@ class SolidLanguageServer(ABC):
file_buffer.contents = new_contents
self.server.notify.did_change_text_document(
{
LSPConstants.TEXT_DOCUMENT: {
LSPConstants.TEXT_DOCUMENT: { # type: ignore
LSPConstants.VERSION: file_buffer.version,
LSPConstants.URI: file_buffer.uri,
},
@@ -667,18 +676,18 @@ class SolidLanguageServer(ABC):
for item in response:
assert isinstance(item, dict)
if LSPConstants.URI in item and LSPConstants.RANGE in item:
new_item: ls_types.Location = {}
new_item: dict = {}
new_item.update(item)
new_item["absolutePath"] = PathUtils.uri_to_path(new_item["uri"])
new_item["relativePath"] = PathUtils.get_relative_path(new_item["absolutePath"], self.repository_root_path)
ret.append(ls_types.Location(new_item))
ret.append(ls_types.Location(**new_item)) # type: ignore
elif LSPConstants.TARGET_URI in item and LSPConstants.TARGET_RANGE in item and LSPConstants.TARGET_SELECTION_RANGE in item:
new_item: ls_types.Location = {}
new_item["uri"] = item[LSPConstants.TARGET_URI]
new_item: dict = {} # type: ignore
new_item["uri"] = item[LSPConstants.TARGET_URI] # type: ignore
new_item["absolutePath"] = PathUtils.uri_to_path(new_item["uri"])
new_item["relativePath"] = PathUtils.get_relative_path(new_item["absolutePath"], self.repository_root_path)
new_item["range"] = item[LSPConstants.TARGET_SELECTION_RANGE]
ret.append(ls_types.Location(**new_item))
new_item["range"] = item[LSPConstants.TARGET_SELECTION_RANGE] # type: ignore
ret.append(ls_types.Location(**new_item)) # type: ignore
else:
assert False, f"Unexpected response from Language Server: {item}"
elif isinstance(response, dict):
@@ -686,11 +695,11 @@ class SolidLanguageServer(ABC):
assert LSPConstants.URI in response
assert LSPConstants.RANGE in response
new_item: ls_types.Location = {}
new_item: dict = {} # type: ignore
new_item.update(response)
new_item["absolutePath"] = PathUtils.uri_to_path(new_item["uri"])
new_item["relativePath"] = PathUtils.get_relative_path(new_item["absolutePath"], self.repository_root_path)
ret.append(ls_types.Location(**new_item))
ret.append(ls_types.Location(**new_item)) # type: ignore
elif response is None:
# Some language servers return None when they cannot find a definition
# This is expected for certain symbol types like generics or types with incomplete information
@@ -759,7 +768,7 @@ class SolidLanguageServer(ABC):
assert LSPConstants.URI in item
assert LSPConstants.RANGE in item
abs_path = PathUtils.uri_to_path(item[LSPConstants.URI])
abs_path = PathUtils.uri_to_path(item[LSPConstants.URI]) # type: ignore
if not Path(abs_path).is_relative_to(self.repository_root_path):
self.logger.log(
"Found a reference in a path outside the repository, probably the LS is parsing things in installed packages or in the standardlib! "
@@ -773,11 +782,11 @@ class SolidLanguageServer(ABC):
self.logger.log(f"Ignoring reference in {rel_path} since it should be ignored", logging.DEBUG)
continue
new_item: ls_types.Location = {}
new_item: dict = {}
new_item.update(item)
new_item["absolutePath"] = str(abs_path)
new_item["relativePath"] = str(rel_path)
ret.append(ls_types.Location(**new_item))
ret.append(ls_types.Location(**new_item)) # type: ignore
return ret
@@ -800,26 +809,26 @@ class SolidLanguageServer(ABC):
with self.open_file(relative_file_path):
response = self.server.send.text_document_diagnostic(
{
LSPConstants.TEXT_DOCUMENT: {
LSPConstants.TEXT_DOCUMENT: { # type: ignore
LSPConstants.URI: pathlib.Path(str(PurePath(self.repository_root_path, relative_file_path))).as_uri()
}
}
)
if response is None:
return []
return [] # type: ignore
assert isinstance(response, dict), f"Unexpected response from Language Server (expected list, got {type(response)}): {response}"
ret: list[ls_types.Diagnostic] = []
for item in response["items"]:
for item in response["items"]: # type: ignore
new_item: ls_types.Diagnostic = {
"uri": pathlib.Path(str(PurePath(self.repository_root_path, relative_file_path))).as_uri(),
"severity": item["severity"],
"message": item["message"],
"range": item["range"],
"code": item["code"],
"code": item["code"], # type: ignore
}
ret.append(ls_types.Diagnostic(new_item))
ret.append(ls_types.Diagnostic(**new_item))
return ret
@@ -878,19 +887,19 @@ class SolidLanguageServer(ABC):
response: list[LSPTypes.CompletionItem] | LSPTypes.CompletionList | None = None
num_retries = 0
while response is None or (response["isIncomplete"] and num_retries < 30):
while response is None or (response["isIncomplete"] and num_retries < 30): # type: ignore
self.completions_available.wait()
response: list[LSPTypes.CompletionItem] | LSPTypes.CompletionList | None = self.server.send.completion(completion_params)
response = self.server.send.completion(completion_params)
if isinstance(response, list):
response = {"items": response, "isIncomplete": False}
num_retries += 1
# TODO: Understand how to appropriately handle `isIncomplete`
if response is None or (response["isIncomplete"] and not (allow_incomplete)):
if response is None or (response["isIncomplete"] and not allow_incomplete): # type: ignore
return []
if "items" in response:
response = response["items"]
response = response["items"] # type: ignore
response = cast(list[LSPTypes.CompletionItem], response)
@@ -908,8 +917,8 @@ class SolidLanguageServer(ABC):
if "label" in item:
completion_item["completionText"] = item["label"]
completion_item["kind"] = item["kind"]
elif "insertText" in item:
completion_item["kind"] = item["kind"] # type: ignore
elif "insertText" in item: # type: ignore
completion_item["completionText"] = item["insertText"]
completion_item["kind"] = item["kind"]
elif "textEdit" in item and "newText" in item["textEdit"]:
@@ -936,12 +945,14 @@ class SolidLanguageServer(ABC):
else:
assert False
completion_item = ls_types.CompletionItem(**completion_item)
completion_item = ls_types.CompletionItem(**completion_item) # type: ignore
completions_list.append(completion_item)
return [json.loads(json_repr) for json_repr in set(json.dumps(item, sort_keys=True) for item in completions_list)]
def _request_document_symbols(self, relative_file_path: str, file_data: LSPFileBuffer | None) -> list[GenericDocumentSymbol]:
def _request_document_symbols(
self, relative_file_path: str, file_data: LSPFileBuffer | None
) -> list[SymbolInformation] | list[DocumentSymbol] | None:
"""
Sends a [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 - or returns a cached result if available.
@@ -951,7 +962,7 @@ class SolidLanguageServer(ABC):
:return: the list of root symbols in the file.
"""
def get_cached_raw_document_symbols(cache_key: str, fd: LSPFileBuffer) -> list[GenericDocumentSymbol] | None:
def get_cached_raw_document_symbols(cache_key: str, fd: LSPFileBuffer) -> list[SymbolInformation] | list[DocumentSymbol] | None:
file_hash_and_result = self._raw_document_symbols_cache.get(cache_key)
if file_hash_and_result is not None:
file_hash, result = file_hash_and_result
@@ -964,7 +975,7 @@ class SolidLanguageServer(ABC):
log.debug("No cache hit for raw document symbols symbols in %s", relative_file_path)
return None
def get_raw_document_symbols(fd: LSPFileBuffer) -> list[GenericDocumentSymbol]:
def get_raw_document_symbols(fd: LSPFileBuffer) -> list[SymbolInformation] | list[DocumentSymbol] | None:
# check for cached result
cache_key = relative_file_path
response = get_cached_raw_document_symbols(cache_key, fd)
@@ -978,7 +989,7 @@ class SolidLanguageServer(ABC):
)
# update cache
self._raw_document_symbols_cache[cache_key] = (file_data.content_hash, response)
self._raw_document_symbols_cache[cache_key] = (fd.content_hash, response)
self._raw_document_symbols_cache_is_modified = True
return response
@@ -1057,9 +1068,9 @@ class SolidLanguageServer(ABC):
item["location"] = tree_location
location = item["location"]
if "absolutePath" not in location:
location["absolutePath"] = absolute_path
location["absolutePath"] = absolute_path # type: ignore
if "relativePath" not in location:
location["relativePath"] = relative_file_path
location["relativePath"] = relative_file_path # type: ignore
if "body" not in item:
item["body"] = self.retrieve_symbol_body(item, file_lines=file_lines)
@@ -1074,16 +1085,17 @@ class SolidLanguageServer(ABC):
return item
def convert_symbols_with_common_parent(
symbols: list[GenericDocumentSymbol], parent: ls_types.UnifiedSymbolInformation | None
symbols: list[DocumentSymbol] | list[SymbolInformation] | list[UnifiedSymbolInformation],
parent: ls_types.UnifiedSymbolInformation | None,
) -> list[ls_types.UnifiedSymbolInformation]:
"""
Converts the given symbols into UnifiedSymbolInformation with proper parent-child relationships,
adding overload indices for symbols with the same name under the same parent.
"""
total_name_counts = defaultdict(lambda: 0)
total_name_counts: dict[str, int] = defaultdict(lambda: 0)
for symbol in symbols:
total_name_counts[symbol["name"]] += 1
name_counts = defaultdict(lambda: 0)
name_counts: dict[str, int] = defaultdict(lambda: 0)
unified_symbols = []
for symbol in symbols:
usymbol = convert_to_unified_symbol(symbol)
@@ -1092,9 +1104,9 @@ class SolidLanguageServer(ABC):
name_counts[usymbol["name"]] += 1
usymbol["parent"] = parent
if "children" in usymbol:
usymbol["children"] = convert_symbols_with_common_parent(usymbol["children"], usymbol)
usymbol["children"] = convert_symbols_with_common_parent(usymbol["children"], usymbol) # type: ignore
else:
usymbol["children"] = []
usymbol["children"] = [] # type: ignore
unified_symbols.append(usymbol)
return unified_symbols
@@ -1222,10 +1234,10 @@ class SolidLanguageServer(ABC):
package_symbol["children"].append(file_symbol)
# TODO: Not sure if this is actually still needed given recent changes to relative path handling
def fix_relative_path(nodes: list[ls_types.UnifiedSymbolInformation]):
def fix_relative_path(nodes: list[ls_types.UnifiedSymbolInformation]) -> None:
for node in nodes:
if "location" in node and "relativePath" in node["location"]:
path = Path(node["location"]["relativePath"])
path = Path(node["location"]["relativePath"]) # type: ignore
if path.is_absolute():
try:
path = path.relative_to(self.repository_root_path)
@@ -1262,7 +1274,7 @@ class SolidLanguageServer(ABC):
result: dict[str, list[UnifiedSymbolInformation]] = defaultdict(list)
# Helper function to process a symbol and its children
def process_symbol(symbol: ls_types.UnifiedSymbolInformation):
def process_symbol(symbol: ls_types.UnifiedSymbolInformation) -> None:
if symbol["kind"] == ls_types.SymbolKind.File:
# For file symbols, process their children (top-level symbols)
for child in symbol["children"]:
@@ -1350,11 +1362,11 @@ class SolidLanguageServer(ABC):
assert isinstance(response, dict)
return ls_types.Hover(**response)
return ls_types.Hover(**response) # type: ignore
def retrieve_symbol_body(
self,
symbol: ls_types.UnifiedSymbolInformation | LSPTypes.DocumentSymbol | LSPTypes.SymbolInformation,
symbol: ls_types.UnifiedSymbolInformation | LSPTypes.SymbolInformation,
file_lines: list[str] | None = None,
file_buffer: LSPFileBuffer | None = None,
) -> str:
@@ -1363,19 +1375,19 @@ class SolidLanguageServer(ABC):
"""
existing_body = symbol.get("body", None)
if existing_body:
return existing_body
return str(existing_body)
assert "location" in symbol
symbol_start_line = symbol["location"]["range"]["start"]["line"]
symbol_end_line = symbol["location"]["range"]["end"]["line"]
assert "relativePath" in symbol["location"]
if file_lines is None:
with self._open_file_context(symbol["location"]["relativePath"], file_buffer) as f:
with self._open_file_context(symbol["location"]["relativePath"], file_buffer) as f: # type: ignore
file_lines = f.split_lines()
symbol_body = "\n".join(file_lines[symbol_start_line : symbol_end_line + 1])
# remove leading indentation
symbol_start_column = symbol["location"]["range"]["start"]["character"]
symbol_start_column = symbol["location"]["range"]["start"]["character"] # type: ignore
symbol_body = symbol_body[symbol_start_column:]
return symbol_body
@@ -1424,6 +1436,7 @@ class SolidLanguageServer(ABC):
incoming_symbol = None
for ref in references:
ref_path = ref["relativePath"]
assert ref_path is not None
ref_line = ref["range"]["start"]["line"]
ref_col = ref["range"]["start"]["character"]
@@ -1653,7 +1666,7 @@ class SolidLanguageServer(ABC):
return symbol["parent"]
assert "location" in symbol, f"Symbol {symbol} has no location and no parent attribute"
return self.request_containing_symbol(
symbol["location"]["relativePath"],
symbol["location"]["relativePath"], # type: ignore
symbol["location"]["range"]["start"]["line"],
symbol["location"]["range"]["start"]["character"],
strict=True,
@@ -1694,6 +1707,7 @@ class SolidLanguageServer(ABC):
# Use the first definition location
definition = definitions[0]
def_path = definition["relativePath"]
assert def_path is not None
def_line = definition["range"]["start"]["line"]
def_col = definition["range"]["start"]["character"]
@@ -1702,7 +1716,7 @@ class SolidLanguageServer(ABC):
return defining_symbol
def _save_raw_document_symbols_cache(self):
def _save_raw_document_symbols_cache(self) -> None:
cache_file = self.cache_dir / self.RAW_DOCUMENT_SYMBOL_CACHE_FILENAME
if not self._raw_document_symbols_cache_is_modified:
@@ -1720,10 +1734,10 @@ class SolidLanguageServer(ABC):
e,
)
def _raw_document_symbols_cache_version(self):
def _raw_document_symbols_cache_version(self) -> tuple[int, Hashable]:
return (self.RAW_DOCUMENT_SYMBOLS_CACHE_VERSION, self._ls_specific_raw_document_symbols_cache_version)
def _load_raw_document_symbols_cache(self):
def _load_raw_document_symbols_cache(self) -> None:
cache_file = self.cache_dir / self.RAW_DOCUMENT_SYMBOL_CACHE_FILENAME
if not cache_file.exists():
@@ -1743,7 +1757,7 @@ class SolidLanguageServer(ABC):
migrated_cache[new_cache_key] = (file_hash, root_symbols)
num_symbols_migrated += len(all_symbols)
log.info("Migrated %d document symbols from legacy cache", num_symbols_migrated)
self._raw_document_symbols_cache = migrated_cache
self._raw_document_symbols_cache = migrated_cache # type: ignore
self._raw_document_symbols_cache_is_modified = True
self._save_raw_document_symbols_cache()
legacy_cache_file.unlink()
@@ -1770,7 +1784,7 @@ class SolidLanguageServer(ABC):
e,
)
def _save_document_symbols_cache(self):
def _save_document_symbols_cache(self) -> None:
cache_file = self.cache_dir / self.DOCUMENT_SYMBOL_CACHE_FILENAME
if not self._document_symbols_cache_is_modified:
@@ -1788,7 +1802,7 @@ class SolidLanguageServer(ABC):
e,
)
def _load_document_symbols_cache(self):
def _load_document_symbols_cache(self) -> None:
cache_file = self.cache_dir / self.DOCUMENT_SYMBOL_CACHE_FILENAME
if cache_file.exists():
log.info("Loading document symbols cache from %s", cache_file)
@@ -1805,7 +1819,7 @@ class SolidLanguageServer(ABC):
e,
)
def save_cache(self):
def save_cache(self) -> None:
self._save_raw_document_symbols_cache()
self._save_document_symbols_cache()
@@ -1832,7 +1846,7 @@ class SolidLanguageServer(ABC):
assert LSPConstants.KIND in item
assert LSPConstants.LOCATION in item
ret.append(ls_types.UnifiedSymbolInformation(**item))
ret.append(ls_types.UnifiedSymbolInformation(**item)) # type: ignore
return ret
@@ -1853,7 +1867,7 @@ class SolidLanguageServer(ABC):
:param new_name: The new name for the symbol
:return: A WorkspaceEdit containing the changes needed to rename the symbol, or None if rename is not supported
"""
params = ls_types.RenameParams(
params = RenameParams(
textDocument=ls_types.TextDocumentIdentifier(
uri=pathlib.Path(os.path.join(self.repository_root_path, relative_file_path)).as_uri()
),
@@ -1892,7 +1906,7 @@ class SolidLanguageServer(ABC):
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_process()
self._start_server_process()
return self
def stop(self, shutdown_timeout: float = 2.0) -> None:
+2 -2
View File
@@ -274,7 +274,7 @@ class Language(str, Enum):
return ZigLanguageServer
case self.NIX:
from solidlsp.language_servers.nixd_ls import NixLanguageServer
from solidlsp.language_servers.nixd_ls import NixLanguageServer # type: ignore
return NixLanguageServer
case self.LUA:
@@ -350,7 +350,7 @@ class LanguageServerConfig:
"""File encoding to use when reading source files"""
@classmethod
def from_dict(cls, env: dict):
def from_dict(cls, env: dict) -> Self:
"""
Create a MultilspyConfig instance from a dictionary
"""
+2 -2
View File
@@ -6,7 +6,7 @@ from solidlsp.ls_config import Language
class SolidLSPException(Exception):
def __init__(self, message: str, cause: Exception | None = None):
def __init__(self, message: str, cause: Exception | None = None) -> None:
"""
Initializes the exception with the given message.
@@ -19,7 +19,7 @@ class SolidLSPException(Exception):
self.cause = cause
super().__init__(message)
def is_language_server_terminated(self):
def is_language_server_terminated(self) -> bool:
"""
:return: True if the exception is caused by the language server having terminated as indicated
by the causing exception being an instance of LanguageServerTerminatedException.
+17 -17
View File
@@ -66,7 +66,7 @@ class Request(ToStringMixin):
self._request_id = request_id
self._method = method
self._status = "pending"
self._result_queue = Queue()
self._result_queue: Queue[Request.Result] = Queue()
def _tostring_includes(self) -> list[str]:
return ["_request_id", "_status", "_method"]
@@ -136,7 +136,7 @@ class SolidLanguageServerHandler:
language: Language,
determine_log_level: Callable[[str], int],
logger: Callable[[str, str, StringDict | str], None] | None = None,
start_independent_lsp_process=True,
start_independent_lsp_process: bool = True,
request_timeout: float | None = None,
) -> None:
self.language = language
@@ -145,15 +145,15 @@ class SolidLanguageServerHandler:
self.notify = LspNotification(self.send_notification)
self.process_launch_info = process_launch_info
self.process: subprocess.Popen | None = None
self.process: subprocess.Popen[bytes] | None = None
self._is_shutting_down = False
self.request_id = 1
self._pending_requests: dict[Any, Request] = {}
self.on_request_handlers = {}
self.on_notification_handlers = {}
self.on_request_handlers: dict[str, Callable[[Any], Any]] = {}
self.on_notification_handlers: dict[str, Callable[[Any], None]] = {}
self.logger = logger
self.tasks = {}
self.tasks: dict[int, Any] = {}
self.task_counter = 0
self.loop = None
self.start_independent_lsp_process = start_independent_lsp_process
@@ -209,7 +209,7 @@ class SolidLanguageServerHandler:
if self.process.returncode is not None:
log.error("Language server has already terminated/could not be started")
# Process has already terminated
stderr_data = self.process.stderr.read()
stderr_data = self.process.stderr.read() if self.process.stderr else b""
error_message = stderr_data.decode("utf-8", errors="replace")
raise RuntimeError(f"Process terminated immediately with code {self.process.returncode}. Error: {error_message}")
@@ -234,7 +234,7 @@ class SolidLanguageServerHandler:
if process:
self._cleanup_process(process)
def _cleanup_process(self, process):
def _cleanup_process(self, process: subprocess.Popen[bytes]) -> None:
"""Clean up a process: close stdin, terminate/kill process, close stdout/stderr."""
# Close stdin first to prevent deadlocks
# See: https://bugs.python.org/issue35539
@@ -251,7 +251,7 @@ class SolidLanguageServerHandler:
self._safely_close_pipe(process.stdout)
self._safely_close_pipe(process.stderr)
def _safely_close_pipe(self, pipe):
def _safely_close_pipe(self, pipe: Any) -> None:
"""Safely close a pipe, ignoring any exceptions."""
if pipe:
try:
@@ -259,12 +259,12 @@ class SolidLanguageServerHandler:
except Exception:
pass
def _terminate_or_kill_process(self, process):
def _terminate_or_kill_process(self, process: subprocess.Popen[bytes]) -> None:
"""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)
def _signal_process_tree(self, process, terminate=True):
def _signal_process_tree(self, process: subprocess.Popen[bytes], terminate: bool = True) -> None:
"""Send signal (terminate or kill) to the process and all its children."""
signal_method = "terminate" if terminate else "kill"
@@ -315,7 +315,7 @@ class SolidLanguageServerHandler:
if self.logger is not None:
self.logger("client", "logger", message)
def _read_bytes_from_process(self, process, stream, num_bytes):
def _read_bytes_from_process(self, process, stream, num_bytes) -> bytes: # type: ignore
"""Read exactly num_bytes from process stdout"""
data = b""
while len(data) < num_bytes:
@@ -385,9 +385,9 @@ class SolidLanguageServerHandler:
line = self.process.stderr.readline()
if not line:
continue
line = line.decode(ENCODING, errors="replace")
level = self._determine_log_level(line)
log.log(level, line)
line_str = line.decode(ENCODING, errors="replace")
level = self._determine_log_level(line_str)
log.log(level, line_str)
except Exception as e:
log.error("Error while reading stderr from language server process: %s", e, exc_info=e)
if not self._is_shutting_down:
@@ -504,13 +504,13 @@ class SolidLanguageServerHandler:
self.logger("client", "logger", f"Failed to write to stdin: {e}")
return
def on_request(self, method: str, cb) -> None:
def on_request(self, method: str, cb: Callable[[Any], Any]) -> None:
"""
Register the callback function to handle requests from the server to the client for the given method
"""
self.on_request_handlers[method] = cb
def on_notification(self, method: str, cb) -> None:
def on_notification(self, method: str, cb: Callable[[Any], None]) -> None:
"""
Register the callback function to handle notifications from the server to the client for the given method
"""
+6 -20
View File
@@ -9,6 +9,8 @@ from typing import NotRequired, Union
from typing_extensions import TypedDict
from solidlsp.lsp_protocol_handler.lsp_types import DiagnosticSeverity
URI = str
DocumentUri = str
Uint = int
@@ -144,6 +146,8 @@ class CompletionItem(TypedDict):
class SymbolKind(IntEnum):
"""A symbol kind."""
# TODO: This is a duplicate of SymbolKind in lsp_types.
File = 1
Module = 2
Namespace = 3
@@ -327,13 +331,6 @@ class Hover(TypedDict):
visualize the hover, e.g. by changing the background color. """
class DiagnosticsSeverity(IntEnum):
ERROR = 1
WARNING = 2
INFORMATION = 3
HINT = 4
class TextDocumentIdentifier(TypedDict):
"""A literal to identify a text document in the client."""
@@ -359,17 +356,6 @@ class WorkspaceEdit(TypedDict):
""" Document changes array for versioned edits. """
class RenameParams(TypedDict):
"""The parameters of a RenameRequest."""
textDocument: TextDocumentIdentifier
""" The document to rename. """
position: Position
""" The position at which this request was sent. """
newName: str
""" The new name of the symbol. """
class Diagnostic(TypedDict):
"""Diagnostic information for a text document."""
@@ -377,7 +363,7 @@ class Diagnostic(TypedDict):
""" The URI of the text document to which the diagnostics apply. """
range: Range
""" The range of the text document to which the diagnostics apply. """
severity: NotRequired[DiagnosticsSeverity]
severity: NotRequired[DiagnosticSeverity]
""" The severity of the diagnostic. """
message: str
""" The diagnostic message. """
@@ -409,4 +395,4 @@ def extract_text_edits(workspace_edit: WorkspaceEdit) -> dict[str, list[TextEdit
changes[uri] = edits
return changes
else:
raise f"Invalid WorkspaceEdit (expected 'changes' or 'documentChanges' key):\n{workspace_edit}"
raise Exception(f"Invalid WorkspaceEdit (expected 'changes' or 'documentChanges' key):\n{workspace_edit}")
+11 -6
View File
@@ -129,10 +129,15 @@ class PathUtils:
from urllib.parse import unquote, urlparse
from urllib.request import url2pathname
except ImportError:
# backwards compatibility
from urllib import unquote, url2pathname
# backwards compatibility (Python 2)
from urllib.parse import unquote as unquote_py2
from urllib.request import url2pathname as url2pathname_py2
from urlparse import urlparse
from urlparse import urlparse as urlparse_py2
unquote = unquote_py2
url2pathname = url2pathname_py2
urlparse = urlparse_py2
parsed = urlparse(uri)
host = f"{os.path.sep}{os.path.sep}{parsed.netloc}{os.path.sep}"
path = os.path.normpath(os.path.join(host, url2pathname(unquote(parsed.path))))
@@ -281,7 +286,7 @@ class PlatformId(str, Enum):
LINUX_MUSL_x64 = "linux-musl-x64"
LINUX_MUSL_arm64 = "linux-musl-arm64"
def is_windows(self):
def is_windows(self) -> bool:
return self.value.startswith("win")
@@ -334,7 +339,7 @@ class PlatformUtils:
raise SolidLSPException(f"Unknown platform: {system=}, {machine=}, {bitness=}")
@staticmethod
def _determine_windows_machine_type():
def _determine_windows_machine_type() -> str:
import ctypes
from ctypes import wintypes
@@ -361,7 +366,7 @@ class PlatformUtils:
_anonymous_ = ("u",)
sys_info = SYSTEM_INFO()
ctypes.windll.kernel32.GetNativeSystemInfo(ctypes.byref(sys_info))
ctypes.windll.kernel32.GetNativeSystemInfo(ctypes.byref(sys_info)) # type: ignore
arch_map = {
9: "AMD64",
@@ -29,13 +29,13 @@ OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
"""
from typing import Union
from typing import Any, Union
from solidlsp.lsp_protocol_handler import lsp_types
class LspRequest:
def __init__(self, send_request):
def __init__(self, send_request: Any) -> None:
self.send_request = send_request
async def implementation(
@@ -421,7 +421,7 @@ class LspRequest:
class LspNotification:
def __init__(self, send_notification):
def __init__(self, send_notification: Any) -> None:
self.send_notification = send_notification
def did_change_workspace_folders(self, params: lsp_types.DidChangeWorkspaceFoldersParams) -> None:
@@ -2478,6 +2478,8 @@ class DocumentSymbol(TypedDict):
""" The range that should be selected and revealed when this symbol is being picked, e.g the name of a function.
Must be contained by the `range`. """
# TODO: I think this type is missing the 'children' field - DJ
class DocumentSymbolRegistrationOptions(TypedDict):
"""Registration options for a {@link DocumentSymbolRequest}."""
+3 -3
View File
@@ -37,7 +37,7 @@ from typing import Any, Union
from .lsp_types import ErrorCodes
StringDict = dict[str, Any]
PayloadLike = Union[list[StringDict], StringDict, None]
PayloadLike = Union[list[StringDict], StringDict, None, bool]
CONTENT_LENGTH = "Content-Length: "
ENCODING = "utf-8"
log = logging.getLogger(__name__)
@@ -95,7 +95,7 @@ class StopLoopException(Exception):
pass
def create_message(payload: PayloadLike):
def create_message(payload: PayloadLike) -> tuple[bytes, bytes, bytes]:
body = json.dumps(payload, check_circular=False, ensure_ascii=False, separators=(",", ":")).encode(ENCODING)
return (
f"Content-Length: {len(body)}\r\n".encode(ENCODING),
@@ -118,5 +118,5 @@ def content_length(line: bytes) -> int | None:
try:
return int(value)
except ValueError:
raise ValueError(f"Invalid Content-Length header: {value}")
raise ValueError(f"Invalid Content-Length header: {value!r}")
return None
+2 -2
View File
@@ -35,12 +35,12 @@ class SolidLSPSettings:
No documentation on options means no options are available.
"""
def __post_init__(self):
def __post_init__(self) -> None:
os.makedirs(str(self.solidlsp_dir), exist_ok=True)
os.makedirs(str(self.ls_resources_dir), exist_ok=True)
@property
def ls_resources_dir(self):
def ls_resources_dir(self) -> str:
return os.path.join(str(self.solidlsp_dir), "language_servers", "static")
class CustomLSSettings(ToStringMixin):
+1 -1
View File
@@ -2,7 +2,7 @@ import platform
import subprocess
def subprocess_kwargs():
def subprocess_kwargs() -> dict:
"""
Returns a dictionary of keyword arguments for subprocess calls, adding platform-specific
flags that we want to use consistently.
+1 -1
View File
@@ -115,7 +115,7 @@ class SafeZipExtractor:
"""
if sys.platform.startswith("win"):
return Path(rf"\\?\{os.path.abspath(path)}")
return path
return path # type: ignore
# Example usage: