lsp: Scaffold next generation server

This commit introduces what hopefully will become the language server
that ships in `esbonio` v1.

Currently, the server only does enough to accept an `initialize` request
from a client while keeping most of the "good bits" from the previous
architecture.

lib/esbonio/esbonio/server/__main__.py

@@ -0,0 +1,5 @@
+import sys
+
+from esbonio.server.cli import main
+
+sys.exit(main())

lib/esbonio/esbonio/server/cli.py

@@ -0,0 +1,103 @@
+import argparse
+import logging
+import sys
+import warnings
+from typing import Optional
+from typing import Sequence
+
+from pygls.protocol import default_converter
+
+from .log import LOG_NAMESPACE
+from .log import MemoryHandler
+from .server import EsbonioLanguageServer
+from .server import __version__
+from .setup import create_language_server
+
+
+def build_parser() -> argparse.ArgumentParser:
+ """Return an argument parser with the default command line options required for
+ main.
+ """
+
+ cli = argparse.ArgumentParser(description="The Esbonio language server")
+ cli.add_argument(
+ "-p",
+ "--port",
+ type=int,
+ default=None,
+ help="start a TCP instance of the language server listening on the given port.",
+ )
+ cli.add_argument(
+ "--version",
+ action="version",
+ version=__version__,
+ help="print the current version and exit.",
+ )
+
+ modules = cli.add_argument_group(
+ "modules", "include/exclude language server modules."
+ )
+ modules.add_argument(
+ "-i",
+ "--include",
+ metavar="MOD",
+ action="append",
+ default=[],
+ dest="included_modules",
+ help="include an additional module in the server configuration, can be given multiple times.",
+ )
+ modules.add_argument(
+ "-e",
+ "--exclude",
+ metavar="MOD",
+ action="append",
+ default=[],
+ dest="excluded_modules",
+ help="exclude a module from the server configuration, can be given multiple times.",
+ )
+
+ return cli
+
+
+def main(argv: Optional[Sequence[str]] = None):
+ """Standard main function for each of the default language servers."""
+
+ # Put these here to avoid circular import issues.
+
+ cli = build_parser()
+ args = cli.parse_args(argv)
+
+ modules = list()
+
+ for mod in args.included_modules:
+ modules.append(mod)
+
+ for mod in args.excluded_modules:
+ if mod in modules:
+ modules.remove(mod)
+
+ # Ensure we can capture warnings.
+ logging.captureWarnings(True)
+ warnlog = logging.getLogger("py.warnings")
+
+ if not sys.warnoptions:
+ warnings.simplefilter("default") # Enable capture of DeprecationWarnings
+
+ # Setup a temporary logging handler that can cache messages until the language server
+ # is ready to forward them onto the client.
+ logger = logging.getLogger(LOG_NAMESPACE)
+ logger.setLevel(logging.DEBUG)
+
+ handler = MemoryHandler()
+ handler.setLevel(logging.DEBUG)
+ logger.addHandler(handler)
+ warnlog.addHandler(handler)
+
+ server = create_language_server(
+ EsbonioLanguageServer, modules, converter_factory=default_converter
+ )
+
+ if args.port:
+ server.start_tcp("localhost", args.port)
+ else:
+ server.start_io()

lib/esbonio/esbonio/server/feature.py

@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+import typing
+
+if typing.TYPE_CHECKING:
+ from .server import EsbonioLanguageServer
+
+
+class LanguageFeature:
+ """Base class for language features."""
+
+ def __init__(self, server: EsbonioLanguageServer):
+ self.server = server
+ self.logger = server.logger.getChild(self.__class__.__name__)

lib/esbonio/esbonio/server/log.py

