Migrated variable annotations to use native syntax

src/asphalt/core/cli.py

@@ -44,7 +44,7 @@ def run(configfile, unsafe: bool, loop: Optional[str], service: Optional[str]):
     yaml.constructor.add_constructor('!BinaryFile', binary_file_constructor)
 
     # Read the configuration from the supplied YAML files
-    config = {}  # type: Dict[str, Any]
+    config: Dict[str, Any] = {}
     for path in configfile:
         config_data = yaml.load(path)
         assert isinstance(config_data, dict), 'the document root element must be a dictionary'

src/asphalt/core/component.py

@@ -57,7 +57,7 @@ class ContainerComponent(Component):
 
     def __init__(self, components: Dict[str, Optional[Dict[str, Any]]] = None) -> None:
         assert check_argument_types()
-        self.child_components = OrderedDict()  # type: Dict[str, Component]
+        self.child_components: OrderedDict[str, Component] = OrderedDict()
         self.component_configs = components or {}
 
     def add_component(self, alias: str, type: Union[str, type] = None, **config):

src/asphalt/core/context.py

@@ -153,10 +153,10 @@ def __init__(self, parent: 'Context' = None) -> None:
         self._parent = parent
         self._loop = getattr(parent, 'loop', None)
         self._closed = False
-        self._resources = {}  # type: Dict[Tuple[type, str], ResourceContainer]
-        self._resource_factories = {}  # type: Dict[Tuple[type, str], ResourceContainer]
-        self._resource_factories_by_context_attr = {}  # type: Dict[str, ResourceContainer]
-        self._teardown_callbacks = []  # type: List[Tuple[Callable, bool]]
+        self._resources: Dict[Tuple[type, str], ResourceContainer] = {}
+        self._resource_factories: Dict[Tuple[type, str], ResourceContainer] = {}
+        self._resource_factories_by_context_attr: Dict[str, ResourceContainer] = {}
+        self._teardown_callbacks: List[Tuple[Callable, bool]] = []
 
     def __getattr__(self, name):
         # First look for a resource factory in the whole context chain
@@ -177,7 +177,7 @@ def __getattr__(self, name):
     def context_chain(self) -> List['Context']:
         """Return a list of contexts starting from this one, its parent and so on."""
         contexts = []
-        ctx = self  # type: Optional[Context]
+        ctx: Optional[Context] = self
         while ctx is not None:
             contexts.append(ctx)
             ctx = ctx.parent
@@ -364,7 +364,7 @@ def add_resource_factory(self, factory_callback: factory_callback_type,
             raise ValueError('"types" must not be empty')
 
         if isinstance(types, type):
-            resource_types = (types,)  # type: Tuple[type, ...]
+            resource_types: Tuple[type, ...] = (types,)
         else:
             resource_types = tuple(types)
 
@@ -432,10 +432,11 @@ def get_resources(self, type: Type[T_Resource]) -> Set[T_Resource]:
         assert check_argument_types()
 
         # Collect all the matching resources from this context
-        resources = {container.name: container.value_or_factory
-                     for container in self._resources.values()
-                     if not container.is_factory and type in container.types
-                     }  # type: Dict[str, T_Resource]
+        resources: Dict[str, T_Resource] = {
+            container.name: container.value_or_factory
+            for container in self._resources.values()
+            if not container.is_factory and type in container.types
+        }
 
         # Next, find all matching resource factories in the context chain and generate resources
         resources.update({container.name: container.generate_value(self)

src/asphalt/core/event.py

@@ -80,10 +80,10 @@ def __init__(self, event_class: Type[T_Event], *, source: Any = None,
         self.topic = topic
         if source is not None:
             self.source = weakref.ref(source)
-            self.listeners = None  # type: Optional[List[Callable]]
+            self.listeners: Optional[List[Callable]] = None
         else:
             assert issubclass(event_class, Event), 'event_class must be a subclass of Event'
-            self.bound_signals = WeakKeyDictionary()  # type: MutableMapping[Any, 'Signal']
+            self.bound_signals: MutableMapping[Any, 'Signal'] = WeakKeyDictionary()
 
     def __get__(self, instance, owner) -> 'Signal':
         if instance is None:
@@ -261,7 +261,7 @@ def cleanup():
             queue = None
 
     assert check_argument_types()
-    queue = Queue(max_queue_size)  # type: Queue[T_Event]
+    queue: Queue[T_Event] = Queue(max_queue_size)
     for signal in signals:
         signal.connect(queue.put_nowait)
 

src/asphalt/core/runner.py

@@ -94,7 +94,7 @@ def run_application(component: Union[Component, Dict[str, Any]], *, event_loop_p
 
         logger.info('Starting application')
         context = Context()
-        exception = None  # type: Optional[BaseException]
+        exception: Optional[BaseException] = None
         exit_code = 0
 
         # Start the root component