summaryrefslogtreecommitdiff
path: root/kvmd/inotify.py
diff options
context:
space:
mode:
authorMaxim Devaev <[email protected]>2022-09-04 18:08:40 +0300
committerMaxim Devaev <[email protected]>2022-09-04 18:08:40 +0300
commitee3e224e396494cd0d69bb6167087a071a20349c (patch)
tree5becd28570e58a03c6e1e231d0db24c264a73f88 /kvmd/inotify.py
parent4b75221e9470b4a009955d7677f16adf8e23e302 (diff)
new typing style
Diffstat (limited to 'kvmd/inotify.py')
-rw-r--r--kvmd/inotify.py23
1 files changed, 9 insertions, 14 deletions
diff --git a/kvmd/inotify.py b/kvmd/inotify.py
index 6317e62d..0511f088 100644
--- a/kvmd/inotify.py
+++ b/kvmd/inotify.py
@@ -31,12 +31,7 @@ import dataclasses
import types
import errno
-from typing import Tuple
-from typing import List
-from typing import Dict
-from typing import Type
from typing import Generator
-from typing import Optional
from .logging import get_logger
@@ -53,7 +48,7 @@ _FS_ENCODING = (sys.getfilesystemencoding() or _FS_FALLBACK_ENCODING)
# =====
-def _inotify_parsed_buffer(data: bytes) -> Generator[Tuple[int, int, int, bytes], None, None]:
+def _inotify_parsed_buffer(data: bytes) -> Generator[tuple[int, int, int, bytes], None, None]:
offset = 0
while offset + _EVENT_HEAD_SIZE <= len(data):
(wd, mask, cookie, length) = struct.unpack_from("iIII", data, offset)
@@ -157,7 +152,7 @@ class InotifyMask:
@classmethod
def to_string(cls, mask: int) -> str:
- flags: List[str] = []
+ flags: list[str] = []
for name in dir(cls):
if (
name[0].isupper()
@@ -188,10 +183,10 @@ class Inotify:
def __init__(self) -> None:
self.__fd = -1
- self.__wd_by_path: Dict[str, int] = {}
- self.__path_by_wd: Dict[int, str] = {}
+ self.__wd_by_path: dict[str, int] = {}
+ self.__path_by_wd: dict[int, str] = {}
- self.__moved: Dict[int, str] = {}
+ self.__moved: dict[int, str] = {}
self.__events_queue: "asyncio.Queue[InotifyEvent]" = asyncio.Queue()
@@ -215,15 +210,15 @@ class Inotify:
# def has_events(self) -> bool:
# return (not self.__events_queue.empty())
- async def get_event(self, timeout: float) -> Optional[InotifyEvent]:
+ async def get_event(self, timeout: float) -> (InotifyEvent | None):
assert timeout > 0
try:
return (await asyncio.wait_for(self.__events_queue.get(), timeout=timeout))
except asyncio.TimeoutError:
return None
- async def get_series(self, timeout: float) -> List[InotifyEvent]:
- series: List[InotifyEvent] = []
+ async def get_series(self, timeout: float) -> list[InotifyEvent]:
+ series: list[InotifyEvent] = []
event = await self.get_event(timeout)
if event:
series.append(event)
@@ -283,7 +278,7 @@ class Inotify:
def __exit__(
self,
- _exc_type: Type[BaseException],
+ _exc_type: type[BaseException],
_exc: BaseException,
_tb: types.TracebackType,
) -> None: