Updated annotations to use Python 3.10 style where possible

.pre-commit-config.yaml

@@ -19,7 +19,7 @@ repos:
     rev: v2.31.1
     hooks:
       - id: pyupgrade
-        args: [ "--py37-plus" ]
+        args: [ "--py37-plus", "--keep-runtime-typing" ]
 
   - repo: https://github.com/psf/black
     rev: 22.3.0

src/asphalt/core/cli.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import os
 from pathlib import Path
 from typing import Any, Dict, Optional

src/asphalt/core/component.py

@@ -1,18 +1,20 @@
+from __future__ import annotations
+
+__all__ = ("Component", "ContainerComponent", "CLIApplicationComponent")
+
 import asyncio
 import sys
 from abc import ABCMeta, abstractmethod
 from collections import OrderedDict
 from traceback import print_exception
-from typing import Any, Dict, Optional, Union
+from typing import Any, Dict, Optional, Type, Union
 from warnings import warn
 
 from typeguard import check_argument_types
 
 from asphalt.core.context import Context
 from asphalt.core.utils import PluginContainer, merge_config, qualified_name
 
-__all__ = ("Component", "ContainerComponent", "CLIApplicationComponent")
-
 
 class Component(metaclass=ABCMeta):
     """This is the base class for all Asphalt components."""
@@ -60,7 +62,7 @@ def __init__(self, components: Dict[str, Optional[Dict[str, Any]]] = None) -> No
         self.child_components: OrderedDict[str, Component] = OrderedDict()
         self.component_configs = components or {}
 
-    def add_component(self, alias: str, type: Union[str, type] = None, **config):
+    def add_component(self, alias: str, type: Union[str, Type] = None, **config):
         """
         Add a child component.
 

src/asphalt/core/concurrent.py

@@ -1,3 +1,7 @@
+from __future__ import annotations
+
+__all__ = ("executor",)
+
 import inspect
 from asyncio import get_running_loop
 from concurrent.futures import Executor
@@ -8,8 +12,6 @@
 
 from asphalt.core import Context
 
-__all__ = ("executor",)
-
 T_Retval = TypeVar("T_Retval")
 WrappedCallable = Callable[..., Awaitable[T_Retval]]
 

src/asphalt/core/context.py

@@ -1,3 +1,15 @@
+from __future__ import annotations
+
+__all__ = (
+    "ResourceEvent",
+    "ResourceConflict",
+    "ResourceNotFound",
+    "TeardownError",
+    "Context",
+    "executor",
+    "context_teardown",
+)
+
 import logging
 import re
 import warnings
@@ -33,16 +45,6 @@
 from asphalt.core.event import Event, Signal, wait_event
 from asphalt.core.utils import callable_name, qualified_name
 
-__all__ = (
-    "ResourceEvent",
-    "ResourceConflict",
-    "ResourceNotFound",
-    "TeardownError",
-    "Context",
-    "executor",
-    "context_teardown",
-)
-
 logger = logging.getLogger(__name__)
 factory_callback_type = Callable[["Context"], Any]
 resource_name_re = re.compile(r"\w+")
@@ -77,7 +79,7 @@ def __init__(
         self.context_attr = context_attr
         self.is_factory = is_factory
 
-    def generate_value(self, ctx: "Context"):
+    def generate_value(self, ctx: Context):
         assert self.is_factory, "generate_value() only works for resource factories"
         value = self.value_or_factory(ctx)
 
@@ -122,7 +124,7 @@ class ResourceEvent(Event):
 
     def __init__(
         self,
-        source: "Context",
+        source: Context,
         topic: str,
         types: Tuple[type, ...],
         name: str,
@@ -196,7 +198,7 @@ class Context:
 
     resource_added = Signal(ResourceEvent)
 
-    def __init__(self, parent: "Context" = None) -> None:
+    def __init__(self, parent: Context = None) -> None:
         assert check_argument_types()
         self._parent = parent
         self._loop = getattr(parent, "loop", None)
@@ -222,7 +224,7 @@ def __getattr__(self, name):
         raise AttributeError(f"no such context variable: {name}")
 
     @property
-    def context_chain(self) -> List["Context"]:
+    def context_chain(self) -> List[Context]:
         """Return a list of contexts starting from this one, its parent and so on."""
         contexts = []
         ctx: Optional[Context] = self
@@ -241,7 +243,7 @@ def loop(self) -> AbstractEventLoop:
         return self._loop
 
     @property
-    def parent(self) -> Optional["Context"]:
+    def parent(self) -> Optional[Context]:
         """Return the parent context, or ``None`` if there is no parent."""
         return self._parent
 

src/asphalt/core/event.py

@@ -1,3 +1,7 @@
+from __future__ import annotations
+
+__all__ = ("Event", "Signal", "wait_event", "stream_events")
+
 import logging
 import sys
 import weakref
@@ -30,8 +34,6 @@
 else:
     from async_generator import aclosing
 
-__all__ = ("Event", "Signal", "wait_event", "stream_events")
-
 logger = logging.getLogger(__name__)
 
 
@@ -102,9 +104,9 @@ def __init__(
             assert issubclass(
                 event_class, Event
             ), "event_class must be a subclass of Event"
-            self.bound_signals: MutableMapping[Any, "Signal"] = WeakKeyDictionary()
+            self.bound_signals: MutableMapping[Any, Signal] = WeakKeyDictionary()
 
-    def __get__(self, instance, owner) -> "Signal":
+    def __get__(self, instance, owner) -> Signal:
         if instance is None:
             return self
 

src/asphalt/core/runner.py

@@ -1,3 +1,7 @@
+from __future__ import annotations
+
+__all__ = ("run_application",)
+
 import asyncio
 import signal
 import sys
@@ -13,8 +17,6 @@
 from asphalt.core.context import Context
 from asphalt.core.utils import PluginContainer, qualified_name
 
-__all__ = ("run_application",)
-
 policies = PluginContainer("asphalt.core.event_loop_policies")
 
 
@@ -30,7 +32,7 @@ def run_application(
     event_loop_policy: str = None,
     max_threads: int = None,
     logging: Union[Dict[str, Any], int, None] = INFO,
-    start_timeout: Union[int, float, None] = 10
+    start_timeout: Union[int, float, None] = 10,
 ):
     """
     Configure logging and start the given root component in the default asyncio event loop.

src/asphalt/core/utils.py

@@ -1,7 +1,17 @@
+from __future__ import annotations
+
+__all__ = (
+    "resolve_reference",
+    "qualified_name",
+    "callable_name",
+    "merge_config",
+    "PluginContainer",
+)
+
 import sys
 from importlib import import_module
 from inspect import isclass
-from typing import Any, Callable, Dict, List, Optional, Union
+from typing import Any, Callable, Dict, List, Optional, Type, Union
 
 from typeguard import check_argument_types
 
@@ -10,14 +20,6 @@
 else:
     from importlib_metadata import EntryPoint, entry_points
 
-__all__ = (
-    "resolve_reference",
-    "qualified_name",
-    "callable_name",
-    "merge_config",
-    "PluginContainer",
-)
-
 
 def resolve_reference(ref):
     """
@@ -157,7 +159,7 @@ def resolve(self, obj):
 
         return value
 
-    def create_object(self, type: Union[type, str], **constructor_kwargs):
+    def create_object(self, type: Union[Type, str], **constructor_kwargs):
         """
         Instantiate a plugin.