Apply black & ruff to solidlsp

This commit is contained in:
Dominik Jain
2025-06-23 15:29:48 +02:00
parent e4e39f5a40
commit f04bd7d65b
16 changed files with 522 additions and 654 deletions
+3 -2
View File
@@ -83,7 +83,6 @@ target-version = [
exclude = '''
/(
src/multilspy
| src/solidlsp
)/
'''
@@ -140,7 +139,6 @@ target-version = "py311"
line-length = 140
exclude = [
"src/multilspy",
"src/solidlsp"
]
[tool.ruff.format]
@@ -240,6 +238,9 @@ ignore = [
"SIM110", # requires use of any(...) instead of for-loop
"G001", # forbids str.format in log statements
"E722", # forbids unspecific except clause
"SIM105", # forbids empty/general except clause
"SIM113", # wants to enforce use of enumerate
"E712", # forbids equality comparison with True/False
]
unfixable = [
"F841",
@@ -13,8 +13,7 @@ from multilspy.lsp_protocol_handler.lsp_types import InitializeParams
from multilspy.lsp_protocol_handler.server import ProcessLaunchInfo
from multilspy.multilspy_config import MultilspyConfig
from multilspy.multilspy_logger import MultilspyLogger
from multilspy.multilspy_utils import FileUtils
from multilspy.multilspy_utils import PlatformUtils
from multilspy.multilspy_utils import FileUtils, PlatformUtils
from solidlsp.ls import SolidLanguageServer
@@ -48,7 +47,7 @@ class ClangdLanguageServer(SolidLanguageServer):
"""
platform_id = PlatformUtils.get_platform_id()
with open(os.path.join(os.path.dirname(__file__), "runtime_dependencies.json"), "r") as f:
with open(os.path.join(os.path.dirname(__file__), "runtime_dependencies.json")) as f:
d = json.load(f)
del d["_description"]
@@ -56,12 +55,12 @@ class ClangdLanguageServer(SolidLanguageServer):
"linux-x64",
"win-x64",
"osx-arm64",
], "Unsupported platform: " + platform_id.value
], (
"Unsupported platform: " + platform_id.value
)
runtime_dependencies = d["runtimeDependencies"]
runtime_dependencies = [
dependency for dependency in runtime_dependencies if dependency["platformId"] == platform_id.value
]
runtime_dependencies = [dependency for dependency in runtime_dependencies if dependency["platformId"] == platform_id.value]
assert len(runtime_dependencies) == 1
# Select dependency matching the current platform
dependency = next((dep for dep in runtime_dependencies if dep["platformId"] == platform_id.value), None)
@@ -75,9 +74,7 @@ class ClangdLanguageServer(SolidLanguageServer):
logger.log(f"Clangd executable not found at {clangd_executable_path}. Downloading from {clangd_url}", logging.INFO)
os.makedirs(clangd_ls_dir, exist_ok=True)
if dependency["archiveType"] == "zip":
FileUtils.download_and_extract_archive(
logger, clangd_url, clangd_ls_dir, dependency["archiveType"]
)
FileUtils.download_and_extract_archive(logger, clangd_url, clangd_ls_dir, dependency["archiveType"])
else:
raise RuntimeError(f"Unsupported archive type: {dependency['archiveType']}")
if not os.path.exists(clangd_executable_path):
@@ -93,7 +90,7 @@ class ClangdLanguageServer(SolidLanguageServer):
"""
Returns the initialize params for the clangd Language Server.
"""
with open(os.path.join(os.path.dirname(__file__), "initialize_params.json"), "r") as f:
with open(os.path.join(os.path.dirname(__file__), "initialize_params.json")) as f:
d = json.load(f)
del d["_description"]
@@ -126,6 +123,7 @@ class ClangdLanguageServer(SolidLanguageServer):
# Shutdown the LanguageServer on exit from scope
# LanguageServer has been shutdown
"""
def register_capability_handler(params):
assert "registrations" in params
for registration in params["registrations"]:
@@ -175,7 +173,7 @@ class ClangdLanguageServer(SolidLanguageServer):
assert init_response["capabilities"]["textDocumentSync"]["change"] == 2
assert "completionProvider" in init_response["capabilities"]
assert init_response["capabilities"]["completionProvider"] == {
"triggerCharacters": ['.', '<', '>', ':', '"', '/', '*'],
"triggerCharacters": [".", "<", ">", ":", '"', "/", "*"],
"resolveProvider": False,
}
@@ -185,4 +183,3 @@ class ClangdLanguageServer(SolidLanguageServer):
# set ready flag
self.server_ready.set()
self.server_ready.wait()
@@ -19,7 +19,6 @@ class DartLanguageServer(SolidLanguageServer):
"""
Creates a DartServer instance. This class is not meant to be instantiated directly. Use LanguageServer.create() instead.
"""
executable_path = self.setup_runtime_dependencies(logger)
super().__init__(
config,
@@ -32,14 +31,12 @@ class DartLanguageServer(SolidLanguageServer):
def setup_runtime_dependencies(self, logger: "MultilspyLogger") -> str:
platform_id = PlatformUtils.get_platform_id()
with open(os.path.join(os.path.dirname(__file__), "runtime_dependencies.json"), "r") as f:
with open(os.path.join(os.path.dirname(__file__), "runtime_dependencies.json")) as f:
d = json.load(f)
del d["_description"]
runtime_dependencies = d["runtimeDependencies"]
runtime_dependencies = [
dependency for dependency in runtime_dependencies if dependency["platformId"] == platform_id.value
]
runtime_dependencies = [dependency for dependency in runtime_dependencies if dependency["platformId"] == platform_id.value]
assert len(runtime_dependencies) == 1
dependency = runtime_dependencies[0]
@@ -49,24 +46,18 @@ class DartLanguageServer(SolidLanguageServer):
if not os.path.exists(dart_ls_dir):
os.makedirs(dart_ls_dir)
FileUtils.download_and_extract_archive(
logger, dependency["url"], dart_ls_dir, dependency["archiveType"]
)
FileUtils.download_and_extract_archive(logger, dependency["url"], dart_ls_dir, dependency["archiveType"])
assert os.path.exists(dart_executable_path)
os.chmod(dart_executable_path, stat.S_IEXEC)
return f"{dart_executable_path} language-server --client-id multilspy.dart --client-version 1.2"
def _get_initialize_params(self, repository_absolute_path: str):
"""
Returns the initialize params for the Dart Language Server.
"""
with open(
os.path.join(os.path.dirname(__file__), "initialize_params.json"), "r"
) as f:
with open(os.path.join(os.path.dirname(__file__), "initialize_params.json")) as f:
d = json.load(f)
del d["_description"]
@@ -79,9 +70,7 @@ class DartLanguageServer(SolidLanguageServer):
d["rootUri"] = pathlib.Path(repository_absolute_path).as_uri()
assert d["workspaceFolders"][0]["uri"] == "$uri"
d["workspaceFolders"][0]["uri"] = pathlib.Path(
repository_absolute_path
).as_uri()
d["workspaceFolders"][0]["uri"] = pathlib.Path(repository_absolute_path).as_uri()
assert d["workspaceFolders"][0]["name"] == "$name"
d["workspaceFolders"][0]["name"] = os.path.basename(repository_absolute_path)
@@ -108,28 +97,20 @@ class DartLanguageServer(SolidLanguageServer):
self.server.on_request("client/registerCapability", do_nothing)
self.server.on_notification("language/status", do_nothing)
self.server.on_notification("window/logMessage", window_log_message)
self.server.on_request(
"workspace/executeClientCommand", execute_client_command_handler
)
self.server.on_request("workspace/executeClientCommand", execute_client_command_handler)
self.server.on_notification("$/progress", do_nothing)
self.server.on_notification("textDocument/publishDiagnostics", do_nothing)
self.server.on_notification("language/actionableNotification", do_nothing)
self.server.on_notification(
"experimental/serverStatus", check_experimental_status
)
self.server.on_notification("experimental/serverStatus", check_experimental_status)
self.logger.log(
"Starting dart-language-server server process", logging.INFO
)
self.logger.log("Starting dart-language-server server process", logging.INFO)
self.server.start()
initialize_params = self._get_initialize_params(self.repository_root_path)
self.logger.log(
"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)
self.logger.log(
f"Received initialize response from dart-language-server: {init_response}",
logging.INFO,
@@ -20,8 +20,7 @@ from multilspy.lsp_protocol_handler.server import ProcessLaunchInfo
from multilspy.multilspy_config import MultilspyConfig
from multilspy.multilspy_logger import MultilspyLogger
from multilspy.multilspy_settings import MultilspySettings
from multilspy.multilspy_utils import FileUtils
from multilspy.multilspy_utils import PlatformUtils
from multilspy.multilspy_utils import FileUtils, PlatformUtils
from solidlsp.ls import SolidLanguageServer
@@ -51,7 +50,6 @@ class EclipseJDTLS(SolidLanguageServer):
Creates a new EclipseJDTLS instance initializing the language server settings appropriately.
This class is not meant to be instantiated directly. Use LanguageServer.create() instead.
"""
runtime_dependency_paths = self.setupRuntimeDependencies(logger, config)
self.runtime_dependency_paths = runtime_dependency_paths
@@ -66,9 +64,7 @@ class EclipseJDTLS(SolidLanguageServer):
)
# shared_cache_location is the global cache used by Eclipse JDTLS across all workspaces
shared_cache_location = str(
PurePath(MultilspySettings.get_global_cache_directory(), "lsp", "EclipseJDTLS", "sharedIndex")
)
shared_cache_location = str(PurePath(MultilspySettings.get_global_cache_directory(), "lsp", "EclipseJDTLS", "sharedIndex"))
jre_path = self.runtime_dependency_paths.jre_path
lombok_jar_path = self.runtime_dependency_paths.lombok_jar_path
@@ -138,7 +134,7 @@ class EclipseJDTLS(SolidLanguageServer):
self.initialize_searcher_command_available = threading.Event()
super().__init__(config, logger, repository_root_path, ProcessLaunchInfo(cmd, proc_env, proc_cwd), "java")
@override
def is_ignored_dirname(self, dirname: str) -> bool:
# Ignore common Java build directories from different build tools:
@@ -148,13 +144,13 @@ class EclipseJDTLS(SolidLanguageServer):
# - IntelliJ IDEA: out, .idea
# - General: classes, dist, lib
return super().is_ignored_dirname(dirname) or dirname in [
"target", # Maven
"build", # Gradle
"bin", # Eclipse
"out", # IntelliJ IDEA
"classes", # General
"dist", # General
"lib" # General
"target", # Maven
"build", # Gradle
"bin", # Eclipse
"out", # IntelliJ IDEA
"classes", # General
"dist", # General
"lib", # General
]
def setupRuntimeDependencies(self, logger: MultilspyLogger, config: MultilspyConfig) -> RuntimeDependencyPaths:
@@ -163,7 +159,7 @@ class EclipseJDTLS(SolidLanguageServer):
"""
platformId = PlatformUtils.get_platform_id()
with open(str(PurePath(os.path.dirname(__file__), "runtime_dependencies.json")), "r", encoding="utf-8") as f:
with open(str(PurePath(os.path.dirname(__file__), "runtime_dependencies.json")), encoding="utf-8") as f:
runtimeDependencies = json.load(f)
del runtimeDependencies["_description"]
@@ -192,9 +188,7 @@ class EclipseJDTLS(SolidLanguageServer):
assert os.path.exists(gradle_path)
dependency = runtimeDependencies["vscode-java"][platformId.value]
vscode_java_path = str(
PurePath(os.path.abspath(os.path.dirname(__file__)), "static", dependency["relative_extraction_path"])
)
vscode_java_path = str(PurePath(os.path.abspath(os.path.dirname(__file__)), "static", dependency["relative_extraction_path"]))
os.makedirs(vscode_java_path, exist_ok=True)
jre_home_path = str(PurePath(vscode_java_path, dependency["jre_home_path"]))
jre_path = str(PurePath(vscode_java_path, dependency["jre_path"]))
@@ -211,9 +205,7 @@ class EclipseJDTLS(SolidLanguageServer):
os.path.exists(jdtls_readonly_config_path),
]
):
FileUtils.download_and_extract_archive(
logger, dependency["url"], vscode_java_path, dependency["archiveType"]
)
FileUtils.download_and_extract_archive(logger, dependency["url"], vscode_java_path, dependency["archiveType"])
os.chmod(jre_path, stat.S_IEXEC)
@@ -238,9 +230,7 @@ class EclipseJDTLS(SolidLanguageServer):
os.path.exists(intellisense_members_path),
]
):
FileUtils.download_and_extract_archive(
logger, dependency["url"], intellicode_directory_path, dependency["archiveType"]
)
FileUtils.download_and_extract_archive(logger, dependency["url"], intellicode_directory_path, dependency["archiveType"])
assert os.path.exists(intellicode_directory_path)
assert os.path.exists(intellicode_jar_path)
@@ -262,7 +252,7 @@ class EclipseJDTLS(SolidLanguageServer):
Returns the initialize parameters for the EclipseJDTLS server.
"""
# Look into https://github.com/eclipse/eclipse.jdt.ls/blob/master/org.eclipse.jdt.ls.core/src/org/eclipse/jdt/ls/core/internal/preferences/Preferences.java to understand all the options available
with open(str(PurePath(os.path.dirname(__file__), "initialize_params.json")), "r", encoding="utf-8") as f:
with open(str(PurePath(os.path.dirname(__file__), "initialize_params.json")), encoding="utf-8") as f:
d: InitializeParams = json.load(f)
del d["_description"]
@@ -307,18 +297,12 @@ class EclipseJDTLS(SolidLanguageServer):
for runtime in d["initializationOptions"]["settings"]["java"]["configuration"]["runtimes"]:
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"
assert os.path.exists(runtime["path"]), f"Runtime required for eclipse_jdtls at path {runtime['path']} does not exist"
assert d["initializationOptions"]["settings"]["java"]["import"]["gradle"]["home"] == "abs(static/gradle-7.3.3)"
d["initializationOptions"]["settings"]["java"]["import"]["gradle"][
"home"
] = self.runtime_dependency_paths.gradle_path
d["initializationOptions"]["settings"]["java"]["import"]["gradle"]["home"] = self.runtime_dependency_paths.gradle_path
d["initializationOptions"]["settings"]["java"]["import"]["gradle"]["java"][
"home"
] = self.runtime_dependency_paths.jre_path
d["initializationOptions"]["settings"]["java"]["import"]["gradle"]["java"]["home"] = self.runtime_dependency_paths.jre_path
return d
@@ -386,9 +370,7 @@ class EclipseJDTLS(SolidLanguageServer):
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"]})
self.intellicode_enable_command_available.wait()
+11 -8
View File
@@ -18,7 +18,7 @@ class Gopls(SolidLanguageServer):
"""
Provides Go specific instantiation of the LanguageServer class using gopls.
"""
@override
def is_ignored_dirname(self, dirname: str) -> bool:
# For Go projects, we should ignore:
@@ -31,7 +31,7 @@ class Gopls(SolidLanguageServer):
def _get_go_version():
"""Get the installed Go version or None if not found."""
try:
result = subprocess.run(['go', 'version'], capture_output=True, text=True)
result = subprocess.run(["go", "version"], capture_output=True, text=True, check=False)
if result.returncode == 0:
return result.stdout.strip()
except FileNotFoundError:
@@ -42,7 +42,7 @@ class Gopls(SolidLanguageServer):
def _get_gopls_version():
"""Get the installed gopls version or None if not found."""
try:
result = subprocess.run(['gopls', 'version'], capture_output=True, text=True)
result = subprocess.run(["gopls", "version"], capture_output=True, text=True, check=False)
if result.returncode == 0:
return result.stdout.strip()
except FileNotFoundError:
@@ -57,8 +57,10 @@ class Gopls(SolidLanguageServer):
"""
go_version = cls._get_go_version()
if not go_version:
raise RuntimeError("Go is not installed. Please install Go from https://golang.org/doc/install and make sure it is added to your PATH.")
raise RuntimeError(
"Go is not installed. Please install Go from https://golang.org/doc/install and make sure it is added to your PATH."
)
gopls_version = cls._get_gopls_version()
if not gopls_version:
raise RuntimeError(
@@ -66,12 +68,12 @@ class Gopls(SolidLanguageServer):
"Please install gopls as described in https://pkg.go.dev/golang.org/x/tools/gopls#section-readme\n\n"
"After installation, make sure it is added to your PATH (it might be installed in a different location than Go)."
)
return True
def __init__(self, config: MultilspyConfig, logger: MultilspyLogger, repository_root_path: str):
self.setup_runtime_dependency()
super().__init__(
config,
logger,
@@ -86,7 +88,7 @@ class Gopls(SolidLanguageServer):
"""
Returns the initialize params for the TypeScript Language Server.
"""
with open(os.path.join(os.path.dirname(__file__), "initialize_params.json"), "r", encoding="utf-8") as f:
with open(os.path.join(os.path.dirname(__file__), "initialize_params.json"), encoding="utf-8") as f:
d = json.load(f)
del d["_description"]
@@ -108,6 +110,7 @@ class Gopls(SolidLanguageServer):
def _start_server(self):
"""Start gopls server process"""
def register_capability_handler(params):
return
@@ -16,7 +16,7 @@ from multilspy.lsp_protocol_handler.lsp_types import DefinitionParams, Initializ
from multilspy.lsp_protocol_handler.server import ProcessLaunchInfo
from multilspy.multilspy_config import MultilspyConfig
from multilspy.multilspy_logger import MultilspyLogger
from multilspy.multilspy_utils import PlatformUtils, PlatformId
from multilspy.multilspy_utils import PlatformId, PlatformUtils
from solidlsp.ls import SolidLanguageServer
@@ -24,14 +24,14 @@ class Intelephense(SolidLanguageServer):
"""
Provides PHP specific instantiation of the LanguageServer class using Intelephense.
"""
@override
def is_ignored_dirname(self, dirname: str) -> bool:
# For PHP projects, we should ignore:
# - vendor: third-party dependencies managed by Composer
# - node_modules: if the project has JavaScript components
# - cache: commonly used for caching
return super().is_ignored_dirname(dirname) or dirname in ["node_modules", "vendor", "cache"]
return super().is_ignored_dirname(dirname) or dirname in ["node_modules", "vendor", "cache"]
def setup_runtime_dependencies(self, logger: MultilspyLogger, config: MultilspyConfig) -> str:
"""
@@ -50,17 +50,17 @@ class Intelephense(SolidLanguageServer):
]
assert platform_id in valid_platforms, f"Platform {platform_id} is not supported for multilspy PHP at the moment"
with open(os.path.join(os.path.dirname(__file__), "runtime_dependencies.json"), "r", encoding="utf-8") as f:
with open(os.path.join(os.path.dirname(__file__), "runtime_dependencies.json"), encoding="utf-8") as f:
d = json.load(f)
del d["_description"]
runtime_dependencies = d.get("runtimeDependencies", [])
intelephense_ls_dir = os.path.join(os.path.dirname(__file__), "static", "php-lsp")
# Verify both node and npm are installed
is_node_installed = shutil.which('node') is not None
is_node_installed = shutil.which("node") is not None
assert is_node_installed, "node is not installed or isn't in PATH. Please install NodeJS and try again."
is_npm_installed = shutil.which('npm') is not None
is_npm_installed = shutil.which("npm") is not None
assert is_npm_installed, "npm is not installed or isn't in PATH. Please install npm and try again."
# Install intelephense if not already installed
@@ -75,11 +75,12 @@ class Intelephense(SolidLanguageServer):
check=True,
cwd=intelephense_ls_dir,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL
stderr=subprocess.DEVNULL,
)
else:
# On Unix-like systems, run as non-root user
import pwd
user = pwd.getpwuid(os.getuid()).pw_name
subprocess.run(
dependency["command"],
@@ -88,32 +89,26 @@ class Intelephense(SolidLanguageServer):
user=user,
cwd=intelephense_ls_dir,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL
stderr=subprocess.DEVNULL,
)
intelephense_executable_path = os.path.join(intelephense_ls_dir, "node_modules", ".bin", "intelephense")
assert os.path.exists(intelephense_executable_path), "intelephense executable not found. Please install intelephense and try again."
return f"{intelephense_executable_path} --stdio"
def __init__(self, config: MultilspyConfig, logger: MultilspyLogger, repository_root_path: str):
# Setup runtime dependencies before initializing
intelephense_cmd = self.setup_runtime_dependencies(logger, config)
super().__init__(
config,
logger,
repository_root_path,
ProcessLaunchInfo(cmd=intelephense_cmd, cwd=repository_root_path),
"php"
)
super().__init__(config, logger, repository_root_path, ProcessLaunchInfo(cmd=intelephense_cmd, cwd=repository_root_path), "php")
self.request_id = 0
def _get_initialize_params(self, repository_absolute_path: str) -> InitializeParams:
"""
Returns the initialize params for the TypeScript Language Server.
"""
with open(os.path.join(os.path.dirname(__file__), "initialize_params.json"), "r", encoding="utf-8") as f:
with open(os.path.join(os.path.dirname(__file__), "initialize_params.json"), encoding="utf-8") as f:
d = json.load(f)
del d["_description"]
@@ -135,6 +130,7 @@ class Intelephense(SolidLanguageServer):
def _start_server(self):
"""Start Intelephense server process"""
def register_capability_handler(params):
return
@@ -184,7 +180,7 @@ class Intelephense(SolidLanguageServer):
# The sleeping doesn't seem to be needed on all systems
sleep(1)
return super()._send_references_request(relative_file_path, line, column)
@override
def _send_definition_request(self, definition_params: DefinitionParams):
# TODO: same as above, also only a problem if the definition is in another file
@@ -32,7 +32,7 @@ class JediServer(SolidLanguageServer):
ProcessLaunchInfo(cmd="jedi-language-server", cwd=repository_root_path),
"python",
)
@override
def is_ignored_dirname(self, dirname: str) -> bool:
return super().is_ignored_dirname(dirname) or dirname in ["venv", "__pycache__"]
@@ -41,7 +41,7 @@ class JediServer(SolidLanguageServer):
"""
Returns the initialize params for the Jedi Language Server.
"""
with open(os.path.join(os.path.dirname(__file__), "initialize_params.json"), "r", encoding="utf-8") as f:
with open(os.path.join(os.path.dirname(__file__), "initialize_params.json"), encoding="utf-8") as f:
d = json.load(f)
del d["_description"]
@@ -13,8 +13,7 @@ from multilspy.lsp_protocol_handler.lsp_types import InitializeParams
from multilspy.lsp_protocol_handler.server import ProcessLaunchInfo
from multilspy.multilspy_config import MultilspyConfig
from multilspy.multilspy_logger import MultilspyLogger
from multilspy.multilspy_utils import FileUtils
from multilspy.multilspy_utils import PlatformUtils
from multilspy.multilspy_utils import FileUtils, PlatformUtils
from solidlsp.ls import SolidLanguageServer
@@ -23,6 +22,7 @@ class KotlinRuntimeDependencyPaths:
"""
Stores the paths to the runtime dependencies of Kotlin Language Server
"""
java_path: str
java_home_path: str
kotlin_executable_path: str
@@ -39,13 +39,13 @@ class KotlinLanguageServer(SolidLanguageServer):
"""
runtime_dependency_paths = self.setup_runtime_dependencies(logger, config)
self.runtime_dependency_paths = runtime_dependency_paths
# Create command to execute the Kotlin Language Server script
cmd = f'"{self.runtime_dependency_paths.kotlin_executable_path}"'
# Set environment variables including JAVA_HOME
proc_env = {"JAVA_HOME": self.runtime_dependency_paths.java_home_path}
super().__init__(
config,
logger,
@@ -61,77 +61,75 @@ class KotlinLanguageServer(SolidLanguageServer):
platform_id = PlatformUtils.get_platform_id()
# Verify platform support
assert platform_id.value.startswith("win-") or platform_id.value.startswith("linux-") or platform_id.value.startswith("osx-"), "Only Windows, Linux and macOS platforms are supported for Kotlin in multilspy at the moment"
assert (
platform_id.value.startswith("win-") or platform_id.value.startswith("linux-") or platform_id.value.startswith("osx-")
), "Only Windows, Linux and macOS platforms are supported for Kotlin in multilspy at the moment"
# Load dependency information
with open(os.path.join(os.path.dirname(__file__), "runtime_dependencies.json"), "r", encoding="utf-8") as f:
with open(os.path.join(os.path.dirname(__file__), "runtime_dependencies.json"), encoding="utf-8") as f:
d = json.load(f)
del d["_description"]
kotlin_dependency = d["runtimeDependency"]
java_dependency = d["java"][platform_id.value]
# Setup paths for dependencies
static_dir = os.path.join(os.path.dirname(__file__), "static")
os.makedirs(static_dir, exist_ok=True)
# Setup Java paths
java_dir = os.path.join(static_dir, "java")
os.makedirs(java_dir, exist_ok=True)
java_home_path = os.path.join(java_dir, java_dependency["java_home_path"])
java_path = os.path.join(java_dir, java_dependency["java_path"])
# Download and extract Java if not exists
if not os.path.exists(java_path):
logger.log(f"Downloading Java for {platform_id.value}...", logging.INFO)
FileUtils.download_and_extract_archive(
logger, java_dependency["url"], java_dir, java_dependency["archiveType"]
)
FileUtils.download_and_extract_archive(logger, java_dependency["url"], java_dir, java_dependency["archiveType"])
# Make Java executable
if not platform_id.value.startswith("win-"):
os.chmod(java_path, 0o755)
assert os.path.exists(java_path), f"Java executable not found at {java_path}"
# Setup Kotlin Language Server paths
kotlin_ls_dir = os.path.join(static_dir, "server")
# Get platform-specific executable script path
if platform_id.value.startswith("win-"):
kotlin_script = os.path.join(kotlin_ls_dir, "bin", "kotlin-language-server.bat")
else:
kotlin_script = os.path.join(kotlin_ls_dir, "bin", "kotlin-language-server")
# 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"])
# Make script executable on Unix platforms
if os.path.exists(kotlin_script) and not platform_id.value.startswith("win-"):
os.chmod(kotlin_script, stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR | stat.S_IRGRP | stat.S_IXGRP | stat.S_IROTH | stat.S_IXOTH)
os.chmod(
kotlin_script, stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR | stat.S_IRGRP | stat.S_IXGRP | stat.S_IROTH | stat.S_IXOTH
)
# Use script file
if os.path.exists(kotlin_script):
kotlin_executable_path = kotlin_script
logger.log(f"Using Kotlin Language Server script at {kotlin_script}", logging.INFO)
else:
raise FileNotFoundError(f"Kotlin Language Server script not found at {kotlin_script}")
return KotlinRuntimeDependencyPaths(
java_path=java_path,
java_home_path=java_home_path,
kotlin_executable_path=kotlin_executable_path
java_path=java_path, java_home_path=java_home_path, kotlin_executable_path=kotlin_executable_path
)
def _get_initialize_params(self, repository_absolute_path: str) -> InitializeParams:
"""
Returns the initialize params for the Kotlin Language Server.
"""
with open(str(pathlib.PurePath(os.path.dirname(__file__), "initialize_params.json")), "r", encoding="utf-8") as f:
with open(str(pathlib.PurePath(os.path.dirname(__file__), "initialize_params.json")), encoding="utf-8") as f:
d: InitializeParams = json.load(f)
del d["_description"]
@@ -152,8 +150,8 @@ class KotlinLanguageServer(SolidLanguageServer):
d["initializationOptions"]["workspaceFolders"] = [pathlib.Path(repository_absolute_path).as_uri()]
assert (
d["workspaceFolders"]
== '[\n {\n "uri": pathlib.Path(repository_absolute_path).as_uri(),\n "name": os.path.basename(repository_absolute_path),\n }\n ]'
d["workspaceFolders"]
== '[\n {\n "uri": pathlib.Path(repository_absolute_path).as_uri(),\n "name": os.path.basename(repository_absolute_path),\n }\n ]'
)
d["workspaceFolders"] = [
{
@@ -168,6 +166,7 @@ class KotlinLanguageServer(SolidLanguageServer):
"""
Starts the Kotlin Language Server
"""
def execute_client_command_handler(params):
return []
@@ -8,7 +8,7 @@ import os
import pathlib
import stat
import threading
from typing import Iterable
from collections.abc import Iterable
from overrides import override
@@ -17,7 +17,7 @@ from multilspy.lsp_protocol_handler.server import ProcessLaunchInfo
from multilspy.multilspy_config import MultilspyConfig
from multilspy.multilspy_exceptions import MultilspyException
from multilspy.multilspy_logger import MultilspyLogger
from multilspy.multilspy_utils import FileUtils, PlatformUtils, PlatformId, DotnetVersion
from multilspy.multilspy_utils import DotnetVersion, FileUtils, PlatformId, PlatformUtils
from solidlsp.ls import SolidLanguageServer
@@ -28,7 +28,7 @@ def breadth_first_file_scan(root) -> Iterable[str]:
"""
dirs = [root]
# while we has dirs to scan
while len(dirs):
while dirs:
next_dirs = []
for parent in dirs:
# scan each dir
@@ -102,14 +102,12 @@ class OmniSharp(SolidLanguageServer):
"formattingOptions:indentationSize=4",
]
)
super().__init__(
config, logger, repository_root_path, ProcessLaunchInfo(cmd=cmd, cwd=repository_root_path), "csharp"
)
super().__init__(config, logger, repository_root_path, ProcessLaunchInfo(cmd=cmd, cwd=repository_root_path), "csharp")
self.server_ready = threading.Event()
self.definition_available = threading.Event()
self.references_available = threading.Event()
@override
def is_ignored_dirname(self, dirname: str) -> bool:
return super().is_ignored_dirname(dirname) or dirname in ["bin", "obj"]
@@ -118,7 +116,7 @@ class OmniSharp(SolidLanguageServer):
"""
Returns the initialize params for the Omnisharp Language Server.
"""
with open(os.path.join(os.path.dirname(__file__), "initialize_params.json"), "r", encoding="utf-8") as f:
with open(os.path.join(os.path.dirname(__file__), "initialize_params.json"), encoding="utf-8") as f:
d = json.load(f)
del d["_description"]
@@ -145,7 +143,7 @@ class OmniSharp(SolidLanguageServer):
platform_id = PlatformUtils.get_platform_id()
dotnet_version = PlatformUtils.get_dotnet_version()
with open(os.path.join(os.path.dirname(__file__), "runtime_dependencies.json"), "r", encoding="utf-8") as f:
with open(os.path.join(os.path.dirname(__file__), "runtime_dependencies.json"), encoding="utf-8") as f:
d = json.load(f)
del d["_description"]
@@ -156,7 +154,7 @@ class OmniSharp(SolidLanguageServer):
assert dotnet_version in [
DotnetVersion.V6,
DotnetVersion.V7,
DotnetVersion.V8
DotnetVersion.V8,
], "Only dotnet version 6 and 7 are supported in multilspy at the moment"
# TODO: Do away with this assumption
@@ -165,13 +163,11 @@ class OmniSharp(SolidLanguageServer):
dotnet_version = DotnetVersion.V6
runtime_dependencies = d["runtimeDependencies"]
runtime_dependencies = [
dependency for dependency in runtime_dependencies if dependency["platformId"] == platform_id.value
]
runtime_dependencies = [dependency for dependency in runtime_dependencies if dependency["platformId"] == platform_id.value]
runtime_dependencies = [
dependency
for dependency in runtime_dependencies
if not ("dotnet_version" in dependency) or dependency["dotnet_version"] == dotnet_version.value
if "dotnet_version" not in dependency or dependency["dotnet_version"] == dotnet_version.value
]
assert len(runtime_dependencies) == 2
runtime_dependencies = {
@@ -185,9 +181,7 @@ class OmniSharp(SolidLanguageServer):
omnisharp_ls_dir = os.path.join(os.path.dirname(__file__), "static", "OmniSharp")
if not os.path.exists(omnisharp_ls_dir):
os.makedirs(omnisharp_ls_dir)
FileUtils.download_and_extract_archive(
logger, runtime_dependencies["OmniSharp"]["url"], omnisharp_ls_dir, "zip"
)
FileUtils.download_and_extract_archive(logger, runtime_dependencies["OmniSharp"]["url"], omnisharp_ls_dir, "zip")
omnisharp_executable_path = os.path.join(omnisharp_ls_dir, runtime_dependencies["OmniSharp"]["binaryName"])
assert os.path.exists(omnisharp_executable_path)
os.chmod(omnisharp_executable_path, stat.S_IEXEC)
@@ -195,12 +189,8 @@ class OmniSharp(SolidLanguageServer):
razor_omnisharp_ls_dir = os.path.join(os.path.dirname(__file__), "static", "RazorOmnisharp")
if not os.path.exists(razor_omnisharp_ls_dir):
os.makedirs(razor_omnisharp_ls_dir)
FileUtils.download_and_extract_archive(
logger, runtime_dependencies["RazorOmnisharp"]["url"], razor_omnisharp_ls_dir, "zip"
)
razor_omnisharp_dll_path = os.path.join(
razor_omnisharp_ls_dir, runtime_dependencies["RazorOmnisharp"]["dll_path"]
)
FileUtils.download_and_extract_archive(logger, runtime_dependencies["RazorOmnisharp"]["url"], razor_omnisharp_ls_dir, "zip")
razor_omnisharp_dll_path = os.path.join(razor_omnisharp_ls_dir, runtime_dependencies["RazorOmnisharp"]["dll_path"])
assert os.path.exists(razor_omnisharp_dll_path)
return omnisharp_executable_path, razor_omnisharp_dll_path
@@ -373,20 +363,12 @@ class OmniSharp(SolidLanguageServer):
)
init_response = self.server.send.initialize(initialize_params)
self.server.notify.initialized({})
with open(os.path.join(os.path.dirname(__file__), "workspace_did_change_configuration.json"), "r", encoding="utf-8") as f:
self.server.notify.workspace_did_change_configuration({
"settings": json.load(f)
})
with open(os.path.join(os.path.dirname(__file__), "workspace_did_change_configuration.json"), encoding="utf-8") as f:
self.server.notify.workspace_did_change_configuration({"settings": json.load(f)})
assert "capabilities" in init_response
if (
"definitionProvider" in init_response["capabilities"]
and init_response["capabilities"]["definitionProvider"]
):
if "definitionProvider" in init_response["capabilities"] and init_response["capabilities"]["definitionProvider"]:
self.definition_available.set()
if (
"referencesProvider" in init_response["capabilities"]
and init_response["capabilities"]["referencesProvider"]
):
if "referencesProvider" in init_response["capabilities"] and init_response["capabilities"]["referencesProvider"]:
self.references_available.set()
self.definition_available.wait()
@@ -22,6 +22,7 @@ class PyrightServer(SolidLanguageServer):
Provides Python specific instantiation of the LanguageServer class using Pyright.
Contains various configurations and settings specific to Python.
"""
def __init__(self, config: MultilspyConfig, logger: MultilspyLogger, repository_root_path: str):
"""
Creates a PyrightServer instance. This class is not meant to be instantiated directly.
@@ -50,7 +51,7 @@ class PyrightServer(SolidLanguageServer):
Returns the initialize params for the Pyright Language Server.
"""
# Create basic initialization parameters
initialize_params: InitializeParams = { # type: ignore
initialize_params: InitializeParams = { # type: ignore
"processId": os.getpid(),
"rootPath": repository_absolute_path,
"rootUri": pathlib.Path(repository_absolute_path).as_uri(),
@@ -15,8 +15,7 @@ from multilspy.lsp_protocol_handler.lsp_types import InitializeParams
from multilspy.lsp_protocol_handler.server import ProcessLaunchInfo
from multilspy.multilspy_config import MultilspyConfig
from multilspy.multilspy_logger import MultilspyLogger
from multilspy.multilspy_utils import FileUtils
from multilspy.multilspy_utils import PlatformUtils
from multilspy.multilspy_utils import FileUtils, PlatformUtils
from solidlsp.ls import SolidLanguageServer
@@ -52,7 +51,7 @@ class RustAnalyzer(SolidLanguageServer):
"""
platform_id = PlatformUtils.get_platform_id()
with open(os.path.join(os.path.dirname(__file__), "runtime_dependencies.json"), "r", encoding="utf-8") as f:
with open(os.path.join(os.path.dirname(__file__), "runtime_dependencies.json"), encoding="utf-8") as f:
d = json.load(f)
del d["_description"]
@@ -62,9 +61,7 @@ class RustAnalyzer(SolidLanguageServer):
# ], "Only linux-x64 and win-x64 platform is supported for in multilspy at the moment"
runtime_dependencies = d["runtimeDependencies"]
runtime_dependencies = [
dependency for dependency in runtime_dependencies if dependency["platformId"] == platform_id.value
]
runtime_dependencies = [dependency for dependency in runtime_dependencies if dependency["platformId"] == platform_id.value]
assert len(runtime_dependencies) == 1
dependency = runtime_dependencies[0]
@@ -73,13 +70,9 @@ class RustAnalyzer(SolidLanguageServer):
if not os.path.exists(rustanalyzer_ls_dir):
os.makedirs(rustanalyzer_ls_dir)
if dependency["archiveType"] == "gz":
FileUtils.download_and_extract_archive(
logger, dependency["url"], rustanalyzer_executable_path, dependency["archiveType"]
)
FileUtils.download_and_extract_archive(logger, dependency["url"], rustanalyzer_executable_path, dependency["archiveType"])
else:
FileUtils.download_and_extract_archive(
logger, dependency["url"], rustanalyzer_ls_dir, dependency["archiveType"]
)
FileUtils.download_and_extract_archive(logger, dependency["url"], rustanalyzer_ls_dir, dependency["archiveType"])
assert os.path.exists(rustanalyzer_executable_path)
os.chmod(rustanalyzer_executable_path, stat.S_IEXEC)
@@ -89,7 +82,7 @@ class RustAnalyzer(SolidLanguageServer):
"""
Returns the initialize params for the Rust Analyzer Language Server.
"""
with open(os.path.join(os.path.dirname(__file__), "initialize_params.json"), "r", encoding="utf-8") as f:
with open(os.path.join(os.path.dirname(__file__), "initialize_params.json"), encoding="utf-8") as f:
d = json.load(f)
del d["_description"]
@@ -51,8 +51,7 @@ class Solargraph(SolidLanguageServer):
"""
Setup runtime dependencies for Solargraph.
"""
with open(os.path.join(os.path.dirname(__file__), "runtime_dependencies.json"), "r", encoding="utf-8") as f:
with open(os.path.join(os.path.dirname(__file__), "runtime_dependencies.json"), encoding="utf-8") as f:
d = json.load(f)
del d["_description"]
@@ -64,38 +63,40 @@ class Solargraph(SolidLanguageServer):
ruby_version = 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}")
except FileNotFoundError:
raise RuntimeError("Ruby is not installed. Please install Ruby before continuing.")
raise RuntimeError(f"Error checking for Ruby installation: {e.stderr}") from e
except FileNotFoundError as e:
raise RuntimeError("Ruby is not installed. Please install Ruby before continuing.") from e
# Check if solargraph is installed
try:
result = subprocess.run(["gem", "list", "^solargraph$", "-i"], check=False, capture_output=True, text=True, cwd=repository_root_path)
result = subprocess.run(
["gem", "list", "^solargraph$", "-i"], check=False, capture_output=True, text=True, cwd=repository_root_path
)
if 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)
gem_path = result.stdout.strip()
bin_dir = os.path.join(os.path.dirname(os.path.dirname(gem_path)), "bin")
executable_path = os.path.join(bin_dir, "solargraph")
if not os.path.exists(executable_path):
raise RuntimeError(f"Solargraph executable not found at {executable_path}")
# Ensure the executable has the right permissions
os.chmod(executable_path, os.stat(executable_path).st_mode | stat.S_IEXEC)
return executable_path
except subprocess.CalledProcessError as e:
raise RuntimeError(f"Failed to check or install Solargraph. {e.stderr}")
raise RuntimeError(f"Failed to check or install Solargraph. {e.stderr}") from e
def _get_initialize_params(self, repository_absolute_path: str) -> InitializeParams:
"""
Returns the initialize params for the Solargraph Language Server.
"""
with open(os.path.join(os.path.dirname(__file__), "initialize_params.json"), "r", encoding="utf-8") as f:
with open(os.path.join(os.path.dirname(__file__), "initialize_params.json"), encoding="utf-8") as f:
d = json.load(f)
del d["_description"]
@@ -21,14 +21,14 @@ from multilspy.multilspy_utils import PlatformId, PlatformUtils
from solidlsp.ls import SolidLanguageServer
# Platform-specific imports
if os.name != 'nt': # Unix-like systems
if os.name != "nt": # Unix-like systems
import pwd
else:
# Dummy pwd module for Windows
class pwd:
@staticmethod
def getpwuid(uid):
return type('obj', (), {'pw_name': os.environ.get('USERNAME', 'unknown')})()
return type("obj", (), {"pw_name": os.environ.get("USERNAME", "unknown")})()
# Conditionally import pwd module (Unix-only)
@@ -91,9 +91,9 @@ class TypeScriptLanguageServer(SolidLanguageServer):
tsserver_executable_path = os.path.join(tsserver_ls_dir, "typescript-language-server")
# Verify both node and npm are installed
is_node_installed = shutil.which('node') is not None
is_node_installed = shutil.which("node") is not None
assert is_node_installed, "node is not installed or isn't in PATH. Please install NodeJS and try again."
is_npm_installed = shutil.which('npm') is not None
is_npm_installed = shutil.which("npm") is not None
assert is_npm_installed, "npm is not installed or isn't in PATH. Please install npm and try again."
# Install typescript and typescript-language-server if not already installed
@@ -108,7 +108,7 @@ class TypeScriptLanguageServer(SolidLanguageServer):
check=True,
cwd=tsserver_ls_dir,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL
stderr=subprocess.DEVNULL,
)
else:
# On Unix-like systems, run as non-root user
@@ -120,12 +120,14 @@ class TypeScriptLanguageServer(SolidLanguageServer):
user=user,
cwd=tsserver_ls_dir,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL
stderr=subprocess.DEVNULL,
)
tsserver_executable_path = os.path.join(tsserver_ls_dir, "node_modules", ".bin", "typescript-language-server")
assert os.path.exists(tsserver_executable_path), "typescript-language-server executable not found. Please install typescript-language-server and try again."
assert os.path.exists(
tsserver_executable_path
), "typescript-language-server executable not found. Please install typescript-language-server and try again."
return f"{tsserver_executable_path} --stdio"
def _get_initialize_params(self, repository_absolute_path: str) -> InitializeParams:
@@ -184,8 +186,7 @@ class TypeScriptLanguageServer(SolidLanguageServer):
def window_log_message(msg):
self.logger.log(f"LSP: window/logMessage: {msg}", logging.INFO)
def check_experimental_status(params):
"""
Also listen for experimental/serverStatus as a backup signal
@@ -215,8 +216,8 @@ class TypeScriptLanguageServer(SolidLanguageServer):
assert init_response["capabilities"]["textDocumentSync"] == 2
assert "completionProvider" in init_response["capabilities"]
assert init_response["capabilities"]["completionProvider"] == {
"triggerCharacters": ['.', '"', "'", '/', '@', '<'],
"resolveProvider": True
"triggerCharacters": [".", '"', "'", "/", "@", "<"],
"resolveProvider": True,
}
self.server.notify.initialized({})
+205 -232
View File
@@ -5,18 +5,19 @@ import pathlib
import pickle
import re
import threading
from abc import abstractmethod, ABC
from abc import ABC, abstractmethod
from collections import defaultdict
from collections.abc import Iterator
from contextlib import contextmanager
from copy import copy
from pathlib import Path, PurePath
from typing import Dict, Iterator, List, Optional, Tuple, Union, cast, Self
from typing import Self, cast
import pathspec
import tqdm
from multilspy import multilspy_types
from multilspy.language_server import LSPFileBuffer, GenericDocumentSymbol, ReferenceInSymbol
from multilspy.language_server import GenericDocumentSymbol, LSPFileBuffer, ReferenceInSymbol
from multilspy.lsp_protocol_handler import lsp_types
from multilspy.lsp_protocol_handler import lsp_types as LSPTypes
from multilspy.lsp_protocol_handler.lsp_constants import LSPConstants
@@ -46,10 +47,12 @@ class SolidLanguageServer(ABC):
A language-specific condition for directories that should always be ignored. For example, venv
in Python and node_modules in JS/TS should be ignored always.
"""
return dirname.startswith('.')
return dirname.startswith(".")
@classmethod
def create(cls, config: MultilspyConfig, logger: MultilspyLogger, repository_root_path: str, timeout: Optional[float] = None) -> "SolidLanguageServer":
def create(
cls, config: MultilspyConfig, logger: MultilspyLogger, repository_root_path: str, timeout: float | None = None
) -> "SolidLanguageServer":
"""
Creates a language specific LanguageServer instance based on the given configuration, and appropriate settings for the programming language.
@@ -66,6 +69,7 @@ class SolidLanguageServer(ABC):
from solidlsp.language_servers.pyright_language_server.pyright_server import (
PyrightServer,
)
return PyrightServer(config, logger, repository_root_path)
# It used to be jedi, but pyright is a bit faster, and also more actively maintained
# Keeping the previous code for reference
@@ -79,48 +83,58 @@ class SolidLanguageServer(ABC):
from solidlsp.language_servers.eclipse_jdtls.eclipse_jdtls import (
EclipseJDTLS,
)
return EclipseJDTLS(config, logger, repository_root_path)
elif config.code_language == Language.KOTLIN:
from solidlsp.language_servers.kotlin_language_server.kotlin_language_server import (
KotlinLanguageServer,
)
return KotlinLanguageServer(config, logger, repository_root_path)
elif config.code_language == Language.RUST:
from solidlsp.language_servers.rust_analyzer.rust_analyzer import (
RustAnalyzer,
)
return RustAnalyzer(config, logger, repository_root_path)
elif config.code_language == Language.CSHARP:
from solidlsp.language_servers.omnisharp.omnisharp import OmniSharp
return OmniSharp(config, logger, repository_root_path)
elif config.code_language in [Language.TYPESCRIPT, Language.JAVASCRIPT]:
from solidlsp.language_servers.typescript_language_server.typescript_language_server import (
TypeScriptLanguageServer,
)
return TypeScriptLanguageServer(config, logger, repository_root_path)
elif config.code_language == Language.GO:
from solidlsp.language_servers.gopls.gopls import Gopls
return Gopls(config, logger, repository_root_path)
elif config.code_language == Language.RUBY:
from solidlsp.language_servers.solargraph.solargraph import Solargraph
return Solargraph(config, logger, repository_root_path)
elif config.code_language == Language.DART:
from solidlsp.language_servers.dart_language_server.dart_language_server import DartLanguageServer
return DartLanguageServer(config, logger, repository_root_path)
elif config.code_language == Language.CPP:
from solidlsp.language_servers.clangd_language_server.clangd_language_server import ClangdLanguageServer
return ClangdLanguageServer(config, logger, repository_root_path)
elif config.code_language == Language.PHP:
from solidlsp.language_servers.intelephense.intelephense import Intelephense
return Intelephense(config, logger, repository_root_path)
else:
@@ -128,12 +142,12 @@ class SolidLanguageServer(ABC):
raise MultilspyException(f"Language {config.code_language} is not supported")
def __init__(
self,
config: MultilspyConfig,
logger: MultilspyLogger,
repository_root_path: str,
process_launch_info: ProcessLaunchInfo,
language_id: str,
self,
config: MultilspyConfig,
logger: MultilspyLogger,
repository_root_path: str,
process_launch_info: ProcessLaunchInfo,
language_id: str,
):
"""
Initializes a LanguageServer instance.
@@ -150,14 +164,19 @@ class SolidLanguageServer(ABC):
"""
self.logger = logger
self.repository_root_path: str = repository_root_path
self.logger.log(f"Creating language server instance for {repository_root_path=} with {language_id=} and process launch info: {process_launch_info}", logging.DEBUG)
self.logger.log(
f"Creating language server instance for {repository_root_path=} with {language_id=} and process launch info: {process_launch_info}",
logging.DEBUG,
)
self.language_id = language_id
self.open_file_buffers: Dict[str, LSPFileBuffer] = {}
self.open_file_buffers: dict[str, LSPFileBuffer] = {}
self.language = Language(language_id)
# load cache first to prevent any racing conditions due to asyncio stuff
self._document_symbols_cache: dict[str, Tuple[str, Tuple[List[multilspy_types.UnifiedSymbolInformation], List[multilspy_types.UnifiedSymbolInformation]]]] = {}
self._document_symbols_cache: dict[
str, tuple[str, tuple[list[multilspy_types.UnifiedSymbolInformation], list[multilspy_types.UnifiedSymbolInformation]]]
] = {}
"""Maps file paths to a tuple of (file_content_hash, result_of_request_document_symbols)"""
self._cache_lock = threading.Lock()
self._cache_has_changed: bool = False
@@ -166,36 +185,35 @@ class SolidLanguageServer(ABC):
self.server_started = False
self.completions_available = threading.Event()
if config.trace_lsp_communication:
def logging_fn(source: str, target: str, msg: StringDict | str):
self.logger.log(f"LSP: {source} -> {target}: {str(msg)[:90]}...", self.logger.logger.level)
else:
logging_fn = None
# cmd is obtained from the child classes, which provide the language specific command to start the language server
# LanguageServerHandler provides the functionality to start the language server and communicate with it
self.logger.log(f"Creating language server instance with {language_id=} and process launch info: {process_launch_info}", logging.DEBUG)
self.logger.log(
f"Creating language server instance with {language_id=} and process launch info: {process_launch_info}", logging.DEBUG
)
self.server = SolidLanguageServerHandler(
process_launch_info,
logger=logging_fn,
start_independent_lsp_process=config.start_independent_lsp_process,
)
# Set up the pathspec matcher for the ignored paths
# for all absolute paths in ignored_paths, convert them to relative paths
processed_patterns = []
for pattern in set(config.ignored_paths):
# Normalize separators (pathspec expects forward slashes)
pattern = pattern.replace(os.path.sep, '/')
pattern = pattern.replace(os.path.sep, "/")
processed_patterns.append(pattern)
self.logger.log(f"Processing {len(processed_patterns)} ignored paths from the config", logging.DEBUG)
# Create a pathspec matcher from the processed patterns
self._ignore_spec = pathspec.PathSpec.from_lines(
pathspec.patterns.GitWildMatchPattern,
processed_patterns
)
self._ignore_spec = pathspec.PathSpec.from_lines(pathspec.patterns.GitWildMatchPattern, processed_patterns)
self._server_context = None
@@ -247,12 +265,12 @@ class SolidLanguageServer(ABC):
# Use pathspec for gitignore-style pattern matching
# Normalize path separators for pathspec (it expects forward slashes)
normalized_path = str(rel_path).replace(os.path.sep, '/')
normalized_path = str(rel_path).replace(os.path.sep, "/")
# pathspec can't handle the matching of directories if they don't end with a slash!
# see https://github.com/cpburnz/python-pathspec/issues/89
if os.path.isdir(os.path.join(self.repository_root_path, normalized_path)) and not normalized_path.endswith('/'):
normalized_path = normalized_path + '/'
if os.path.isdir(os.path.join(self.repository_root_path, normalized_path)) and not normalized_path.endswith("/"):
normalized_path = normalized_path + "/"
# Use the pathspec matcher to check if the path matches any ignore pattern
if self.get_ignore_spec().match_file(normalized_path):
@@ -271,7 +289,6 @@ class SolidLanguageServer(ABC):
self.logger.log(f"Initiating final robust shutdown with a {timeout}s timeout...", logging.INFO)
process = self.server.process
reader_tasks = list(self.server.tasks.values())
# --- Main Shutdown Logic ---
# Stage 1: Graceful Termination Request
@@ -281,50 +298,11 @@ class SolidLanguageServer(ABC):
if process.stdin and not process.stdin.is_closing():
process.stdin.close()
except Exception:
pass # Ignore errors here, we are proceeding to terminate anyway.
pass # Ignore errors here, we are proceeding to terminate anyway.
# Stage 2: Terminate and Concurrently Drain stdout/stderr
process.terminate()
"""
except asyncio.TimeoutError:
# Stage 3: Forceful Kill
self.logger.log("Graceful termination failed. Forcefully killing process...", logging.WARNING)
if self.server.is_running():
try:
process.kill()
# Wait for the killed process to be reaped by the OS.
await process.wait()
except Exception as e:
self.logger.log(f"Error during forceful kill: {e}", logging.ERROR)
except Exception as e:
self.logger.log(f"An unexpected error occurred during shutdown logic: {e}", logging.ERROR)
finally:
# === STAGE 4: EXPLICIT TASK & PIPE CLEANUP ===
self.logger.log("Performing final task cancellation and pipe handle cleanup...", logging.DEBUG)
# 1. Cancel any lingering reader tasks.
for task in reader_tasks:
if not task.done():
task.cancel()
# Wait for cancellations to complete.
await asyncio.gather(*reader_tasks, return_exceptions=True)
self.server.tasks = {}
# 2. Explicitly close each pipe to release OS handles.
for pipe in [process.stdin, process.stdout, process.stderr]:
if pipe and hasattr(pipe, 'is_closing') and not pipe.is_closing():
try:
pipe.close()
except Exception:
pass
# 3. Null out the process object in the handler.
self.server.process = None
self.logger.log("Shutdown sequence fully finished.", logging.DEBUG)
"""
@contextmanager
def start_server(self) -> Iterator["SolidLanguageServer"]:
self.start()
@@ -393,10 +371,10 @@ class SolidLanguageServer(ABC):
del self.open_file_buffers[uri]
def insert_text_at_position(
self, relative_file_path: str, line: int, column: int, text_to_be_inserted: str
self, relative_file_path: str, line: int, column: int, text_to_be_inserted: str
) -> multilspy_types.Position:
"""
Insert text at the given line and column in the given file and return
Insert text at the given line and column in the given file and return
the updated cursor position after inserting the text.
:param relative_file_path: The relative path of the file to open.
@@ -442,10 +420,10 @@ class SolidLanguageServer(ABC):
return multilspy_types.Position(line=new_l, character=new_c)
def delete_text_between_positions(
self,
relative_file_path: str,
start: multilspy_types.Position,
end: multilspy_types.Position,
self,
relative_file_path: str,
start: multilspy_types.Position,
end: multilspy_types.Position,
) -> str:
"""
Delete text between the given start and end positions in the given file and return the deleted text.
@@ -465,7 +443,9 @@ class SolidLanguageServer(ABC):
file_buffer = self.open_file_buffers[uri]
file_buffer.version += 1
new_contents, deleted_text = TextUtils.delete_text_between_positions(file_buffer.contents, start_line=start["line"], start_col=start["character"], end_line=end["line"], end_col=end["character"])
new_contents, deleted_text = TextUtils.delete_text_between_positions(
file_buffer.contents, start_line=start["line"], start_col=start["character"], end_line=end["line"], end_col=end["character"]
)
file_buffer.contents = new_contents
self.server.notify.did_change_text_document(
{
@@ -478,12 +458,10 @@ class SolidLanguageServer(ABC):
)
return deleted_text
def _send_definition_request(self, definition_params: DefinitionParams) -> Union[Definition, List[LocationLink], None]:
def _send_definition_request(self, definition_params: DefinitionParams) -> Definition | list[LocationLink] | None:
return self.server.send.definition(definition_params)
def request_definition(
self, relative_file_path: str, line: int, column: int
) -> List[multilspy_types.Location]:
def request_definition(self, relative_file_path: str, line: int, column: int) -> list[multilspy_types.Location]:
"""
Raise a [textDocument/definition](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#textDocument_definition) request to the Language Server
for the symbol at the given line and column in the given file. Wait for the response and return the result.
@@ -494,7 +472,6 @@ class SolidLanguageServer(ABC):
:return List[multilspy_types.Location]: A list of locations where the symbol is defined
"""
if not self.server_started:
self.logger.log(
"find_function_definition called before Language Server started",
@@ -504,20 +481,21 @@ class SolidLanguageServer(ABC):
with self.open_file(relative_file_path):
# sending request to the language server and waiting for response
definition_params = cast(DefinitionParams, {
LSPConstants.TEXT_DOCUMENT: {
LSPConstants.URI: pathlib.Path(
str(PurePath(self.repository_root_path, relative_file_path))
).as_uri()
definition_params = cast(
DefinitionParams,
{
LSPConstants.TEXT_DOCUMENT: {
LSPConstants.URI: pathlib.Path(str(PurePath(self.repository_root_path, relative_file_path))).as_uri()
},
LSPConstants.POSITION: {
LSPConstants.LINE: line,
LSPConstants.CHARACTER: column,
},
},
LSPConstants.POSITION: {
LSPConstants.LINE: line,
LSPConstants.CHARACTER: column,
},
})
)
response = self._send_definition_request(definition_params)
ret: List[multilspy_types.Location] = []
ret: list[multilspy_types.Location] = []
if isinstance(response, list):
# response is either of type Location[] or LocationLink[]
for item in response:
@@ -529,10 +507,10 @@ class SolidLanguageServer(ABC):
new_item["relativePath"] = PathUtils.get_relative_path(new_item["absolutePath"], self.repository_root_path)
ret.append(multilspy_types.Location(new_item))
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
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: multilspy_types.Location = {}
new_item["uri"] = item[LSPConstants.TARGET_URI]
@@ -565,7 +543,7 @@ class SolidLanguageServer(ABC):
return ret
# Some LS cause problems with this, so the call is isolated from the rest to allow overriding in subclasses
def _send_references_request(self, relative_file_path: str, line: int, column: int) -> List[lsp_types.Location] | None:
def _send_references_request(self, relative_file_path: str, line: int, column: int) -> list[lsp_types.Location] | None:
return self.server.send.references(
{
"textDocument": {"uri": PathUtils.path_to_uri(os.path.join(self.repository_root_path, relative_file_path))},
@@ -574,9 +552,7 @@ class SolidLanguageServer(ABC):
}
)
def request_references(
self, relative_file_path: str, line: int, column: int
) -> List[multilspy_types.Location]:
def request_references(self, relative_file_path: str, line: int, column: int) -> list[multilspy_types.Location]:
"""
Raise a [textDocument/references](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#textDocument_references) request to the Language Server
to find references to the symbol at the given line and column in the given file. Wait for the response and return the result.
@@ -588,7 +564,6 @@ class SolidLanguageServer(ABC):
:return: A list of locations where the symbol is referenced (excluding ignored directories)
"""
if not self.server_started:
self.logger.log(
"request_references called before Language Server started",
@@ -596,13 +571,12 @@ class SolidLanguageServer(ABC):
)
raise MultilspyException("Language Server not started")
with self.open_file(relative_file_path):
try:
response = self._send_references_request(relative_file_path, line=line, column=column)
except Exception as e:
# Catch LSP internal error (-32603) and raise a more informative exception
if isinstance(e, Error) and getattr(e, 'code', None) == -32603:
if isinstance(e, Error) and getattr(e, "code", None) == -32603:
raise RuntimeError(
f"LSP internal error (-32603) when requesting references for {relative_file_path}:{line}:{column}. "
"This often occurs when requesting references for a symbol not referenced in the expected way. "
@@ -611,7 +585,7 @@ class SolidLanguageServer(ABC):
if response is None:
return []
ret: List[multilspy_types.Location] = []
ret: list[multilspy_types.Location] = []
assert isinstance(response, list), f"Unexpected response from Language Server (expected list, got {type(response)}): {response}"
for item in response:
assert isinstance(item, dict), f"Unexpected response from Language Server (expected dict, got {type(item)}): {item}"
@@ -633,8 +607,8 @@ class SolidLanguageServer(ABC):
return ret
def request_references_with_content(
self, relative_file_path: str, line: int, column: int, context_lines_before: int = 0, context_lines_after: int = 0
) -> List[MatchedConsecutiveLines]:
self, relative_file_path: str, line: int, column: int, context_lines_before: int = 0, context_lines_after: int = 0
) -> list[MatchedConsecutiveLines]:
"""
Like request_references, but returns the content of the lines containing the references, not just the locations.
@@ -647,7 +621,10 @@ class SolidLanguageServer(ABC):
:return: A list of MatchedConsecutiveLines objects, one for each reference.
"""
references = self.request_references(relative_file_path, line, column)
return [self.retrieve_content_around_line(ref["relativePath"], ref["range"]["start"]["line"], context_lines_before, context_lines_after) for ref in references]
return [
self.retrieve_content_around_line(ref["relativePath"], ref["range"]["start"]["line"], context_lines_before, context_lines_after)
for ref in references
]
def retrieve_full_file_content(self, relative_file_path: str) -> str:
"""
@@ -656,7 +633,9 @@ class SolidLanguageServer(ABC):
with self.open_file(relative_file_path) as file_data:
return file_data.contents
def retrieve_content_around_line(self, relative_file_path: str, line: int, context_lines_before: int = 0, context_lines_after: int = 0) -> MatchedConsecutiveLines:
def retrieve_content_around_line(
self, relative_file_path: str, line: int, context_lines_before: int = 0, context_lines_after: int = 0
) -> MatchedConsecutiveLines:
"""
Retrieve the content of the given file around the given line.
@@ -669,11 +648,17 @@ class SolidLanguageServer(ABC):
"""
with self.open_file(relative_file_path) as file_data:
file_contents = file_data.contents
return MatchedConsecutiveLines.from_file_contents(file_contents, line=line, context_lines_before=context_lines_before, context_lines_after=context_lines_after, source_file_path=relative_file_path)
return MatchedConsecutiveLines.from_file_contents(
file_contents,
line=line,
context_lines_before=context_lines_before,
context_lines_after=context_lines_after,
source_file_path=relative_file_path,
)
def request_completions(
self, relative_file_path: str, line: int, column: int, allow_incomplete: bool = False
) -> List[multilspy_types.CompletionItem]:
self, relative_file_path: str, line: int, column: int, allow_incomplete: bool = False
) -> list[multilspy_types.CompletionItem]:
"""
Raise a [textDocument/completion](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#textDocument_completion) request to the Language Server
to find completions at the given line and column in the given file. Wait for the response and return the result.
@@ -685,39 +670,35 @@ class SolidLanguageServer(ABC):
:return List[multilspy_types.CompletionItem]: A list of completions
"""
with self.open_file(relative_file_path):
open_file_buffer = self.open_file_buffers[
pathlib.Path(os.path.join(self.repository_root_path, relative_file_path)).as_uri()
]
open_file_buffer = self.open_file_buffers[pathlib.Path(os.path.join(self.repository_root_path, relative_file_path)).as_uri()]
completion_params: LSPTypes.CompletionParams = {
"position": {"line": line, "character": column},
"textDocument": {"uri": open_file_buffer.uri},
"context": {"triggerKind": LSPTypes.CompletionTriggerKind.Invoked},
}
response: Union[List[LSPTypes.CompletionItem], LSPTypes.CompletionList, None] = None
response: list[LSPTypes.CompletionItem] | LSPTypes.CompletionList | None = None
num_retries = 0
while response is None or (response["isIncomplete"] and num_retries < 30):
self.completions_available.wait()
response: Union[
List[LSPTypes.CompletionItem], LSPTypes.CompletionList, None
] = self.server.send.completion(completion_params)
response: list[LSPTypes.CompletionItem] | LSPTypes.CompletionList | None = 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)):
return []
if "items" in response:
response = response["items"]
response: List[LSPTypes.CompletionItem] = response
response = cast(list[LSPTypes.CompletionItem], response)
# TODO: Handle the case when the completion is a keyword
items = [item for item in response if item["kind"] != LSPTypes.CompletionItemKind.Keyword]
completions_list: List[multilspy_types.CompletionItem] = []
completions_list: list[multilspy_types.CompletionItem] = []
for item in items:
assert "insertText" in item or "textEdit" in item
@@ -745,8 +726,7 @@ class SolidLanguageServer(ABC):
item["textEdit"]["range"]["start"]["line"] == new_dot_lineno,
item["textEdit"]["range"]["start"]["character"] == new_dot_colno,
item["textEdit"]["range"]["start"]["line"] == item["textEdit"]["range"]["end"]["line"],
item["textEdit"]["range"]["start"]["character"]
== item["textEdit"]["range"]["end"]["character"],
item["textEdit"]["range"]["start"]["character"] == item["textEdit"]["range"]["end"]["character"],
)
)
@@ -760,12 +740,11 @@ class SolidLanguageServer(ABC):
completion_item = multilspy_types.CompletionItem(**completion_item)
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])
]
return [json.loads(json_repr) for json_repr in set(json.dumps(item, sort_keys=True) for item in completions_list)]
def request_document_symbols(self, relative_file_path: str, include_body: bool = False) -> Tuple[List[multilspy_types.UnifiedSymbolInformation], List[multilspy_types.UnifiedSymbolInformation]]:
def request_document_symbols(
self, relative_file_path: str, include_body: bool = False
) -> tuple[list[multilspy_types.UnifiedSymbolInformation], list[multilspy_types.UnifiedSymbolInformation]]:
"""
Raise a [textDocument/documentSymbol](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#textDocument_documentSymbol) request to the Language Server
to find symbols in the given file. Wait for the response and return the result.
@@ -797,13 +776,12 @@ class SolidLanguageServer(ABC):
self.logger.log(f"Requesting document symbols for {relative_file_path} from the Language Server", logging.DEBUG)
response = self.server.send.document_symbol(
{
"textDocument": {
"uri": pathlib.Path(os.path.join(self.repository_root_path, relative_file_path)).as_uri()
}
}
{"textDocument": {"uri": pathlib.Path(os.path.join(self.repository_root_path, relative_file_path)).as_uri()}}
)
self.logger.log(
f"Received {len(response) if response is not None else None} document symbols for {relative_file_path} from the Language Server",
logging.DEBUG,
)
self.logger.log(f"Received {len(response) if response is not None else None} document symbols for {relative_file_path} from the Language Server", logging.DEBUG)
def turn_item_into_symbol_with_children(item: GenericDocumentSymbol):
item = cast(multilspy_types.UnifiedSymbolInformation, item)
@@ -815,18 +793,18 @@ class SolidLanguageServer(ABC):
assert "range" in item
tree_location = multilspy_types.Location(
uri=uri,
range=item['range'],
range=item["range"],
absolutePath=absolute_path,
relativePath=relative_file_path,
)
item['location'] = tree_location
item["location"] = tree_location
location = item["location"]
if "absolutePath" not in location:
location["absolutePath"] = absolute_path
if "relativePath" not in location:
location["relativePath"] = relative_file_path
if include_body:
item['body'] = self.retrieve_symbol_body(item)
item["body"] = self.retrieve_symbol_body(item)
# handle missing selectionRange
if "selectionRange" not in item:
if "range" in item:
@@ -838,9 +816,9 @@ class SolidLanguageServer(ABC):
child["parent"] = item
item[LSPConstants.CHILDREN] = children
flat_all_symbol_list: List[multilspy_types.UnifiedSymbolInformation] = []
flat_all_symbol_list: list[multilspy_types.UnifiedSymbolInformation] = []
assert isinstance(response, list), f"Unexpected response from Language Server: {response}"
root_nodes: List[multilspy_types.UnifiedSymbolInformation] = []
root_nodes: list[multilspy_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]:
@@ -861,9 +839,9 @@ class SolidLanguageServer(ABC):
if LSPConstants.CHILDREN in root_symbol:
# TODO: l_tree should be a list of TreeRepr. Define the following function to return TreeRepr as well
def visit_tree_nodes_and_build_tree_repr(node: GenericDocumentSymbol) -> List[multilspy_types.UnifiedSymbolInformation]:
def visit_tree_nodes_and_build_tree_repr(node: GenericDocumentSymbol) -> list[multilspy_types.UnifiedSymbolInformation]:
node = cast(multilspy_types.UnifiedSymbolInformation, node)
l: List[multilspy_types.UnifiedSymbolInformation] = []
l: list[multilspy_types.UnifiedSymbolInformation] = []
turn_item_into_symbol_with_children(node)
assert LSPConstants.CHILDREN in node
children = node[LSPConstants.CHILDREN]
@@ -883,9 +861,11 @@ class SolidLanguageServer(ABC):
self._cache_has_changed = True
return result
def request_full_symbol_tree(self, within_relative_path: str | None = None, include_body: bool = False) -> List[multilspy_types.UnifiedSymbolInformation]:
def request_full_symbol_tree(
self, within_relative_path: str | None = None, include_body: bool = False
) -> list[multilspy_types.UnifiedSymbolInformation]:
"""
Will go through all files in the project or within a relative path and build a tree of symbols.
Will go through all files in the project or within a relative path and build a tree of symbols.
Note: this may be slow the first time it is called, especially if `within_relative_path` is not used to restrict the search.
For each file, a symbol of kind File (2) will be created. For directories, a symbol of kind Package (4) will be created.
@@ -902,21 +882,23 @@ class SolidLanguageServer(ABC):
:return: A list of root symbols representing the top-level packages/modules in the project.
"""
if within_relative_path is not None:
within_abs_path = os.path.join(self.repository_root_path, within_relative_path)
if not os.path.exists(within_abs_path):
raise FileNotFoundError(f"File or directory not found: {within_abs_path}")
if os.path.isfile(within_abs_path):
if self.is_ignored_path(within_relative_path):
self.logger.log(f"You passed a file explicitly, but it is ignored. This is probably an error. File: {within_relative_path}", logging.ERROR)
self.logger.log(
f"You passed a file explicitly, but it is ignored. This is probably an error. File: {within_relative_path}",
logging.ERROR,
)
return []
else:
_, root_nodes = self.request_document_symbols(within_relative_path, include_body=include_body)
return root_nodes
# Helper function to recursively process directories
def process_directory(rel_dir_path: str) -> List[multilspy_types.UnifiedSymbolInformation]:
def process_directory(rel_dir_path: str) -> list[multilspy_types.UnifiedSymbolInformation]:
abs_dir_path = self.repository_root_path if rel_dir_path == "." else os.path.join(self.repository_root_path, rel_dir_path)
abs_dir_path = os.path.realpath(abs_dir_path)
@@ -931,7 +913,7 @@ class SolidLanguageServer(ABC):
return []
# Create package symbol for directory
package_symbol = multilspy_types.UnifiedSymbolInformation( # type: ignore
package_symbol = multilspy_types.UnifiedSymbolInformation( # type: ignore
name=os.path.basename(abs_dir_path),
kind=multilspy_types.SymbolKind.Package,
location=multilspy_types.Location(
@@ -940,7 +922,7 @@ class SolidLanguageServer(ABC):
absolutePath=str(abs_dir_path),
relativePath=str(Path(abs_dir_path).resolve().relative_to(self.repository_root_path)),
),
children=[]
children=[],
)
result.append(package_symbol)
@@ -964,7 +946,7 @@ class SolidLanguageServer(ABC):
file_rel_path = str(Path(contained_dir_or_file_abs_path).resolve().relative_to(self.repository_root_path))
with self.open_file(file_rel_path) as file_data:
fileRange = self._get_range_from_file_content(file_data.contents)
file_symbol = multilspy_types.UnifiedSymbolInformation( # type: ignore
file_symbol = multilspy_types.UnifiedSymbolInformation( # type: ignore
name=os.path.splitext(contained_dir_or_file_name)[0],
kind=multilspy_types.SymbolKind.File,
range=fileRange,
@@ -985,7 +967,7 @@ 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[multilspy_types.UnifiedSymbolInformation]):
def fix_relative_path(nodes: list[multilspy_types.UnifiedSymbolInformation]):
for node in nodes:
if "location" in node and "relativePath" in node["location"]:
path = Path(node["location"]["relativePath"])
@@ -1015,8 +997,7 @@ class SolidLanguageServer(ABC):
end_line = len(lines)
end_column = len(lines[-1])
return multilspy_types.Range(
start=multilspy_types.Position(line=0, character=0),
end=multilspy_types.Position(line=end_line, character=end_column)
start=multilspy_types.Position(line=0, character=0), end=multilspy_types.Position(line=end_line, character=end_column)
)
def request_dir_overview(self, relative_dir_path: str) -> dict[str, list[tuple[str, multilspy_types.SymbolKind, int, int]]]:
@@ -1038,12 +1019,14 @@ class SolidLanguageServer(ABC):
assert "location" in child
assert "selectionRange" in child
path = Path(child["location"]["absolutePath"]).resolve().relative_to(self.repository_root_path)
result[str(path)].append((
child["name"],
child["kind"],
child["selectionRange"]["start"]["line"],
child["selectionRange"]["start"]["character"]
))
result[str(path)].append(
(
child["name"],
child["kind"],
child["selectionRange"]["start"]["line"],
child["selectionRange"]["start"]["character"],
)
)
# For package/directory symbols, process their children
for child in symbol["children"]:
process_symbol(child)
@@ -1063,15 +1046,10 @@ class SolidLanguageServer(ABC):
for root in document_roots:
try:
result.append(
( root["name"],
root["kind"],
root["selectionRange"]["start"]["line"],
root["selectionRange"]["start"]["character"],)
(root["name"], root["kind"], root["selectionRange"]["start"]["line"], root["selectionRange"]["start"]["character"])
)
except KeyError as e:
raise KeyError(
f"Could not process symbol of name {root.get('name', 'unknown')} in {relative_file_path=}"
) from e
raise KeyError(f"Could not process symbol of name {root.get('name', 'unknown')} in {relative_file_path=}") from e
return result
def request_overview(self, within_relative_path: str) -> dict[str, list[tuple[str, multilspy_types.SymbolKind, int, int]]]:
@@ -1091,7 +1069,7 @@ class SolidLanguageServer(ABC):
else:
return self.request_dir_overview(within_relative_path)
def request_hover(self, relative_file_path: str, line: int, column: int) -> Union[multilspy_types.Hover, None]:
def request_hover(self, relative_file_path: str, line: int, column: int) -> multilspy_types.Hover | None:
"""
Raise a [textDocument/hover](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#textDocument_hover) request to the Language Server
to find the hover information at the given line and column in the given file. Wait for the response and return the result.
@@ -1105,9 +1083,7 @@ class SolidLanguageServer(ABC):
with self.open_file(relative_file_path):
response = self.server.send.hover(
{
"textDocument": {
"uri": pathlib.Path(os.path.join(self.repository_root_path, relative_file_path)).as_uri()
},
"textDocument": {"uri": pathlib.Path(os.path.join(self.repository_root_path, relative_file_path)).as_uri()},
"position": {
"line": line,
"character": column,
@@ -1124,7 +1100,9 @@ class SolidLanguageServer(ABC):
# ----------------------------- FROM HERE ON MODIFICATIONS BY MISCHA --------------------
def retrieve_symbol_body(self, symbol: multilspy_types.UnifiedSymbolInformation | LSPTypes.DocumentSymbol | LSPTypes.SymbolInformation) -> str:
def retrieve_symbol_body(
self, symbol: multilspy_types.UnifiedSymbolInformation | LSPTypes.DocumentSymbol | LSPTypes.SymbolInformation
) -> str:
"""
Load the body of the given symbol. If the body is already contained in the symbol, just return it.
"""
@@ -1138,7 +1116,7 @@ class SolidLanguageServer(ABC):
assert "relativePath" in symbol["location"]
symbol_file = self.retrieve_full_file_content(symbol["location"]["relativePath"])
symbol_lines = symbol_file.split("\n")
symbol_body = "\n".join(symbol_lines[symbol_start_line:symbol_end_line+1])
symbol_body = "\n".join(symbol_lines[symbol_start_line : symbol_end_line + 1])
# remove leading indentation
symbol_start_column = symbol["location"]["range"]["start"]["character"]
@@ -1163,12 +1141,12 @@ class SolidLanguageServer(ABC):
return rel_file_paths
def search_files_for_pattern(
self,
pattern: re.Pattern | str,
context_lines_before: int = 0,
context_lines_after: int = 0,
paths_include_glob: str | None = None,
paths_exclude_glob: str | None = None,
self,
pattern: re.Pattern | str,
context_lines_before: int = 0,
context_lines_after: int = 0,
paths_include_glob: str | None = None,
paths_exclude_glob: str | None = None,
) -> list[MatchedConsecutiveLines]:
"""
Search for a pattern across all files analyzed by the Language Server.
@@ -1191,19 +1169,19 @@ class SolidLanguageServer(ABC):
context_lines_before=context_lines_before,
context_lines_after=context_lines_after,
paths_include_glob=paths_include_glob,
paths_exclude_glob=paths_exclude_glob
paths_exclude_glob=paths_exclude_glob,
)
def request_referencing_symbols(
self,
relative_file_path: str,
line: int,
column: int,
include_imports: bool = True,
include_self: bool = False,
include_body: bool = False,
include_file_symbols: bool = False,
) -> List[ReferenceInSymbol]:
self,
relative_file_path: str,
line: int,
column: int,
include_imports: bool = True,
include_self: bool = False,
include_body: bool = False,
include_file_symbols: bool = False,
) -> list[ReferenceInSymbol]:
"""
Finds all symbols that reference the symbol at the given location.
This is similar to request_references but filters to only include symbols
@@ -1244,9 +1222,7 @@ class SolidLanguageServer(ABC):
with self.open_file(ref_path) as file_data:
# Get the containing symbol for this reference
containing_symbol = self.request_containing_symbol(
ref_path, ref_line, ref_col, include_body=include_body
)
containing_symbol = self.request_containing_symbol(ref_path, ref_line, ref_col, include_body=include_body)
if containing_symbol is None:
# TODO: HORRIBLE HACK! I don't know how to do it better for now...
# THIS IS BOUND TO BREAK IN MANY CASES! IT IS ALSO SPECIFIC TO PYTHON!
@@ -1276,7 +1252,7 @@ class SolidLanguageServer(ABC):
if containing_symbol is None and include_file_symbols:
self.logger.log(
f"Could not find containing symbol for {ref_path}:{ref_line}:{ref_col}. Returning file symbol instead",
logging.WARNING
logging.WARNING,
)
fileRange = self._get_range_from_file_content(file_data.contents)
location = multilspy_types.Location(
@@ -1301,7 +1277,7 @@ class SolidLanguageServer(ABC):
children=[],
body=body,
)
if containing_symbol is None or not include_file_symbols and containing_symbol["kind"] == multilspy_types.SymbolKind.File:
if containing_symbol is None or (not include_file_symbols and containing_symbol["kind"] == multilspy_types.SymbolKind.File):
continue
assert "location" in containing_symbol
@@ -1309,31 +1285,31 @@ class SolidLanguageServer(ABC):
# Checking for self-reference
if (
containing_symbol["location"]["relativePath"] == relative_file_path
and containing_symbol["selectionRange"]["start"]["line"] == ref_line
and containing_symbol["selectionRange"]["start"]["character"] == ref_col
containing_symbol["location"]["relativePath"] == relative_file_path
and containing_symbol["selectionRange"]["start"]["line"] == ref_line
and containing_symbol["selectionRange"]["start"]["character"] == ref_col
):
incoming_symbol = containing_symbol
if include_self:
result.append(ReferenceInSymbol(symbol=containing_symbol, line=ref_line, character=ref_col))
continue
else:
self.logger.log(f"Found self-reference for {incoming_symbol['name']}, skipping it since {include_self=}", logging.DEBUG)
continue
self.logger.log(f"Found self-reference for {incoming_symbol['name']}, skipping it since {include_self=}", logging.DEBUG)
continue
# checking whether reference is an import
# This is neither really safe nor elegant, but if we don't do it,
# there is no way to distinguish between definitions and imports as import is not a symbol-type
# and we get the type referenced symbol resulting from imports...
if (not include_imports \
and incoming_symbol is not None \
and containing_symbol["name"] == incoming_symbol["name"] \
and containing_symbol["kind"] == incoming_symbol["kind"] \
):
if (
not include_imports
and incoming_symbol is not None
and containing_symbol["name"] == incoming_symbol["name"]
and containing_symbol["kind"] == incoming_symbol["kind"]
):
self.logger.log(
f"Found import of referenced symbol {incoming_symbol['name']}"
f"in {containing_symbol['location']['relativePath']}, skipping",
logging.DEBUG
logging.DEBUG,
)
continue
@@ -1342,12 +1318,12 @@ class SolidLanguageServer(ABC):
return result
def request_containing_symbol(
self,
relative_file_path: str,
line: int,
column: Optional[int] = None,
strict: bool = False,
include_body: bool = False,
self,
relative_file_path: str,
line: int,
column: int | None = None,
strict: bool = False,
include_body: bool = False,
) -> multilspy_types.UnifiedSymbolInformation | None:
"""
Finds the first symbol containing the position for the given file.
@@ -1377,9 +1353,7 @@ class SolidLanguageServer(ABC):
"""
# checking if the line is empty, unfortunately ugly and duplicating code, but I don't want to refactor
with self.open_file(relative_file_path):
absolute_file_path = str(
PurePath(self.repository_root_path, relative_file_path)
)
absolute_file_path = str(PurePath(self.repository_root_path, relative_file_path))
content = FileUtils.read_file(self.logger, absolute_file_path)
if content.split("\n")[line].strip() == "":
self.logger.log(
@@ -1411,11 +1385,7 @@ class SolidLanguageServer(ABC):
location["uri"] = Path(absolute_file_path).as_uri()
# Allowed container kinds, currently only for Python
container_symbol_kinds = {
multilspy_types.SymbolKind.Method,
multilspy_types.SymbolKind.Function,
multilspy_types.SymbolKind.Class
}
container_symbol_kinds = {multilspy_types.SymbolKind.Method, multilspy_types.SymbolKind.Function, multilspy_types.SymbolKind.Class}
def is_position_in_range(line: int, range_d: multilspy_types.Range) -> bool:
start = range_d["start"]
@@ -1434,11 +1404,11 @@ class SolidLanguageServer(ABC):
# Only consider containers that are not one-liners (otherwise we may get imports)
candidate_containers = [
s for s in symbols if s["kind"] in container_symbol_kinds and s["location"]["range"]["start"]["line"] != s["location"]["range"]["end"]["line"]
]
var_containers = [
s for s in symbols if s["kind"] == multilspy_types.SymbolKind.Variable
s
for s in symbols
if s["kind"] in container_symbol_kinds and s["location"]["range"]["start"]["line"] != s["location"]["range"]["end"]["line"]
]
var_containers = [s for s in symbols if s["kind"] == multilspy_types.SymbolKind.Variable]
candidate_containers.extend(var_containers)
if not candidate_containers:
@@ -1461,7 +1431,9 @@ class SolidLanguageServer(ABC):
else:
return None
def request_container_of_symbol(self, symbol: multilspy_types.UnifiedSymbolInformation, include_body: bool = False) -> multilspy_types.UnifiedSymbolInformation | None:
def request_container_of_symbol(
self, symbol: multilspy_types.UnifiedSymbolInformation, include_body: bool = False
) -> multilspy_types.UnifiedSymbolInformation | None:
"""
Finds the container of the given symbol if there is one. If the parent attribute is present, the parent is returned
without further searching.
@@ -1482,12 +1454,12 @@ class SolidLanguageServer(ABC):
)
def request_defining_symbol(
self,
relative_file_path: str,
line: int,
column: int,
include_body: bool = False,
) -> Optional[multilspy_types.UnifiedSymbolInformation]:
self,
relative_file_path: str,
line: int,
column: int,
include_body: bool = False,
) -> multilspy_types.UnifiedSymbolInformation | None:
"""
Finds the symbol that defines the symbol at the given location.
@@ -1519,9 +1491,7 @@ class SolidLanguageServer(ABC):
def_col = definition["range"]["start"]["character"]
# Find the symbol at or containing this location
defining_symbol = self.request_containing_symbol(
def_path, def_line, def_col, strict=False, include_body=include_body
)
defining_symbol = self.request_containing_symbol(def_path, def_line, def_col, strict=False, include_body=include_body)
return defining_symbol
@@ -1587,7 +1557,7 @@ class SolidLanguageServer(ABC):
logging.ERROR,
)
def request_workspace_symbol(self, query: str) -> Union[List[multilspy_types.UnifiedSymbolInformation], None]:
def request_workspace_symbol(self, query: str) -> list[multilspy_types.UnifiedSymbolInformation] | None:
"""
Raise a [workspace/symbol](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#workspace_symbol) request to the Language Server
to find symbols across the whole workspace. Wait for the response and return the result.
@@ -1602,7 +1572,7 @@ class SolidLanguageServer(ABC):
assert isinstance(response, list)
ret: List[multilspy_types.UnifiedSymbolInformation] = []
ret: list[multilspy_types.UnifiedSymbolInformation] = []
for item in response:
assert isinstance(item, dict)
@@ -1620,7 +1590,10 @@ class SolidLanguageServer(ABC):
:return: self for method chaining
"""
self.logger.log(f"Starting language server with language {self.language_server.language} for {self.language_server.repository_root_path}", logging.INFO)
self.logger.log(
f"Starting language server with language {self.language_server.language} for {self.language_server.repository_root_path}",
logging.INFO,
)
self._server_context = self._start_server_process()
return self
+37 -21
View File
@@ -5,16 +5,29 @@ import os
import subprocess
import threading
import time
from collections.abc import Callable
from dataclasses import dataclass
from queue import Queue
from typing import Any, Callable, Dict, Optional
from typing import Any
import psutil
from multilspy.lsp_protocol_handler.lsp_requests import LspNotification
from multilspy.lsp_protocol_handler.lsp_types import ErrorCodes
from multilspy.lsp_protocol_handler.server import ProcessLaunchInfo, StringDict, content_length, ENCODING, Error, MessageType, \
make_notification, PayloadLike, make_response, make_error_response, make_request, create_message
from multilspy.lsp_protocol_handler.server import (
ENCODING,
Error,
MessageType,
PayloadLike,
ProcessLaunchInfo,
StringDict,
content_length,
create_message,
make_error_response,
make_notification,
make_request,
make_response,
)
from multilspy.multilspy_exceptions import MultilspyException
from solidlsp.lsp_request import SolidLspRequest
@@ -25,8 +38,8 @@ class Request:
@dataclass
class Result:
payload: Optional[PayloadLike] = None
error: Optional[Error] = None
payload: PayloadLike | None = None
error: Error | None = None
def is_error(self) -> bool:
return self.error is not None
@@ -81,13 +94,14 @@ class SolidLanguageServerHandler:
language server process in an independent process group. Default is `True`. Setting it to
`False` means that the language server process will be in the same process group as the
the current process, and any SIGINT and SIGTERM signals will be sent to both processes.
"""
def __init__(
self,
process_launch_info: ProcessLaunchInfo,
logger: Optional[Callable[[str, str, StringDict | str], None]] = None,
start_independent_lsp_process=True,
self,
process_launch_info: ProcessLaunchInfo,
logger: Callable[[str, str, StringDict | str], None] | None = None,
start_independent_lsp_process=True,
) -> None:
"""
Params:
@@ -103,7 +117,7 @@ class SolidLanguageServerHandler:
self._received_shutdown = False
self.request_id = 1
self._response_handlers: Dict[Any, Request] = {}
self._response_handlers: dict[Any, Request] = {}
self.on_request_handlers = {}
self.on_notification_handlers = {}
self.logger = logger
@@ -141,7 +155,7 @@ class SolidLanguageServerHandler:
env=child_proc_env,
cwd=self.process_launch_info.cwd,
start_new_session=self.start_independent_lsp_process,
shell=True
shell=True,
)
# Check if process terminated immediately
@@ -149,7 +163,7 @@ class SolidLanguageServerHandler:
log.error("Language server has already terminated/could not be started")
# Process has already terminated
stderr_data = self.process.stderr.read()
error_message = stderr_data.decode('utf-8', errors='replace')
error_message = stderr_data.decode("utf-8", errors="replace")
raise RuntimeError(f"Process terminated immediately with code {self.process.returncode}. Error: {error_message}")
# start threads to read stdout and stderr of the process
@@ -284,7 +298,7 @@ class SolidLanguageServerHandler:
# Process has terminated, check if we can still read
pass
data = b''
data = b""
while len(data) < num_bytes:
chunk = stream.read(num_bytes - len(data))
if not chunk:
@@ -332,7 +346,7 @@ class SolidLanguageServerHandler:
line = self.process.stderr.readline()
if not line:
continue
self._log("LSP stderr: " + line.decode(ENCODING, errors='replace'))
self._log("LSP stderr: " + line.decode(ENCODING, errors="replace"))
except (BrokenPipeError, ConnectionResetError):
pass
@@ -342,7 +356,7 @@ class SolidLanguageServerHandler:
"""
try:
self._receive_payload(json.loads(body))
except IOError as ex:
except OSError as ex:
self._log(f"malformed {ENCODING}: {ex}")
except UnicodeDecodeError as ex:
self._log(f"malformed {ENCODING}: {ex}")
@@ -368,7 +382,7 @@ class SolidLanguageServerHandler:
except Exception as err:
self._log(f"Error handling server payload: {err}")
def send_notification(self, method: str, params: Optional[dict] = None) -> None:
def send_notification(self, method: str, params: dict | None = None) -> None:
"""
Send notification pertaining to the given method to the server with the given parameters
"""
@@ -387,7 +401,7 @@ class SolidLanguageServerHandler:
# Use lock to prevent race conditions on tasks and task_counter
self._send_payload(make_error_response(request_id, err))
def send_request(self, method: str, params: Optional[dict] = None) -> PayloadLike:
def send_request(self, method: str, params: dict | None = None) -> PayloadLike:
"""
Send request to the server, register the request id, and wait for the response
"""
@@ -406,10 +420,12 @@ class SolidLanguageServerHandler:
self._log(f"Waiting for response to request {method} with params:\n{params}")
result = request.get_result()
self._log(f"Processing result")
self._log("Processing result")
if result.is_error():
raise MultilspyException(f"Could not process request {method} with params:\n{params}.\n Language server error: {result.error}") from result.error
raise MultilspyException(
f"Could not process request {method} with params:\n{params}.\n Language server error: {result.error}"
) from result.error
self._log(f"Returning non-error result, which is:\n{result.payload}")
return result.payload
@@ -472,7 +488,7 @@ class SolidLanguageServerHandler:
request_id,
Error(
ErrorCodes.MethodNotFound,
"method '{}' not handled on client.".format(method),
f"method '{method}' not handled on client.",
),
)
return
+129 -187
View File
@@ -1,4 +1,5 @@
from typing import List, Union
from typing import Union
from multilspy.lsp_protocol_handler import lsp_types
@@ -6,258 +7,230 @@ class SolidLspRequest:
def __init__(self, send_request):
self.send_request = send_request
def implementation(
self, params: lsp_types.ImplementationParams
) -> Union["lsp_types.Definition", List["lsp_types.LocationLink"], None]:
def implementation(self, params: lsp_types.ImplementationParams) -> Union["lsp_types.Definition", list["lsp_types.LocationLink"], None]:
"""A request to resolve the implementation locations of a symbol at a given text
document position. The request's parameter is of type [TextDocumentPositionParams]
(#TextDocumentPositionParams) the response is of type {@link Definition} or a
Thenable that resolves to such."""
Thenable that resolves to such.
"""
return self.send_request("textDocument/implementation", params)
def type_definition(
self, params: lsp_types.TypeDefinitionParams
) -> Union["lsp_types.Definition", List["lsp_types.LocationLink"], None]:
self, params: lsp_types.TypeDefinitionParams
) -> Union["lsp_types.Definition", list["lsp_types.LocationLink"], None]:
"""A request to resolve the type definition locations of a symbol at a given text
document position. The request's parameter is of type [TextDocumentPositionParams]
(#TextDocumentPositionParams) the response is of type {@link Definition} or a
Thenable that resolves to such."""
Thenable that resolves to such.
"""
return self.send_request("textDocument/typeDefinition", params)
def document_color(
self, params: lsp_types.DocumentColorParams
) -> List["lsp_types.ColorInformation"]:
def document_color(self, params: lsp_types.DocumentColorParams) -> list["lsp_types.ColorInformation"]:
"""A request to list all color symbols found in a given text document. The request's
parameter is of type {@link DocumentColorParams} the
response is of type {@link ColorInformation ColorInformation[]} or a Thenable
that resolves to such."""
that resolves to such.
"""
return self.send_request("textDocument/documentColor", params)
def color_presentation(
self, params: lsp_types.ColorPresentationParams
) -> List["lsp_types.ColorPresentation"]:
def color_presentation(self, params: lsp_types.ColorPresentationParams) -> list["lsp_types.ColorPresentation"]:
"""A request to list all presentation for a color. The request's
parameter is of type {@link ColorPresentationParams} the
response is of type {@link ColorInformation ColorInformation[]} or a Thenable
that resolves to such."""
that resolves to such.
"""
return self.send_request("textDocument/colorPresentation", params)
def folding_range(
self, params: lsp_types.FoldingRangeParams
) -> Union[List["lsp_types.FoldingRange"], None]:
def folding_range(self, params: lsp_types.FoldingRangeParams) -> list["lsp_types.FoldingRange"] | None:
"""A request to provide folding ranges in a document. The request's
parameter is of type {@link FoldingRangeParams}, the
response is of type {@link FoldingRangeList} or a Thenable
that resolves to such."""
that resolves to such.
"""
return self.send_request("textDocument/foldingRange", params)
def declaration(
self, params: lsp_types.DeclarationParams
) -> Union["lsp_types.Declaration", List["lsp_types.LocationLink"], None]:
def declaration(self, params: lsp_types.DeclarationParams) -> Union["lsp_types.Declaration", list["lsp_types.LocationLink"], None]:
"""A request to resolve the type definition locations of a symbol at a given text
document position. The request's parameter is of type [TextDocumentPositionParams]
(#TextDocumentPositionParams) the response is of type {@link Declaration}
or a typed array of {@link DeclarationLink} or a Thenable that resolves
to such."""
to such.
"""
return self.send_request("textDocument/declaration", params)
def selection_range(
self, params: lsp_types.SelectionRangeParams
) -> Union[List["lsp_types.SelectionRange"], None]:
def selection_range(self, params: lsp_types.SelectionRangeParams) -> list["lsp_types.SelectionRange"] | None:
"""A request to provide selection ranges in a document. The request's
parameter is of type {@link SelectionRangeParams}, the
response is of type {@link SelectionRange SelectionRange[]} or a Thenable
that resolves to such."""
that resolves to such.
"""
return self.send_request("textDocument/selectionRange", params)
def prepare_call_hierarchy(
self, params: lsp_types.CallHierarchyPrepareParams
) -> Union[List["lsp_types.CallHierarchyItem"], None]:
def prepare_call_hierarchy(self, params: lsp_types.CallHierarchyPrepareParams) -> list["lsp_types.CallHierarchyItem"] | None:
"""A request to result a `CallHierarchyItem` in a document at a given position.
Can be used as an input to an incoming or outgoing call hierarchy.
@since 3.16.0"""
@since 3.16.0
"""
return self.send_request("textDocument/prepareCallHierarchy", params)
def incoming_calls(
self, params: lsp_types.CallHierarchyIncomingCallsParams
) -> Union[List["lsp_types.CallHierarchyIncomingCall"], None]:
def incoming_calls(self, params: lsp_types.CallHierarchyIncomingCallsParams) -> list["lsp_types.CallHierarchyIncomingCall"] | None:
"""A request to resolve the incoming calls for a given `CallHierarchyItem`.
@since 3.16.0"""
@since 3.16.0
"""
return self.send_request("callHierarchy/incomingCalls", params)
def outgoing_calls(
self, params: lsp_types.CallHierarchyOutgoingCallsParams
) -> Union[List["lsp_types.CallHierarchyOutgoingCall"], None]:
def outgoing_calls(self, params: lsp_types.CallHierarchyOutgoingCallsParams) -> list["lsp_types.CallHierarchyOutgoingCall"] | None:
"""A request to resolve the outgoing calls for a given `CallHierarchyItem`.
@since 3.16.0"""
@since 3.16.0
"""
return self.send_request("callHierarchy/outgoingCalls", params)
def semantic_tokens_full(
self, params: lsp_types.SemanticTokensParams
) -> Union["lsp_types.SemanticTokens", None]:
def semantic_tokens_full(self, params: lsp_types.SemanticTokensParams) -> Union["lsp_types.SemanticTokens", None]:
"""@since 3.16.0"""
return self.send_request("textDocument/semanticTokens/full", params)
def semantic_tokens_delta(
self, params: lsp_types.SemanticTokensDeltaParams
self, params: lsp_types.SemanticTokensDeltaParams
) -> Union["lsp_types.SemanticTokens", "lsp_types.SemanticTokensDelta", None]:
"""@since 3.16.0"""
return self.send_request("textDocument/semanticTokens/full/delta", params)
def semantic_tokens_range(
self, params: lsp_types.SemanticTokensRangeParams
) -> Union["lsp_types.SemanticTokens", None]:
def semantic_tokens_range(self, params: lsp_types.SemanticTokensRangeParams) -> Union["lsp_types.SemanticTokens", None]:
"""@since 3.16.0"""
return self.send_request("textDocument/semanticTokens/range", params)
def linked_editing_range(
self, params: lsp_types.LinkedEditingRangeParams
) -> Union["lsp_types.LinkedEditingRanges", None]:
def linked_editing_range(self, params: lsp_types.LinkedEditingRangeParams) -> Union["lsp_types.LinkedEditingRanges", None]:
"""A request to provide ranges that can be edited together.
@since 3.16.0"""
@since 3.16.0
"""
return self.send_request("textDocument/linkedEditingRange", params)
def will_create_files(
self, params: lsp_types.CreateFilesParams
) -> Union["lsp_types.WorkspaceEdit", None]:
def will_create_files(self, params: lsp_types.CreateFilesParams) -> Union["lsp_types.WorkspaceEdit", None]:
"""The will create files request is sent from the client to the server before files are actually
created as long as the creation is triggered from within the client.
@since 3.16.0"""
@since 3.16.0
"""
return self.send_request("workspace/willCreateFiles", params)
def will_rename_files(
self, params: lsp_types.RenameFilesParams
) -> Union["lsp_types.WorkspaceEdit", None]:
def will_rename_files(self, params: lsp_types.RenameFilesParams) -> Union["lsp_types.WorkspaceEdit", None]:
"""The will rename files request is sent from the client to the server before files are actually
renamed as long as the rename is triggered from within the client.
@since 3.16.0"""
@since 3.16.0
"""
return self.send_request("workspace/willRenameFiles", params)
def will_delete_files(
self, params: lsp_types.DeleteFilesParams
) -> Union["lsp_types.WorkspaceEdit", None]:
def will_delete_files(self, params: lsp_types.DeleteFilesParams) -> Union["lsp_types.WorkspaceEdit", None]:
"""The did delete files notification is sent from the client to the server when
files were deleted from within the client.
@since 3.16.0"""
@since 3.16.0
"""
return self.send_request("workspace/willDeleteFiles", params)
def moniker(
self, params: lsp_types.MonikerParams
) -> Union[List["lsp_types.Moniker"], None]:
def moniker(self, params: lsp_types.MonikerParams) -> list["lsp_types.Moniker"] | None:
"""A request to get the moniker of a symbol at a given text document position.
The request parameter is of type {@link TextDocumentPositionParams}.
The response is of type {@link Moniker Moniker[]} or `null`."""
The response is of type {@link Moniker Moniker[]} or `null`.
"""
return self.send_request("textDocument/moniker", params)
def prepare_type_hierarchy(
self, params: lsp_types.TypeHierarchyPrepareParams
) -> Union[List["lsp_types.TypeHierarchyItem"], None]:
def prepare_type_hierarchy(self, params: lsp_types.TypeHierarchyPrepareParams) -> list["lsp_types.TypeHierarchyItem"] | None:
"""A request to result a `TypeHierarchyItem` in a document at a given position.
Can be used as an input to a subtypes or supertypes type hierarchy.
@since 3.17.0"""
@since 3.17.0
"""
return self.send_request("textDocument/prepareTypeHierarchy", params)
def type_hierarchy_supertypes(
self, params: lsp_types.TypeHierarchySupertypesParams
) -> Union[List["lsp_types.TypeHierarchyItem"], None]:
def type_hierarchy_supertypes(self, params: lsp_types.TypeHierarchySupertypesParams) -> list["lsp_types.TypeHierarchyItem"] | None:
"""A request to resolve the supertypes for a given `TypeHierarchyItem`.
@since 3.17.0"""
@since 3.17.0
"""
return self.send_request("typeHierarchy/supertypes", params)
def type_hierarchy_subtypes(
self, params: lsp_types.TypeHierarchySubtypesParams
) -> Union[List["lsp_types.TypeHierarchyItem"], None]:
def type_hierarchy_subtypes(self, params: lsp_types.TypeHierarchySubtypesParams) -> list["lsp_types.TypeHierarchyItem"] | None:
"""A request to resolve the subtypes for a given `TypeHierarchyItem`.
@since 3.17.0"""
@since 3.17.0
"""
return self.send_request("typeHierarchy/subtypes", params)
def inline_value(
self, params: lsp_types.InlineValueParams
) -> Union[List["lsp_types.InlineValue"], None]:
def inline_value(self, params: lsp_types.InlineValueParams) -> list["lsp_types.InlineValue"] | None:
"""A request to provide inline values in a document. The request's parameter is of
type {@link InlineValueParams}, the response is of type
{@link InlineValue InlineValue[]} or a Thenable that resolves to such.
@since 3.17.0"""
@since 3.17.0
"""
return self.send_request("textDocument/inlineValue", params)
def inlay_hint(
self, params: lsp_types.InlayHintParams
) -> Union[List["lsp_types.InlayHint"], None]:
def inlay_hint(self, params: lsp_types.InlayHintParams) -> list["lsp_types.InlayHint"] | None:
"""A request to provide inlay hints in a document. The request's parameter is of
type {@link InlayHintsParams}, the response is of type
{@link InlayHint InlayHint[]} or a Thenable that resolves to such.
@since 3.17.0"""
@since 3.17.0
"""
return self.send_request("textDocument/inlayHint", params)
def resolve_inlay_hint(
self, params: lsp_types.InlayHint
) -> "lsp_types.InlayHint":
def resolve_inlay_hint(self, params: lsp_types.InlayHint) -> "lsp_types.InlayHint":
"""A request to resolve additional properties for an inlay hint.
The request's parameter is of type {@link InlayHint}, the response is
of type {@link InlayHint} or a Thenable that resolves to such.
@since 3.17.0"""
@since 3.17.0
"""
return self.send_request("inlayHint/resolve", params)
def text_document_diagnostic(
self, params: lsp_types.DocumentDiagnosticParams
) -> "lsp_types.DocumentDiagnosticReport":
def text_document_diagnostic(self, params: lsp_types.DocumentDiagnosticParams) -> "lsp_types.DocumentDiagnosticReport":
"""The document diagnostic request definition.
@since 3.17.0"""
@since 3.17.0
"""
return self.send_request("textDocument/diagnostic", params)
def workspace_diagnostic(
self, params: lsp_types.WorkspaceDiagnosticParams
) -> "lsp_types.WorkspaceDiagnosticReport":
def workspace_diagnostic(self, params: lsp_types.WorkspaceDiagnosticParams) -> "lsp_types.WorkspaceDiagnosticReport":
"""The workspace diagnostic request definition.
@since 3.17.0"""
@since 3.17.0
"""
return self.send_request("workspace/diagnostic", params)
def initialize(
self, params: lsp_types.InitializeParams
) -> "lsp_types.InitializeResult":
def initialize(self, params: lsp_types.InitializeParams) -> "lsp_types.InitializeResult":
"""The initialize request is sent from the client to the server.
It is sent once as the request after starting up the server.
The requests parameter is of type {@link InitializeParams}
the response if of type {@link InitializeResult} of a Thenable that
resolves to such."""
resolves to such.
"""
return self.send_request("initialize", params)
def shutdown(self) -> None:
"""A shutdown request is sent from the client to the server.
It is sent once when the client decides to shutdown the
server. The only notification that is sent after a shutdown request
is the exit event."""
is the exit event.
"""
return self.send_request("shutdown")
def will_save_wait_until(
self, params: lsp_types.WillSaveTextDocumentParams
) -> Union[List["lsp_types.TextEdit"], 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
the document is actually saved. The request can return an array of TextEdits
which will be applied to the text document before it is saved. Please note that
clients might drop results if computing the text edits took too long or if a
server constantly fails on this request. This is done to keep the save fast and
reliable."""
reliable.
"""
return self.send_request("textDocument/willSaveWaitUntil", params)
def completion(
self, params: lsp_types.CompletionParams
) -> Union[List["lsp_types.CompletionItem"], "lsp_types.CompletionList", None]:
def completion(self, params: lsp_types.CompletionParams) -> Union[list["lsp_types.CompletionItem"], "lsp_types.CompletionList", None]:
"""Request to request completion at a given text document position. The request's
parameter is of type {@link TextDocumentPosition} the response
is of type {@link CompletionItem CompletionItem[]} or {@link CompletionList}
@@ -270,85 +243,72 @@ class SolidLspRequest:
"""
return self.send_request("textDocument/completion", params)
def resolve_completion_item(
self, params: lsp_types.CompletionItem
) -> "lsp_types.CompletionItem":
def resolve_completion_item(self, params: lsp_types.CompletionItem) -> "lsp_types.CompletionItem":
"""Request to resolve additional information for a given completion item.The request's
parameter is of type {@link CompletionItem} the response
is of type {@link CompletionItem} or a Thenable that resolves to such."""
is of type {@link CompletionItem} or a Thenable that resolves to such.
"""
return self.send_request("completionItem/resolve", params)
def hover(
self, params: lsp_types.HoverParams
) -> Union["lsp_types.Hover", None]:
def hover(self, params: lsp_types.HoverParams) -> Union["lsp_types.Hover", None]:
"""Request to request hover information at a given text document position. The request's
parameter is of type {@link TextDocumentPosition} the response is of
type {@link Hover} or a Thenable that resolves to such."""
type {@link Hover} or a Thenable that resolves to such.
"""
return self.send_request("textDocument/hover", params)
def signature_help(
self, params: lsp_types.SignatureHelpParams
) -> Union["lsp_types.SignatureHelp", None]:
def signature_help(self, params: lsp_types.SignatureHelpParams) -> Union["lsp_types.SignatureHelp", None]:
return self.send_request("textDocument/signatureHelp", params)
def definition(
self, params: lsp_types.DefinitionParams
) -> Union["lsp_types.Definition", List["lsp_types.LocationLink"], None]:
def definition(self, params: lsp_types.DefinitionParams) -> Union["lsp_types.Definition", list["lsp_types.LocationLink"], None]:
"""A request to resolve the definition location of a symbol at a given text
document position. The request's parameter is of type [TextDocumentPosition]
(#TextDocumentPosition) the response is of either type {@link Definition}
or a typed array of {@link DefinitionLink} or a Thenable that resolves
to such."""
to such.
"""
return self.send_request("textDocument/definition", params)
def references(
self, params: lsp_types.ReferenceParams
) -> Union[List["lsp_types.Location"], None]:
def references(self, params: lsp_types.ReferenceParams) -> list["lsp_types.Location"] | None:
"""A request to resolve project-wide references for the symbol denoted
by the given text document position. The request's parameter is of
type {@link ReferenceParams} the response is of type
{@link Location Location[]} or a Thenable that resolves to such."""
{@link Location Location[]} or a Thenable that resolves to such.
"""
return self.send_request("textDocument/references", params)
def document_highlight(
self, params: lsp_types.DocumentHighlightParams
) -> Union[List["lsp_types.DocumentHighlight"], None]:
def document_highlight(self, params: lsp_types.DocumentHighlightParams) -> list["lsp_types.DocumentHighlight"] | None:
"""Request to resolve a {@link DocumentHighlight} for a given
text document position. The request's parameter is of type [TextDocumentPosition]
(#TextDocumentPosition) the request response is of type [DocumentHighlight[]]
(#DocumentHighlight) or a Thenable that resolves to such."""
(#DocumentHighlight) or a Thenable that resolves to such.
"""
return self.send_request("textDocument/documentHighlight", params)
def document_symbol(
self, params: lsp_types.DocumentSymbolParams
) -> Union[
List["lsp_types.SymbolInformation"], List["lsp_types.DocumentSymbol"], None
]:
self, params: lsp_types.DocumentSymbolParams
) -> list["lsp_types.SymbolInformation"] | list["lsp_types.DocumentSymbol"] | None:
"""A request to list all symbols found in a given text document. The request's
parameter is of type {@link TextDocumentIdentifier} the
response is of type {@link SymbolInformation SymbolInformation[]} or a Thenable
that resolves to such."""
that resolves to such.
"""
return self.send_request("textDocument/documentSymbol", params)
def code_action(
self, params: lsp_types.CodeActionParams
) -> Union[List[Union["lsp_types.Command", "lsp_types.CodeAction"]], None]:
def code_action(self, params: lsp_types.CodeActionParams) -> list[Union["lsp_types.Command", "lsp_types.CodeAction"]] | None:
"""A request to provide commands for the given text document and range."""
return self.send_request("textDocument/codeAction", params)
def resolve_code_action(
self, params: lsp_types.CodeAction
) -> "lsp_types.CodeAction":
def resolve_code_action(self, params: lsp_types.CodeAction) -> "lsp_types.CodeAction":
"""Request to resolve additional information for a given code action.The request's
parameter is of type {@link CodeAction} the response
is of type {@link CodeAction} or a Thenable that resolves to such."""
is of type {@link CodeAction} or a Thenable that resolves to such.
"""
return self.send_request("codeAction/resolve", params)
def workspace_symbol(
self, params: lsp_types.WorkspaceSymbolParams
) -> Union[
List["lsp_types.SymbolInformation"], List["lsp_types.WorkspaceSymbol"], None
]:
self, params: lsp_types.WorkspaceSymbolParams
) -> list["lsp_types.SymbolInformation"] | list["lsp_types.WorkspaceSymbol"] | None:
"""A request to list project-wide symbols matching the query string given
by the {@link WorkspaceSymbolParams}. The response is
of type {@link SymbolInformation SymbolInformation[]} or a Thenable that
@@ -360,76 +320,58 @@ class SolidLspRequest:
"""
return self.send_request("workspace/symbol", params)
def resolve_workspace_symbol(
self, params: lsp_types.WorkspaceSymbol
) -> "lsp_types.WorkspaceSymbol":
def resolve_workspace_symbol(self, params: lsp_types.WorkspaceSymbol) -> "lsp_types.WorkspaceSymbol":
"""A request to resolve the range inside the workspace
symbol's location.
@since 3.17.0"""
@since 3.17.0
"""
return self.send_request("workspaceSymbol/resolve", params)
def code_lens(
self, params: lsp_types.CodeLensParams
) -> Union[List["lsp_types.CodeLens"], None]:
def code_lens(self, params: lsp_types.CodeLensParams) -> list["lsp_types.CodeLens"] | None:
"""A request to provide code lens for the given text document."""
return self.send_request("textDocument/codeLens", params)
def resolve_code_lens(
self, params: lsp_types.CodeLens
) -> "lsp_types.CodeLens":
def resolve_code_lens(self, params: lsp_types.CodeLens) -> "lsp_types.CodeLens":
"""A request to resolve a command for a given code lens."""
return self.send_request("codeLens/resolve", params)
def document_link(
self, params: lsp_types.DocumentLinkParams
) -> Union[List["lsp_types.DocumentLink"], None]:
def document_link(self, params: lsp_types.DocumentLinkParams) -> list["lsp_types.DocumentLink"] | None:
"""A request to provide document links"""
return self.send_request("textDocument/documentLink", params)
def resolve_document_link(
self, params: lsp_types.DocumentLink
) -> "lsp_types.DocumentLink":
def resolve_document_link(self, params: lsp_types.DocumentLink) -> "lsp_types.DocumentLink":
"""Request to resolve additional information for a given document link. The request's
parameter is of type {@link DocumentLink} the response
is of type {@link DocumentLink} or a Thenable that resolves to such."""
is of type {@link DocumentLink} or a Thenable that resolves to such.
"""
return self.send_request("documentLink/resolve", params)
def formatting(
self, params: lsp_types.DocumentFormattingParams
) -> Union[List["lsp_types.TextEdit"], None]:
def formatting(self, params: lsp_types.DocumentFormattingParams) -> list["lsp_types.TextEdit"] | None:
"""A request to to format a whole document."""
return self.send_request("textDocument/formatting", params)
def range_formatting(
self, params: lsp_types.DocumentRangeFormattingParams
) -> Union[List["lsp_types.TextEdit"], None]:
def range_formatting(self, params: lsp_types.DocumentRangeFormattingParams) -> list["lsp_types.TextEdit"] | None:
"""A request to to format a range in a document."""
return self.send_request("textDocument/rangeFormatting", params)
def on_type_formatting(
self, params: lsp_types.DocumentOnTypeFormattingParams
) -> Union[List["lsp_types.TextEdit"], None]:
def on_type_formatting(self, params: lsp_types.DocumentOnTypeFormattingParams) -> list["lsp_types.TextEdit"] | None:
"""A request to format a document on type."""
return self.send_request("textDocument/onTypeFormatting", params)
def rename(
self, params: lsp_types.RenameParams
) -> Union["lsp_types.WorkspaceEdit", None]:
def rename(self, params: lsp_types.RenameParams) -> Union["lsp_types.WorkspaceEdit", None]:
"""A request to rename a symbol."""
return self.send_request("textDocument/rename", params)
def prepare_rename(
self, params: lsp_types.PrepareRenameParams
) -> Union["lsp_types.PrepareRenameResult", None]:
def prepare_rename(self, params: lsp_types.PrepareRenameParams) -> Union["lsp_types.PrepareRenameResult", None]:
"""A request to test and perform the setup necessary for a rename.
@since 3.16 - support for default behavior"""
@since 3.16 - support for default behavior
"""
return self.send_request("textDocument/prepareRename", params)
def execute_command(
self, params: lsp_types.ExecuteCommandParams
) -> Union["lsp_types.LSPAny", None]:
def execute_command(self, params: lsp_types.ExecuteCommandParams) -> Union["lsp_types.LSPAny", None]:
"""A request send from the client to the server to execute a command. The request might return
a workspace edit which the client will apply to the workspace."""
a workspace edit which the client will apply to the workspace.
"""
return self.send_request("workspace/executeCommand", params)