FastAPI, terminal input, background task, all in one...

asynchello.py

import asyncio
import os
import signal
import sys
import time

from fastapi import FastAPI
from aioconsole import ainput
from threading import Thread, Event
import uvicorn

app = FastAPI()

# Global variables
global_variable = "Hello, World!"
exit_event = Event()


# Function to get the value of the global variable
def get_global_variable():
    return global_variable


# FastAPI endpoint to get the value of the global variable
@app.get("/api/v1/demo")
async def read_demo():
    return {"value": get_global_variable()}


# Function to run the command loop
async def command_loop():
    while True:
        command = await ainput("Enter a command (type 'exit' to stop the loop): ")
        if command.lower() == 'exit':
            print("Exiting command loop...")
            exit_event.set()
            break
        else:
            print(f"Command not recognized: {command}")


# Start the FastAPI app asynchronously in a separate thread
def run_uv():
    server = uvicorn.Server(uvicorn.Config(app, loop="asyncio", host="127.0.0.1", port=8000))
    loop = asyncio.new_event_loop()

    try:
        loop.run_until_complete(server.serve())
    except asyncio.CancelledError:
        pass


# Run the asynchronous event loop
async def run_app():
    await asyncio.gather(
        command_loop()
    )


# Background task
def run_background():
    try:
        while not exit_event.is_set():
            print("Running background task...")
            time.sleep(10.0)

        os.kill(os.getpid(), signal.SIGINT)
    except asyncio.CancelledError:
        pass


if __name__ == "__main__":
    # Start uvicorn in a separate thread
    uvicorn_thread = Thread(target=run_uv)
    uvicorn_thread.start()

    # Start a separate thread to run a background task
    background_thread = Thread(target=run_background)
    background_thread.start()

    # Run the command loop in the main thread
    asyncio.run(run_app())

    # Signal the background thread to exit
    exit_event.set()

    # Wait for the uvicorn thread to finish
    try:
        uvicorn_thread.join()
    except (KeyboardInterrupt, SystemExit) as _:
        pass

    # Wait for the background thread to finish
    background_thread.join()

    print("Exiting main thread...")

    # Exit the main thread
    sys.exit(0)