FastAPI signal handler

README.md

Useful if you need to have some sort of child process stop handling something on received termination signal, rather than when the service is shutting down

fastapi-signals.py

import asyncio
import signal
from types import FrameType

from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI(title="example-service")
setattr(app, "_shutting_down", False)


def append_signal() -> None:
    old = signal.getsignal(signal.SIGINT)

    def helper(sig: int, frame: FrameType | None) -> None:
        setattr(app, "_shutting_down", True)
        if callable(old):
            old(sig, frame)

    signal.signal(signal.SIGINT, helper)


@app.on_event("startup")
def startup() -> None:
    append_signal()


@app.get("/")
async def root() -> JSONResponse:
    for _ in range(15):
        print(getattr(app, "_shutting_down"))
        await asyncio.sleep(1)
    return JSONResponse({"status": "ok"})