Asynchronous Programming with GLib and PyGObject¶

Asynchronous programming is an essential paradigm for handling tasks like I/O operations, ensuring that your application remains responsive. GLib supports asynchronous operations alongside its synchronous counterparts. These asynchronous functions typically have a _async suffix and execute tasks in the background, invoking a callback or returning a future-like object upon completion or cancellation.

PyGObject offers two primary ways to implement asynchronous programming:

Using Python’s `asyncio` module (available since PyGObject 3.50)
Using callbacks (the traditional approach)

Asynchronous Programming with `asyncio`¶

Attention

Asyncio support for PyGObject is experimental. Feel free to explore its integration, but note that the API is subject to change as potential issues are addressed.

Overview of asyncio Integration¶

PyGObject integrates seamlessly with Python’s asyncio module by providing:

1. An asyncio event loop implementation, enabling normal operation of Python’s asynchronous code. 2. Awaitable objects for Gio’s asynchronous functions, allowing await syntax within Python coroutines.

Event Loop Integration¶

To use the asyncio event loop with GLib, set up the GLib-based event loop policy:

import asyncio
from gi.events import GLibEventLoopPolicy

# Set up the GLib event loop
policy = GLibEventLoopPolicy()
asyncio.set_event_loop_policy(policy)

Now, fetch the event loop and submit tasks:

loop = policy.get_event_loop()

async def do_some_work():
    await asyncio.sleep(2)
    print("Done working!")

task = loop.create_task(do_some_work())

Note

Keep a reference to the tasks you create, as asyncio only maintains weak references to them.

Gio Asynchronous Function Integration¶

If the callback parameter of a Gio asynchronous function is omitted, PyGObject automatically returns an awaitable object (similar to asyncio.Future). This allows you to use await and cancel operations from within a coroutine.

loop = policy.get_event_loop()


async def list_files():
    f = Gio.file_new_for_path("/")
    for info in await f.enumerate_children_async(
        "standard::*", 0, GLib.PRIORITY_DEFAULT
    ):
        print(info.get_display_name())


task = loop.create_task(list_files())

Example: Download Window with Async Feedback¶

Here is a full example illustrating asynchronous programming with asyncio for a download window with async feedback.

import asyncio
import time
import gi

gi.require_version("Gtk", "4.0")
from gi.repository import Gio, GLib, Gtk
from gi.events import GLibEventLoopPolicy


class DownloadWindow(Gtk.ApplicationWindow):

    def __init__(self, *args, **kwargs):
        super().__init__(
            *args,
            **kwargs,
            default_width=500,
            default_height=400,
            title="Async I/O Example",
        )

        self.background_task = None

        self.cancel_button = Gtk.Button(label="Cancel")
        self.cancel_button.connect("clicked", self.on_cancel_clicked)
        self.cancel_button.set_sensitive(False)

        self.start_button = Gtk.Button(label="Load")
        self.start_button.connect("clicked", self.on_start_clicked)

        textview = Gtk.TextView(vexpand=True)
        self.textbuffer = textview.get_buffer()
        scrolled = Gtk.ScrolledWindow()
        scrolled.set_child(textview)

        box = Gtk.Box(
            orientation=Gtk.Orientation.VERTICAL,
            spacing=6,
            margin_start=12,
            margin_end=12,
            margin_top=12,
            margin_bottom=12,
        )
        box.append(self.start_button)
        box.append(self.cancel_button)
        box.append(scrolled)

        self.set_child(box)

    def append_text(self, text):
        iter_ = self.textbuffer.get_end_iter()
        self.textbuffer.insert(iter_, f"[{time.time()}] {text}\n")

    def on_start_clicked(self, button):
        button.set_sensitive(False)
        self.cancel_button.set_sensitive(True)
        self.append_text("Start clicked...")

        self.background_task = asyncio.create_task(self.download())

    def on_cancel_clicked(self, button):
        self.append_text("Cancel clicked...")
        self.background_task.cancel()

    async def download(self):
        file_ = Gio.File.new_for_uri("https://pygobject.gnome.org/")

        try:
            succes, content, etag = await file_.load_contents_async()
        except GLib.GError as e:
            self.append_text(f"Error: {e.message}")
        else:
            content_text = content[:100].decode("utf-8")
            self.append_text(f"Got content: {content_text}...")
        finally:
            self.cancel_button.set_sensitive(False)
            self.start_button.set_sensitive(True)


class Application(Gtk.Application):

    def do_activate(self):
        window = DownloadWindow(application=self)
        window.present()


def main():
    asyncio.set_event_loop_policy(GLibEventLoopPolicy())
    app = Application()
    app.run()


if __name__ == "__main__":
    main()

Key Considerations¶

