diff --git a/.github/workflows/pytest.yml b/.github/workflows/pytest.yml index 22209e48..4c36b51a 100644 --- a/.github/workflows/pytest.yml +++ b/.github/workflows/pytest.yml @@ -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 diff --git a/pyproject.toml b/pyproject.toml index 6f353f7a..618d1b3f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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" diff --git a/src/solidlsp/language_servers/al_language_server.py b/src/solidlsp/language_servers/al_language_server.py index 2bc1b1ad..b98551bd 100644 --- a/src/solidlsp/language_servers/al_language_server.py +++ b/src/solidlsp/language_servers/al_language_server.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 diff --git a/src/solidlsp/language_servers/bash_language_server.py b/src/solidlsp/language_servers/bash_language_server.py index 904d51fe..6759b5a6 100644 --- a/src/solidlsp/language_servers/bash_language_server.py +++ b/src/solidlsp/language_servers/bash_language_server.py @@ -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", "") diff --git a/src/solidlsp/language_servers/clangd_language_server.py b/src/solidlsp/language_servers/clangd_language_server.py index 2a1f9e8e..d4f67e9b 100644 --- a/src/solidlsp/language_servers/clangd_language_server.py +++ b/src/solidlsp/language_servers/clangd_language_server.py @@ -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": [".", "<", ">", ":", '"', "/", "*"], diff --git a/src/solidlsp/language_servers/clojure_lsp.py b/src/solidlsp/language_servers/clojure_lsp.py index 36583e85..241e57a0 100644 --- a/src/solidlsp/language_servers/clojure_lsp.py +++ b/src/solidlsp/language_servers/clojure_lsp.py @@ -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 diff --git a/src/solidlsp/language_servers/common.py b/src/solidlsp/language_servers/common.py index 8505141e..c1cea673 100644 --- a/src/solidlsp/language_servers/common.py +++ b/src/solidlsp/language_servers/common.py @@ -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) diff --git a/src/solidlsp/language_servers/csharp_language_server.py b/src/solidlsp/language_servers/csharp_language_server.py index 09b2570d..2382b761 100644 --- a/src/solidlsp/language_servers/csharp_language_server.py +++ b/src/solidlsp/language_servers/csharp_language_server.py @@ -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): diff --git a/src/solidlsp/language_servers/dart_language_server.py b/src/solidlsp/language_servers/dart_language_server.py index 297294a4..0135ae7a 100644 --- a/src/solidlsp/language_servers/dart_language_server.py +++ b/src/solidlsp/language_servers/dart_language_server.py @@ -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, diff --git a/src/solidlsp/language_servers/eclipse_jdtls.py b/src/solidlsp/language_servers/eclipse_jdtls.py index cfe7d28c..7c718c07 100644 --- a/src/solidlsp/language_servers/eclipse_jdtls.py +++ b/src/solidlsp/language_servers/eclipse_jdtls.py @@ -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) ", 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: diff --git a/src/solidlsp/language_servers/elixir_tools/elixir_tools.py b/src/solidlsp/language_servers/elixir_tools/elixir_tools.py index b5b8643e..fa14d3e0 100644 --- a/src/solidlsp/language_servers/elixir_tools/elixir_tools.py +++ b/src/solidlsp/language_servers/elixir_tools/elixir_tools.py @@ -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. diff --git a/src/solidlsp/language_servers/elm_language_server.py b/src/solidlsp/language_servers/elm_language_server.py index 6f5a6baa..f75af40c 100644 --- a/src/solidlsp/language_servers/elm_language_server.py +++ b/src/solidlsp/language_servers/elm_language_server.py @@ -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) diff --git a/src/solidlsp/language_servers/erlang_language_server.py b/src/solidlsp/language_servers/erlang_language_server.py index aaa85f1c..36651688 100644 --- a/src/solidlsp/language_servers/erlang_language_server.py +++ b/src/solidlsp/language_servers/erlang_language_server.py @@ -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: diff --git a/src/solidlsp/language_servers/fortran_language_server.py b/src/solidlsp/language_servers/fortran_language_server.py index 782df7b3..993c21fa 100644 --- a/src/solidlsp/language_servers/fortran_language_server.py +++ b/src/solidlsp/language_servers/fortran_language_server.py @@ -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 diff --git a/src/solidlsp/language_servers/gopls.py b/src/solidlsp/language_servers/gopls.py index 5b16c2dc..9f9aec7a 100644 --- a/src/solidlsp/language_servers/gopls.py +++ b/src/solidlsp/language_servers/gopls.py @@ -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) diff --git a/src/solidlsp/language_servers/haskell_language_server.py b/src/solidlsp/language_servers/haskell_language_server.py index 1db2dc6a..f1353a34 100644 --- a/src/solidlsp/language_servers/haskell_language_server.py +++ b/src/solidlsp/language_servers/haskell_language_server.py @@ -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) diff --git a/src/solidlsp/language_servers/intelephense.py b/src/solidlsp/language_servers/intelephense.py index aabcb583..dc62a70e 100644 --- a/src/solidlsp/language_servers/intelephense.py +++ b/src/solidlsp/language_servers/intelephense.py @@ -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) diff --git a/src/solidlsp/language_servers/jedi_server.py b/src/solidlsp/language_servers/jedi_server.py index 956e00b9..f08cd09a 100644 --- a/src/solidlsp/language_servers/jedi_server.py +++ b/src/solidlsp/language_servers/jedi_server.py @@ -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": [".", "'", '"'], diff --git a/src/solidlsp/language_servers/julia_server.py b/src/solidlsp/language_servers/julia_server.py index 05a0cbbb..178ac91b 100644 --- a/src/solidlsp/language_servers/julia_server.py +++ b/src/solidlsp/language_servers/julia_server.py @@ -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) diff --git a/src/solidlsp/language_servers/kotlin_language_server.py b/src/solidlsp/language_servers/kotlin_language_server.py index 0ea2b4da..c1c6b975 100644 --- a/src/solidlsp/language_servers/kotlin_language_server.py +++ b/src/solidlsp/language_servers/kotlin_language_server.py @@ -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) diff --git a/src/solidlsp/language_servers/lua_ls.py b/src/solidlsp/language_servers/lua_ls.py index c0f64b37..0ff6b7c4 100644 --- a/src/solidlsp/language_servers/lua_ls.py +++ b/src/solidlsp/language_servers/lua_ls.py @@ -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) diff --git a/src/solidlsp/language_servers/marksman.py b/src/solidlsp/language_servers/marksman.py index c35d2ca6..85919113 100644 --- a/src/solidlsp/language_servers/marksman.py +++ b/src/solidlsp/language_servers/marksman.py @@ -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) diff --git a/src/solidlsp/language_servers/nixd_ls.py b/src/solidlsp/language_servers/nixd_ls.py index beea7c6b..edf092e3 100644 --- a/src/solidlsp/language_servers/nixd_ls.py +++ b/src/solidlsp/language_servers/nixd_ls.py @@ -1,3 +1,4 @@ +# type: ignore """ Provides Nix specific instantiation of the LanguageServer class using nixd (Nix Language Server). diff --git a/src/solidlsp/language_servers/omnisharp.py b/src/solidlsp/language_servers/omnisharp.py index e43287c5..5c5be862 100644 --- a/src/solidlsp/language_servers/omnisharp.py +++ b/src/solidlsp/language_servers/omnisharp.py @@ -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 [ { diff --git a/src/solidlsp/language_servers/perl_language_server.py b/src/solidlsp/language_servers/perl_language_server.py index 0552c834..5937b6a4 100644 --- a/src/solidlsp/language_servers/perl_language_server.py +++ b/src/solidlsp/language_servers/perl_language_server.py @@ -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, "."], diff --git a/src/solidlsp/language_servers/pyright_server.py b/src/solidlsp/language_servers/pyright_server.py index c760ebc1..10cc7886 100644 --- a/src/solidlsp/language_servers/pyright_server.py +++ b/src/solidlsp/language_servers/pyright_server.py @@ -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 """ diff --git a/src/solidlsp/language_servers/r_language_server.py b/src/solidlsp/language_servers/r_language_server.py index 91aa6935..467cc144 100644 --- a/src/solidlsp/language_servers/r_language_server.py +++ b/src/solidlsp/language_servers/r_language_server.py @@ -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 diff --git a/src/solidlsp/language_servers/ruby_lsp.py b/src/solidlsp/language_servers/ruby_lsp.py index 0bf2c207..95c9f7b6 100644 --- a/src/solidlsp/language_servers/ruby_lsp.py +++ b/src/solidlsp/language_servers/ruby_lsp.py @@ -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. """ diff --git a/src/solidlsp/language_servers/rust_analyzer.py b/src/solidlsp/language_servers/rust_analyzer.py index c8125f85..aa5617c5 100644 --- a/src/solidlsp/language_servers/rust_analyzer.py +++ b/src/solidlsp/language_servers/rust_analyzer.py @@ -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, diff --git a/src/solidlsp/language_servers/scala_language_server.py b/src/solidlsp/language_servers/scala_language_server.py index 5a7ca14c..5911c453 100644 --- a/src/solidlsp/language_servers/scala_language_server.py +++ b/src/solidlsp/language_servers/scala_language_server.py @@ -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 """ diff --git a/src/solidlsp/language_servers/solargraph.py b/src/solidlsp/language_servers/solargraph.py index 6ba7375a..61249cde 100644 --- a/src/solidlsp/language_servers/solargraph.py +++ b/src/solidlsp/language_servers/solargraph.py @@ -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) diff --git a/src/solidlsp/language_servers/sourcekit_lsp.py b/src/solidlsp/language_servers/sourcekit_lsp.py index 2c7ba472..d21eac4d 100644 --- a/src/solidlsp/language_servers/sourcekit_lsp.py +++ b/src/solidlsp/language_servers/sourcekit_lsp.py @@ -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) diff --git a/src/solidlsp/language_servers/terraform_ls.py b/src/solidlsp/language_servers/terraform_ls.py index e83c6cb6..66fd01a1 100644 --- a/src/solidlsp/language_servers/terraform_ls.py +++ b/src/solidlsp/language_servers/terraform_ls.py @@ -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) diff --git a/src/solidlsp/language_servers/typescript_language_server.py b/src/solidlsp/language_servers/typescript_language_server.py index 51517474..2ebcb869 100644 --- a/src/solidlsp/language_servers/typescript_language_server.py +++ b/src/solidlsp/language_servers/typescript_language_server.py @@ -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 """ diff --git a/src/solidlsp/language_servers/vts_language_server.py b/src/solidlsp/language_servers/vts_language_server.py index b07e4c29..51922e70 100644 --- a/src/solidlsp/language_servers/vts_language_server.py +++ b/src/solidlsp/language_servers/vts_language_server.py @@ -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 """ diff --git a/src/solidlsp/language_servers/yaml_language_server.py b/src/solidlsp/language_servers/yaml_language_server.py index 16ffa979..6e2dc959 100644 --- a/src/solidlsp/language_servers/yaml_language_server.py +++ b/src/solidlsp/language_servers/yaml_language_server.py @@ -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) diff --git a/src/solidlsp/language_servers/zls.py b/src/solidlsp/language_servers/zls.py index 893a4681..9ab8daac 100644 --- a/src/solidlsp/language_servers/zls.py +++ b/src/solidlsp/language_servers/zls.py @@ -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) diff --git a/src/solidlsp/ls.py b/src/solidlsp/ls.py index 03c63dbe..fcb0734f 100644 --- a/src/solidlsp/ls.py +++ b/src/solidlsp/ls.py @@ -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: diff --git a/src/solidlsp/ls_config.py b/src/solidlsp/ls_config.py index 41f61f7b..7a09588f 100644 --- a/src/solidlsp/ls_config.py +++ b/src/solidlsp/ls_config.py @@ -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 """ diff --git a/src/solidlsp/ls_exceptions.py b/src/solidlsp/ls_exceptions.py index 61751eff..a7c7edb4 100644 --- a/src/solidlsp/ls_exceptions.py +++ b/src/solidlsp/ls_exceptions.py @@ -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. diff --git a/src/solidlsp/ls_handler.py b/src/solidlsp/ls_handler.py index ee1dd2a7..e8ddbcae 100644 --- a/src/solidlsp/ls_handler.py +++ b/src/solidlsp/ls_handler.py @@ -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 """ diff --git a/src/solidlsp/ls_types.py b/src/solidlsp/ls_types.py index bbd3cbad..a1baab88 100644 --- a/src/solidlsp/ls_types.py +++ b/src/solidlsp/ls_types.py @@ -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}") diff --git a/src/solidlsp/ls_utils.py b/src/solidlsp/ls_utils.py index 0cec5994..08e6a279 100644 --- a/src/solidlsp/ls_utils.py +++ b/src/solidlsp/ls_utils.py @@ -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", diff --git a/src/solidlsp/lsp_protocol_handler/lsp_requests.py b/src/solidlsp/lsp_protocol_handler/lsp_requests.py index 383e4566..78f308b2 100644 --- a/src/solidlsp/lsp_protocol_handler/lsp_requests.py +++ b/src/solidlsp/lsp_protocol_handler/lsp_requests.py @@ -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: diff --git a/src/solidlsp/lsp_protocol_handler/lsp_types.py b/src/solidlsp/lsp_protocol_handler/lsp_types.py index 5488c358..4c1753b5 100644 --- a/src/solidlsp/lsp_protocol_handler/lsp_types.py +++ b/src/solidlsp/lsp_protocol_handler/lsp_types.py @@ -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}.""" diff --git a/src/solidlsp/lsp_protocol_handler/server.py b/src/solidlsp/lsp_protocol_handler/server.py index 62d50e7f..219135b7 100644 --- a/src/solidlsp/lsp_protocol_handler/server.py +++ b/src/solidlsp/lsp_protocol_handler/server.py @@ -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 diff --git a/src/solidlsp/settings.py b/src/solidlsp/settings.py index dff5df9a..ad77eb22 100644 --- a/src/solidlsp/settings.py +++ b/src/solidlsp/settings.py @@ -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): diff --git a/src/solidlsp/util/subprocess_util.py b/src/solidlsp/util/subprocess_util.py index 718d7afa..c0735f0c 100644 --- a/src/solidlsp/util/subprocess_util.py +++ b/src/solidlsp/util/subprocess_util.py @@ -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. diff --git a/src/solidlsp/util/zip.py b/src/solidlsp/util/zip.py index d93d9421..bf014923 100644 --- a/src/solidlsp/util/zip.py +++ b/src/solidlsp/util/zip.py @@ -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: