diff --git a/pygls/lsp/_base_client.py b/pygls/lsp/_base_client.py new file mode 100644 index 00000000..eb2f58c8 --- /dev/null +++ b/pygls/lsp/_base_client.py @@ -0,0 +1,989 @@ +# GENERATED FROM scripts/genenerate_base_lsp_classes.py -- DO NOT EDIT +# flake8: noqa +import asyncio +from lsprotocol import types +from pygls.client._native import JsonRPCClient +from pygls.protocol import next as json_rpc +from typing import Any +from typing import Callable +from typing import List +from typing import Optional +from typing import Union + + +class BaseLanguageClient(JsonRPCClient): + + def __init__( + **kwargs, + ): + super().__init__(**kwargs) + + def call_hierarchy_incoming_calls( + self, + params: types.CallHierarchyIncomingCallsParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[List[types.CallHierarchyIncomingCall]]], None]] = None, + ) -> asyncio.Task[Optional[List[types.CallHierarchyIncomingCall]]]: + """Make a :lsp:`callHierarchy/incomingCalls` request. + + A request to resolve the incoming calls for a given `CallHierarchyItem`. + + @since 3.16.0 + """ + return self.send_request("callHierarchy/incomingCalls", params, msg_id=msg_id, callback=callback) + + def call_hierarchy_outgoing_calls( + self, + params: types.CallHierarchyOutgoingCallsParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[List[types.CallHierarchyOutgoingCall]]], None]] = None, + ) -> asyncio.Task[Optional[List[types.CallHierarchyOutgoingCall]]]: + """Make a :lsp:`callHierarchy/outgoingCalls` request. + + A request to resolve the outgoing calls for a given `CallHierarchyItem`. + + @since 3.16.0 + """ + return self.send_request("callHierarchy/outgoingCalls", params, msg_id=msg_id, callback=callback) + + def code_action_resolve( + self, + params: types.CodeAction, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[types.CodeAction], None]] = None, + ) -> asyncio.Task[types.CodeAction]: + """Make a :lsp:`codeAction/resolve` request. + + 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. + """ + return self.send_request("codeAction/resolve", params, msg_id=msg_id, callback=callback) + + def code_lens_resolve( + self, + params: types.CodeLens, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[types.CodeLens], None]] = None, + ) -> asyncio.Task[types.CodeLens]: + """Make a :lsp:`codeLens/resolve` request. + + A request to resolve a command for a given code lens. + """ + return self.send_request("codeLens/resolve", params, msg_id=msg_id, callback=callback) + + def completion_item_resolve( + self, + params: types.CompletionItem, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[types.CompletionItem], None]] = None, + ) -> asyncio.Task[types.CompletionItem]: + """Make a :lsp:`completionItem/resolve` request. + + 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. + """ + return self.send_request("completionItem/resolve", params, msg_id=msg_id, callback=callback) + + def document_link_resolve( + self, + params: types.DocumentLink, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[types.DocumentLink], None]] = None, + ) -> asyncio.Task[types.DocumentLink]: + """Make a :lsp:`documentLink/resolve` request. + + 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. + """ + return self.send_request("documentLink/resolve", params, msg_id=msg_id, callback=callback) + + def initialize( + self, + params: types.InitializeParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[types.InitializeResult], None]] = None, + ) -> asyncio.Task[types.InitializeResult]: + """Make a :lsp:`initialize` request. + + 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. + """ + return self.send_request("initialize", params, msg_id=msg_id, callback=callback) + + def inlay_hint_resolve( + self, + params: types.InlayHint, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[types.InlayHint], None]] = None, + ) -> asyncio.Task[types.InlayHint]: + """Make a :lsp:`inlayHint/resolve` request. + + 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 + """ + return self.send_request("inlayHint/resolve", params, msg_id=msg_id, callback=callback) + + def shutdown( + self, + params: None, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[None], None]] = None, + ) -> asyncio.Task[None]: + """Make a :lsp:`shutdown` request. + + 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. + """ + return self.send_request("shutdown", params, msg_id=msg_id, callback=callback) + + def text_document_code_action( + self, + params: types.CodeActionParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[List[Union[types.Command, types.CodeAction]]]], None]] = None, + ) -> asyncio.Task[Optional[List[Union[types.Command, types.CodeAction]]]]: + """Make a :lsp:`textDocument/codeAction` request. + + A request to provide commands for the given text document and range. + """ + return self.send_request("textDocument/codeAction", params, msg_id=msg_id, callback=callback) + + def text_document_code_lens( + self, + params: types.CodeLensParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[List[types.CodeLens]]], None]] = None, + ) -> asyncio.Task[Optional[List[types.CodeLens]]]: + """Make a :lsp:`textDocument/codeLens` request. + + A request to provide code lens for the given text document. + """ + return self.send_request("textDocument/codeLens", params, msg_id=msg_id, callback=callback) + + def text_document_color_presentation( + self, + params: types.ColorPresentationParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[List[types.ColorPresentation]], None]] = None, + ) -> asyncio.Task[List[types.ColorPresentation]]: + """Make a :lsp:`textDocument/colorPresentation` request. + + 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. + """ + return self.send_request("textDocument/colorPresentation", params, msg_id=msg_id, callback=callback) + + def text_document_completion( + self, + params: types.CompletionParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Union[List[types.CompletionItem], types.CompletionList, None]], None]] = None, + ) -> asyncio.Task[Union[List[types.CompletionItem], types.CompletionList, None]]: + """Make a :lsp:`textDocument/completion` request. + + 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} + or a Thenable that resolves to such. + + The request can delay the computation of the {@link CompletionItem.detail `detail`} + and {@link CompletionItem.documentation `documentation`} properties to the `completionItem/resolve` + request. However, properties that are needed for the initial sorting and filtering, like `sortText`, + `filterText`, `insertText`, and `textEdit`, must not be changed during resolve. + """ + return self.send_request("textDocument/completion", params, msg_id=msg_id, callback=callback) + + def text_document_declaration( + self, + params: types.DeclarationParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Union[types.Location, List[types.Location], List[types.LocationLink], None]], None]] = None, + ) -> asyncio.Task[Union[types.Location, List[types.Location], List[types.LocationLink], None]]: + """Make a :lsp:`textDocument/declaration` request. + + A request to resolve the type definition locations of a symbol at a given text + document position. The request's parameter is of type {@link TextDocumentPositionParams} + the response is of type {@link Declaration} or a typed array of {@link DeclarationLink} + or a Thenable that resolves to such. + """ + return self.send_request("textDocument/declaration", params, msg_id=msg_id, callback=callback) + + def text_document_definition( + self, + params: types.DefinitionParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Union[types.Location, List[types.Location], List[types.LocationLink], None]], None]] = None, + ) -> asyncio.Task[Union[types.Location, List[types.Location], List[types.LocationLink], None]]: + """Make a :lsp:`textDocument/definition` request. + + A request to resolve the definition location of a symbol at a given text + document position. The request's parameter is of type {@link TextDocumentPosition} + the response is of either type {@link Definition} or a typed array of + {@link DefinitionLink} or a Thenable that resolves to such. + """ + return self.send_request("textDocument/definition", params, msg_id=msg_id, callback=callback) + + def text_document_diagnostic( + self, + params: types.DocumentDiagnosticParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Union[types.RelatedFullDocumentDiagnosticReport, types.RelatedUnchangedDocumentDiagnosticReport]], None]] = None, + ) -> asyncio.Task[Union[types.RelatedFullDocumentDiagnosticReport, types.RelatedUnchangedDocumentDiagnosticReport]]: + """Make a :lsp:`textDocument/diagnostic` request. + + The document diagnostic request definition. + + @since 3.17.0 + """ + return self.send_request("textDocument/diagnostic", params, msg_id=msg_id, callback=callback) + + def text_document_document_color( + self, + params: types.DocumentColorParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[List[types.ColorInformation]], None]] = None, + ) -> asyncio.Task[List[types.ColorInformation]]: + """Make a :lsp:`textDocument/documentColor` request. + + 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. + """ + return self.send_request("textDocument/documentColor", params, msg_id=msg_id, callback=callback) + + def text_document_document_highlight( + self, + params: types.DocumentHighlightParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[List[types.DocumentHighlight]]], None]] = None, + ) -> asyncio.Task[Optional[List[types.DocumentHighlight]]]: + """Make a :lsp:`textDocument/documentHighlight` request. + + Request to resolve a {@link DocumentHighlight} for a given + text document position. The request's parameter is of type {@link TextDocumentPosition} + the request response is an array of type {@link DocumentHighlight} + or a Thenable that resolves to such. + """ + return self.send_request("textDocument/documentHighlight", params, msg_id=msg_id, callback=callback) + + def text_document_document_link( + self, + params: types.DocumentLinkParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[List[types.DocumentLink]]], None]] = None, + ) -> asyncio.Task[Optional[List[types.DocumentLink]]]: + """Make a :lsp:`textDocument/documentLink` request. + + A request to provide document links + """ + return self.send_request("textDocument/documentLink", params, msg_id=msg_id, callback=callback) + + def text_document_document_symbol( + self, + params: types.DocumentSymbolParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Union[List[types.SymbolInformation], List[types.DocumentSymbol], None]], None]] = None, + ) -> asyncio.Task[Union[List[types.SymbolInformation], List[types.DocumentSymbol], None]]: + """Make a :lsp:`textDocument/documentSymbol` request. + + 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. + """ + return self.send_request("textDocument/documentSymbol", params, msg_id=msg_id, callback=callback) + + def text_document_folding_range( + self, + params: types.FoldingRangeParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[List[types.FoldingRange]]], None]] = None, + ) -> asyncio.Task[Optional[List[types.FoldingRange]]]: + """Make a :lsp:`textDocument/foldingRange` request. + + 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. + """ + return self.send_request("textDocument/foldingRange", params, msg_id=msg_id, callback=callback) + + def text_document_formatting( + self, + params: types.DocumentFormattingParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[List[types.TextEdit]]], None]] = None, + ) -> asyncio.Task[Optional[List[types.TextEdit]]]: + """Make a :lsp:`textDocument/formatting` request. + + A request to format a whole document. + """ + return self.send_request("textDocument/formatting", params, msg_id=msg_id, callback=callback) + + def text_document_hover( + self, + params: types.HoverParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[types.Hover]], None]] = None, + ) -> asyncio.Task[Optional[types.Hover]]: + """Make a :lsp:`textDocument/hover` request. + + 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. + """ + return self.send_request("textDocument/hover", params, msg_id=msg_id, callback=callback) + + def text_document_implementation( + self, + params: types.ImplementationParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Union[types.Location, List[types.Location], List[types.LocationLink], None]], None]] = None, + ) -> asyncio.Task[Union[types.Location, List[types.Location], List[types.LocationLink], None]]: + """Make a :lsp:`textDocument/implementation` request. + + A request to resolve the implementation locations of a symbol at a given text + document position. The request's parameter is of type {@link TextDocumentPositionParams} + the response is of type {@link Definition} or a Thenable that resolves to such. + """ + return self.send_request("textDocument/implementation", params, msg_id=msg_id, callback=callback) + + def text_document_inlay_hint( + self, + params: types.InlayHintParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[List[types.InlayHint]]], None]] = None, + ) -> asyncio.Task[Optional[List[types.InlayHint]]]: + """Make a :lsp:`textDocument/inlayHint` request. + + 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 + """ + return self.send_request("textDocument/inlayHint", params, msg_id=msg_id, callback=callback) + + def text_document_inline_completion( + self, + params: types.InlineCompletionParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Union[types.InlineCompletionList, List[types.InlineCompletionItem], None]], None]] = None, + ) -> asyncio.Task[Union[types.InlineCompletionList, List[types.InlineCompletionItem], None]]: + """Make a :lsp:`textDocument/inlineCompletion` request. + + A request to provide inline completions in a document. The request's parameter is of + type {@link InlineCompletionParams}, the response is of type + {@link InlineCompletion InlineCompletion[]} or a Thenable that resolves to such. + + @since 3.18.0 + @proposed + """ + return self.send_request("textDocument/inlineCompletion", params, msg_id=msg_id, callback=callback) + + def text_document_inline_value( + self, + params: types.InlineValueParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[List[Union[types.InlineValueText, types.InlineValueVariableLookup, types.InlineValueEvaluatableExpression]]]], None]] = None, + ) -> asyncio.Task[Optional[List[Union[types.InlineValueText, types.InlineValueVariableLookup, types.InlineValueEvaluatableExpression]]]]: + """Make a :lsp:`textDocument/inlineValue` request. + + 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 + """ + return self.send_request("textDocument/inlineValue", params, msg_id=msg_id, callback=callback) + + def text_document_linked_editing_range( + self, + params: types.LinkedEditingRangeParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[types.LinkedEditingRanges]], None]] = None, + ) -> asyncio.Task[Optional[types.LinkedEditingRanges]]: + """Make a :lsp:`textDocument/linkedEditingRange` request. + + A request to provide ranges that can be edited together. + + @since 3.16.0 + """ + return self.send_request("textDocument/linkedEditingRange", params, msg_id=msg_id, callback=callback) + + def text_document_moniker( + self, + params: types.MonikerParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[List[types.Moniker]]], None]] = None, + ) -> asyncio.Task[Optional[List[types.Moniker]]]: + """Make a :lsp:`textDocument/moniker` request. + + 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`. + """ + return self.send_request("textDocument/moniker", params, msg_id=msg_id, callback=callback) + + def text_document_on_type_formatting( + self, + params: types.DocumentOnTypeFormattingParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[List[types.TextEdit]]], None]] = None, + ) -> asyncio.Task[Optional[List[types.TextEdit]]]: + """Make a :lsp:`textDocument/onTypeFormatting` request. + + A request to format a document on type. + """ + return self.send_request("textDocument/onTypeFormatting", params, msg_id=msg_id, callback=callback) + + def text_document_prepare_call_hierarchy( + self, + params: types.CallHierarchyPrepareParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[List[types.CallHierarchyItem]]], None]] = None, + ) -> asyncio.Task[Optional[List[types.CallHierarchyItem]]]: + """Make a :lsp:`textDocument/prepareCallHierarchy` request. + + 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 + """ + return self.send_request("textDocument/prepareCallHierarchy", params, msg_id=msg_id, callback=callback) + + def text_document_prepare_rename( + self, + params: types.PrepareRenameParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Union[types.Range, types.PrepareRenameResult_Type1, types.PrepareRenameResult_Type2, None]], None]] = None, + ) -> asyncio.Task[Union[types.Range, types.PrepareRenameResult_Type1, types.PrepareRenameResult_Type2, None]]: + """Make a :lsp:`textDocument/prepareRename` request. + + A request to test and perform the setup necessary for a rename. + + @since 3.16 - support for default behavior + """ + return self.send_request("textDocument/prepareRename", params, msg_id=msg_id, callback=callback) + + def text_document_prepare_type_hierarchy( + self, + params: types.TypeHierarchyPrepareParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[List[types.TypeHierarchyItem]]], None]] = None, + ) -> asyncio.Task[Optional[List[types.TypeHierarchyItem]]]: + """Make a :lsp:`textDocument/prepareTypeHierarchy` request. + + 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 + """ + return self.send_request("textDocument/prepareTypeHierarchy", params, msg_id=msg_id, callback=callback) + + def text_document_ranges_formatting( + self, + params: types.DocumentRangesFormattingParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[List[types.TextEdit]]], None]] = None, + ) -> asyncio.Task[Optional[List[types.TextEdit]]]: + """Make a :lsp:`textDocument/rangesFormatting` request. + + A request to format ranges in a document. + + @since 3.18.0 + @proposed + """ + return self.send_request("textDocument/rangesFormatting", params, msg_id=msg_id, callback=callback) + + def text_document_range_formatting( + self, + params: types.DocumentRangeFormattingParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[List[types.TextEdit]]], None]] = None, + ) -> asyncio.Task[Optional[List[types.TextEdit]]]: + """Make a :lsp:`textDocument/rangeFormatting` request. + + A request to format a range in a document. + """ + return self.send_request("textDocument/rangeFormatting", params, msg_id=msg_id, callback=callback) + + def text_document_references( + self, + params: types.ReferenceParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[List[types.Location]]], None]] = None, + ) -> asyncio.Task[Optional[List[types.Location]]]: + """Make a :lsp:`textDocument/references` request. + + 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. + """ + return self.send_request("textDocument/references", params, msg_id=msg_id, callback=callback) + + def text_document_rename( + self, + params: types.RenameParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[types.WorkspaceEdit]], None]] = None, + ) -> asyncio.Task[Optional[types.WorkspaceEdit]]: + """Make a :lsp:`textDocument/rename` request. + + A request to rename a symbol. + """ + return self.send_request("textDocument/rename", params, msg_id=msg_id, callback=callback) + + def text_document_selection_range( + self, + params: types.SelectionRangeParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[List[types.SelectionRange]]], None]] = None, + ) -> asyncio.Task[Optional[List[types.SelectionRange]]]: + """Make a :lsp:`textDocument/selectionRange` request. + + 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. + """ + return self.send_request("textDocument/selectionRange", params, msg_id=msg_id, callback=callback) + + def text_document_semantic_tokens_full( + self, + params: types.SemanticTokensParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[types.SemanticTokens]], None]] = None, + ) -> asyncio.Task[Optional[types.SemanticTokens]]: + """Make a :lsp:`textDocument/semanticTokens/full` request. + + @since 3.16.0 + """ + return self.send_request("textDocument/semanticTokens/full", params, msg_id=msg_id, callback=callback) + + def text_document_semantic_tokens_full_delta( + self, + params: types.SemanticTokensDeltaParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Union[types.SemanticTokens, types.SemanticTokensDelta, None]], None]] = None, + ) -> asyncio.Task[Union[types.SemanticTokens, types.SemanticTokensDelta, None]]: + """Make a :lsp:`textDocument/semanticTokens/full/delta` request. + + @since 3.16.0 + """ + return self.send_request("textDocument/semanticTokens/full/delta", params, msg_id=msg_id, callback=callback) + + def text_document_semantic_tokens_range( + self, + params: types.SemanticTokensRangeParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[types.SemanticTokens]], None]] = None, + ) -> asyncio.Task[Optional[types.SemanticTokens]]: + """Make a :lsp:`textDocument/semanticTokens/range` request. + + @since 3.16.0 + """ + return self.send_request("textDocument/semanticTokens/range", params, msg_id=msg_id, callback=callback) + + def text_document_signature_help( + self, + params: types.SignatureHelpParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[types.SignatureHelp]], None]] = None, + ) -> asyncio.Task[Optional[types.SignatureHelp]]: + """Make a :lsp:`textDocument/signatureHelp` request. + + + """ + return self.send_request("textDocument/signatureHelp", params, msg_id=msg_id, callback=callback) + + def text_document_type_definition( + self, + params: types.TypeDefinitionParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Union[types.Location, List[types.Location], List[types.LocationLink], None]], None]] = None, + ) -> asyncio.Task[Union[types.Location, List[types.Location], List[types.LocationLink], None]]: + """Make a :lsp:`textDocument/typeDefinition` request. + + A request to resolve the type definition locations of a symbol at a given text + document position. The request's parameter is of type {@link TextDocumentPositionParams} + the response is of type {@link Definition} or a Thenable that resolves to such. + """ + return self.send_request("textDocument/typeDefinition", params, msg_id=msg_id, callback=callback) + + def text_document_will_save_wait_until( + self, + params: types.WillSaveTextDocumentParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[List[types.TextEdit]]], None]] = None, + ) -> asyncio.Task[Optional[List[types.TextEdit]]]: + """Make a :lsp:`textDocument/willSaveWaitUntil` request. + + 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. + """ + return self.send_request("textDocument/willSaveWaitUntil", params, msg_id=msg_id, callback=callback) + + def type_hierarchy_subtypes( + self, + params: types.TypeHierarchySubtypesParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[List[types.TypeHierarchyItem]]], None]] = None, + ) -> asyncio.Task[Optional[List[types.TypeHierarchyItem]]]: + """Make a :lsp:`typeHierarchy/subtypes` request. + + A request to resolve the subtypes for a given `TypeHierarchyItem`. + + @since 3.17.0 + """ + return self.send_request("typeHierarchy/subtypes", params, msg_id=msg_id, callback=callback) + + def type_hierarchy_supertypes( + self, + params: types.TypeHierarchySupertypesParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[List[types.TypeHierarchyItem]]], None]] = None, + ) -> asyncio.Task[Optional[List[types.TypeHierarchyItem]]]: + """Make a :lsp:`typeHierarchy/supertypes` request. + + A request to resolve the supertypes for a given `TypeHierarchyItem`. + + @since 3.17.0 + """ + return self.send_request("typeHierarchy/supertypes", params, msg_id=msg_id, callback=callback) + + def workspace_diagnostic( + self, + params: types.WorkspaceDiagnosticParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[types.WorkspaceDiagnosticReport], None]] = None, + ) -> asyncio.Task[types.WorkspaceDiagnosticReport]: + """Make a :lsp:`workspace/diagnostic` request. + + The workspace diagnostic request definition. + + @since 3.17.0 + """ + return self.send_request("workspace/diagnostic", params, msg_id=msg_id, callback=callback) + + def workspace_execute_command( + self, + params: types.ExecuteCommandParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[Any]], None]] = None, + ) -> asyncio.Task[Optional[Any]]: + """Make a :lsp:`workspace/executeCommand` request. + + 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. + """ + return self.send_request("workspace/executeCommand", params, msg_id=msg_id, callback=callback) + + def workspace_symbol( + self, + params: types.WorkspaceSymbolParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Union[List[types.SymbolInformation], List[types.WorkspaceSymbol], None]], None]] = None, + ) -> asyncio.Task[Union[List[types.SymbolInformation], List[types.WorkspaceSymbol], None]]: + """Make a :lsp:`workspace/symbol` request. + + 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 + resolves to such. + + @since 3.17.0 - support for WorkspaceSymbol in the returned data. Clients + need to advertise support for WorkspaceSymbols via the client capability + `workspace.symbol.resolveSupport`. + """ + return self.send_request("workspace/symbol", params, msg_id=msg_id, callback=callback) + + def workspace_symbol_resolve( + self, + params: types.WorkspaceSymbol, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[types.WorkspaceSymbol], None]] = None, + ) -> asyncio.Task[types.WorkspaceSymbol]: + """Make a :lsp:`workspaceSymbol/resolve` request. + + A request to resolve the range inside the workspace + symbol's location. + + @since 3.17.0 + """ + return self.send_request("workspaceSymbol/resolve", params, msg_id=msg_id, callback=callback) + + def workspace_will_create_files( + self, + params: types.CreateFilesParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[types.WorkspaceEdit]], None]] = None, + ) -> asyncio.Task[Optional[types.WorkspaceEdit]]: + """Make a :lsp:`workspace/willCreateFiles` request. + + 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. + + The request can return a `WorkspaceEdit` which will be applied to workspace before the + files are created. Hence the `WorkspaceEdit` can not manipulate the content of the file + to be created. + + @since 3.16.0 + """ + return self.send_request("workspace/willCreateFiles", params, msg_id=msg_id, callback=callback) + + def workspace_will_delete_files( + self, + params: types.DeleteFilesParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[types.WorkspaceEdit]], None]] = None, + ) -> asyncio.Task[Optional[types.WorkspaceEdit]]: + """Make a :lsp:`workspace/willDeleteFiles` request. + + 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 + """ + return self.send_request("workspace/willDeleteFiles", params, msg_id=msg_id, callback=callback) + + def workspace_will_rename_files( + self, + params: types.RenameFilesParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[types.WorkspaceEdit]], None]] = None, + ) -> asyncio.Task[Optional[types.WorkspaceEdit]]: + """Make a :lsp:`workspace/willRenameFiles` request. + + 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 + """ + return self.send_request("workspace/willRenameFiles", params, msg_id=msg_id, callback=callback) + + def cancel_request(self, params: types.CancelParams) -> None: + """Send a :lsp:`$/cancelRequest` notification. + + + """ + + self.send_notification("$/cancelRequest", params) + + def exit(self, params: None) -> None: + """Send a :lsp:`exit` notification. + + The exit event is sent from the client to the server to + ask the server to exit its process. + """ + + self.send_notification("exit", params) + + def initialized(self, params: types.InitializedParams) -> None: + """Send a :lsp:`initialized` notification. + + The initialized notification is sent from the client to the + server after the client is fully initialized and the server + is allowed to send requests from the server to the client. + """ + + self.send_notification("initialized", params) + + def notebook_document_did_change(self, params: types.DidChangeNotebookDocumentParams) -> None: + """Send a :lsp:`notebookDocument/didChange` notification. + + + """ + + self.send_notification("notebookDocument/didChange", params) + + def notebook_document_did_close(self, params: types.DidCloseNotebookDocumentParams) -> None: + """Send a :lsp:`notebookDocument/didClose` notification. + + A notification sent when a notebook closes. + + @since 3.17.0 + """ + + self.send_notification("notebookDocument/didClose", params) + + def notebook_document_did_open(self, params: types.DidOpenNotebookDocumentParams) -> None: + """Send a :lsp:`notebookDocument/didOpen` notification. + + A notification sent when a notebook opens. + + @since 3.17.0 + """ + + self.send_notification("notebookDocument/didOpen", params) + + def notebook_document_did_save(self, params: types.DidSaveNotebookDocumentParams) -> None: + """Send a :lsp:`notebookDocument/didSave` notification. + + A notification sent when a notebook document is saved. + + @since 3.17.0 + """ + + self.send_notification("notebookDocument/didSave", params) + + def progress(self, params: types.ProgressParams) -> None: + """Send a :lsp:`$/progress` notification. + + + """ + + self.send_notification("$/progress", params) + + def set_trace(self, params: types.SetTraceParams) -> None: + """Send a :lsp:`$/setTrace` notification. + + + """ + + self.send_notification("$/setTrace", params) + + def text_document_did_change(self, params: types.DidChangeTextDocumentParams) -> None: + """Send a :lsp:`textDocument/didChange` notification. + + The document change notification is sent from the client to the server to signal + changes to a text document. + """ + + self.send_notification("textDocument/didChange", params) + + def text_document_did_close(self, params: types.DidCloseTextDocumentParams) -> None: + """Send a :lsp:`textDocument/didClose` notification. + + The document close notification is sent from the client to the server when + the document got closed in the client. The document's truth now exists where + the document's uri points to (e.g. if the document's uri is a file uri the + truth now exists on disk). As with the open notification the close notification + is about managing the document's content. Receiving a close notification + doesn't mean that the document was open in an editor before. A close + notification requires a previous open notification to be sent. + """ + + self.send_notification("textDocument/didClose", params) + + def text_document_did_open(self, params: types.DidOpenTextDocumentParams) -> None: + """Send a :lsp:`textDocument/didOpen` notification. + + The document open notification is sent from the client to the server to signal + newly opened text documents. The document's truth is now managed by the client + and the server must not try to read the document's truth using the document's + uri. Open in this sense means it is managed by the client. It doesn't necessarily + mean that its content is presented in an editor. An open notification must not + be sent more than once without a corresponding close notification send before. + This means open and close notification must be balanced and the max open count + is one. + """ + + self.send_notification("textDocument/didOpen", params) + + def text_document_did_save(self, params: types.DidSaveTextDocumentParams) -> None: + """Send a :lsp:`textDocument/didSave` notification. + + The document save notification is sent from the client to the server when + the document got saved in the client. + """ + + self.send_notification("textDocument/didSave", params) + + def text_document_will_save(self, params: types.WillSaveTextDocumentParams) -> None: + """Send a :lsp:`textDocument/willSave` notification. + + A document will save notification is sent from the client to the server before + the document is actually saved. + """ + + self.send_notification("textDocument/willSave", params) + + def window_work_done_progress_cancel(self, params: types.WorkDoneProgressCancelParams) -> None: + """Send a :lsp:`window/workDoneProgress/cancel` notification. + + The `window/workDoneProgress/cancel` notification is sent from the client to the server to cancel a progress + initiated on the server side. + """ + + self.send_notification("window/workDoneProgress/cancel", params) + + def workspace_did_change_configuration(self, params: types.DidChangeConfigurationParams) -> None: + """Send a :lsp:`workspace/didChangeConfiguration` notification. + + The configuration change notification is sent from the client to the server + when the client's configuration has changed. The notification contains + the changed configuration as defined by the language client. + """ + + self.send_notification("workspace/didChangeConfiguration", params) + + def workspace_did_change_watched_files(self, params: types.DidChangeWatchedFilesParams) -> None: + """Send a :lsp:`workspace/didChangeWatchedFiles` notification. + + The watched files notification is sent from the client to the server when + the client detects changes to file watched by the language client. + """ + + self.send_notification("workspace/didChangeWatchedFiles", params) + + def workspace_did_change_workspace_folders(self, params: types.DidChangeWorkspaceFoldersParams) -> None: + """Send a :lsp:`workspace/didChangeWorkspaceFolders` notification. + + The `workspace/didChangeWorkspaceFolders` notification is sent from the client to the server when the workspace + folder configuration changes. + """ + + self.send_notification("workspace/didChangeWorkspaceFolders", params) + + def workspace_did_create_files(self, params: types.CreateFilesParams) -> None: + """Send a :lsp:`workspace/didCreateFiles` notification. + + The did create files notification is sent from the client to the server when + files were created from within the client. + + @since 3.16.0 + """ + + self.send_notification("workspace/didCreateFiles", params) + + def workspace_did_delete_files(self, params: types.DeleteFilesParams) -> None: + """Send a :lsp:`workspace/didDeleteFiles` notification. + + The will delete files request is sent from the client to the server before files are actually + deleted as long as the deletion is triggered from within the client. + + @since 3.16.0 + """ + + self.send_notification("workspace/didDeleteFiles", params) + + def workspace_did_rename_files(self, params: types.RenameFilesParams) -> None: + """Send a :lsp:`workspace/didRenameFiles` notification. + + The did rename files notification is sent from the client to the server when + files were renamed from within the client. + + @since 3.16.0 + """ + + self.send_notification("workspace/didRenameFiles", params) diff --git a/pygls/lsp/_base_server.py b/pygls/lsp/_base_server.py new file mode 100644 index 00000000..f42ea335 --- /dev/null +++ b/pygls/lsp/_base_server.py @@ -0,0 +1,253 @@ +# GENERATED FROM scripts/genenerate_base_lsp_classes.py -- DO NOT EDIT +# flake8: noqa +import asyncio +from lsprotocol import types +from pygls.protocol import next as json_rpc +from pygls.server._native import JsonRPCServer +from typing import Any +from typing import Callable +from typing import List +from typing import Optional + + +class BaseLanguageServer(JsonRPCServer): + + def __init__( + **kwargs, + ): + super().__init__(**kwargs) + + def client_register_capability( + self, + params: types.RegistrationParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[None], None]] = None, + ) -> asyncio.Task[None]: + """Make a :lsp:`client/registerCapability` request. + + The `client/registerCapability` request is sent from the server to the client to register a new capability + handler on the client side. + """ + return self.send_request("client/registerCapability", params, msg_id=msg_id, callback=callback) + + def client_unregister_capability( + self, + params: types.UnregistrationParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[None], None]] = None, + ) -> asyncio.Task[None]: + """Make a :lsp:`client/unregisterCapability` request. + + The `client/unregisterCapability` request is sent from the server to the client to unregister a previously registered capability + handler on the client side. + """ + return self.send_request("client/unregisterCapability", params, msg_id=msg_id, callback=callback) + + def window_show_document( + self, + params: types.ShowDocumentParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[types.ShowDocumentResult], None]] = None, + ) -> asyncio.Task[types.ShowDocumentResult]: + """Make a :lsp:`window/showDocument` request. + + A request to show a document. This request might open an + external program depending on the value of the URI to open. + For example a request to open `https://code.visualstudio.com/` + will very likely open the URI in a WEB browser. + + @since 3.16.0 + """ + return self.send_request("window/showDocument", params, msg_id=msg_id, callback=callback) + + def window_show_message_request( + self, + params: types.ShowMessageRequestParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[types.MessageActionItem]], None]] = None, + ) -> asyncio.Task[Optional[types.MessageActionItem]]: + """Make a :lsp:`window/showMessageRequest` request. + + The show message request is sent from the server to the client to show a message + and a set of options actions to the user. + """ + return self.send_request("window/showMessageRequest", params, msg_id=msg_id, callback=callback) + + def window_work_done_progress_create( + self, + params: types.WorkDoneProgressCreateParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[None], None]] = None, + ) -> asyncio.Task[None]: + """Make a :lsp:`window/workDoneProgress/create` request. + + The `window/workDoneProgress/create` request is sent from the server to the client to initiate progress + reporting from the server. + """ + return self.send_request("window/workDoneProgress/create", params, msg_id=msg_id, callback=callback) + + def workspace_apply_edit( + self, + params: types.ApplyWorkspaceEditParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[types.ApplyWorkspaceEditResult], None]] = None, + ) -> asyncio.Task[types.ApplyWorkspaceEditResult]: + """Make a :lsp:`workspace/applyEdit` request. + + A request sent from the server to the client to modified certain resources. + """ + return self.send_request("workspace/applyEdit", params, msg_id=msg_id, callback=callback) + + def workspace_code_lens_refresh( + self, + params: None, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[None], None]] = None, + ) -> asyncio.Task[None]: + """Make a :lsp:`workspace/codeLens/refresh` request. + + A request to refresh all code actions + + @since 3.16.0 + """ + return self.send_request("workspace/codeLens/refresh", params, msg_id=msg_id, callback=callback) + + def workspace_configuration( + self, + params: types.ConfigurationParams, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[List[Optional[Any]]], None]] = None, + ) -> asyncio.Task[List[Optional[Any]]]: + """Make a :lsp:`workspace/configuration` request. + + The 'workspace/configuration' request is sent from the server to the client to fetch a certain + configuration setting. + + This pull model replaces the old push model were the client signaled configuration change via an + event. If the server still needs to react to configuration changes (since the server caches the + result of `workspace/configuration` requests) the server should register for an empty configuration + change event and empty the cache if such an event is received. + """ + return self.send_request("workspace/configuration", params, msg_id=msg_id, callback=callback) + + def workspace_diagnostic_refresh( + self, + params: None, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[None], None]] = None, + ) -> asyncio.Task[None]: + """Make a :lsp:`workspace/diagnostic/refresh` request. + + The diagnostic refresh request definition. + + @since 3.17.0 + """ + return self.send_request("workspace/diagnostic/refresh", params, msg_id=msg_id, callback=callback) + + def workspace_inlay_hint_refresh( + self, + params: None, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[None], None]] = None, + ) -> asyncio.Task[None]: + """Make a :lsp:`workspace/inlayHint/refresh` request. + + @since 3.17.0 + """ + return self.send_request("workspace/inlayHint/refresh", params, msg_id=msg_id, callback=callback) + + def workspace_inline_value_refresh( + self, + params: None, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[None], None]] = None, + ) -> asyncio.Task[None]: + """Make a :lsp:`workspace/inlineValue/refresh` request. + + @since 3.17.0 + """ + return self.send_request("workspace/inlineValue/refresh", params, msg_id=msg_id, callback=callback) + + def workspace_semantic_tokens_refresh( + self, + params: None, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[None], None]] = None, + ) -> asyncio.Task[None]: + """Make a :lsp:`workspace/semanticTokens/refresh` request. + + @since 3.16.0 + """ + return self.send_request("workspace/semanticTokens/refresh", params, msg_id=msg_id, callback=callback) + + def workspace_workspace_folders( + self, + params: None, + msg_id: Optional[json_rpc.MsgId] = None, + callback: Optional[Callable[[Optional[List[types.WorkspaceFolder]]], None]] = None, + ) -> asyncio.Task[Optional[List[types.WorkspaceFolder]]]: + """Make a :lsp:`workspace/workspaceFolders` request. + + The `workspace/workspaceFolders` is sent from the server to the client to fetch the open workspace folders. + """ + return self.send_request("workspace/workspaceFolders", params, msg_id=msg_id, callback=callback) + + def cancel_request(self, params: types.CancelParams) -> None: + """Send a :lsp:`$/cancelRequest` notification. + + + """ + + self.send_notification("$/cancelRequest", params) + + def log_trace(self, params: types.LogTraceParams) -> None: + """Send a :lsp:`$/logTrace` notification. + + + """ + + self.send_notification("$/logTrace", params) + + def progress(self, params: types.ProgressParams) -> None: + """Send a :lsp:`$/progress` notification. + + + """ + + self.send_notification("$/progress", params) + + def telemetry_event(self, params: typing.Optional) -> None: + """Send a :lsp:`telemetry/event` notification. + + The telemetry event notification is sent from the server to the client to ask + the client to log telemetry data. + """ + + self.send_notification("telemetry/event", params) + + def text_document_publish_diagnostics(self, params: types.PublishDiagnosticsParams) -> None: + """Send a :lsp:`textDocument/publishDiagnostics` notification. + + Diagnostics notification are sent from the server to the client to signal + results of validation runs. + """ + + self.send_notification("textDocument/publishDiagnostics", params) + + def window_log_message(self, params: types.LogMessageParams) -> None: + """Send a :lsp:`window/logMessage` notification. + + The log message notification is sent from the server to the client to ask + the client to log a particular message. + """ + + self.send_notification("window/logMessage", params) + + def window_show_message(self, params: types.ShowMessageParams) -> None: + """Send a :lsp:`window/showMessage` notification. + + The show message notification is sent from a server to a client to ask + the client to display a particular message in the user interface. + """ + + self.send_notification("window/showMessage", params) diff --git a/scripts/generate_base_lsp_classes.py b/scripts/generate_base_lsp_classes.py new file mode 100644 index 00000000..927a7120 --- /dev/null +++ b/scripts/generate_base_lsp_classes.py @@ -0,0 +1,303 @@ +"""Script to automatically generate a lanaguge client from `lsprotocol` type definitons +""" +import argparse +import inspect +import pathlib +import re +import sys +import textwrap +from typing import Optional +from typing import Set +from typing import Tuple +from typing import Type +from typing import Union + +from lsprotocol._hooks import _resolve_forward_references +from lsprotocol.types import METHOD_TO_TYPES +from lsprotocol.types import message_direction + +cli = argparse.ArgumentParser( + description="generate a base language client and server from lsprotocol types." +) +cli.add_argument( + "-o", "--outdir", default=None, help="the directory to save the generated files to" +) + + +def write_imports(imports: Set[Union[str, Tuple[str, str]]]) -> str: + lines = [] + + for import_ in sorted(list(imports), key=lambda i: (i[0], i[1])): + if isinstance(import_, tuple): + mod, name = import_ + lines.append(f"from {mod} import {name}") + continue + + lines.append(f"import {import_}") + + return "\n".join(lines) + + +def to_snake_case(string: str) -> str: + return "".join(f"_{c.lower()}" if c.isupper() else c for c in string) + + +def write_notification( + method: str, + notification: Type, + params: Optional[Type], +) -> str: + """Write a method that sends the given notification message + + Parameters + ---------- + method + The notification method name + + notification + The class representing the notification message's overall structure + + params + The class representing the notification message's parameters + + Returns + ------- + str + The method definition + """ + python_name = to_snake_case(method).replace("/", "_").replace("$_", "") + + if params is None: + param_name = "None" + param_mod = "" + else: + param_mod, param_name = params.__module__, params.__name__ + param_mod = param_mod.replace("lsprotocol.types", "types") + "." + + return "\n".join( + [ + f"def {python_name}(self, params: {param_mod}{param_name}) -> None:", + f' """Send a :lsp:`{method}` notification.', + "", + textwrap.indent(inspect.getdoc(notification) or "", " "), + ' """', + "", + f' self.send_notification("{method}", params)', + "", + ] + ) + + +def get_response_type(response: Type, imports: Set[Tuple[str, str]]) -> str: + """Given a response message type, return the corresponsing result type annotation. + + Parameters + ---------- + response + The class describing the response message + + imports + The set of objects to import, this function may add new elements. + + Returns + ------- + str + The type annotation to use for the response type. + """ + # Find the response type. + result_field = [f for f in response.__attrs_attrs__ if f.name == "result"][0] + result = re.sub(r"", r"\1", str(result_field.type)) + result = re.sub(r"ForwardRef\('([\w.]+)'\)", r"lsprotocol.types.\1", result) + result = result.replace("NoneType", "None") + + # Replace any typing imports with their short name. + for match in re.finditer(r"typing.([\w]+)", result): + imports.add(("typing", match.group(1))) + + result = result.replace("lsprotocol.types.", "types.") + result = result.replace("typing.", "") + + return result + + +def write_request( + method: str, + request: Type, + params: Optional[Type], + response: Type, + imports: Set[Tuple[str, str]], +) -> str: + """Write a method that sends the given request + + Parameters + ---------- + method + The notification method name + + request + The class representing the request message's overall structure + + params + The class representing the request message's parameters + + response + The class representing the respone message's overall structure + + imports + The set of objects to import, this function may add new elements + + Returns + ------- + str + The method definition + """ + + python_name = to_snake_case(method).replace("/", "_").replace("$_", "") + + if params is None: + param_name = "None" + param_mod = "" + else: + param_mod, param_name = params.__module__, params.__name__ + param_mod = param_mod.replace("lsprotocol.types", "types") + "." + + result_type = get_response_type(response, imports) + + return "\n".join( + [ + f"def {python_name}(", + " self,", + f" params: {param_mod}{param_name},", + f" msg_id: Optional[json_rpc.MsgId] = None,", + f" callback: Optional[Callable[[{result_type}], None]] = None,", + f") -> asyncio.Task[{result_type}]:", + f' """Make a :lsp:`{method}` request.', + "", + textwrap.indent(inspect.getdoc(request) or "", " "), + ' """', + f' return self.send_request("{method}", params, msg_id=msg_id, callback=callback)', + "", + ] + ) + + +def generate_client() -> str: + """Generate a base language client derived from ``lsprotocol`` type definitions. + + Returns + ------- + str + The generated base client. + """ + methods = [] + imports = { + "asyncio", + ("lsprotocol", "types"), + ("pygls.client._native", "JsonRPCClient"), + ("pygls.protocol", "next as json_rpc"), + ("typing", "Callable"), + ("typing", "Optional"), + } + + for method_name, types in METHOD_TO_TYPES.items(): + # Skip any requests that come from the server. + if message_direction(method_name) == "serverToClient": + continue + + request, response, params, _ = types + + if response is None: + method = write_notification(method_name, request, params) + else: + method = write_request(method_name, request, params, response, imports) + + methods.append(textwrap.indent(method, " ")) + + code = [ + "# GENERATED FROM scripts/genenerate_base_lsp_classes.py -- DO NOT EDIT", + "# flake8: noqa", + write_imports(imports), + "", + "", + "class BaseLanguageClient(JsonRPCClient):", + "", + " def __init__(", + " **kwargs,", + " ):", + " super().__init__(**kwargs)", + "", + *methods, + ] + return "\n".join(code) + + +def generate_server() -> str: + """Generate a base language server derived from ``lsprotocol`` type definitions. + + Returns + ------- + str + The generated base server. + """ + methods = [] + imports = { + "asyncio", + ("lsprotocol", "types"), + ("pygls.server._native", "JsonRPCServer"), + ("pygls.protocol", "next as json_rpc"), + ("typing", "Callable"), + ("typing", "Optional"), + } + + for method_name, types in METHOD_TO_TYPES.items(): + # Skip any requests that come from the client. + if message_direction(method_name) == "clientToServer": + continue + + request, response, params, _ = types + + if response is None: + method = write_notification(method_name, request, params) + else: + method = write_request(method_name, request, params, response, imports) + + methods.append(textwrap.indent(method, " ")) + + code = [ + "# GENERATED FROM scripts/genenerate_base_lsp_classes.py -- DO NOT EDIT", + "# flake8: noqa", + write_imports(imports), + "", + "", + "class BaseLanguageServer(JsonRPCServer):", + "", + " def __init__(", + " **kwargs,", + " ):", + " super().__init__(**kwargs)", + "", + *methods, + ] + return "\n".join(code) + + +def main(): + args = cli.parse_args() + + # Make sure all the type annotations in lsprotocol are resolved correctly. + _resolve_forward_references() + client = generate_client() + server = generate_server() + + if args.outdir is None: + sys.stdout.write(client) + sys.stdout.write(server) + else: + outdir = pathlib.Path(args.outdir) + + (outdir / "_base_client.py").write_text(client) + (outdir / "_base_server.py").write_text(server) + + +if __name__ == "__main__": + main()