🔨 Upgrade typing to 3.10+ (#108)

Author: roma-glushko

hyx/bulkhead/api.py

@@ -1,6 +1,7 @@
 import functools
+from collections.abc import Sequence
 from types import TracebackType
-from typing import Any, Optional, Sequence, Type, cast
+from typing import Any, cast
 
 from hyx.bulkhead.events import _BULKHEAD_LISTENERS, BulkheadListener
 from hyx.bulkhead.manager import BulkheadManager
@@ -29,9 +30,9 @@ def __init__(
         max_concurrency: int,
         max_capacity: int,
         *,
-        name: Optional[str] = None,
-        listeners: Optional[Sequence[BulkheadListener]] = None,
-        event_manager: Optional["EventManager"] = None,
+        name: str | None = None,
+        listeners: Sequence[BulkheadListener] | None = None,
+        event_manager: "EventManager | None" = None,
     ) -> None:
         event_dispatcher = EventDispatcher[BulkheadManager, BulkheadListener](
             listeners,
@@ -55,10 +56,10 @@ async def __aenter__(self) -> "bulkhead":
 
     async def __aexit__(
         self,
-        exc_type: Optional[Type[BaseException]],
-        exc_val: Optional[BaseException],
-        exc_tb: Optional[TracebackType],
-    ) -> Optional[bool]:
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: TracebackType | None,
+    ) -> bool | None:
         await self._manager.release()
 
         return None

hyx/bulkhead/events.py

@@ -1,4 +1,4 @@
-from typing import TYPE_CHECKING, Union
+from typing import TYPE_CHECKING
 
 from hyx.events import ListenerFactoryT, ListenerRegistry
 
@@ -12,7 +12,7 @@ class BulkheadListener:
     async def on_bulkhead_full(self, bulkhead: "BulkheadManager") -> None: ...
 
 
-def register_bulkhead_listener(listener: Union[BulkheadListener, ListenerFactoryT]) -> None:
+def register_bulkhead_listener(listener: BulkheadListener | ListenerFactoryT) -> None:
     """
     Register a listener that will listen to all fallback components in the system
     """

hyx/bulkhead/manager.py

@@ -1,5 +1,5 @@
 import asyncio
-from typing import Any, Optional
+from typing import Any
 
 from hyx.bulkhead.events import BulkheadListener
 from hyx.bulkhead.exceptions import BulkheadFull
@@ -18,7 +18,7 @@ def __init__(
         max_concurrency: int,
         max_capacity: int,
         event_dispatcher: BulkheadListener,
-        name: Optional[str] = None,
+        name: str | None = None,
     ) -> None:
         if max_concurrency <= 0:
             raise ValueError(f'max_concurrency should be greater than zero ("{max_concurrency}" given)')

hyx/circuitbreaker/api.py

@@ -1,6 +1,7 @@
 import functools
+from collections.abc import Sequence
 from types import TracebackType
-from typing import Any, Optional, Sequence, Type, cast
+from typing import Any, cast
 
 from hyx.circuitbreaker.events import _BREAKER_LISTENERS, BreakerListener
 from hyx.circuitbreaker.managers import ConsecutiveCircuitBreaker
@@ -48,9 +49,9 @@ def __init__(
         failure_threshold: int = 5,
         recovery_time_secs: DelayT = 30,
         recovery_threshold: int = 3,
-        listeners: Optional[Sequence[BreakerListener]] = None,
-        name: Optional[str] = None,
-        event_manager: Optional["EventManager"] = None,
+        listeners: Sequence[BreakerListener] | None = None,
+        name: str | None = None,
+        event_manager: "EventManager | None" = None,
     ) -> None:
         event_dispatcher = EventDispatcher[ConsecutiveCircuitBreaker, BreakerListener](
             listeners,
@@ -80,10 +81,10 @@ async def __aenter__(self) -> "consecutive_breaker":
 
     async def __aexit__(
         self,
-        exc_type: Optional[Type[BaseException]],
-        exc_val: Optional[BaseException],
-        exc_tb: Optional[TracebackType],
-    ) -> Optional[bool]:
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: TracebackType | None,
+    ) -> bool | None:
         await self._manager.release(exc_val)
 
         return None

hyx/circuitbreaker/context.py

@@ -1,5 +1,5 @@
 import dataclasses
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING
 
 from hyx.circuitbreaker.typing import DelayT
 from hyx.typing import ExceptionsT
@@ -10,7 +10,7 @@
 
 @dataclasses.dataclass
 class BreakerContext:
-    breaker_name: Optional[str]
+    breaker_name: str | None
     exceptions: ExceptionsT
     failure_threshold: int
     recovery_time_secs: DelayT

hyx/circuitbreaker/events.py

@@ -1,4 +1,4 @@
-from typing import TYPE_CHECKING, Union
+from typing import TYPE_CHECKING
 
 from hyx.circuitbreaker.context import BreakerContext
 from hyx.circuitbreaker.managers import ConsecutiveCircuitBreaker
@@ -37,7 +37,7 @@ async def on_failing(
     async def on_success(self, context: BreakerContext, state: "BreakerState") -> None: ...
 
 
-def register_breaker_listener(listener: Union[BreakerListener, ListenerFactoryT]) -> None:
+def register_breaker_listener(listener: BreakerListener | ListenerFactoryT) -> None:
     """
     Register a listener that will listen to all circuit breaker components in the system
     """

hyx/circuitbreaker/managers.py

@@ -1,4 +1,4 @@
-from typing import TYPE_CHECKING, Any, Optional
+from typing import TYPE_CHECKING, Any
 
 from hyx.circuitbreaker.context import BreakerContext
 from hyx.circuitbreaker.states import BreakerState, WorkingState
@@ -48,7 +48,7 @@ async def _transit_state(self, new_state: BreakerState) -> None:
     async def acquire(self) -> None:
         await self._transit_state(await self._state.before_execution())
 
-    async def release(self, exception: Optional[BaseException]) -> None:
+    async def release(self, exception: BaseException | None) -> None:
         if exception and isinstance(exception, self._context.exceptions):
             await self._transit_state(await self._state.on_exception())
             raise exception

hyx/circuitbreaker/states.py

@@ -1,5 +1,4 @@
 from datetime import datetime, timedelta
-from typing import Optional
 
 from hyx.circuitbreaker.context import BreakerContext
 from hyx.circuitbreaker.exceptions import BreakerFailing
@@ -110,7 +109,7 @@ def until(self) -> datetime:
         return self._failing_until
 
     @property
-    def remain(self) -> Optional[timedelta]:
+    def remain(self) -> timedelta | None:
         """
         Remaining time the breaker is going to fail
         """

hyx/circuitbreaker/typing.py

@@ -1,3 +1 @@
-from typing import Union
-
-DelayT = Union[float, int]
+DelayT = float | int

hyx/events.py

@@ -1,12 +1,13 @@
 import asyncio
 import traceback
 import weakref
-from typing import Callable, Generic, List, Optional, Protocol, Sequence, TypeVar, Union, cast, runtime_checkable
+from collections.abc import Callable, Sequence
+from typing import Generic, Protocol, TypeVar, cast, runtime_checkable
 
 ComponentT = TypeVar("ComponentT")
 ListenerT = TypeVar("ListenerT")
 
-_EVENT_MANAGER: Optional["EventManager"] = None
+_EVENT_MANAGER: "EventManager | None" = None
 
 
 class EventManager:
@@ -61,13 +62,13 @@ class ListenerRegistry(Generic[ComponentT, ListenerT]):
     __slots__ = ("_listeners",)
 
     def __init__(self) -> None:
-        self._listeners: list[Union[ListenerT, ListenerFactoryT]] = []
+        self._listeners: list[ListenerT | ListenerFactoryT] = []
 
     @property
-    def listeners(self) -> list[Union[ListenerT, ListenerFactoryT]]:
+    def listeners(self) -> list[ListenerT | ListenerFactoryT]:
         return self._listeners
 
-    def register(self, listener: Union[ListenerT, ListenerFactoryT]) -> None:
+    def register(self, listener: ListenerT | ListenerFactoryT) -> None:
         self._listeners.append(listener)
 
 
@@ -87,17 +88,17 @@ class EventDispatcher(Generic[ComponentT, ListenerT]):
 
     def __init__(
         self,
-        local_listeners: Optional[Sequence[Union[ListenerT, ListenerFactoryT]]] = None,
-        global_listener_registry: Optional[ListenerRegistry] = None,
-        event_manager: Optional["EventManager"] = None,
+        local_listeners: Sequence[ListenerT | ListenerFactoryT] | None = None,
+        global_listener_registry: ListenerRegistry | None = None,
+        event_manager: "EventManager | None" = None,
     ) -> None:
         self._event_manager = event_manager if event_manager else _EVENT_MANAGER
 
         self._local_listeners = local_listeners or []
         self._global_listener_registry = global_listener_registry
 
-        self._component: Optional[ComponentT] = None
-        self._inited_listeners: Optional[List[ListenerT]] = None
+        self._component: ComponentT | None = None
+        self._inited_listeners: list[ListenerT] | None = None
 
     @property
     def as_listener(self) -> ListenerT:
@@ -148,7 +149,7 @@ async def execute_listeners(self, event_handler_name: str, *args, **kwargs) -> N
 
         await asyncio.gather(*listeners_to_wakeup)
 
-    async def _get_or_init_listeners(self) -> List[ListenerT]:
+    async def _get_or_init_listeners(self) -> list[ListenerT]:
         if self._inited_listeners is not None:
             return self._inited_listeners
 
@@ -171,7 +172,7 @@ async def _get_or_init_listeners(self) -> List[ListenerT]:
         return self._inited_listeners
 
 
-def get_default_name(func: Optional[Callable] = None) -> str:
+def get_default_name(func: Callable | None = None) -> str:
     """
     Get the default name of the component based on code context where it's being used
     """