lsp: pyupgrade --py39-plus

lib/esbonio/esbonio/server/_configuration.py

@@ -17,14 +17,9 @@
 T = TypeVar("T")
 
 if typing.TYPE_CHECKING:
+    from collections.abc import Awaitable
     from typing import Any
-    from typing import Awaitable
     from typing import Callable
-    from typing import Dict
-    from typing import List
-    from typing import Optional
-    from typing import Set
-    from typing import Type
     from typing import Union
 
     from .server import EsbonioLanguageServer
@@ -47,7 +42,7 @@ class Subscription(Generic[T]):
     section: str
     """The configuration section."""
 
-    spec: Type[T]
+    spec: type[T]
     """The subscription's class definition."""
 
     callback: ConfigurationCallback
@@ -67,7 +62,7 @@ class ConfigChangeEvent(Generic[T]):
     value: T
     """The latest configuration value."""
 
-    previous: Optional[T] = None
+    previous: T | None = None
     """The previous configuration value, (if any)."""
 
 
@@ -90,13 +85,13 @@ def scope(self) -> str:
         return max([self.file_scope, self.workspace_scope], key=len)
 
     @property
-    def scope_path(self) -> Optional[str]:
+    def scope_path(self) -> str | None:
         """The scope uri as a path."""
         uri = Uri.parse(self.scope)
         return uri.path
 
     @property
-    def scope_fs_path(self) -> Optional[str]:
+    def scope_fs_path(self) -> str | None:
         """The scope uri as an fs path."""
         uri = Uri.parse(self.scope)
         return uri.fs_path
@@ -158,16 +153,16 @@ def __init__(self, server: EsbonioLanguageServer):
         self.logger = server.logger.getChild("Configuration")
         """The logger instance to use"""
 
-        self._initialization_options: Dict[str, Any] = {}
+        self._initialization_options: dict[str, Any] = {}
         """The received initializaion options (if any)"""
 
-        self._workspace_config: Dict[str, Dict[str, Any]] = {}
+        self._workspace_config: dict[str, dict[str, Any]] = {}
         """The cached workspace configuration."""
 
-        self._file_config: Dict[str, Dict[str, Any]] = {}
+        self._file_config: dict[str, dict[str, Any]] = {}
         """The cached configuration coming from configuration files."""
 
-        self._subscriptions: Dict[Subscription, Any] = {}
+        self._subscriptions: dict[Subscription, Any] = {}
         """Subscriptions and their last known value"""
 
     @property
@@ -203,9 +198,9 @@ def supports_workspace_config(self):
     def subscribe(
         self,
         section: str,
-        spec: Type[T],
+        spec: type[T],
         callback: ConfigurationCallback,
-        scope: Optional[Uri] = None,
+        scope: Uri | None = None,
     ):
         """Subscribe to updates to the given configuration section.
 
@@ -276,7 +271,7 @@ def _notify_subscriptions(self, *args):
                     exc_info=True,
                 )
 
-    def get(self, section: str, spec: Type[T], scope: Optional[Uri] = None) -> T:
+    def get(self, section: str, spec: type[T], scope: Uri | None = None) -> T:
         """Get the requested configuration section.
 
         Parameters
@@ -325,7 +320,7 @@ def scope_for(self, uri: Uri) -> str:
     def _get_config(
         self,
         section: str,
-        spec: Type[T],
+        spec: type[T],
         context: ConfigurationContext,
     ) -> T:
         """Get the requested configuration section."""
@@ -364,19 +359,19 @@ def _get_config(
             )
             return spec()
 
-    def _uri_to_file_scope(self, uri: Optional[Uri]) -> str:
+    def _uri_to_file_scope(self, uri: Uri | None) -> str:
         folder_uris = list(self._file_config.keys())
         return _uri_to_scope(folder_uris, uri)
 
-    def _uri_to_workspace_scope(self, uri: Optional[Uri]) -> str:
+    def _uri_to_workspace_scope(self, uri: Uri | None) -> str:
         folder_uris = [f.uri for f in self.workspace.folders.values()]
 
         if (root_uri := self.workspace.root_uri) is not None:
             folder_uris.append(root_uri)
 
         return _uri_to_scope(folder_uris, uri)
 
