Test

web_server.py

import hypercorn
import hypercorn.trio
import trio
from quart import websocket
from quart_trio import QuartTrio

from web_app import static

quart_app = QuartTrio(__name__, static_folder=static.__path__[0])


@quart_app.route("/")
async def index():
    return await quart_app.send_static_file("index.html")


@quart_app.websocket("/")
async def ws() -> None:
    await websocket.accept()
    async with trio.open_nursery() as nursery:
        nursery.start_soon(presenter)
        nursery.start_soon(controller)


async def presenter() -> None:
    while True:
        await websocket.send_json({"app_name": "app_name"})
        await trio.sleep(1.0)


async def controller() -> None:
    while True:
        action = await websocket.receive_json()
        print(f"{action=}")


async def main() -> None:
    async with trio.open_nursery() as nursery:
        port = 8_000
        config = hypercorn.Config.from_mapping(
            bind=[f"0.0.0.0:{port}"],
            accesslog="-",
            errorlog="-",
            worker_class="trio",
        )
        urls = await nursery.start(hypercorn.trio.serve, quart_app, config)
        print(f"{urls=}")


if __name__ == "__main__":
    trio.run(main)