Coverage for src / lilbee / cli / tui / thread_safe.py: 100%

1"""Thread-safe helpers for posting from @work(thread=True) workers to the main thread. 

2 

3Textual's call_from_thread raises OSError when the app's message queue 

4has already been closed during shutdown. Since workers run in daemon 

5threads, they can outlive the app. This module provides a drop-in 

6wrapper that silently drops calls when the app is gone. 

7""" 

8 

9from __future__ import annotations 

10 

11import logging 

12from typing import Any 

13 

14from textual.dom import DOMNode 

15 

16log = logging.getLogger(__name__) 

17 

18 

19def call_from_thread(node: DOMNode, fn: Any, *args: Any, **kwargs: Any) -> None: 

20 """Post *fn* to the main thread via the app. 

21 

22 Drops the call (does not crash the worker) when the target node's app 

23 is no longer reachable, e.g. during shutdown or after a screen was 

24 replaced. Logs at debug so the drop is discoverable without leaking 

25 warning text into the TUI render (textual's log handler routes 

26 stderr into the rendered frame). Long-running workers that must 

27 survive a screen switch should own their state on the app 

28 (TaskBarController pattern in widgets/task_bar.py) rather than 

29 relying on this wrapper. 

30 """ 

31 try: 

32 node.app.call_from_thread(fn, *args, **kwargs) 

33 except Exception as exc: 

34 log.debug( 

35 "call_from_thread dropped %s: %s", 

36 getattr(fn, "__name__", fn), 

37 exc, 

38 )