Advice on working with ListView and ListItem

[MuongKimhong]

What I'm doing:
I have an Input. every key press, I make requests to server for some data and update my ListView

Currently, I'm doing it this way and it's a bit unresponsive when response data has length of 20 or 30.

  def on_input_changed(self, event: Input.Changed) -> None:
      self.search(event.value)

  def search(self, search_text: str) -> None:
      data_list_view = self.app.query_one("#data-list-view")
      data_list_view.clear()

      if search_text.strip() != "":
          res = requests.get(...) # make requests
          list_items = [CustomListItem(data["key_one"], data["key_two"]) for data in res["data"]]
          data_list_view.extend(list_items)

I tried https://textual.textualize.io/guide/workers/#work-decorator, but i dont see any changes in performance.

[TomJGooding]

Notce that the example uses an async API like httpx, whereas requests is blocking. Perhaps you should have another read over the docs, in particular for the difference between async and thread workers.

[MuongKimhong]

When I tried thread worker, I always ran into Error when I try to update data_list_view by calling self.app.call_from_thread(data_list_view.extend(list_items))

[TomJGooding]

Could you show the code where you tried to use a thread worker? And the exact error you were getting?

[davep]

@MuongKimhong You're calling call_from_thread with the result of calling ListView.extend, what you should be doing is passing a reference to the function you want to call. For example:

self.app.call_from_thread(data_list_view.extend, list_items)

[MuongKimhong]

Textual and its community are just Awesome! Thanks you.

For future reference, here is the working code

from textual.worker import get_current_worker

async def on_input_changed(self, event: Input.Changed) -> None:
    self.search(event.value)

@work(exclusive=True, thread=True)
def search(self, search_text: str) -> None:
    worker = get_current_worker()
    data_list_view = self.app.query_one("#data-list-view")

    if not worker.is_cancelled:
        self.app.call_from_thread(data_list_view.clear)

    if search_text.strip() != "":
        res = requests.get(...) # make requests
        list_items = [CustomListItem(data["key_one"], data["key_two"]) for data in res["data"]]

        if not worker.is_cancalled:
            self.app.call_from_thread(data_list_view.extend, list_items)