best practise when I want to have a uvicorn server besides with tkinter gui

[wanghaisheng]

how to integrate this wonderful lib with my need

I want to start a fastapi server in the background for files upload task and have tkinter gui do other tasks.

and if I hit the quit menu in the system tray after close gui windows, it should exit both tkinter and fastapi server.

my code kinda of freezing tkinter though

import sys
import threading
from fastapi import FastAPI
from fastapi.responses import FileResponse
from tkinter import *

from fastapi.staticfiles import StaticFiles
from fastapi.middleware.cors import CORSMiddleware
from pystray import MenuItem as item
import pystray
from PIL import Image, ImageTk
import asyncio
import tkinter as tk
from asyncio import CancelledError
from contextlib import suppress
import random
from async_tkinter_loop import async_handler, async_mainloop

app = FastAPI()
# Allow all origins
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # You can replace this with specific origins if needed
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


async def one_url(url):
    """One task."""
    print(f'run one_url: {url}')  # for debug
    sec = random.randint(1, 8)
    await asyncio.sleep(sec)
    return "url: {}\tsec: {}".format(url, sec)


async def do_urls():
    """Creating and starting 10 tasks."""
    tasks = [one_url(url) for url in range(10)]
    completed, pending = await asyncio.wait(tasks)
    results = [task.result() for task in completed]
    print("\n".join(results))


def do_tasks():
    """Button-Event-Handler starting the asyncio part."""
    asyncio.ensure_future(do_urls())


def start(lang, root=None):
    global mainwindow, canvas

    # root.resizable(width=True, height=True)
    root.iconbitmap("assets/icon.ico")
    root.title('tkinter asyncio demo')
    Button(master=root, text="Asyncio Tasks", command=async_handler(do_tasks())).pack()

    root.update_idletasks()



async def quit_window(icon, item):
    global loop, fastapi_thread

    print('Shutdown icon')
    icon.stop()

    print('Shutdown server')
    server.should_exit = True
    server.force_exit = True
    await server.shutdown()
    print('Shutdown root')
    root.quit()

    print('Stop loop')
    # loop.stop()




def show_window(icon, item):
    icon.stop()
    root.after(0, root.deiconify)


def withdraw_window():
    root.withdraw()
    image = Image.open("assets/icon.ico")
    # menu = (item("Quit", lambda icon, item: threading.Thread(target=quit_window, args=(icon, item)).start()),
    #         item("Show", show_window))
    menu = (item("Quit",async_handler(quit_window)),
            item("Show", show_window))

    
    icon = pystray.Icon("name", image, "title", menu)
    # icon.run_detached()
    icon.run()


def start_fastapi_server(loop):
    import uvicorn
    global server
    config = uvicorn.Config(app, loop=loop, host="0.0.0.0", port=8000)
    server = uvicorn.Server(config)
    try:
        loop.run_until_complete(server.serve())
    except KeyboardInterrupt:
        print("Received Ctrl+C. Stopping gracefully...")
        # Cancel all running tasks
        for task in asyncio.Task.all_tasks():
            task.cancel()
        # Optionally: Close any open resources (sockets, files, etc.)
        # Cleanup code here
    # finally:
    #     loop.close()


def start_tkinter_app():
    global root, settings, db, canvas, locale
    root = tk.Tk()

    locale = 'en'
    start(locale, root)

    root.protocol('WM_DELETE_WINDOW', withdraw_window)

    # root.mainloop()
    async_mainloop(root)


if __name__ == "__main__":
    global loop, fastapi_thread
    loop = None

    if sys.platform == "win32" and (3, 8, 0) <= sys.version_info < (3, 9, 0):
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())


    if sys.platform == 'win32':

        asyncio.get_event_loop().close()
        # On Windows, the default event loop is SelectorEventLoop, which does
        # not support subprocesses. ProactorEventLoop should be used instead.
        # Source: https://docs.python.org/3/library/asyncio-subprocess.html
        loop = asyncio.ProactorEventLoop()
        asyncio.set_event_loop(loop)
    else:
        loop = asyncio.get_event_loop()

    # Start FastAPI server in a separate thread
    fastapi_thread = threading.Thread(target=start_fastapi_server, args=(loop,)).start()

    start_tkinter_app()

[insolor]

I'd suggest you not to put everithing in one application. Just make the server a separate application and run it from the GUI using subprocess (asyncio subprocess in this case). When the gui app shut down, also shut down the fastapi service.

[insolor]

Yeah, and also here command=async_handler(do_urls) you should pass do_urls without parenthesis, just it's name

[wanghaisheng]

my real code do_urls got some input variables,

    b_upload = tk.Button(
        operationframe,
        text=settings[locale]["uploadview"]["b_uploadAll"],
        command=async_handler(
            do_urls(
                frame=result_frame,
                status=task_status_var.get(),
                platform=platform_var.get(),
                username=channelname.get(),
                vid=vid.get(),
                vtitle=vtitle.get(),
                schedule_at=schedule_at_var.get(),
                pageno=None,
                pagecount=50,
                sortby=sortby_var.get(),
            ))
    )
``

Exception in Tkinter callback
Traceback (most recent call last):
  File "/Users/wenke/miniconda3/lib/python3.9/tkinter/__init__.py", line 1892, in __call__
    return self.func(*args)
  File "/Users/wenke/miniconda3/lib/python3.9/site-packages/async_tkinter_loop/async_tkinter_loop.py", line 104, in wrapper
    get_event_loop().create_task(async_function(*handler_args, *args, **kwargs))
TypeError: 'coroutine' object is not callable


if. i use the following 

platform_var = tk.StringVar()

b_upload = tk.Button(
    operationframe,
    text=settings[locale]["uploadview"]["b_uploadAll"],
    command=async_handler(
        runupload,
            frame=result_frame,
            status=task_status_var.get(),
            platform=platform_var.get(),
            username=channelname.get(),
            vid=vid.get(),
            vtitle=vtitle.get(),
            schedule_at=schedule_at_var.get(),
            pageno=None,
            pagecount=50,
            sortby=sortby_var.get(),
        )
)

platform_var.get() will always not get the right value

it seems 

[insolor]

Create a separate async function, which will get values from entry fields and then call do_urls. This function specify as a command handler (wrapped in async_handler).

[wanghaisheng]

Create a separate async function, which will get values from entry fields and then call do_urls

but in this way. I need create a bunch of global variable to get values...

async def do_urls(i=10):
    """Creating and starting 10 tasks."""
    tasks = [one_url(url) for url in range(i)]
    completed, pending = await asyncio.wait(tasks)
    results = [task.result() for task in completed]
    print("\n".join(results))

async def do_task():
    print('get variable value')
    platform_value=platform_var.get()
    await do_urls(i=platform_value)
def start(lang, root=None):
    global mainwindow, canvas

    # root.resizable(width=True, height=True)
    root.iconbitmap("assets/icon.ico")
    root.title('tkinter asyncio demo')
    global platform_var
    platform_var = tk.IntVar()
    platform_var.set(5)
    Button(master=root, text="Asyncio Tasks", command=async_handler(do_task)).pack()
    Button(master=root, text="Start Server", command=start_fastapi_server).pack(side=tk.LEFT)

    Button(master=root, text="Stop Server", command=stop).pack(side=tk.LEFT)

    root.update_idletasks()

[insolor]

There are several ways to deal with this.

1. Use OOP, create a class for your window, create this window, and make direct handlers (like do_tasks) its methods
2. Put do_tasks into start as a nested function, so it can access to local variables of start