Deliver ContextManager to child widgets

[iscas-zac]

Hello, I'm writing a TUI module submitting and getting messages from other modules. I have been trying to register the message exchange function in some main module dynamically, like

from tui import TextualApp
from contexts import message_exchange_function
# if name is main
app = TextualApp()
app.register_on_input_submitted(lambda input: message_exchange_function(input))
app.run()

I've consulted docs of worker and skimmed some issues, but they only gave examples when the function is a class/instance method. Is this the only way recommended? Is there some way to decouple the tui and contexts module somewhat?

Thanks!

[davep]

It's not very clear from your question what it is you actually want to do, or what the connection is to workers. Do you think you could try rewording your question and adding some more detail?

[iscas-zac]

OK, the message_exchange_function is handling a game's TCP connection, and it will retrieve response from the game server from time to time. I write the function in a module named contexts, and want to make the function be invoked as reaction to some input events.

In my naive model, I want to

• init a TCP connection on app's mount,
• close it on app's unmount,
• send client messages through the connection on my input
• and receive server messages periodically.

I wrap the functionality in some other modules, but can't figure out how to make incoming messages trigger textual app's update and how to call outside functions in the app's lifecycle.

I'm just learning asynchronous programming, and not knowing what the best practice is like.
Maybe there is a better idiom to do this, and if so, please give me some links to show that.

(If you see previous responds by another account, please don't be confused. It's me, using a wrong account)

[davep]

Unfortunately I'm struggling to better understand the question based on that; hopefully someone who can comprehend what you're asking will see this.

[iscas-zac]

Hello again @davep and very sorry for disturbance as well as any confusion in my amateurish expression. I'm actually thinking about something like such a situation:

class TaskPanel(Widget):
    def __init__(
        self,
        logger=None,
        *,
        name: str | None = None,
        id: str | None = None,
        classes: str | None = None,
        disabled: bool = False,
    ) -> None:
        super().__init__(name=name, id=id, classes=classes, disabled=disabled)
        self.logger = logger
        self.input_queue = []

    def on_exit(self):
        self.logger.__exit__()
    ...

class Tabb(App):
    def compose(self) -> ComposeResult:
        yield TabbedContent()
        yield Input()

    @on(Input.Submitted)
    def on_input_message_submitted(self, event: Input.Submitted) -> None:
        if event.value == "aaa":
            self.new_tab()

    def on_mount(self):
        self.logger = open("test.txt", "w")

    def new_tab(self):
        pane = TabPane("Front Door", id="front_door")
        test_conn = open("aaa.txt", "w")
        pane.mount(TaskPanel(logger=self.logger))
        tabbed = self.query_one(TabbedContent)
        tabbed.add_pane(pane)
        tabbed.active = "front_door"

, in which I try to mount something dynamically onto a tab with a ContextManager like a TextIOWrapper or a TCP connection. I want it to handle the context just like in a with statement, where it calls __exit__() when encountering an error or exiting normally. Is it enough to just put the __exit__() inside the on_exit() method of the widget?

[davep]

There is no Exit event in textual so I'm unclear what on_exit would be in the above code.

[iscas-zac]

Ok, so if I want to handle the context correctly when encountering an error or exiting normally like in a with statement, how can I write the code?

[davep]

You appear to be wanting to close a file you've opened for writing at the app level, it's not clear to me why you wouldn't simply call close on the resulting handle. As for how and where you'd do that: it's also unclear to me why you'd open it at the app level, but then want to close it down at an individual widget level.

I'm still not really clear what it is you're asking or why the mention of context managers and with; but given the code you've shown, suggesting you want a resource open for the lifetime of the application and you want to tidy up when you leave, honestly I think I'd personally create/clean outside of the application. For example:

from io import TextIOWrapper
from textual import on
from textual.app import App, ComposeResult
from textual.widgets import Button

class OpenCloseFileApp(App[None]):

    def __init__(self, my_file: TextIOWrapper) -> None:
        super().__init__()
        self.my_file = my_file

    def compose(self) -> ComposeResult:
        yield Button("Write a thing and close me")

    @on(Button.Pressed)
    def write_and_close(self) -> None:
        self.my_file.write("And we're done!\n")
        self.exit()

if __name__ == "__main__":
    with open("foo.txt", "a", encoding="utf-8") as my_file:
        OpenCloseFileApp(my_file).run()

[iscas-zac]

OK, I see, so if I would open a file in an app, I should close it as the app exits.

A last question (though a little beside the point) if it wouldn't bother you too much: is there any convenient builtin way to output the logs to a file? I use the from textual import log way to print messages in the console, but the console would not keep long history.

[iscas-zac]

This question was opened when I was confused with workers and didn't understand wait syntax. Later I was confused how to pass arguments to apps and widgets until I came across app init FAQ and its peers.

Apologies again for my amateurish expressions. I am improving my understanding for the textual project, and hopefully getting some insight for how to write better code patterns. Before that, I think it better to mark this question as answered. Thanks for your patience after all.