imprv: asyncio-related stubs in events and Gio

src/gi-stubs/events.pyi

@@ -1,31 +1,74 @@
-from typing import Any
 from typing_extensions import Self
+from typing_extensions import TypeVar
+from typing_extensions import TypeVarTuple
+from typing_extensions import Unpack
 
+import asyncio
+import sys
 from collections.abc import Callable
-from signal import Signals
+from collections.abc import Generator
+from contextlib import contextmanager
+from types import TracebackType
 
 from gi.repository import GLib
 
-class GLibTask:
-    def __init__(self, *args: object, **kwargs: object) -> None: ...
+if sys.version_info >= (3, 14):
+    from asyncio.events import _AbstractEventLoopPolicy
+else:
+    from asyncio.events import AbstractEventLoopPolicy as _AbstractEventLoopPolicy
+
+_T_co = TypeVar("_T_co", covariant=True)
+_Ts = TypeVarTuple("_Ts")
+
+class GLibTask(asyncio.Task[_T_co]):
     def set_priority(self, priority: int) -> None: ...
     def get_priority(self) -> int: ...
 
-class GLibEventLoop:
-    def __init__(self, main_context: GLib.MainContext) -> None: ...
-    def add_signal_handler(
-        self,
-        sig: Signals,
-        callback: Callable[..., Any],
-        *args: object,
-    ) -> None: ...
-    def remove_signal_handler(self, sig: Signals) -> bool: ...
-    def close(self) -> None: ...
+class _GLibEventLoopMixin:
+    def __init__(self, main_context: GLib.MainContext | None) -> None: ...
+    @contextmanager
+    def paused(self) -> Generator[None]: ...
+    @contextmanager
+    def running(self, quit_func: Callable[[], object]) -> Generator[None]: ...
+    def stop(self) -> None: ...
+    def time(self) -> float: ...
+
+class _GLibEventLoopRunMixin:
+    def run_forever(self) -> None: ...
 
-class GLibEventLoopPolicy:
+if sys.platform == "win32":
+    class GLibEventLoop(
+        _GLibEventLoopMixin, _GLibEventLoopRunMixin, asyncio.ProactorEventLoop
+    ):
+        def __init__(self, main_context: GLib.MainContext | None = None) -> None: ...
+
+else:
+    class GLibEventLoop(
+        _GLibEventLoopMixin, _GLibEventLoopRunMixin, asyncio.SelectorEventLoop
+    ):
+        def __init__(self, main_context: GLib.MainContext | None = None) -> None: ...
+        def add_signal_handler(
+            self,
+            sig: int,
+            callback: Callable[[Unpack[_Ts]], object],
+            *args: Unpack[_Ts],
+        ) -> None: ...
+        def remove_signal_handler(self, sig: int) -> bool: ...
+        def close(self) -> None: ...
+
+class GLibEventLoopPolicy(_AbstractEventLoopPolicy):
     def __init__(self) -> None: ...
     def __enter__(self) -> Self: ...
-    def __exit__(self, exc_type, exc_value, traceback) -> None: ...
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        traceback: TracebackType | None,
+    ) -> None: ...
     def get_event_loop(self) -> GLibEventLoop: ...
     def get_event_loop_for_context(self, ctx: GLib.MainContext) -> GLibEventLoop: ...
-    def set_event_loop(self, loop: GLibEventLoop) -> None: ...
+    def set_event_loop(self, loop: asyncio.AbstractEventLoop | None) -> None: ...
+    def new_event_loop(self) -> GLibEventLoop: ...
+
+    if sys.platform != "win32" and sys.version_info < (3, 12):
+        def get_child_watcher(self) -> asyncio.AbstractChildWatcher: ...

src/gi-stubs/repository/Gio.pyi

@@ -12,6 +12,7 @@ from typing_extensions import TypeVar
 from typing_extensions import TypeVarTuple
 from typing_extensions import Unpack
 
+import asyncio
 from collections.abc import Callable
 from collections.abc import Iterator
 from collections.abc import Sequence
@@ -21,6 +22,7 @@ from gi.repository import GioUnix
 from gi.repository import GLib
 from gi.repository import GObject
 
+_T = TypeVar("_T")
 ObjectItemType = TypeVar("ObjectItemType", bound=GObject.Object, default=Any)
 ObjectPropsItemType = TypeVar("ObjectPropsItemType", bound=GObject.Object, default=Any)
 _DataTs = TypeVarTuple("_DataTs", default=Unpack[tuple[()]])
@@ -1230,7 +1232,8 @@ class Application(GObject.Object, ActionGroup, ActionMap):
     def add_main_option_entries(self, entries: Sequence[GLib.OptionEntry]) -> None: ...
     def add_option_group(self, group: GLib.OptionGroup) -> None: ...
     def bind_busy_property(self, object: GObject.Object, property: str) -> None: ...
-    def create_asyncio_task(self, coro):
+    # override
+    def create_asyncio_task(self, coro: asyncio._CoroutineLike[_T]) -> asyncio.Task[_T]:
         """
         Safely create an asyncio task. The application will not quit until the
         task completes. For potentially longer running tasks, you should add
@@ -1245,7 +1248,7 @@ class Application(GObject.Object, ActionGroup, ActionMap):
         You can deal with this by either only storing a weak reference to the
         Task, by explicitly collecting the result, or by only cancelling it if
         it is not done already.
-        """  # FIXME: Override is missing typing annotation
+        """
     def do_activate(self) -> None: ...
     def do_add_platform_data(self, builder: GLib.VariantBuilder, /) -> None: ...
     def do_after_emit(self, platform_data: GLib.Variant, /) -> None: ...
@@ -1293,7 +1296,7 @@ class Application(GObject.Object, ActionGroup, ActionMap):
     def register(self, cancellable: Cancellable | None = None) -> bool: ...
     def release(self) -> None: ...
     # override
-    def run(self, argv: list[str] | None) -> int: ...
+    def run(self, argv: list[str] | None = None) -> int: ...
     def send_notification(self, id: str | None, notification: Notification) -> None: ...
     def set_action_group(self, action_group: ActionGroup | None = None) -> None: ...
     def set_application_id(self, application_id: str | None = None) -> None: ...