@@ -0,0 +1,182 @@
+from __future__ import annotations
+
+import logging
+import pathlib
+import traceback
+import typing
+from typing import List
+from typing import Tuple
+
+import pygls.uris as uri
+from lsprotocol.types import Diagnostic
+from lsprotocol.types import DiagnosticSeverity
+from lsprotocol.types import DiagnosticTag
+from lsprotocol.types import Position
+from lsprotocol.types import Range
+
+if typing.TYPE_CHECKING:
+ from .server import EsbonioLanguageServer
+ from .server import ServerConfig
+
+
+LOG_NAMESPACE = "esbonio"
+LOG_LEVELS = {
+ "debug": logging.DEBUG,
+ "error": logging.ERROR,
+ "info": logging.INFO,
+}
+
+
+class LogFilter(logging.Filter):
+ """A log filter that accepts message from any of the listed logger names."""
+
+ def __init__(self, names):
+ self.names = names
+
+ def filter(self, record):
+ return any(record.name == name for name in self.names)
+
+
+class MemoryHandler(logging.Handler):
+ """A logging handler that caches messages in memory."""
+
+ def __init__(self):
+ super().__init__()
+ self.records: List[logging.LogRecord] = []
+
+ def emit(self, record: logging.LogRecord) -> None:
+ self.records.append(record)
+
+
+class LspHandler(logging.Handler):
+ """A logging handler that will send log records to an LSP client."""
+
+ def __init__(
+ self, server: EsbonioLanguageServer, show_deprecation_warnings: bool = False
+ ):
+ super().__init__()
+ self.server = server
+ self.show_deprecation_warnings = show_deprecation_warnings
+
+ def get_warning_path(self, warning: str) -> Tuple[str, List[str]]:
+ """Determine the filepath that the warning was emitted from."""
+
+ path, *parts = warning.split(":")
+
+ # On windows the rest of the path will be in the first element of parts.
+ if pathlib.Path(warning).drive:
+ path += f":{parts.pop(0)}"
+
+ return path, parts
+
+ def handle_warning(self, record: logging.LogRecord):
+ """Publish warnings to the client as diagnostics."""
+
+ if not isinstance(record.args, tuple):
+ self.server.logger.debug(
+ "Unable to handle warning, expected tuple got: %s", record.args
+ )
+ return
+
+ # The way warnings are logged is different in Python 3.11+
+ if len(record.args) == 0:
+ argument = record.msg
+ else:
+ argument = record.args[0] # type: ignore
+
+ if not isinstance(argument, str):
+ self.server.logger.debug(
+ "Unable to handle warning, expected string got: %s", argument
+ )
+ return
+
+ warning, *_ = argument.split("\n")
+ path, (linenum, category, *msg) = self.get_warning_path(warning)
+
+ category = category.strip()
+ message = ":".join(msg).strip()
+
+ try:
+ line = int(linenum)
+ except ValueError:
+ line = 1
+ self.server.logger.debug(
+ "Unable to parse line number: '%s'\n%s", linenum, traceback.format_exc()
+ )
+
+ tags = []
+ if category == "DeprecationWarning":
+ tags.append(DiagnosticTag.Deprecated)
+
+ diagnostic = Diagnostic(
+ range=Range(
+ start=Position(line=line - 1, character=0),
+ end=Position(line=line, character=0),
+ ),
+ message=message,
+ severity=DiagnosticSeverity.Warning,
+ tags=tags,
+ )
+
+ self.server.add_diagnostics("esbonio", uri.from_fs_path(path), diagnostic)
+ self.server.sync_diagnostics()
+
+ def emit(self, record: logging.LogRecord) -> None:
+ """Sends the record to the client."""
+
+ # To avoid infinite recursions, it's simpler to just ignore all log records
+ # coming from pygls...
+ if "pygls" in record.name:
+ return
+
+ if record.name == "py.warnings":
+ if not self.show_deprecation_warnings:
+ return
+
+ self.handle_warning(record)
+
+ log = self.format(record).strip()
+ self.server.show_message_log(log)
+
+
+def setup_logging(server: EsbonioLanguageServer, config: ServerConfig):
+ """Setup logging to route log messages to the language client as
+ ``window/logMessage`` messages.
+
+ Parameters
+ ----------
+ server
+ The server to use to send messages
+
+ config
+ The configuration to use
+ """
+
+ level = LOG_LEVELS[config.log_level]
+
+ warnlog = logging.getLogger("py.warnings")
+ logger = logging.getLogger(LOG_NAMESPACE)
+ logger.setLevel(level)
+
+ lsp_handler = LspHandler(server, config.show_deprecation_warnings)
+ lsp_handler.setLevel(level)
+
+ if len(config.log_filter) > 0:
+ lsp_handler.addFilter(LogFilter(config.log_filter))
+
+ formatter = logging.Formatter("[%(name)s] %(message)s")
+ lsp_handler.setFormatter(formatter)
+
+ # Look to see if there are any cached messages we should forward to the client.
+ for handler in logger.handlers:
+ if not isinstance(handler, MemoryHandler):
+ continue
+
+ for record in handler.records:
+ if logger.isEnabledFor(record.levelno):
+ lsp_handler.emit(record)
+
+ logger.removeHandler(handler)
+
+ logger.addHandler(lsp_handler)
+ warnlog.addHandler(lsp_handler)