-    def _discover_config_files(self) -> List[pathlib.Path]:
+    def _discover_config_files(self) -> list[pathlib.Path]:
         """Scan the workspace for available configuration files."""
         folder_uris = {f.uri for f in self.workspace.folders.values()}
 
@@ -395,7 +390,7 @@ def _discover_config_files(self) -> List[pathlib.Path]:
 
         return paths
 
-    def update_file_configuration(self, paths: Optional[List[pathlib.Path]] = None):
+    def update_file_configuration(self, paths: list[pathlib.Path] | None = None):
         """Update the internal cache of configuration coming from files.
 
         Parameters
@@ -466,7 +461,7 @@ async def update_workspace_configuration(self):
         self._notify_subscriptions()
 
 
-def _uri_to_scope(known_scopes: List[str], uri: Optional[Uri]) -> str:
+def _uri_to_scope(known_scopes: list[str], uri: Uri | None) -> str:
     """Convert the given uri to a scope or the empty string if none could be found.
 
     Parameters
@@ -496,13 +491,13 @@ def _uri_to_scope(known_scopes: List[str], uri: Optional[Uri]) -> str:
     return sorted(candidates, key=len, reverse=True)[0]
 
 
-def _merge_configs(*configs: Dict[str, Any]):
+def _merge_configs(*configs: dict[str, Any]):
     """Recursively merge all the given configuration sources together.
 
     The last config given takes precedence.
     """
     final = {}
-    all_keys: Set[str] = set()
+    all_keys: set[str] = set()
 
     for c in configs:
         all_keys.update(c.keys())

lib/esbonio/esbonio/server/cli.py

@@ -2,9 +2,9 @@
 import logging
 import sys
 import warnings
+from collections.abc import Sequence
 from logging.handlers import MemoryHandler
 from typing import Optional
-from typing import Sequence
 
 from pygls.protocol import default_converter
 

lib/esbonio/esbonio/server/events.py

@@ -9,9 +9,6 @@
 
 if typing.TYPE_CHECKING:
     from typing import Any
-    from typing import Dict
-    from typing import Optional
-    from typing import Set
 
 
 class EventSource:
@@ -20,14 +17,14 @@ class EventSource:
     # TODO: It might be nice to do some fancy typing here so that type checkers
     # etc know which events are possible etc.
 
-    def __init__(self, logger: Optional[logging.Logger] = None):
+    def __init__(self, logger: logging.Logger | None = None):
         self.logger = logger or logging.getLogger(__name__)
         """The logging instance to use."""
 
-        self.handlers: Dict[str, set] = {}
+        self.handlers: dict[str, set] = {}
         """Collection of handlers for various events."""
 
-        self._tasks: Set[asyncio.Task] = set()
+        self._tasks: set[asyncio.Task] = set()
         """Holds tasks that are currently executing an async event handler."""
 
     def add_listener(self, event: str, handler):

lib/esbonio/esbonio/server/feature.py

@@ -12,23 +12,21 @@
 
 if typing.TYPE_CHECKING:
     import re
+    from collections.abc import Coroutine
     from typing import Any
-    from typing import Coroutine
-    from typing import List
     from typing import Optional
-    from typing import Set
     from typing import Union
 
     from .server import EsbonioLanguageServer
 
     CompletionResult = Union[
-        Optional[List[types.CompletionItem]],
-        Coroutine[Any, Any, Optional[List[types.CompletionItem]]],
+        Optional[list[types.CompletionItem]],
+        Coroutine[Any, Any, Optional[list[types.CompletionItem]]],
     ]
 
     DocumentSymbolResult = Union[
-        Optional[List[types.DocumentSymbol]],
-        Coroutine[Any, Any, Optional[List[types.DocumentSymbol]]],
+        Optional[list[types.DocumentSymbol]],
+        Coroutine[Any, Any, Optional[list[types.DocumentSymbol]]],
     ]
 
     MaybeAsyncNone = Union[
@@ -37,8 +35,8 @@
     ]
 
     WorkspaceSymbolResult = Union[
-        Optional[List[types.WorkspaceSymbol]],
-        Coroutine[Any, Any, Optional[List[types.WorkspaceSymbol]]],
+        Optional[list[types.WorkspaceSymbol]],
+        Coroutine[Any, Any, Optional[list[types.WorkspaceSymbol]]],
     ]
 
 
