asyncio hell?

[hemna]

Hey guys,
I don't know squat about asyncio, but it sure seems like it requires a doctorate to figure out.

I'm writing a terminal chat app with Textual for ham radio that happens to use the python kiss3 package, which evidently is asyncio. I'm getting all sorts of problems even trying to connect the client to the remote from inside the textual APP, and I have no clue how to get it to work.

I created a sample small app the reproduces the problem below with a screen recording with asciinema to show it's failure.

https://asciinema.org/a/QVmqGAPBPUj09Ny70NZ5zmxlk

# You can run this script using uv
# https://docs.astral.sh/uv/getting-started/installation/

# uv run fail.py

# /// script
# dependencies = [
#   "textual",
#   "kiss3",
#   "ax253"
# ]
# ///

import asyncio
from textual.app import App, ComposeResult
from textual.widgets import Footer, Header

import kiss
from ax253 import Frame


class StopwatchApp(App):
    """A Textual app to manage stopwatches."""

    BINDINGS = [("d", "toggle_dark", "Toggle dark mode")]

    # Create a KISS client so we can tx/rx packets
    # the kiss client is asyncio as well.
    # asyncio is a pain in the ass.
    def on_mount(self) -> None:
        client = kiss.TCPKISS(
                host="192.168.1.21",
                port=8001,
                strip_df_start=True
                )

        # this doesn't work as it fails due to event loop already running
        # raise RuntimeError(                                                                                                                                                                        │
        #│    587 │   │   │   │   'Cannot run the event loop while another loop is running')
        # client.start()
        # same error.
        asyncio.create_task(client.start())

    def compose(self) -> ComposeResult:
        """Create child widgets for the app."""
        yield Header()
        yield Footer()

    def action_toggle_dark(self) -> None:
        """An action to toggle dark mode."""
        self.theme = (
            "textual-dark" if self.theme == "textual-light" else "textual-light"
        )


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

[r-dgreen]

You can place all of your async functionality inside an async function decorated with @work

Inside your app you can call this work function like normal, and Textual will handle adding it to the eventloop. There's a good blog post about Await (Maybe) that describes why this works: https://textual.textualize.io/blog/2023/03/15/no-async-async-with-python/

I've included an example with aiohttp you should be able to easily modify to support your own code.

from textual.app import App, ComposeResult
from textual.widgets import Footer, Header, Label
from textual import work

import aiohttp


class StopwatchApp(App):
    """A Textual app to manage stopwatches."""

    BINDINGS = [("d", "toggle_dark", "Toggle dark mode")]

    def on_mount(self) -> None:
        self.async_hell()
    
    # place all of your async functionality in a decorated work function.
    @work
    async def async_hell(self):
        label = self.query_one('#main_label')
        async with aiohttp.ClientSession() as session:
            async with session.get('http://python.org') as response:
                status = f'Status: [cyan]{response.status}[/]'
                content_type = f'Content-type: {response.headers["content-type"]}'
                html = await response.text()
                body = f'Body: {html[:20]}...'
        label.update(f'{status}\n{content_type}\n{body}')          


    def compose(self) -> ComposeResult:
        """Create child widgets for the app."""
        yield Header()
        yield Label(id='main_label')
        yield Footer()


    def action_toggle_dark(self) -> None:
        """An action to toggle dark mode."""
        self.theme = (
            "textual-dark" if self.theme == "textual-light" else "textual-light"
        )


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