Fixed mypy errors

src/asphalt/core/component.py

@@ -58,13 +58,15 @@ class ContainerComponent(Component):
 
     __slots__ = "child_components", "component_configs"
 
-    def __init__(self, components: Dict[str, Optional[Dict[str, Any]]] = None) -> None:
+    def __init__(
+        self, components: Optional[Dict[str, Optional[Dict[str, Any]]]] = None
+    ) -> None:
         assert check_argument_types()
         self.child_components: OrderedDict[str, Component] = OrderedDict()
         self.component_configs = components or {}
 
     def add_component(
-        self, alias: str, type: Union[str, Type] = None, **config
+        self, alias: str, type: Union[str, Type, None] = None, **config
     ) -> None:
         """
         Add a child component.

src/asphalt/core/concurrent.py

@@ -81,6 +81,7 @@ async def request_handler(ctx):
         instance or the name of an :class:`~concurrent.futures.Executor` resource
 
     """
+    executor: Executor | str | None = None
 
     def outer(
         func: Callable[Concatenate[Context, P], T_Retval]
@@ -109,7 +110,6 @@ def wrapper(
         ), "Cannot wrap coroutine functions to be run in an executor"
         return wraps(func)(wrapper)
 
-    executor: Executor | str | None = None
     if isinstance(func_or_executor, (str, Executor)):
         executor = func_or_executor
         return outer

src/asphalt/core/context.py

@@ -784,7 +784,7 @@ def call_in_executor(
         self,
         func: Callable[..., T_Retval],
         *args,
-        executor: Union[Executor, str] = None,
+        executor: Union[Executor, str, None] = None,
         **kwargs,
     ) -> Awaitable[T_Retval]:
         """
@@ -1091,7 +1091,7 @@ def resolve_forward_refs() -> None:
         forward_refs_resolved = True
 
     @wraps(func)
-    def sync_wrapper(*args, **kwargs) -> T_Retval:
+    def sync_wrapper(*args, **kwargs) -> Any:
         if not forward_refs_resolved:
             resolve_forward_refs()
 
@@ -1108,7 +1108,7 @@ def sync_wrapper(*args, **kwargs) -> T_Retval:
         return func(*args, **kwargs, **resources)
 
     @wraps(func)
-    async def async_wrapper(*args, **kwargs) -> T_Retval:
+    async def async_wrapper(*args, **kwargs) -> Any:
         if not forward_refs_resolved:
             resolve_forward_refs()
 

src/asphalt/core/event.py

@@ -249,21 +249,24 @@ def dispatch(self, *args, **kwargs) -> Awaitable[bool]:
         return self.dispatch_raw(event)
 
     def wait_event(
-        self, filter: Callable[[T_Event], bool] = None
+        self, filter: Callable[[T_Event], bool] | None = None
     ) -> Awaitable[T_Event]:
         """Shortcut for calling :func:`wait_event` with this signal in the first argument."""
         return wait_event([self], filter)
 
     def stream_events(
-        self, filter: Callable[[T_Event], bool] = None, *, max_queue_size: int = 0
+        self,
+        filter: Callable[[T_Event], bool] | None = None,
+        *,
+        max_queue_size: int = 0,
     ) -> AsyncIterator[T_Event]:
         """Shortcut for calling :func:`stream_events` with this signal in the first argument."""
         return stream_events([self], filter, max_queue_size=max_queue_size)
 
 
 def stream_events(
     signals: Sequence[Signal[T_Event]],
-    filter: Callable[[T_Event], bool] = None,
+    filter: Optional[Callable[[T_Event], bool]] = None,
     *,
     max_queue_size: int = 0,
 ) -> AsyncIterator[T_Event]:
@@ -309,7 +312,8 @@ def cleanup() -> None:
 
 
 async def wait_event(
-    signals: Sequence[Signal[T_Event]], filter: Callable[[T_Event], bool] = None
+    signals: Sequence[Signal[T_Event]],
+    filter: Optional[Callable[[T_Event], bool]] = None,
 ) -> T_Event:
     """
     Wait until any of the given signals dispatches an event that satisfies the filter (if any).
@@ -322,8 +326,6 @@ async def wait_event(
     :return: the event that was dispatched
 
     """
-    if sys.version_info >= (3, 5, 3):
-        assert check_argument_types()
-
+    assert check_argument_types()
     async with aclosing(stream_events(signals, filter)) as events:
         return await events.asend(None)

src/asphalt/core/runner.py

@@ -100,7 +100,7 @@ def run_application(
         logger.info("Starting application")
         context = Context()
         exception: Optional[BaseException] = None
-        exit_code = 0
+        exit_code: str | int = 0
 
         # Start the root component
         token = _current_context.set(context)
@@ -134,7 +134,7 @@ def run_application(
                 except KeyboardInterrupt:
                     pass
                 except SystemExit as e:
-                    exit_code = e.code
+                    exit_code = e.code or 0
 
             # Close the root context
             logger.info("Stopping application")

src/asphalt/core/utils.py

@@ -126,7 +126,7 @@ class PluginContainer:
 
     __slots__ = "namespace", "base_class", "_entrypoints"
 
-    def __init__(self, namespace: str, base_class: type = None) -> None:
+    def __init__(self, namespace: str, base_class: type | None = None) -> None:
         self.namespace = namespace
         self.base_class = base_class
         group = entry_points().select(group=namespace)  # type: ignore[attr-defined]