@@ -82,7 +80,7 @@ def document_open(self, params: types.DidOpenTextDocumentParams) -> MaybeAsyncNo
     def document_save(self, params: types.DidSaveTextDocumentParams) -> MaybeAsyncNone:
         """Called when a text document is saved."""
 
-    completion_trigger: Optional[CompletionTrigger] = None
+    completion_trigger: CompletionTrigger | None = None
 
     def completion(self, context: CompletionContext) -> CompletionResult:
         """Called when a completion request matches one of the specified triggers."""
@@ -102,16 +100,16 @@ def workspace_symbol(
 class CompletionTrigger:
     """Define when the feature's completion method should be called."""
 
-    patterns: List[re.Pattern]
+    patterns: list[re.Pattern]
     """A list of regular expressions to try"""
 
-    languages: Set[str] = attrs.field(factory=set)
+    languages: set[str] = attrs.field(factory=set)
     """Languages in which the completion trigger should fire.
 
     If empty, the document's language will be ignored.
     """
 
-    characters: Set[str] = attrs.field(factory=set)
+    characters: set[str] = attrs.field(factory=set)
     """Characters which, when typed, should trigger a completion request.
 
     If empty, this trigger will ignore any trigger characters.
@@ -124,7 +122,7 @@ def __call__(
         document: TextDocument,
         language: str,
         client_capabilities: types.ClientCapabilities,
-    ) -> Optional[CompletionContext]:
+    ) -> CompletionContext | None:
         """Determine if this completion trigger should fire.
 
         Parameters
@@ -254,7 +252,7 @@ def deprecated_support(self) -> bool:
         )
 
     @property
-    def documentation_formats(self) -> List[types.MarkupKind]:
+    def documentation_formats(self) -> list[types.MarkupKind]:
         """The list of documentation formats supported by the client."""
         return get_capability(
             self.capabilities,
@@ -291,7 +289,7 @@ def snippet_support(self) -> bool:
         )
 
     @property
-    def supported_tags(self) -> List[types.CompletionItemTag]:
+    def supported_tags(self) -> list[types.CompletionItemTag]:
         """The list of ``CompletionItemTags`` supported by the client."""
         capabilities = get_capability(
             self.capabilities,

lib/esbonio/esbonio/server/features/directives/__init__.py

@@ -8,12 +8,8 @@
 from esbonio import server
 
 if typing.TYPE_CHECKING:
+    from collections.abc import Coroutine
     from typing import Any
-    from typing import Coroutine
-    from typing import Dict
-    from typing import List
-    from typing import Optional
-    from typing import Union
 
 
 @attrs.define
@@ -23,7 +19,7 @@ class Directive:
     name: str
     """The name of the directive, as the user would type in an rst file."""
 
-    implementation: Optional[str]
+    implementation: str | None
     """The dotted name of the directive's implementation."""
 
 
@@ -32,9 +28,7 @@ class DirectiveProvider:
 
     def suggest_directives(
         self, context: server.CompletionContext
-    ) -> Union[
-        Optional[List[Directive]], Coroutine[Any, Any, Optional[List[Directive]]]
-    ]:
+    ) -> list[Directive] | None | Coroutine[Any, Any, list[Directive] | None]:
         """Given a completion context, suggest directives that may be used."""
         return None
 
@@ -49,7 +43,7 @@ class DirectiveFeature(server.LanguageFeature):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
-        self._providers: Dict[int, DirectiveProvider] = {}
+        self._providers: dict[int, DirectiveProvider] = {}
 
     def add_provider(self, provider: DirectiveProvider):
         """Register a directive provider.
@@ -63,19 +57,19 @@ def add_provider(self, provider: DirectiveProvider):
 
     async def suggest_directives(
         self, context: server.CompletionContext
-    ) -> List[Directive]:
+    ) -> list[Directive]:
         """Suggest directives that may be used, given a completion context.
 
         Parameters
         ----------
         context
            The completion context.
         """
-        items: List[Directive] = []
+        items: list[Directive] = []
 
         for provider in self._providers.values():
             try:
-                result: Optional[List[Directive]] = None
+                result: list[Directive] | None = None
 
                 aresult = provider.suggest_directives(context)
                 if inspect.isawaitable(aresult):