Function serve_development_app

src/py/reactpy/reactpy/backend/flask.py:93–130

Run a development server for FastAPI

(
    app: Flask,
    host: str,
    port: int,
    started: asyncio.Event | None = None,
)

# BackendType.serve_development_app
async def serve_development_app(
    app: Flask,
    host: str,
    port: int,
    started: asyncio.Event | None = None,
) -> None:
    """Run a development server for FastAPI"""
    loop = asyncio.get_running_loop()
    stopped = asyncio.Event()

    server: Ref[BaseWSGIServer] = Ref()

    def run_server() -> None:
        server.current = make_server(host, port, app, threaded=True)
        if started:
            loop.call_soon_threadsafe(started.set)
        try:
            server.current.serve_forever()  # type: ignore
        finally:
            loop.call_soon_threadsafe(stopped.set)

    thread = Thread(target=run_server, daemon=True)
    thread.start()

    if started:
        await started.wait()

    try:
        await stopped.wait()
    finally:
        # we may have exited because this task was cancelled
        server.current.shutdown()
        # the thread should eventually join
        thread.join(timeout=3)
        # just double check it happened
        if thread.is_alive():  # nocov
            msg = "Failed to shutdown server."
            raise RuntimeError(msg)


def use_websocket() -> WebSocket: