uvicorn: uvicorn eats SIGINTs, does not propagate exceptions

With the above, the CancelledError happens.

This is not sufficient for an asyncio application. KeyboardInterrupt is needed to signal to the event loop to shut down all tasks, not just those manually tied to the uvicorn server; an application is impossible to shut down gracefully otherwise. It is also needed to signal to the orchestration of the program (be this a shell, systemd, or similar) that shutdown occurred properly.

What is the issue you are trying to solve? I mean real case scenario.

I’m not sure what kind of answer you are expecting to this. “run uvicorn as part of a larger async application” is a realistic use-case in my book. This requires being able to shut down the application using standard means; right now uvicorn prevents this (at least without extensive workarounds).

In our specific case, we use uvicorn to implement an (optional) REST API of a highly concurrent resource management application. Mishandling SIGINT means that the application could not properly shut down and needed a hard kill, leaking other resources.

You can just override install_signal_handlers function:

# FastAPI Uvicorn override
class Server(uvicorn.Server):

    # Override
    def install_signal_handlers(self) -> None:

        # Do nothing
        pass

And then when you catch interrupt elsewhere you can just signal server to stop:

self.server.should_exit = True

This has turned out to be a real problem for running an optional uvicorn based server inside an existing asyncio application. Since Server.serve finishes gracefully and the original signal handlers are never restored there is no way to ^C the application itself.

It looks okay’ish for Server.serve to capture signals for graceful shutdown, but it should at least restore the original signal handlers and ideally also reproduce the original behaviour as well. From the looks of it, Server.serve should raise KeyboardInterrupt when done and only Server.run should suppress it.

I don’t know if this could help but this is my workaround for my use case:

import signal
from types import FrameType

from fastapi import FastAPI 

app = FastAPI()


@app.on_event("startup")
async def startup_event() -> None:
    default_sigint_handler = signal.getsignal(signal.SIGINT)
    def terminate_now(signum: int, frame: FrameType = None):
        # do whatever you need to unblock your own tasks

        default_sigint_handler(signum, frame)
    signal.signal(signal.SIGINT, terminate_now)

You can just override install_signal_handlers function:

# FastAPI Uvicorn override
class Server(uvicorn.Server):

    # Override
    def install_signal_handlers(self) -> None:

        # Do nothing
        pass

And then when you catch interrupt elsewhere you can just signal server to stop:

self.server.should_exit = True

Nice! one more option is to override the handle_exit:

on_exit = lambda: logging.getLogger(__name__).info("Exiting...") 
class Server(uvicorn.Server):
        """Override uvicorn.Server to handle exit."""

        def handle_exit(self, sig, frame):
            # type: (int, Optional[FrameType]) -> None
            """Handle exit."""
            on_exit()
            super().handle_exit(sig=sig, frame=frame)

Hi all,

To chime in with another example.

I have a fairly trivial application where in a websocket handler, I’m using an async queue to push messages to clients. On shutdown, I’m using an event and a None message to signal we are closing down. But the fastapi shutdown handler is never applied and I cannot trigger these sentinel messages. It’s ina deadlock position as far as I can tell.

@Kludex Even when adding such code to manually handle the exit, the exit code is wrong for any naive handling. The program shown exits with 0; it should propagate the SIGINT if killed by that, though. A proper exit is via KeyboardInterrupt iff the application was killed by SIGINT.

That needs quite some extra scaffolding to do right.

^ courtesy ping for @Kludex , who seems to be the most active maintainer

Well, for my problem I’ve uninstalled watchfiles so uvicorn falls back to polling which works.

Here’s a very hacky workaround

from uvicorn import Server

class MyCustomUvicornServer(Server):
    def install_signal_handlers(self):
        pass