diff options
author | Maxim Devaev <[email protected]> | 2022-09-04 18:08:40 +0300 |
---|---|---|
committer | Maxim Devaev <[email protected]> | 2022-09-04 18:08:40 +0300 |
commit | ee3e224e396494cd0d69bb6167087a071a20349c (patch) | |
tree | 5becd28570e58a03c6e1e231d0db24c264a73f88 /kvmd/aiomulti.py | |
parent | 4b75221e9470b4a009955d7677f16adf8e23e302 (diff) |
new typing style
Diffstat (limited to 'kvmd/aiomulti.py')
-rw-r--r-- | kvmd/aiomulti.py | 13 |
1 files changed, 5 insertions, 8 deletions
diff --git a/kvmd/aiomulti.py b/kvmd/aiomulti.py index 72ec5cda..2222b9be 100644 --- a/kvmd/aiomulti.py +++ b/kvmd/aiomulti.py @@ -23,12 +23,9 @@ import multiprocessing import queue -from typing import Tuple -from typing import Dict from typing import Type from typing import TypeVar from typing import Generic -from typing import Optional from . import aiotools @@ -40,7 +37,7 @@ _QueueItemT = TypeVar("_QueueItemT") async def queue_get_last( # pylint: disable=invalid-name q: "multiprocessing.Queue[_QueueItemT]", timeout: float, -) -> Tuple[bool, Optional[_QueueItemT]]: +) -> tuple[bool, (_QueueItemT | None)]: return (await aiotools.run_async(queue_get_last_sync, q, timeout)) @@ -48,7 +45,7 @@ async def queue_get_last( # pylint: disable=invalid-name def queue_get_last_sync( # pylint: disable=invalid-name q: "multiprocessing.Queue[_QueueItemT]", timeout: float, -) -> Tuple[bool, Optional[_QueueItemT]]: +) -> tuple[bool, (_QueueItemT | None)]: try: item = q.get(timeout=timeout) @@ -79,7 +76,7 @@ _SharedFlagT = TypeVar("_SharedFlagT", int, bool) class AioSharedFlags(Generic[_SharedFlagT]): def __init__( self, - initial: Dict[str, _SharedFlagT], + initial: dict[str, _SharedFlagT], notifier: AioProcessNotifier, type: Type[_SharedFlagT]=bool, # pylint: disable=redefined-builtin ) -> None: @@ -105,10 +102,10 @@ class AioSharedFlags(Generic[_SharedFlagT]): if changed: self.__notifier.notify() - async def get(self) -> Dict[str, _SharedFlagT]: + async def get(self) -> dict[str, _SharedFlagT]: return (await aiotools.run_async(self.__inner_get)) - def __inner_get(self) -> Dict[str, _SharedFlagT]: + def __inner_get(self) -> dict[str, _SharedFlagT]: with self.__lock: return { key: self.__type(shared.value) |