How to stop FastAPI app after raising an Exception?

Question

When handling exceptions in FastAPI, is there a way to stop the application after raising an HTTPException?

An example of what I am trying to achieve:

@api.route("/")
def index():
    try:
        do_something()
    except Exception as e:
        raise HTTPException(status_code=500, detail="Doing something failed!")
        sys.exit(1)

if __name__ == "__main__":
    uvicorn.run(api)

Raising the HTTPException alone won't stop my program, and every line of code after the raise won't be executed.

Is there a good way to do something like this, or something similar with the same result?

Answer 1

As you already know how to solve part of raising an exception and executing the code, last part is to stop the loop. You cannot do it with sys.exit(), you need to call stop directly:

@api.route("/")
def stop():
    loop = asyncio.get_event_loop()
    loop.stop()

Or kill the gunicorn process with subprocess.run and kill/pkill if for some reason loop cannot be stopped gracefully.

Be careful of the concurrency here!

Answer 2

As described in the comments earlier, you can follow a similar approach described here, as well as here and here. Once an exception is raised, you can use a custom handler, in which you can stop the currently running event loop, using a Background Task (see Starlette's documentation as well). It is not necessary to do this inside a background task, as when stop() is called, the loop will run all scheduled tasks and then exit; however, if you do so (as in the example below), make sure you define the background task function with async def, as normal def endpoints/functions, as described in this answer, run in an external threadpool, and you, otherwise, wouldn't be able to get the running event loop. Using this approach, any operations you need to be executed when the application is shutting down, using a shutdown event handler, they will do so.

Example

For demo purposes, accessing http://127.0.0.1:8000/hi in the example below will cause the app to terminate after returning the response.

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPException
from starlette.background import BackgroundTask
import asyncio

app = FastAPI()

@app.on_event('shutdown')
def shutdown_event():
    print('Shutting down...!')
    
async def exit_app():
    loop = asyncio.get_running_loop()
    loop.stop()
    
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    task = BackgroundTask(exit_app)
    return PlainTextResponse(str(exc.detail), status_code=exc.status_code, background=task)
 
@app.get('/{msg}')
def main(msg: str):
    if msg == 'hi':
        raise HTTPException(status_code=500, detail='Something went wrong')

    return {'msg': msg}
    
if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=8000)