From 3a05cac93e463960faea96ea35c47cb8c31eac58 Mon Sep 17 00:00:00 2001 From: MischaPanch Date: Sat, 26 Jul 2025 13:49:20 +0200 Subject: [PATCH] Make mypy run through on solidlsp --- pyproject.toml | 2 +- .../clangd_language_server.py | 19 +-- src/solidlsp/language_servers/clojure_lsp.py | 26 ++-- src/solidlsp/language_servers/common.py | 3 + .../csharp_language_server.py | 34 +++--- .../language_servers/dart_language_server.py | 23 ++-- .../language_servers/eclipse_jdtls.py | 29 ++--- .../elixir_tools/elixir_tools.py | 21 ++-- src/solidlsp/language_servers/gopls.py | 17 +-- src/solidlsp/language_servers/intelephense.py | 18 +-- src/solidlsp/language_servers/jedi_server.py | 15 +-- .../kotlin_language_server.py | 15 +-- src/solidlsp/language_servers/omnisharp.py | 18 +-- .../language_servers/pyright_server.py | 15 +-- .../language_servers/rust_analyzer.py | 19 +-- src/solidlsp/language_servers/solargraph.py | 25 ++-- src/solidlsp/language_servers/terraform_ls.py | 14 ++- .../typescript_language_server.py | 22 ++-- .../language_servers/vts_language_server.py | 20 +-- src/solidlsp/ls.py | 114 +++++++++--------- src/solidlsp/ls_config.py | 2 +- src/solidlsp/ls_handler.py | 28 ++--- src/solidlsp/ls_request.py | 7 +- src/solidlsp/ls_utils.py | 15 ++- .../lsp_protocol_handler/lsp_requests.py | 6 +- src/solidlsp/lsp_protocol_handler/server.py | 4 +- src/solidlsp/settings.py | 4 +- 27 files changed, 291 insertions(+), 244 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index e79567c2..24797bf2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -119,7 +119,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"] [tool.ruff] diff --git a/src/solidlsp/language_servers/clangd_language_server.py b/src/solidlsp/language_servers/clangd_language_server.py index a770d78e..03cb3360 100644 --- a/src/solidlsp/language_servers/clangd_language_server.py +++ b/src/solidlsp/language_servers/clangd_language_server.py @@ -7,6 +7,7 @@ import os import pathlib import stat import threading +from typing import Any, cast from solidlsp.ls import SolidLanguageServer from solidlsp.ls_config import LanguageServerConfig @@ -126,9 +127,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. @@ -142,7 +143,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": @@ -150,24 +151,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: Any) -> None: self.logger.log(f"LSP: window/logMessage: {msg}", logging.INFO) self.server.on_request("client/registerCapability", register_capability_handler) @@ -188,7 +189,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 f973db15..e1098e6f 100644 --- a/src/solidlsp/language_servers/clojure_lsp.py +++ b/src/solidlsp/language_servers/clojure_lsp.py @@ -9,6 +9,7 @@ import shutil import stat import subprocess import threading +from typing import cast from solidlsp.ls import SolidLanguageServer from solidlsp.ls_config import LanguageServerConfig @@ -26,7 +27,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) @@ -125,7 +126,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, @@ -153,9 +154,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": @@ -163,24 +165,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) @@ -202,11 +204,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 a0973e56..68be342f 100644 --- a/src/solidlsp/language_servers/common.py +++ b/src/solidlsp/language_servers/common.py @@ -94,6 +94,9 @@ class RuntimeDependencyCollection: @staticmethod def _install_from_url(dep: RuntimeDependency, logger: LanguageServerLogger, target_dir: str) -> None: + if not dep.url: + raise ValueError(f"Dependency {dep.id} has no URL") + if dep.archive_type == "gz" 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 88bf9bc4..080ea88a 100644 --- a/src/solidlsp/language_servers/csharp_language_server.py +++ b/src/solidlsp/language_servers/csharp_language_server.py @@ -13,8 +13,9 @@ import tarfile import threading import urllib.request import zipfile +from collections.abc import Iterable from pathlib import Path -from typing import cast +from typing import Any, cast from overrides import override @@ -23,7 +24,7 @@ from solidlsp.ls_config import LanguageServerConfig from solidlsp.ls_exceptions import LanguageServerException 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 @@ -126,7 +127,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. @@ -148,7 +149,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. @@ -296,6 +297,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(): @@ -304,6 +306,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 @@ -423,6 +427,7 @@ class CSharpLanguageServer(SolidLanguageServer): # Download .NET 9 runtime using config dotnet_dir = Path(cls.ls_resources_dir(solidlsp_settings)) / "dotnet-runtime-9.0" + assert runtime_dep.binary_name is not None dotnet_exe = dotnet_dir / runtime_dep.binary_name if dotnet_exe.exists(): @@ -434,6 +439,7 @@ class CSharpLanguageServer(SolidLanguageServer): dotnet_dir.mkdir(parents=True, exist_ok=True) url = runtime_dep.url + assert url is not None archive_type = runtime_dep.archive_type # Download the runtime @@ -519,11 +525,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 @@ -533,7 +539,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", {}) @@ -567,10 +573,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", "") @@ -613,12 +619,12 @@ 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 @@ -683,13 +689,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 c9cfcad1..901fc3f4 100644 --- a/src/solidlsp/language_servers/dart_language_server.py +++ b/src/solidlsp/language_servers/dart_language_server.py @@ -2,12 +2,15 @@ import logging import os import pathlib import stat +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 @@ -16,7 +19,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. """ @@ -89,7 +94,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. """ @@ -115,23 +120,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) @@ -150,7 +155,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 bf87aab7..1f43d0c8 100644 --- a/src/solidlsp/language_servers/eclipse_jdtls.py +++ b/src/solidlsp/language_servers/eclipse_jdtls.py @@ -11,6 +11,7 @@ import stat import threading import uuid from pathlib import PurePath +from typing import cast from overrides import override @@ -625,29 +626,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": @@ -665,22 +666,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) @@ -700,13 +701,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() diff --git a/src/solidlsp/language_servers/elixir_tools/elixir_tools.py b/src/solidlsp/language_servers/elixir_tools/elixir_tools.py index 5f6a9794..b596bba2 100644 --- a/src/solidlsp/language_servers/elixir_tools/elixir_tools.py +++ b/src/solidlsp/language_servers/elixir_tools/elixir_tools.py @@ -5,6 +5,7 @@ import stat import subprocess import threading import time +from typing import Any, cast from overrides import override @@ -47,7 +48,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 | None: """Override to filter out Next LS internal files from references.""" from solidlsp.ls_utils import PathUtils @@ -70,7 +71,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) @@ -145,10 +146,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 @@ -243,15 +246,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) @@ -262,10 +265,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 @@ -280,7 +283,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/gopls.py b/src/solidlsp/language_servers/gopls.py index c21663b7..d74b0f7d 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 @@ -28,7 +29,7 @@ class Gopls(SolidLanguageServer): return super().is_ignored_dirname(dirname) or dirname in ["vendor", "node_modules", "dist", "build"] @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) @@ -39,7 +40,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) @@ -50,7 +51,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. @@ -117,18 +118,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/intelephense.py b/src/solidlsp/language_servers/intelephense.py index 2d1d48e2..25b7e5d6 100644 --- a/src/solidlsp/language_servers/intelephense.py +++ b/src/solidlsp/language_servers/intelephense.py @@ -7,6 +7,7 @@ import os import pathlib import shutil from time import sleep +from typing import cast from overrides import override @@ -14,10 +15,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 @@ -124,18 +126,18 @@ class Intelephense(SolidLanguageServer): ], } - return initialize_params + return cast(InitializeParams, initialize_params) - 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) @@ -170,7 +172,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. @@ -180,7 +182,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/kotlin_language_server.py b/src/solidlsp/language_servers/kotlin_language_server.py index 5d0d2083..56ba98f4 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/omnisharp.py b/src/solidlsp/language_servers/omnisharp.py index a23a58a8..9392f1df 100644 --- a/src/solidlsp/language_servers/omnisharp.py +++ b/src/solidlsp/language_servers/omnisharp.py @@ -22,7 +22,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. @@ -48,7 +48,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 @@ -205,12 +205,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": @@ -220,7 +220,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? @@ -228,17 +228,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/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/rust_analyzer.py b/src/solidlsp/language_servers/rust_analyzer.py index 48bae28a..4c448990 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 stat import threading +from typing import cast from overrides import override @@ -580,14 +581,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": @@ -595,24 +596,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) @@ -633,7 +634,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/solargraph.py b/src/solidlsp/language_servers/solargraph.py index 56a1b7f7..73c46ba0 100644 --- a/src/solidlsp/language_servers/solargraph.py +++ b/src/solidlsp/language_servers/solargraph.py @@ -10,6 +10,7 @@ import pathlib import stat import subprocess import threading +from typing import cast from overrides import override @@ -71,7 +72,7 @@ class Solargraph(SolidLanguageServer): # Check if Ruby is installed try: result = subprocess.run(["ruby", "--version"], check=True, capture_output=True, cwd=repository_root_path) - ruby_version = result.stdout.strip() + ruby_version = str(result.stdout).strip() logger.log(f"Ruby version: {ruby_version}", logging.INFO) except subprocess.CalledProcessError as e: raise RuntimeError(f"Error checking for Ruby installation: {e.stderr}") from e @@ -81,16 +82,16 @@ class Solargraph(SolidLanguageServer): # Check if solargraph is installed try: result = subprocess.run( - ["gem", "list", "^solargraph$", "-i"], check=False, capture_output=True, text=True, cwd=repository_root_path + ["gem", "list", "^solargraph$", "-i"], check=False, capture_output=True, text=True, cwd=repository_root_path # type: ignore ) - if result.stdout.strip() == "false": + if str(result.stdout).strip() == "false": logger.log("Installing Solargraph...", logging.INFO) subprocess.run(dependency["installCommand"].split(), check=True, capture_output=True, cwd=repository_root_path) # Get the gem executable path directly - result = subprocess.run(["gem", "which", "solargraph"], check=True, capture_output=True, text=True, cwd=repository_root_path) + result = subprocess.run(["gem", "which", "solargraph"], check=True, capture_output=True, text=True, cwd=repository_root_path) # type: ignore gem_path = result.stdout.strip() - bin_dir = os.path.join(os.path.dirname(os.path.dirname(gem_path)), "bin") + bin_dir = os.path.join(os.path.dirname(os.path.dirname(gem_path)), "bin") # type: ignore executable_path = os.path.join(bin_dir, "solargraph") if not os.path.exists(executable_path): @@ -122,14 +123,14 @@ class Solargraph(SolidLanguageServer): } ], } - return initialize_params + return cast(InitializeParams, initialize_params) - 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": @@ -137,20 +138,20 @@ class Solargraph(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 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/terraform_ls.py b/src/solidlsp/language_servers/terraform_ls.py index e40d9af1..1c5339ac 100644 --- a/src/solidlsp/language_servers/terraform_ls.py +++ b/src/solidlsp/language_servers/terraform_ls.py @@ -3,6 +3,7 @@ import os import shutil import stat import threading +from typing import cast from overrides import override @@ -27,7 +28,7 @@ class TerraformLS(SolidLanguageServer): return super().is_ignored_dirname(dirname) or dirname in [".terraform", "terraform.tfstate.d"] @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 @@ -141,7 +142,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, @@ -166,17 +167,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 9709cc8f..21af4f15 100644 --- a/src/solidlsp/language_servers/typescript_language_server.py +++ b/src/solidlsp/language_servers/typescript_language_server.py @@ -8,6 +8,7 @@ import pathlib import shutil import threading from time import sleep +from typing import Any, cast from overrides import override @@ -19,6 +20,7 @@ from solidlsp.lsp_protocol_handler.lsp_types import InitializeParams 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 # Platform-specific imports @@ -26,9 +28,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")})() @@ -170,9 +172,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. @@ -186,7 +188,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": @@ -196,16 +198,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 """ @@ -249,7 +251,7 @@ class TypeScriptLanguageServer(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. diff --git a/src/solidlsp/language_servers/vts_language_server.py b/src/solidlsp/language_servers/vts_language_server.py index c8596b89..1fc42736 100644 --- a/src/solidlsp/language_servers/vts_language_server.py +++ b/src/solidlsp/language_servers/vts_language_server.py @@ -10,6 +10,7 @@ import pathlib import shutil import threading from time import sleep +from typing import cast from overrides import override @@ -21,6 +22,7 @@ from solidlsp.lsp_protocol_handler.lsp_types import InitializeParams 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 @@ -145,9 +147,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. @@ -161,30 +163,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 """ @@ -233,6 +235,6 @@ class VtsLanguageServer(SolidLanguageServer): # VTS may need longer to process references, so we include a sleep similar to TypeScript LS @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: sleep(1) return super()._send_references_request(relative_file_path, line, column) diff --git a/src/solidlsp/ls.py b/src/solidlsp/ls.py index 415f4139..5a2b0d6a 100644 --- a/src/solidlsp/ls.py +++ b/src/solidlsp/ls.py @@ -14,7 +14,7 @@ from collections.abc import Iterator from contextlib import contextmanager from copy import copy from pathlib import Path, PurePath -from typing import Self, Union, cast +from typing import Any, Self, Union, cast import pathspec @@ -72,7 +72,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() @@ -227,7 +227,7 @@ class SolidLanguageServer(ABC): ls = TerraformLS(config, logger, repository_root_path, solidlsp_settings=solidlsp_settings) else: - logger.log(f"Language {config.code_language} is not supported", logging.ERROR) + logger.log(f"Language {config.code_language} is not supported", logging.ERROR) # type: ignore raise LanguageServerException(f"Language {config.code_language} is not supported") ls.set_request_timeout(timeout) @@ -280,11 +280,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}: {str(msg)[:90]}...", 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 @@ -366,7 +366,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. @@ -377,6 +377,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 @@ -394,7 +397,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: @@ -428,12 +431,12 @@ class SolidLanguageServer(ABC): yield self self.stop() - def _start_server_process(self) -> None: + def _start_server_process(self) -> Any | None: self.server_started = True - self._start_server() + return self._start_server() @abstractmethod - def _start_server(self): + def _start_server(self) -> None: pass @contextmanager @@ -468,7 +471,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, @@ -482,7 +485,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, } } @@ -519,7 +522,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, }, @@ -566,7 +569,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, }, @@ -618,23 +621,23 @@ 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.ORIGIN_SELECTION_RANGE in item and 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 = {} + 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): @@ -642,11 +645,11 @@ class SolidLanguageServer(ABC): assert LSPConstants.URI in response assert LSPConstants.RANGE in response - new_item: ls_types.Location = {} + new_item = {} 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 @@ -709,7 +712,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! " @@ -723,11 +726,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 @@ -786,19 +789,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) @@ -816,8 +819,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"]: @@ -844,7 +847,7 @@ 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)] @@ -899,7 +902,7 @@ class SolidLanguageServer(ABC): logging.DEBUG, ) - def turn_item_into_symbol_with_children(item: GenericDocumentSymbol): + def turn_item_into_symbol_with_children(item: GenericDocumentSymbol) -> None: item = cast(ls_types.UnifiedSymbolInformation, item) absolute_path = os.path.join(self.repository_root_path, relative_file_path) @@ -916,9 +919,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 include_body: item["body"] = self.retrieve_symbol_body(item) # handle missing selectionRange @@ -927,16 +930,16 @@ class SolidLanguageServer(ABC): item["selectionRange"] = item["range"] else: item["selectionRange"] = item["location"]["range"] - children = item.get(LSPConstants.CHILDREN, []) - for child in children: + children = item.get(LSPConstants.CHILDREN, []) # type: ignore + for child in children: # type: ignore child["parent"] = item - item[LSPConstants.CHILDREN] = children + item[LSPConstants.CHILDREN] = children # type: ignore flat_all_symbol_list: list[ls_types.UnifiedSymbolInformation] = [] root_nodes: list[ls_types.UnifiedSymbolInformation] = [] for root_item in response: if "range" not in root_item and "location" not in root_item: - if root_item["kind"] in [SymbolKind.File, SymbolKind.Module]: + if root_item["kind"] in [SymbolKind.File, SymbolKind.Module]: # type: ignore ... # mutation is more convenient than creating a new dict, @@ -959,7 +962,7 @@ class SolidLanguageServer(ABC): l: list[ls_types.UnifiedSymbolInformation] = [] turn_item_into_symbol_with_children(node) assert LSPConstants.CHILDREN in node - children = node[LSPConstants.CHILDREN] + children = node[LSPConstants.CHILDREN] # type: ignore l.append(node) for child in children: l.extend(visit_tree_nodes_and_build_tree_repr(child)) @@ -1082,10 +1085,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) @@ -1125,7 +1128,7 @@ class SolidLanguageServer(ABC): result: dict[str, list[tuple[str, ls_types.SymbolKind, int, int]]] = 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"]: @@ -1209,7 +1212,7 @@ 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) -> str: """ @@ -1217,18 +1220,18 @@ 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"] - symbol_file = self.retrieve_full_file_content(symbol["location"]["relativePath"]) + symbol_start_line = symbol["location"]["range"]["start"]["line"] # type: ignore + symbol_end_line = symbol["location"]["range"]["end"]["line"] # type: ignore + assert "relativePath" in symbol["location"] # type: ignore + symbol_file = self.retrieve_full_file_content(symbol["location"]["relativePath"]) # type: ignore symbol_lines = symbol_file.split("\n") symbol_body = "\n".join(symbol_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 @@ -1277,6 +1280,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"] @@ -1300,6 +1304,7 @@ class SolidLanguageServer(ABC): ref_text = file_data.contents.split("\n")[ref_line] if "." in ref_text: containing_symbol_name = ref_text.split(".")[0] + assert ref_path is not None all_symbols, _ = self.request_document_symbols(ref_path) for symbol in all_symbols: if symbol["name"] == containing_symbol_name and symbol["kind"] == ls_types.SymbolKind.Variable: @@ -1506,7 +1511,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, @@ -1547,6 +1552,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"] @@ -1562,7 +1568,7 @@ class SolidLanguageServer(ABC): """ return Path(self.repository_root_path) / ".serena" / "cache" / self.language_id / "document_symbols_cache_v23-06-25.pkl" - def save_cache(self): + def save_cache(self) -> None: with self._cache_lock: if not self._cache_has_changed: self.logger.log("No changes to document symbols cache, skipping save", logging.DEBUG) @@ -1581,7 +1587,7 @@ class SolidLanguageServer(ABC): logging.ERROR, ) - def load_cache(self): + def load_cache(self) -> None: if not self.cache_path.exists(): return @@ -1622,7 +1628,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 diff --git a/src/solidlsp/ls_config.py b/src/solidlsp/ls_config.py index efb55a87..2d4ebb92 100644 --- a/src/solidlsp/ls_config.py +++ b/src/solidlsp/ls_config.py @@ -120,7 +120,7 @@ class LanguageServerConfig: """Paths, dirs or glob-like patterns. The matching will follow the same logic as for .gitignore entries""" @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_handler.py b/src/solidlsp/ls_handler.py index 731952f3..ab50a938 100644 --- a/src/solidlsp/ls_handler.py +++ b/src/solidlsp/ls_handler.py @@ -46,7 +46,7 @@ class Request: return self.error is not None def __init__(self) -> None: - self._result_queue = Queue() + self._result_queue: Queue[Request.Result] = Queue() def on_result(self, params: PayloadLike) -> None: self._result_queue.put(Request.Result(payload=params)) @@ -101,22 +101,22 @@ class SolidLanguageServerHandler: self, process_launch_info: ProcessLaunchInfo, 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.send = LanguageServerRequest(self.send_request) self.notify = LspNotification(self.send_notification) self.process_launch_info = process_launch_info - self.process = None + self.process: subprocess.Popen[bytes] | None = None self._received_shutdown = False self.request_id = 1 self._response_handlers: 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 @@ -170,7 +170,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}") @@ -195,7 +195,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 @@ -212,7 +212,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: @@ -220,7 +220,7 @@ 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) @@ -240,7 +240,7 @@ class SolidLanguageServerHandler: pass """ - 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" @@ -300,7 +300,7 @@ class SolidLanguageServerHandler: self.logger("client", "logger", message) @staticmethod - def _read_bytes_from_process(process, stream, num_bytes): + def _read_bytes_from_process(process: subprocess.Popen[bytes], stream: Any, num_bytes: int) -> bytes: """Read exactly num_bytes from process stdout""" if process.poll() is not None: # Process has terminated, check if we can still read @@ -457,13 +457,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_request.py b/src/solidlsp/ls_request.py index 209a3f04..d76ffa5d 100644 --- a/src/solidlsp/ls_request.py +++ b/src/solidlsp/ls_request.py @@ -1,10 +1,11 @@ -from typing import Union +from collections.abc import Callable +from typing import Any, Union from solidlsp.lsp_protocol_handler import lsp_types class LanguageServerRequest: - def __init__(self, send_request): + def __init__(self, send_request: Callable[[str, Any], Any]) -> None: self.send_request = send_request def implementation(self, params: lsp_types.ImplementationParams) -> Union["lsp_types.Definition", list["lsp_types.LocationLink"], None]: @@ -218,7 +219,7 @@ class LanguageServerRequest: server. The only notification that is sent after a shutdown request is the exit event. """ - return self.send_request("shutdown") + return self.send_request("shutdown", None) def will_save_wait_until(self, params: lsp_types.WillSaveTextDocumentParams) -> list["lsp_types.TextEdit"] | None: """A document will save request is sent from the client to the server before diff --git a/src/solidlsp/ls_utils.py b/src/solidlsp/ls_utils.py index 5110fe86..aa6b4885 100644 --- a/src/solidlsp/ls_utils.py +++ b/src/solidlsp/ls_utils.py @@ -125,10 +125,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)))) @@ -290,7 +295,7 @@ class PlatformUtils: raise LanguageServerException(f"Unknown platform: {system=}, {machine=}, {bitness=}") @staticmethod - def _determine_windows_machine_type(): + def _determine_windows_machine_type() -> str: import ctypes from ctypes import wintypes @@ -317,7 +322,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/server.py b/src/solidlsp/lsp_protocol_handler/server.py index 0cb59490..24bc2bcd 100644 --- a/src/solidlsp/lsp_protocol_handler/server.py +++ b/src/solidlsp/lsp_protocol_handler/server.py @@ -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 0b956bff..3b4dbff0 100644 --- a/src/solidlsp/settings.py +++ b/src/solidlsp/settings.py @@ -11,10 +11,10 @@ from dataclasses import dataclass class SolidLSPSettings: solidlsp_dir: str = str(pathlib.Path.home() / ".solidlsp") - 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")