Make mypy run through on solidlsp

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]

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": [".", "<", ">", ":", '"', "/", "*"],

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

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)

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):

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,

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()
 

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.

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)

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)

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": [".", "'", '"'],

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)

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 [
                 {

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
             """

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,

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)

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)

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.

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)

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
 

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
         """

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
         """

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

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",

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:

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

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")