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
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
|
############################################################################
# Copyright(c) Open Law Library. All rights reserved. #
# See ThirdPartyNotices.txt in the project root for additional notices. #
# #
# Licensed under the Apache License, Version 2.0 (the "License") #
# you may not use this file except in compliance with the License. #
# You may obtain a copy of the License at #
# #
# http: // www.apache.org/licenses/LICENSE-2.0 #
# #
# Unless required by applicable law or agreed to in writing, software #
# distributed under the License is distributed on an "AS IS" BASIS, #
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. #
# See the License for the specific language governing permissions and #
# limitations under the License. #
############################################################################
import asyncio
import functools
import inspect
import itertools
import logging
from typing import Any, Callable, Dict, Optional, get_type_hints
from pygls.constants import (
ATTR_COMMAND_TYPE,
ATTR_EXECUTE_IN_THREAD,
ATTR_FEATURE_TYPE,
ATTR_REGISTERED_NAME,
ATTR_REGISTERED_TYPE,
PARAM_LS,
)
from pygls.exceptions import (
CommandAlreadyRegisteredError,
FeatureAlreadyRegisteredError,
ThreadDecoratorError,
ValidationError,
)
from pygls.lsp import get_method_options_type, is_instance
logger = logging.getLogger(__name__)
def assign_help_attrs(f, reg_name, reg_type):
setattr(f, ATTR_REGISTERED_NAME, reg_name)
setattr(f, ATTR_REGISTERED_TYPE, reg_type)
def assign_thread_attr(f):
setattr(f, ATTR_EXECUTE_IN_THREAD, True)
def get_help_attrs(f):
return getattr(f, ATTR_REGISTERED_NAME, None), getattr(
f, ATTR_REGISTERED_TYPE, None
)
def has_ls_param_or_annotation(f, annotation):
"""Returns true if callable has first parameter named `ls` or type of
annotation"""
try:
sig = inspect.signature(f)
first_p = next(itertools.islice(sig.parameters.values(), 0, 1))
return first_p.name == PARAM_LS or get_type_hints(f)[first_p.name] == annotation
except Exception:
return False
def is_thread_function(f):
return getattr(f, ATTR_EXECUTE_IN_THREAD, False)
def wrap_with_server(f, server):
"""Returns a new callable/coroutine with server as first argument."""
if not has_ls_param_or_annotation(f, type(server)):
return f
if asyncio.iscoroutinefunction(f):
async def wrapped(*args, **kwargs):
return await f(server, *args, **kwargs)
else:
wrapped = functools.partial(f, server)
if is_thread_function(f):
assign_thread_attr(wrapped)
return wrapped
class FeatureManager:
"""A class for managing server features.
Attributes:
_builtin_features(dict): Predefined set of lsp methods
_feature_options(dict): Registered feature's options
_features(dict): Registered features
_commands(dict): Registered commands
server(LanguageServer): Reference to the language server
If passed, server will be passed to registered
features/commands with first parameter:
1. ls - parameter naming convention
2. name: LanguageServer - add typings
"""
def __init__(self, server=None, converter=None):
self._builtin_features = {}
self._feature_options = {}
self._features = {}
self._commands = {}
self.server = server
self.converter = converter
def add_builtin_feature(self, feature_name: str, func: Callable) -> None:
"""Registers builtin (predefined) feature."""
self._builtin_features[feature_name] = func
logger.info("Registered builtin feature %s", feature_name)
@property
def builtin_features(self) -> Dict:
"""Returns server builtin features."""
return self._builtin_features
def command(self, command_name: str) -> Callable:
"""Decorator used to register custom commands.
Example:
@ls.command('myCustomCommand')
"""
def decorator(f):
# Validate
if command_name is None or command_name.strip() == "":
logger.error("Missing command name.")
raise ValidationError("Command name is required.")
# Check if not already registered
if command_name in self._commands:
logger.error('Command "%s" is already registered.', command_name)
raise CommandAlreadyRegisteredError(command_name)
assign_help_attrs(f, command_name, ATTR_COMMAND_TYPE)
wrapped = wrap_with_server(f, self.server)
# Assign help attributes for thread decorator
assign_help_attrs(wrapped, command_name, ATTR_COMMAND_TYPE)
self._commands[command_name] = wrapped
logger.info('Command "%s" is successfully registered.', command_name)
return f
return decorator
@property
def commands(self) -> Dict:
"""Returns registered custom commands."""
return self._commands
def feature(
self,
feature_name: str,
options: Optional[Any] = None,
) -> Callable:
"""Decorator used to register LSP features.
Example:
@ls.feature('textDocument/completion', CompletionItems(trigger_characters=['.']))
"""
def decorator(f):
# Validate
if feature_name is None or feature_name.strip() == "":
logger.error("Missing feature name.")
raise ValidationError("Feature name is required.")
# Add feature if not exists
if feature_name in self._features:
logger.error('Feature "%s" is already registered.', feature_name)
raise FeatureAlreadyRegisteredError(feature_name)
assign_help_attrs(f, feature_name, ATTR_FEATURE_TYPE)
wrapped = wrap_with_server(f, self.server)
# Assign help attributes for thread decorator
assign_help_attrs(wrapped, feature_name, ATTR_FEATURE_TYPE)
self._features[feature_name] = wrapped
if options:
options_type = get_method_options_type(feature_name)
if options_type and not is_instance(
self.converter, options, options_type
):
raise TypeError(
(
f'Options of method "{feature_name}"'
f" should be instance of type {options_type}"
)
)
self._feature_options[feature_name] = options
logger.info('Registered "%s" with options "%s"', feature_name, options)
return f
return decorator
@property
def feature_options(self) -> Dict:
"""Returns feature options for registered features."""
return self._feature_options
@property
def features(self) -> Dict:
"""Returns registered features"""
return self._features
def thread(self) -> Callable:
"""Decorator that mark function to execute it in a thread."""
def decorator(f):
if asyncio.iscoroutinefunction(f):
raise ThreadDecoratorError(
f'Thread decorator cannot be used with async functions "{f.__name__}"'
)
# Allow any decorator order
try:
reg_name = getattr(f, ATTR_REGISTERED_NAME)
reg_type = getattr(f, ATTR_REGISTERED_TYPE)
if reg_type is ATTR_FEATURE_TYPE:
assign_thread_attr(self.features[reg_name])
elif reg_type is ATTR_COMMAND_TYPE:
assign_thread_attr(self.commands[reg_name])
except AttributeError:
assign_thread_attr(f)
return f
return decorator
|