Async tasks use GLib.PRIORITY_DEFAULT. For background tasks, consider using a lower priority to avoid affecting the responsiveness of your GTK UI. Please see the PRIORITY_* GLib Constants <https://docs.gtk.org/glib/index.html#constants> for other settings.
Prefer starting your application using Gio.Application or Gtk.Application instead of asyncio.run(), which is incompatible.

Asynchronous Programming with Callbacks¶

The traditional callback approach is a robust alternative for asynchronous programming in PyGObject. Consider this example of downloading a web page and displaying its source in a text field. The operation can also be canceled mid-execution.

import time
import gi

gi.require_version("Gtk", "4.0")
from gi.repository import Gio, GLib, Gtk


class DownloadWindow(Gtk.ApplicationWindow):

    def __init__(self, *args, **kwargs):
        super().__init__(
            *args,
            **kwargs,
            default_width=500,
            default_height=400,
            title="Async I/O Example",
        )

        self.cancellable = Gio.Cancellable()

        self.cancel_button = Gtk.Button(label="Cancel")
        self.cancel_button.connect("clicked", self.on_cancel_clicked)
        self.cancel_button.set_sensitive(False)

        self.start_button = Gtk.Button(label="Load")
        self.start_button.connect("clicked", self.on_start_clicked)

        textview = Gtk.TextView(vexpand=True)
        self.textbuffer = textview.get_buffer()
        scrolled = Gtk.ScrolledWindow()
        scrolled.set_child(textview)

        box = Gtk.Box(
            orientation=Gtk.Orientation.VERTICAL,
            spacing=6,
            margin_start=12,
            margin_end=12,
            margin_top=12,
            margin_bottom=12,
        )
        box.append(self.start_button)
        box.append(self.cancel_button)
        box.append(scrolled)

        self.set_child(box)

    def append_text(self, text):
        iter_ = self.textbuffer.get_end_iter()
        self.textbuffer.insert(iter_, f"[{time.time()}] {text}\n")

    def on_start_clicked(self, button):
        button.set_sensitive(False)
        self.cancel_button.set_sensitive(True)
        self.append_text("Start clicked...")

        file_ = Gio.File.new_for_uri("https://pygobject.gnome.org/")
        file_.load_contents_async(self.cancellable, self.on_ready_callback, None)

    def on_cancel_clicked(self, button):
        self.append_text("Cancel clicked...")
        self.cancellable.cancel()

    def on_ready_callback(self, source_object, result, user_data):
        try:
            succes, content, etag = source_object.load_contents_finish(result)
        except GLib.GError as e:
            self.append_text(f"Error: {e.message}")
        else:
            content_text = content[:100].decode("utf-8")
            self.append_text(f"Got content: {content_text}...")
        finally:
            self.cancellable.reset()
            self.cancel_button.set_sensitive(False)
            self.start_button.set_sensitive(True)


class Application(Gtk.Application):

    def do_activate(self):
        window = DownloadWindow(application=self)
        window.present()


def main():
    app = Application()
    app.run()


if __name__ == "__main__":
    main()

Synchronous Comparison¶

Before exploring the asynchronous method, let’s review the simpler blocking approach:

file = Gio.File.new_for_uri(
    "https://developer.gnome.org/documentation/tutorials/beginners.html"
)
try:
    status, contents, etag_out = file.load_contents(None)
except GLib.GError:
    print("Error!")
else:
    print(contents)

Asynchronous Workflow with Callbacks¶

For asynchronous tasks, you’ll need:

A Gio.Cancellable to allow task cancellation.
A Gio.AsyncReadyCallback function to handle the result upon task completion.

The example setup includes:

Start Button: The handler calls Gio.File.load_contents_async() with a cancellable and a callback function.
Cancel Button: The handler calls Gio.Cancellable.cancel() to stop the operation.

Once the operation completes—whether successfully, due to an error, or because it was canceled—the on_ready_callback() function is invoked. This callback receives two arguments: the Gio.File instance and a Gio.AsyncResult instance containing the operation’s result.

To retrieve the result, call Gio.File.load_contents_finish(). This method behaves like Gio.File.load_contents(), but since the operation has already completed, it returns immediately without blocking.

After handling the result, call Gio.Cancellable.reset() to prepare the Gio.Cancellable for reuse in future operations. This ensures that the “Load” button can be clicked again to initiate another task. The application enforces that only one operation is active at a time by disabling the “Load” button during an ongoing task using Gtk.Widget.set_sensitive().

Asynchronous Programming with GLib and PyGObject¶

Asynchronous Programming with asyncio¶

Overview of asyncio Integration¶

Event Loop Integration¶

Gio Asynchronous Function Integration¶

Example: Download Window with Async Feedback¶

Key Considerations¶

Asynchronous Programming with Callbacks¶

Synchronous Comparison¶

Asynchronous Workflow with Callbacks¶

Asynchronous Programming with `asyncio`¶