gtk_async_to_sync.py

#!/usr/bin/env python
import pygtk
import gtk
import gobject
from threading import Event


def async_fetch(callback):
    assert callable(callback)
    gobject.timeout_add(100, lambda: callback('The Result'))


class GtkTest(object):

    def get_result(self):
        async_fetch(callback=self._result_loaded)
        self._loaded_event.wait(1)
        return self._result

    def _result_loaded(self, result):
        print('result loaded')
        self._result = result
        self._loaded_event.set()

    def btn_pressed(self, widget, data=None):
        result = self.get_result()
        if result:
            self._alert('Got result: {}'.format(result), error=False)
        else:
            self._alert('Event wait timed out!', error=True)

    def __init__(self):
        self._loaded_event = Event()
        self._result = None
        self.window = gtk.Dialog(title='GTK Async Test', parent=None, flags=0, buttons=None)
        self.window.set_frame_dimensions(left=1000, right=100, top=10, bottom=10)
        self._add_btn('Run Test', self.btn_pressed)
        self._add_btn('Cancel', gtk.main_quit)
        self.window.show()

    def _add_btn(self, label, callback):
        btn = gtk.Button(label)
        btn.connect("clicked", callback, None)
        self.window.action_area.pack_start(btn, True, True, 0)
        btn.show()

    def _alert(self, message, error=False):
        md = gtk.MessageDialog(self.window,
            type=gtk.MESSAGE_ERROR if error else gtk.MESSAGE_INFO,
            buttons=gtk.BUTTONS_OK)
        md.set_markup(message)
        md.run()
        md.destroy()
        print('Quitting')
        gtk.main_quit()


if __name__ == "__main__":
    pygtk.require('2.0')
    GtkTest()
    gtk.main()