summaryrefslogtreecommitdiff
path: root/kvmd/aiomulti.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/aiomulti.py
parent4b75221e9470b4a009955d7677f16adf8e23e302 (diff)
new typing style
Diffstat (limited to 'kvmd/aiomulti.py')
-rw-r--r--kvmd/aiomulti.py13
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)