How to update UI before slow action?

[nnmm]

If some slow function is being called on button press like this:

from textual.app import App, ComposeResult
from textual.widgets import Button, Static


class MyApp(App):

    CSS = '.hidden { display: none }'

    def compose(self) -> ComposeResult:
        yield Button('Do something slow', variant='success', id='btn')
        yield Static('Please be patient', classes='hidden', id='msg1')
        yield Static('Done', classes='hidden', id='msg2')

    async def on_button_pressed(self, event: Button.Pressed) -> None:
        self.query_one('#msg1').remove_class('hidden')
        self.slow_func()
        self.query_one('#msg2').remove_class('hidden')

    def slow_func(self):
        import time
        time.sleep(3)

if __name__ == '__main__':
    MyApp().run()

both messages will not be shown until after on_button_pressed() has returned. Is there a clean way to update the UI before doing something slow?

My current workaround is to do

from textual.app import App, ComposeResult
from textual.widgets import Button, Static
import asyncio


class MyApp(App):

    CSS = '.hidden { display: none }'

    def compose(self) -> ComposeResult:
        yield Button('Do something slow', variant='success', id='btn')
        yield Static('Please be patient', classes='hidden', id='msg1')
        yield Static('Done', classes='hidden', id='msg2')

    async def on_button_pressed(self, event: Button.Pressed) -> None:
        self.query_one('#msg1').remove_class('hidden')
        asyncio.create_task(self.slow_func_background())

    def slow_func(self):
        import time
        time.sleep(3)

    async def slow_func_background(self):
        loop = asyncio.get_running_loop()
        await loop.run_in_executor(None, lambda: self.slow_func())
        self.query_one('#msg2').remove_class('hidden')

if __name__ == '__main__':
    MyApp().run()

which achieves the desired timing of updates, but any exceptions raised in slow_func() get swallowed, the UI just hangs instead of showing the backtrace. This makes development and debugging difficult. Also, of course the button state goes back to normal after the button press handler returns, which is not what I'd want in this case.

[willmcgugan]

The run_in_executor function returns a asyncio.Future object. To get the result of slow_func and to raise any errors you will need to call future.result()

I think you will need to do something along these lines:

future = await loop.run_in_executor(None, lambda: self.slow_func())
result = future.result()

We are looking in to ways of simplifying concurrency with Textual, to hide such details. In the future we might have some kind of decorator to do this.

[nnmm]

@willmcgugan Thanks – it seems await already takes care of calling result(), but your response put me on the right track. I have this now, which polls the background tasks every second:

from textual.app import App, ComposeResult
from textual.widgets import Button, Static
import asyncio


class MyApp(App):

    CSS = '.hidden { display: none }'

    def on_mount(self) -> None:
        self.timer = self.set_interval(1, self.check_background_tasks, pause=True)
        self.background_tasks = set()

    def compose(self) -> ComposeResult:
        yield Button('Do something slow', variant='success', id='btn')
        yield Static('Please be patient', classes='hidden', id='msg1')
        yield Static('Done', classes='hidden', id='msg2')

    async def on_button_pressed(self, event: Button.Pressed) -> None:
        self.query_one('#msg1').remove_class('hidden')
        task = asyncio.create_task(self.slow_func_background())
        self.background_tasks.add(task)
        self.timer.resume()

    def check_background_tasks(self):
        background_tasks = self.background_tasks
        self.background_tasks = set()
        for task in background_tasks:
            try:
                task.result()
            except asyncio.InvalidStateError:
                # not done yet
                self.background_tasks.add(task)

    def slow_func(self):
        import time
        time.sleep(3)
        raise RuntimeError('oh no')

    async def slow_func_background(self):
        loop = asyncio.get_running_loop()
        result = await loop.run_in_executor(None, lambda: self.slow_func())
        self.query_one('#msg2').remove_class('hidden')

if __name__ == '__main__':
    MyApp().run()