Not able to access API endpoints and swagger URL when asyncio.run() used in main-CodePudding

not able to access the swagger endpoint http://127.0.0.1:8000/docs when using asyncio.run(startup()) in main but background tasks start running in the background.

app = FastAPI()

monitor_data = [
    {"url": "https://endpoint_1/", "type": "https", "Interval": 30},
    {"url": "https://endpoint_2/", "type": "https", "Interval": 60}
]

@app.get('/')
async def amialive():
    return "Live"

async def monitor(url, interval):
    while True:
        response = requests.get(url)
        print(f"{url}: {response.status_code})")
        await asyncio.sleep(interval)

async def startup():
    tasks = []
    for data in monitor_data:
        url = data["url"]
        interval = data["Interval"]
        task = asyncio.create_task(monitor(url, interval))
        tasks.append(task)
    await asyncio.gather(*tasks)

if __name__ == "__main__":
    asyncio.run(startup())
    # asyncio.create_task(startup()) 
    uvicorn.run("network_monitoring:app", host="0.0.0.0", port=8000 ,reload=True)

And when I remove asyncio.run(startup()) from the main, able to access the API endpoint as well as swagger.

is there a way where I can run the background tasks on the startup of the script and use the endpoint as well at the same time? thanks in advance.

CodePudding user response：

Your function startup awaits monitor, which is an infinite loop, which means asyncio.run(startup()) will block and run forever.

To remove the blocking behavior, simply remove await asyncio.gather(*tasks) at the end of monitor. Since its purpose is to create background tasks, and asyncio.create_task does that, it's not needed.

However, that's not enough. When calling uvicorn.run, that will start a new event loop (even a new process), and you want to run your background tasks in the same event loop. To do this, you can use FastAPI's startup event: add the startup decorator @app.on_event("startup") to your startup function

@app.on_event("startup")
async def startup():
    ...

if __name__ == "__main__":
    uvicorn.run("network_monitoring:app", host="0.0.0.0", port=8000, reload=True)