结合使用 call_soon_threadsafe 和 GLib.idle_add 是从 asyncio 安全调用 GTK 代码的正确方法吗，还是有点过度？

Question

我有一个在 asyncio 事件循环中运行的应用程序，我想添加一个与我的 Gnome Shell 交互的功能，特别是发送通知。由于 GTK 有自己的（我理解是不兼容的）事件循环，我决定将其放入Gio.Application后台线程。我的问题是关于从我的 asyncio 事件循环运行“GTK 代码”的适当方法，例如创建通知并从我的应用程序中触发它。

以下面的代码为例，我使用以下代码行来调度“添加通知”方法：

asyncio.get_event_loop().call_soon_threadsafe(GLib.idle_add,
    app.add_notification, "You have a new email")

这“有效”，但我想知道call_soon_threadsafe和是否idle_add都是多余的。如果我简单地调用它也可以有效

app.add_notification("You have a new email")

从我的协程，然而PyGObject 文档表明这是线程不安全的。推荐的安全方法是什么？下面是从 asyncio 协程启动通知的最小概念验证。

import asyncio
import threading

import gi
gi.require_version('Gio', '2.0')
from gi.repository import Gio, GLib


class ThreadsafeEvent(asyncio.Event):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        if self._loop is None:
            self._loop = asyncio.get_event_loop()

    def set(self):
        self._loop.call_soon_threadsafe(super().set)

    def clear(self):
        self._loop.call_soon_threadsafe(super().clear)

class Application(Gio.Application):
    APP_ID = "org.something"

    def __init__(self, activated_event: threading.Event):
        flags = Gio.ApplicationFlags.FLAGS_NONE
        super().__init__(application_id=Application.APP_ID, flags=flags)
        self.activated_event = activated_event
        self.activated = False

    def do_activate(self):
        if not self.activated:
            self.hold()
            self.activated = True
            self.activated_event.set()

    def add_notification(self, message):
        note = Gio.Notification.new(message)
        self.send_notification(None, note)


async def main():
    event = ThreadsafeEvent()
    app = Application(event)

    def start(app: Application):
        app.register()
        app.run()

    thread = threading.Thread(target=start, args=(app,))
    thread.setDaemon(True)
    thread.start()

    try:
        # When the event is set, we know the Application
        # is activated and safe to send notifications through
        await asyncio.wait_for(event.wait(), 10)
    except asyncio.TimeoutError:
        print('Timed out before App received its activation.')

    asyncio.get_event_loop().call_soon_threadsafe(GLib.idle_add,
        app.add_notification, "You have a new email")

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    asyncio.ensure_future(main(), loop=loop)
    loop.run_forever()