DataTable dynamic update (several times in one function)

[nickkitovsky]

Tell me a way to update fields in the DataTable within one function (this works great if you hang it, for example, on a Button.Pressed handler), but in one function sequentially.

from time import sleep

from textual import on
from textual.app import App, ComposeResult
from textual.coordinate import Coordinate
from textual.widgets import Button, DataTable


class TableApp(App):
    cnt = 0

    def compose(self) -> ComposeResult:
        yield Button()
        yield DataTable(cursor_type='row')

    def on_mount(self) -> None:
        self.table = self.query_one(DataTable)
        self.table.add_column('A')
        self.table.add_row('0')

    def change_value(self):
        self.table.update_cell_at(
            coordinate=Coordinate(row=0, column=0),
            value=self.cnt,
            update_width=True,
        )
        self.cnt += 1

    @on(message_type=Button.Pressed)
    def input_typing_handler(self, event: Button.Pressed) -> None:
        self.change_value()
        sleep(3)  # some hard work
        self.change_value()
app = TableApp()
if __name__ == '__main__':
    app.run()

I don't see two consecutive changes using the change_value function with a break for some external function, I immediately get the last result after the work is done (speed(3) in the example.
I tried to create custom message class and run

self.post_message(MyCustomMessage()) 
sleep(3)  # some hard work
self.post_message(MyCustomMessage())

But the result was the same.
The same applies to different cells; changes are not applied one by one, but only after the function is completely completed

[davep]

If you're going to be doing non-async-friendly busy work, and you want to update the UI as the work takes place, you should have a read about workers and, in your case, especially thread workers.

[nickkitovsky]

Thank you, it really works like I wanted. I'll leave the function code here so people can quickly resolve their issue.
UPD: Use custom message and handle it, for thread-safety code.

import time

from textual import on, work
from textual.app import App, ComposeResult
from textual.coordinate import Coordinate
from textual.message import Message
from textual.widgets import Button, DataTable


class TableApp(App):
    class HardWorkMessage(Message):
        def __init__(self, cell_coordinates: Coordinate, cell_value: str) -> None:
            self.cell_coordinates = cell_coordinates
            self.cell_value = cell_value
            super().__init__()

    @work(thread=True)
    def do_hardwork(self):
        self.post_message(self.HardWorkMessage(cell_coordinates=Coordinate(0, 0), cell_value='one'))
        time.sleep(1)  # simulate first hard work
        self.post_message(self.HardWorkMessage(cell_coordinates=Coordinate(0, 0), cell_value='two'))
        time.sleep(1)  # simulate second hard work
        self.post_message(self.HardWorkMessage(cell_coordinates=Coordinate(0, 0), cell_value='three'))

    def compose(self) -> ComposeResult:
        yield Button()
        yield DataTable(cursor_type='row')

    def on_mount(self) -> None:
        self.table = self.query_one(DataTable)
        self.table.add_column('A')
        self.table.add_row('zero')

    @on(message_type=HardWorkMessage)
    def set_cell_value(self, message: HardWorkMessage):
        self.table.update_cell_at(
            coordinate=message.cell_coordinates,
            value=message.cell_value,
            update_width=True,
        )

    @on(message_type=Button.Pressed)
    def start_hardwork(self, event: Button.Pressed) -> None:
        self.do_hardwork()


app = TableApp()
if __name__ == '__main__':
    app.run()

[davep]

Close, but I would suggest a re-read of the final paragraph that talks about thread-safety. So in your worker you would either want to post a custom message back to your main code, and handle it there and do the update, or make do_hardwork more like this:

def do_hardwork(self):
    self.call_from_thread(self.set_cell_value, 'one')
    sleep(3)  # simulate first hard work
    self.call_from_thread(self.set_cell_value, 'two')
    sleep(3)  # simulate second hard work
    self.call_from_thread(self.set_cell_value, 'three')