2

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?

Chris
  • 18,724
  • 6
  • 46
  • 80
noah
  • 312
  • 2
  • 13
  • What do you mean by *"stopping the API"*? – JPG Nov 21 '22 at 10:15
  • Make sure the process is stopped and/or no endpoint can't be reached. – noah Nov 21 '22 at 10:16
  • We need to know how do you run it in first place. – kosciej16 Nov 21 '22 at 10:28
  • @kosciej16 it is run via `uvicorn.run(api)` – noah Nov 21 '22 at 10:29
  • Does this answer your question? [How to add background tasks when request fails and HTTPException is raised in FastAPI?](https://stackoverflow.com/questions/73282411/how-to-add-background-tasks-when-request-fails-and-httpexception-is-raised-in-fa) – Chris Nov 21 '22 at 10:29
  • You could use a background task, as described in [this answer](https://stackoverflow.com/a/73283272/17865804). You could also create a custom exception, as described [here](https://stackoverflow.com/a/72833284/17865804), or handle a known exception as you wish, as described [here](https://stackoverflow.com/a/71800464/17865804). – Chris Nov 21 '22 at 10:31
  • @Chris none of this is related to the question... – kosciej16 Nov 21 '22 at 10:32
  • @Chris I've tried with a background Task, unfortunately that didn't work. Custom exception handlers would, I suppose, only make the same Problem a lot more difficult to understand. – noah Nov 21 '22 at 10:36
  • @Chris indeed I focued on it's other part so much I forgotted about the second problem. Sorry! – kosciej16 Nov 21 '22 at 10:49

2 Answers2

1

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)
Chris
  • 18,724
  • 6
  • 46
  • 80
  • After porting this to work with my app structure, I get the following error: `RuntimeError: Event loop stopped before Future completed.`. As far as I understand, it could have somthing to do with the issue described [here](https://stackoverflow.com/questions/64282224/intermittent-event-loop-stopped-before-future-completed) – noah Nov 21 '22 at 12:50
  • After a small brainstrom with my colleagues, I came to the conclusion that it is not really useful to stop the API at that point, so that kind of makes my question useless. Thanks for your quick help tho! If I find a minute to spare, im going to verify your answer because it might still be useful to others. – noah Nov 22 '22 at 08:05
  • there is a lot of company-sensitive data, I can't really provide the full code. But I will provide an example soon and, if you wan't to, I can keep you updated – noah Nov 22 '22 at 13:21
0

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!

kosciej16
  • 6,294
  • 1
  • 18
  • 29