1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
|
# Copyright 2017-2020 Palantir Technologies, Inc.
# Copyright 2021- Python Language Server Contributors.
import logging
import sys
from functools import lru_cache
from typing import List, Mapping, Sequence, Union
import pluggy
from pluggy._hooks import HookImpl
from pylsp import PYLSP, _utils, hookspecs, uris
# See compatibility note on `group` keyword:
# https://docs.python.org/3/library/importlib.metadata.html#entry-points
if sys.version_info < (3, 10): # pragma: no cover
from importlib_metadata import entry_points
else: # pragma: no cover
from importlib.metadata import entry_points
log = logging.getLogger(__name__)
# Sources of config, first source overrides next source
DEFAULT_CONFIG_SOURCES = ["pycodestyle"]
class PluginManager(pluggy.PluginManager):
def _hookexec(
self,
hook_name: str,
methods: Sequence[HookImpl],
kwargs: Mapping[str, object],
firstresult: bool,
) -> Union[object, List[object]]:
# called from all hookcaller instances.
# enable_tracing will set its own wrapping function at self._inner_hookexec
try:
return self._inner_hookexec(hook_name, methods, kwargs, firstresult)
except Exception as e:
log.warning(f"Failed to load hook {hook_name}: {e}", exc_info=True)
return []
class Config:
def __init__(self, root_uri, init_opts, process_id, capabilities) -> None:
self._root_path = uris.to_fs_path(root_uri)
self._root_uri = root_uri
self._init_opts = init_opts
self._process_id = process_id
self._capabilities = capabilities
self._settings = {}
self._plugin_settings = {}
self._config_sources = {}
try:
from .flake8_conf import Flake8Config
self._config_sources["flake8"] = Flake8Config(self._root_path)
except ImportError:
pass
try:
from .pycodestyle_conf import PyCodeStyleConfig
self._config_sources["pycodestyle"] = PyCodeStyleConfig(self._root_path)
except ImportError:
pass
self._pm = PluginManager(PYLSP)
self._pm.trace.root.setwriter(log.debug)
self._pm.enable_tracing()
self._pm.add_hookspecs(hookspecs)
# Pluggy will skip loading a plugin if it throws a DistributionNotFound exception.
# However I don't want all plugins to have to catch ImportError and re-throw. So here we'll filter
# out any entry points that throw ImportError assuming one or more of their dependencies isn't present.
for entry_point in entry_points(group=PYLSP):
try:
entry_point.load()
except Exception as e:
log.info(
"Failed to load %s entry point '%s': %s", PYLSP, entry_point.name, e
)
self._pm.set_blocked(entry_point.name)
# Load the entry points into pluggy, having blocked any failing ones.
# Despite the API name, recent Pluggy versions will use ``importlib_metadata``.
self._pm.load_setuptools_entrypoints(PYLSP)
for name, plugin in self._pm.list_name_plugin():
if plugin is not None:
log.info("Loaded pylsp plugin %s from %s", name, plugin)
for plugin_conf in self._pm.hook.pylsp_settings(config=self):
self._plugin_settings = _utils.merge_dicts(
self._plugin_settings, plugin_conf
)
self._plugin_settings = _utils.merge_dicts(
self._plugin_settings, self._init_opts.get("pylsp", {})
)
self._update_disabled_plugins()
@property
def disabled_plugins(self):
return self._disabled_plugins
@property
def plugin_manager(self):
return self._pm
@property
def init_opts(self):
return self._init_opts
@property
def root_uri(self):
return self._root_uri
@property
def process_id(self):
return self._process_id
@property
def capabilities(self):
return self._capabilities
@lru_cache(maxsize=32)
def settings(self, document_path=None):
"""Settings are constructed from a few sources:
1. User settings, found in user's home directory
2. Plugin settings, reported by PyLS plugins
3. LSP settings, given to us from didChangeConfiguration
4. Project settings, found in config files in the current project.
Since this function is nondeterministic, it is important to call
settings.cache_clear() when the config is updated
"""
settings = {}
sources = self._settings.get("configurationSources", DEFAULT_CONFIG_SOURCES)
# Plugin configuration
settings = _utils.merge_dicts(settings, self._plugin_settings)
# LSP configuration
settings = _utils.merge_dicts(settings, self._settings)
# User configuration
for source_name in reversed(sources):
source = self._config_sources.get(source_name)
if not source:
continue
source_conf = source.user_config()
log.debug(
"Got user config from %s: %s", source.__class__.__name__, source_conf
)
settings = _utils.merge_dicts(settings, source_conf)
# Project configuration
for source_name in reversed(sources):
source = self._config_sources.get(source_name)
if not source:
continue
source_conf = source.project_config(document_path or self._root_path)
log.debug(
"Got project config from %s: %s", source.__class__.__name__, source_conf
)
settings = _utils.merge_dicts(settings, source_conf)
log.debug("With configuration: %s", settings)
return settings
def find_parents(self, path, names):
root_path = uris.to_fs_path(self._root_uri)
return _utils.find_parents(root_path, path, names)
def plugin_settings(self, plugin, document_path=None):
return (
self.settings(document_path=document_path)
.get("plugins", {})
.get(plugin, {})
)
def update(self, settings) -> None:
"""Recursively merge the given settings into the current settings."""
self.settings.cache_clear()
self._settings = settings
log.info("Updated settings to %s", self._settings)
self._update_disabled_plugins()
def _update_disabled_plugins(self) -> None:
# All plugins default to enabled
self._disabled_plugins = [
plugin
for name, plugin in self.plugin_manager.list_name_plugin()
if not self.settings().get("plugins", {}).get(name, {}).get("enabled", True)
]
log.info("Disabled plugins: %s", self._disabled_plugins)
|
