new typing style

7 files changed, 33 insertions, 56 deletions

diff --git a/kvmd/apps/vnc/__init__.py b/kvmd/apps/vnc/__init__.py
index d7e2c68a..3653f416 100644
--- a/kvmd/apps/vnc/__init__.py
+++ b/kvmd/apps/vnc/__init__.py
@@ -20,9 +20,6 @@
 # ========================================================================== #
 
 
-from typing import List
-from typing import Optional
-
 from ...clients.kvmd import KvmdClient
 from ...clients.streamer import StreamFormats
 from ...clients.streamer import BaseStreamerClient
@@ -38,7 +35,7 @@ from .server import VncServer
 
 
 # =====
-def main(argv: Optional[List[str]]=None) -> None:
+def main(argv: (list[str] | None)=None) -> None:
     config = init(
         prog="kvmd-vnc",
         description="VNC to KVMD proxy",
@@ -48,12 +45,12 @@ def main(argv: Optional[List[str]]=None) -> None:
 
     user_agent = htclient.make_user_agent("KVMD-VNC")
 
-    def make_memsink_streamer(name: str, fmt: int) -> Optional[MemsinkStreamerClient]:
+    def make_memsink_streamer(name: str, fmt: int) -> (MemsinkStreamerClient | None):
         if getattr(config.memsink, name).sink:
             return MemsinkStreamerClient(name.upper(), fmt, **getattr(config.memsink, name)._unpack())
         return None
 
-    streamers: List[BaseStreamerClient] = list(filter(None, [
+    streamers: list[BaseStreamerClient] = list(filter(None, [
         make_memsink_streamer("h264", StreamFormats.H264),
         make_memsink_streamer("jpeg", StreamFormats.JPEG),
         HttpStreamerClient(name="JPEG", user_agent=user_agent, **config.streamer._unpack()),
diff --git a/kvmd/apps/vnc/rfb/__init__.py b/kvmd/apps/vnc/rfb/__init__.py
index 059d4d1d..f715f9b7 100644
--- a/kvmd/apps/vnc/rfb/__init__.py
+++ b/kvmd/apps/vnc/rfb/__init__.py
@@ -23,9 +23,6 @@
 import asyncio
 import ssl
 
-from typing import Tuple
-from typing import List
-from typing import Dict
 from typing import Callable
 from typing import Coroutine
 
@@ -66,7 +63,7 @@ class RfbClient(RfbClientStream):  # pylint: disable=too-many-instance-attribute
         width: int,
         height: int,
         name: str,
-        vnc_passwds: List[str],
+        vnc_passwds: list[str],
         vencrypt: bool,
         none_auth_only: bool,
     ) -> None:
@@ -103,7 +100,7 @@ class RfbClient(RfbClientStream):  # pylint: disable=too-many-instance-attribute
         finally:
             await aiotools.shield_fg(self.__cleanup(tasks))
 
-    async def __cleanup(self, tasks: List[asyncio.Task]) -> None:
+    async def __cleanup(self, tasks: list[asyncio.Task]) -> None:
         for task in tasks:
             task.cancel()
         await asyncio.gather(*tasks, return_exceptions=True)
@@ -150,7 +147,7 @@ class RfbClient(RfbClientStream):  # pylint: disable=too-many-instance-attribute
     async def _on_ext_key_event(self, code: int, state: bool) -> None:
         raise NotImplementedError
 
-    async def _on_pointer_event(self, buttons: Dict[str, bool], wheel: Dict[str, int], move: Dict[str, int]) -> None:
+    async def _on_pointer_event(self, buttons: dict[str, bool], wheel: dict[str, int], move: dict[str, int]) -> None:
         raise NotImplementedError
 
     async def _on_cut_event(self, text: str) -> None:
@@ -232,7 +229,7 @@ class RfbClient(RfbClientStream):  # pylint: disable=too-many-instance-attribute
     # =====
 
     async def __handshake_security(self) -> None:
-        sec_types: Dict[int, Tuple[str, Callable]] = {}
+        sec_types: dict[int, tuple[str, Callable]] = {}
         if self.__vencrypt and self.__rfb_version > 3:
             sec_types[19] = ("VeNCrypt", self.__handshake_security_vencrypt)
         if self.__none_auth_only:
diff --git a/kvmd/apps/vnc/rfb/crypto.py b/kvmd/apps/vnc/rfb/crypto.py
index 854dceb3..8953bef5 100644
--- a/kvmd/apps/vnc/rfb/crypto.py
+++ b/kvmd/apps/vnc/rfb/crypto.py
@@ -22,8 +22,6 @@
 
 import os
 
-from typing import List
-
 import passlib.crypto.des
 
 
@@ -43,7 +41,7 @@ def rfb_encrypt_challenge(challenge: bytes, passwd: bytes) -> bytes:
 
 def _make_key(passwd: bytes) -> bytes:
     passwd = (passwd + b"\0" * 8)[:8]
-    key: List[int] = []
+    key: list[int] = []
     for ch in passwd:
         btgt = 0
         for index in range(8):
diff --git a/kvmd/apps/vnc/rfb/encodings.py b/kvmd/apps/vnc/rfb/encodings.py
index de51dde5..e153b5e8 100644
--- a/kvmd/apps/vnc/rfb/encodings.py
+++ b/kvmd/apps/vnc/rfb/encodings.py
@@ -22,10 +22,6 @@
 
 import dataclasses
 
-from typing import List
-from typing import Dict
-from typing import FrozenSet
-from typing import Union
 from typing import Any
 
 
@@ -45,13 +41,13 @@ class RfbEncodings:
     H264 = 50  # Open H.264 Encoding
 
 
-def _make_meta(variants: Union[int, FrozenSet[int]]) -> Dict:
+def _make_meta(variants: (int | frozenset[int])) -> dict:
     return {"variants": (frozenset([variants]) if isinstance(variants, int) else variants)}
 
 
 @dataclasses.dataclass(frozen=True)
 class RfbClientEncodings:  # pylint: disable=too-many-instance-attributes
-    encodings: FrozenSet[int]
+    encodings: frozenset[int]
 
     has_resize: bool =		    dataclasses.field(default=False, metadata=_make_meta(RfbEncodings.RESIZE))  # noqa: E224
     has_rename: bool =		    dataclasses.field(default=False, metadata=_make_meta(RfbEncodings.RENAME))  # noqa: E224
@@ -63,8 +59,8 @@ class RfbClientEncodings:  # pylint: disable=too-many-instance-attributes
 
     has_h264: bool =			dataclasses.field(default=False, metadata=_make_meta(RfbEncodings.H264))  # noqa: E224
 
-    def get_summary(self) -> List[str]:
-        summary: List[str] = [f"encodings -- {sorted(self.encodings)}"]
+    def get_summary(self) -> list[str]:
+        summary: list[str] = [f"encodings -- {sorted(self.encodings)}"]
         for field in dataclasses.fields(self):
             if field.name != "encodings":
                 found = ", ".join(map(str, sorted(map(int, self.__get_found(field)))))
@@ -80,7 +76,7 @@ class RfbClientEncodings:  # pylint: disable=too-many-instance-attributes
     def __set_value(self, key: str, value: Any) -> None:
         object.__setattr__(self, key, value)
 
-    def __get_found(self, field: dataclasses.Field) -> FrozenSet[int]:
+    def __get_found(self, field: dataclasses.Field) -> frozenset[int]:
         return self.encodings.intersection(field.metadata["variants"])
 
     def __get_tight_jpeg_quality(self) -> int:
diff --git a/kvmd/apps/vnc/rfb/stream.py b/kvmd/apps/vnc/rfb/stream.py
index ee42436e..dfa6cb13 100644
--- a/kvmd/apps/vnc/rfb/stream.py
+++ b/kvmd/apps/vnc/rfb/stream.py
@@ -24,9 +24,6 @@ import asyncio
 import ssl
 import struct
 
-from typing import Tuple
-from typing import Union
-
 from .... import aiotools
 
 from .errors import RfbConnectionError
@@ -57,7 +54,7 @@ class RfbClientStream:
         except (ConnectionError, asyncio.IncompleteReadError) as err:
             raise RfbConnectionError(f"Can't read {msg}", err)
 
-    async def _read_struct(self, msg: str, fmt: str) -> Tuple[int, ...]:
+    async def _read_struct(self, msg: str, fmt: str) -> tuple[int, ...]:
         assert len(fmt) > 1
         try:
             fmt = f">{fmt}"
@@ -73,7 +70,7 @@ class RfbClientStream:
 
     # =====
 
-    async def _write_struct(self, msg: str, fmt: str, *values: Union[int, bytes], drain: bool=True) -> None:
+    async def _write_struct(self, msg: str, fmt: str, *values: (int | bytes), drain: bool=True) -> None:
         try:
             if not fmt:
                 for value in values:
diff --git a/kvmd/apps/vnc/server.py b/kvmd/apps/vnc/server.py
index 7fac9aa4..c6ea076a 100644
--- a/kvmd/apps/vnc/server.py
+++ b/kvmd/apps/vnc/server.py
@@ -26,11 +26,6 @@ import socket
 import dataclasses
 import contextlib
 
-from typing import List
-from typing import Dict
-from typing import Union
-from typing import Optional
-
 import aiohttp
 
 from ...logging import get_logger
@@ -83,12 +78,12 @@ class _Client(RfbClient):  # pylint: disable=too-many-instance-attributes
 
         desired_fps: int,
         keymap_name: str,
-        symmap: Dict[int, Dict[int, str]],
+        symmap: dict[int, dict[int, str]],
 
         kvmd: KvmdClient,
-        streamers: List[BaseStreamerClient],
+        streamers: list[BaseStreamerClient],
 
-        vnc_credentials: Dict[str, VncAuthKvmdCredentials],
+        vnc_credentials: dict[str, VncAuthKvmdCredentials],
         vencrypt: bool,
         none_auth_only: bool,
         shared_params: _SharedParams,
@@ -122,15 +117,15 @@ class _Client(RfbClient):  # pylint: disable=too-many-instance-attributes
         self.__stage2_encodings_accepted = aiotools.AioStage()
         self.__stage3_ws_connected = aiotools.AioStage()
 
-        self.__kvmd_session: Optional[KvmdClientSession] = None
-        self.__kvmd_ws: Optional[KvmdClientWs] = None
+        self.__kvmd_session: (KvmdClientSession | None) = None
+        self.__kvmd_ws: (KvmdClientWs | None) = None
 
         self.__fb_notifier = aiotools.AioNotifier()
-        self.__fb_queue: "asyncio.Queue[Dict]" = asyncio.Queue()
+        self.__fb_queue: "asyncio.Queue[dict]" = asyncio.Queue()
 
         # Эти состояния шарить не обязательно - бекенд исключает дублирующиеся события.
         # Все это нужно только чтобы не посылать лишние жсоны в сокет KVMD
-        self.__mouse_buttons: Dict[str, Optional[bool]] = dict.fromkeys(["left", "right", "middle"], None)
+        self.__mouse_buttons: dict[str, (bool | None)] = dict.fromkeys(["left", "right", "middle"], None)
         self.__mouse_move = {"x": -1, "y": -1}
 
         self.__lock = asyncio.Lock()
@@ -175,7 +170,7 @@ class _Client(RfbClient):  # pylint: disable=too-many-instance-attributes
         finally:
             self.__kvmd_ws = None
 
-    async def __process_ws_event(self, event_type: str, event: Dict) -> None:
+    async def __process_ws_event(self, event_type: str, event: dict) -> None:
         if event_type == "info_meta_state":
             try:
                 host = event["server"]["host"]
@@ -225,7 +220,7 @@ class _Client(RfbClient):  # pylint: disable=too-many-instance-attributes
             StreamFormats.JPEG: "has_tight",
             StreamFormats.H264: "has_h264",
         }
-        streamer: Optional[BaseStreamerClient] = None
+        streamer: (BaseStreamerClient | None) = None
         for streamer in self.__streamers:
             if getattr(self._encodings, formats[streamer.get_format()]):
                 get_logger(0).info("%s [streamer]: Using preferred %s", self._remote, streamer)
@@ -237,12 +232,12 @@ class _Client(RfbClient):  # pylint: disable=too-many-instance-attributes
         get_logger(0).info("%s [streamer]: Using default %s", self._remote, streamer)
         return streamer
 
-    async def __queue_frame(self, frame: Union[Dict, str]) -> None:
+    async def __queue_frame(self, frame: (dict | str)) -> None:
         if isinstance(frame, str):
             frame = await self.__make_text_frame(frame)
         self.__fb_queue.put_nowait(frame)
 
-    async def __make_text_frame(self, text: str) -> Dict:
+    async def __make_text_frame(self, text: str) -> dict:
         return {
             "data": (await make_text_jpeg(self._width, self._height, self._encodings.tight_jpeg_quality, text)),
             "width": self._width,
@@ -252,7 +247,7 @@ class _Client(RfbClient):  # pylint: disable=too-many-instance-attributes
 
     async def __fb_sender_task_loop(self) -> None:  # pylint: disable=too-many-branches
         has_h264_key = False
-        last: Optional[Dict] = None
+        last: (dict | None) = None
         while True:
             await self.__fb_notifier.wait()
 
@@ -351,7 +346,7 @@ class _Client(RfbClient):  # pylint: disable=too-many-instance-attributes
             if self.__kvmd_ws:
                 await self.__kvmd_ws.send_key_event(web_key, state)
 
-    def __switch_modifiers(self, key: Union[int, str], state: bool) -> bool:
+    def __switch_modifiers(self, key: (int | str), state: bool) -> bool:
         mod = 0
         if key in X11Modifiers.SHIFTS or key in WebModifiers.SHIFTS:
             mod = SymmapModifiers.SHIFT
@@ -367,7 +362,7 @@ class _Client(RfbClient):  # pylint: disable=too-many-instance-attributes
             self.__modifiers &= ~mod
         return True
 
-    async def _on_pointer_event(self, buttons: Dict[str, bool], wheel: Dict[str, int], move: Dict[str, int]) -> None:
+    async def _on_pointer_event(self, buttons: dict[str, bool], wheel: dict[str, int], move: dict[str, int]) -> None:
         if self.__kvmd_ws:
             for (button, state) in buttons.items():
                 if self.__mouse_buttons[button] != state:
@@ -434,7 +429,7 @@ class VncServer:  # pylint: disable=too-many-instance-attributes
         keymap_path: str,
 
         kvmd: KvmdClient,
-        streamers: List[BaseStreamerClient],
+        streamers: list[BaseStreamerClient],
         vnc_auth_manager: VncAuthManager,
     ) -> None:
 
diff --git a/kvmd/apps/vnc/vncauth.py b/kvmd/apps/vnc/vncauth.py
index b0f702d1..c7c04864 100644
--- a/kvmd/apps/vnc/vncauth.py
+++ b/kvmd/apps/vnc/vncauth.py
@@ -22,9 +22,6 @@
 
 import dataclasses
 
-from typing import Tuple
-from typing import Dict
-
 from ...logging import get_logger
 
 from ... import aiofs
@@ -53,7 +50,7 @@ class VncAuthManager:
         self.__path = path
         self.__enabled = enabled
 
-    async def read_credentials(self) -> Tuple[Dict[str, VncAuthKvmdCredentials], bool]:
+    async def read_credentials(self) -> tuple[dict[str, VncAuthKvmdCredentials], bool]:
         if self.__enabled:
             try:
                 return (await self.__inner_read_credentials(), True)
@@ -63,10 +60,10 @@ class VncAuthManager:
                 get_logger(0).exception("Unhandled exception while reading VNCAuth passwd file")
         return ({}, (not self.__enabled))
 
-    async def __inner_read_credentials(self) -> Dict[str, VncAuthKvmdCredentials]:
+    async def __inner_read_credentials(self) -> dict[str, VncAuthKvmdCredentials]:
         lines = (await aiofs.read(self.__path)).split("\n")
 
-        credentials: Dict[str, VncAuthKvmdCredentials] = {}
+        credentials: dict[str, VncAuthKvmdCredentials] = {}
         for (lineno, line) in enumerate(lines):
             if len(line.strip()) == 0 or line.lstrip().startswith("#"):
                 continue