First attempt at asyncio integration of the Tk event loop

tkasyncio.py

#!/usr/bin/env python3

"Testing the idea of providing an off-the-shelf Tk event loop for asyncio"

import asyncio
import random
from tkinter import Tk, Button

class AsyncTk(Tk):
    "Basic Tk with an asyncio-compatible event loop"
    def __init__(self):
        super().__init__()
        self.running = True
        self.runners = [self.tk_loop()]

    async def tk_loop(self):
        "asyncio 'compatible' tk event loop?"
        # Is there a better way to trigger loop exit than using a state vrbl?
        while self.running:
            self.update()
            await asyncio.sleep(0.05) # obviously, sleep time could be parameterized

    def stop(self):
        self.running = False

    async def run(self):
        await asyncio.gather(*self.runners)


class App(AsyncTk):
    "User's app"
    def __init__(self):
        super().__init__()
        self.create_interface()
        self.runners.append(self.counter())

    def create_interface(self):
        b1 = Button(master=self, text='Random Float',
                    command=lambda: print("your wish, as they say...", random.random()))
        b1.pack()
        b2 = Button(master=self, text='Quit', command=self.stop)
        b2.pack()

    async def counter(self):
        "sample async worker... (with apologies to Lawrence Welk)"
        i = 1
        while self.running:
            print("and a", i)
            await asyncio.sleep(1)
            i += 1

async def main():
    app = App()
    await app.run()

if __name__ == '__main__':
    asyncio.run(main())

Hello Skip,
Thanks for the code ideas here. I am trying to extend this code so that you can call a coro on button press so for the class method:

async def foo(self):
    print(f"IO task foo has started")
    await asyncio.sleep(1)
    print(f"IO task foo has finished")

and add a third button to the UI with

    b3 = Button(master=self, text='Foo', command=self.foo)
    b3.pack()

generates a somewhat expected error:

RuntimeWarning: coroutine 'App.foo' was never awaited

But you cant do something like:

b3 = Button(master=self, text='Foo', command=lambda: await asyncio.create_task(self.foo()))

because you are not calling foo from a coro, so what do you do? Something like

def inject_task(self, task):
    self.more_runners.append(task)
    self.run_more()

But that is no good either. There needs to be some way to await in the button press method but the container method for the button press cant be a coro.

I was on a path and eventually I got there. The trick to running a coro on a button press is to add another list of tasks and check in the run() method if any new tasks have been added and if so, await them. New code below (I don't know how to make it format correctly it looks correct in this window but the preview is all messed up ???):

import asyncio
import random
from tkinter import Tk, Button

class AsyncTk(Tk):
"""Basic Tk with an asyncio-compatible event loop"""
def init(self):
super().init()
self.running = True
self.runners = [self.tk_loop()]
self.button_presses = []

async def tk_loop(self):
    """asyncio 'compatible' tk event loop?"""
    # Is there a better way to trigger loop exit than using a state vrbl?
    while self.running:
        self.update()
        await asyncio.sleep(0.05) # obviously, sleep time could be parameterized
        if len(self.button_presses) > 0:
            await self.button_presses.pop(0)

def stop(self):
    self.running = False

async def run(self):
    await asyncio.gather(*self.runners)

def add_button_coro(self, coro):
    task = asyncio.create_task(coro)
    self.button_presses.append(task)

class App(AsyncTk):
"""User's app"""
def init(self):
super().init()
self.create_interface()
self.runners.append(self.counter())

def create_interface(self):
    b1 = Button(master=self, text='Random Float',
                command=lambda: print("your wish, as they say...", random.random()))
    b1.pack()
    b2 = Button(master=self, text='Quit', command=self.stop)
    b2.pack()
    b3 = Button(master=self, text='Foo', command=lambda: self.add_button_coro(self.foo()))
    b3.pack()

async def counter(self):
    """sample async worker... (with apologies to Lawrence Welk)"""
    i = 1
    while self.running:
        print("and a", i)
        await asyncio.sleep(1)
        i += 1

async def foo(self):
    print(f"IO task foo has started")
    await asyncio.sleep(1)
    print(f"IO task foo has finished")

async def main():
app = App()
await app.run()

if name == 'main':
asyncio.run(main())

made a fork with the async button HERE

Thanks. Sorry not to have replied previously. Sounds like you have a working solution.

Maybe I should make this more than just a sideline hack (aka gist)? That way it would have a bit more visibility and be able to do PRs and such.

https://github.com/smontanaro/tk-asyncio

I merged your changes into the above repo.

Did you notice this thread on discuss.python.org? Seems like it might be related.

Thanks, looks interesting