How to capture function output?

[Tim-Cao]

I would like to know what's progress of my function.

For example, when I click the submit button. It will call

    def create_test_fix_ticket(self) -> None:
        assigned = self.query_one("#assign-to-me").value
        case_result_id = self.query_one("#case-result-id").value
        label = self.query_one("#add-label").value
        project_key = self.query_one("#project-key").value

        output = create_test_fix_ticket.main(assigned, case_result_id, label, project_key)
        self.query_one(RichLog).write(output)

The create_test_fix_ticket.main will execute for a while

I use print to show the progress in create_test_fix_ticket.py file different with app file

def main(assigned, case_result_id, label, project_key):
    testray_connection = get_testray_connection()

    case_result = get_case_result(testray_connection, case_result_id)

    case_name = get_case_name(case_result)

    case_error = get_case_error(case_result)

    print("Connecting")

    jira_connection = get_jira_connection()

    print("Seaching")

    component = get_relevant_jira_component(case_result, jira_connection, project_key)

    print("Creating")

    issue = add_test_fix_ticket(case_error, case_name, case_result_id, component, jira_connection, project_key)

How to print the progress info on Textual?

[TomJGooding]

Since your function "will execute for a while", for starters I'd recommend having a look at the Workers guide in the docs.

I've seen a few questions like this about redirecting stdout in Textual, so I think this really needs an 'official' answer from a maintainer. Perhaps it might even be worth adding to the FAQ at some point?

After some quick research, here's my attempt at this which uses contextlib.redirect_stdout.

I do not recommend using this, as I have no idea if it is thread safe!

import contextlib
from time import sleep

from textual import work
from textual.app import App, ComposeResult
from textual.widgets import Footer, RichLog


def long_blocking_process() -> None:
    print("Connecting")
    sleep(3)
    print("Searching")
    sleep(3)
    print("Creating")


class OutputLog:
    def __init__(self, log: RichLog) -> None:
        self.log = log

    def write(self, text: str) -> None:
        if text.strip():
            app = self.log.app
            app.call_from_thread(self.log.write, text)

    def flush(self) -> None:
        pass


class WorkerWithRedirectStdoutApp(App):
    BINDINGS = [
        ("space", "run_long_blocking_process", "Run long blocking process"),
    ]

    def compose(self) -> ComposeResult:
        yield RichLog()
        yield Footer()

    @work(exclusive=True, thread=True)
    def action_run_long_blocking_process(self) -> None:
        log = self.query_one(RichLog)
        with contextlib.redirect_stdout(OutputLog(log)):  # type: ignore[type-var]
            long_blocking_process()


if __name__ == "__main__":
    app = WorkerWithRedirectStdoutApp()
    app.run()

[Tim-Cao]

It seems that it is an issue of begin_capture_print

[TomJGooding]

Here's a slightly modified example which hopefully demonstrates what is happening here.

Also notice my original example has a if text.strip() condition. This is because logs in Textual will emit an empty line if asked to write an empty string.

from time import sleep

from textual import on, work
from textual.app import App, ComposeResult, events
from textual.widgets import Footer, RichLog


def long_blocking_process() -> None:
    print("Connecting", end="...")
    sleep(3)
    print("Searching", end="...")
    sleep(3)
    print("Creating", end="...")


class OutputLog(RichLog):
    @on(events.Print)
    def on_print(self, event: events.Print) -> None:
        # Prevent blank lines in the log
        if event.text.strip():
            self.write(event.text)


class WorkerWithRedirectStdoutApp(App):
    BINDINGS = [
        ("space", "run_long_blocking_process", "Run long blocking process"),
    ]

    def compose(self) -> ComposeResult:
        yield OutputLog()
        yield Footer()

    @work(exclusive=True, thread=True)
    def action_run_long_blocking_process(self) -> None:
        log = self.query_one(OutputLog)
        log.begin_capture_print()
        long_blocking_process()
        log.end_capture_print()


if __name__ == "__main__":
    app = WorkerWithRedirectStdoutApp()
    app.run()

[willmcgugan]

I think this may work better with Log rather than RichLog. The Log widget will handle arbitrary writes, rather than single lines.

[TomJGooding]

Thanks Will, yup that works better. Here's the full example with that tweak just for future reference:

from time import sleep

from textual import on, work
from textual.app import App, ComposeResult, events
from textual.widgets import Footer, Log


def long_blocking_process() -> None:
    print("Connecting")
    sleep(3)
    print("Searching")
    sleep(3)
    print("Creating")


class OutputLog(Log):
    @on(events.Print)
    def on_print(self, event: events.Print) -> None:
        self.write(event.text)


class WorkerWithCapturePrintApp(App):
    BINDINGS = [
        ("space", "run_long_blocking_process", "Run long blocking process"),
    ]

    def compose(self) -> ComposeResult:
        yield OutputLog()
        yield Footer()

    @work(exclusive=True, thread=True)
    def action_run_long_blocking_process(self) -> None:
        log = self.query_one(OutputLog)
        log.begin_capture_print()
        long_blocking_process()
        log.end_capture_print()


if __name__ == "__main__":
    app = WorkerWithCapturePrintApp()
    app.run()

[Tim-Cao]

I see. Many thanks @willmcgugan and @TomJGooding