An example of asynchronous programming with TraitsUI using QT5 backend with Quamash event loop to work with asyncio.

test_traitsui_quamash.py

"""
This code was helpful,
https://stackoverflow.com/questions/32141623/pyqt5-and-asyncio-yield-from-never-finishes

"""
import os
import sys
import time
import quamash
import asyncio
import traceback
import PyQt5
import random
from concurrent.futures import ProcessPoolExecutor
from functools import partial, wraps



# Select QT5
# https://github.com/enthought/traitsui/issues/407
os.environ['ETS_TOOLKIT'] = 'qt'
os.environ['QT_API'] = 'pyqt5'
from traits.api import HasTraits, Button, Instance, Int, Bool, on_trait_change
from traitsui.api import ModelView, View, Item, Label
from pyface.api import error as pyface_error


def trait_change_coroutine(f):
    """A wrapper that lets coroutines behave as trait-change handlers.
    """
    def log_error(future):
        try:
            future.result()
        except Exception:
            traceback.print_exc()
            pyface_error(None, "IT BROKE!!")

    @wraps(f)
    def wrapper(self):
        future = loop.create_task(f(self))
        future.add_done_callback(log_error)

    return wrapper


class App(HasTraits):
    """ Traits application model.
    """

    running = Bool(False)

    async def test_task(self):
        """ Test task that takes 1s to complete.
        """
        self.running = True

        print('Running task.')
        try:
            await asyncio.sleep(1.0)

            if random.randint(0, 1):
                raise ValueError("FAIL")

        finally:
            print('Task finished.')
            self.running = False


def sub_task():
    time.sleep(2.)
    return random.randint(1, 10)

class AppView(ModelView):
    """ Traits application view.
    """

    model = Instance(App)

    test_task_btn = Button('Test Task')

    subprocess_task_btn = Button('Subprocess task')

    subprocess_active = Bool(False)

    subprocess_result = Int

    def close(self, info, is_ok):
        print('Close')
        return True

    @trait_change_coroutine
    async def _test_task_btn_fired(self):
        await self.model.test_task()

    @on_trait_change("subprocess_task_btn")
    @trait_change_coroutine
    async def _launch_subprocess(self):
        self.subprocess_active = True
        try:
            result = await loop.run_in_executor(pool, sub_task)
            self.subprocess_result = result
        finally:
            self.subprocess_active = False

    def default_traits_view(self):
        view = View(
                    Label('Hello, I am a TraitsUI App using the QT event loop with Quamash.'),
                    Item('test_task_btn', show_label=False, enabled_when='not model.running'),
                    Item('subprocess_task_btn', show_label=False, enabled_when='not subprocess_active'),
                    Item('subprocess_result', style="readonly"),
                    resizable=True,
                    title='Async TraitsUI Example'
                    )
        return view

if __name__ == "__main__":
    # Required setup to get event loop working
    app = PyQt5.QtWidgets.QApplication(sys.argv)
    loop = quamash.QEventLoop(app)
    asyncio.set_event_loop(loop)
    loop.set_debug(True)  # optional

    with loop:
        print('Launching app.')
        pool = ProcessPoolExecutor()
        model = App()
        view = AppView(model=model)
        view.edit_traits()
        loop.run_forever()
    print('App closed.')