How to gracefully stop FastAPI app ?

Explanation of the problem

The issue at hand pertains to achieving a graceful stop of a FastAPI application. The concept of a “graceful stop” is of utmost importance in modern web applications as it enables the prevention of data loss or interruption of in-flight requests during application restarts or shutdowns, particularly in scenarios involving rolling updates. While frameworks like Go and Spring Boot offer built-in solutions for graceful stop functionality, the specific approach for achieving this in a FastAPI application has proven elusive. Despite an extensive search through the FastAPI forum and general online resources, comprehensive information on implementing graceful stop in a FastAPI application remains scarce. Notably, the official “uvicorn-gunicorn-fastapi-docker” repository mentions a “graceful_timeout” parameter within the “gunicorn” server configuration, but the uncertainty arises when using the “uvicorn” command to start FastAPI within a custom Docker environment, instead of relying on “gunicorn”.

Troubleshooting with the Lightrun Developer Observability Platform

Getting a sense of what’s actually happening inside a live application is a frustrating experience, one that relies mostly on querying and observing whatever logs were written during development.
Lightrun is a Developer Observability Platform, allowing developers to add telemetry to live applications in real-time, on-demand, and right from the IDE.

• Instantly add logs to, set metrics in, and take snapshots of live applications
• Insights delivered straight to your IDE or CLI
• Works where you do: dev, QA, staging, CI/CD, and production

Start for free today

Problem solution for: How to gracefully stop FastAPI app ?

To achieve a graceful stop in a FastAPI application when using the “uvicorn” command instead of “gunicorn”, one possible solution is to implement a custom signal handler. This approach involves capturing the termination signal (SIGTERM) and performing the necessary cleanup tasks before shutting down the application.

Here’s an example of how you can implement the signal handler in your FastAPI application code:

import signal
import sys
from fastapi import FastAPI

app = FastAPI()

# Define the signal handler function
def graceful_shutdown(signum, frame):
    # Perform cleanup tasks here (e.g., closing database connections, saving state, etc.)
    # ...

    # Exit the application
    sys.exit(0)

# Register the signal handler for SIGTERM
signal.signal(signal.SIGTERM, graceful_shutdown)

# Your FastAPI routes and logic here
# ...

if __name__ == "__main__":
    # Start the FastAPI application using uvicorn
    uvicorn.run("app:app", host='0.0.0.0', port=8000, reload=True, debug=True, workers=3)

In the code snippet above, we import the signal and sys modules to handle the termination signal and exit the application gracefully. The graceful_shutdown function is defined to encapsulate the necessary cleanup tasks. You can customize this function to include any specific actions you need to perform before shutting down the application. Finally, we register the signal handler for the SIGTERM signal using signal.signal(signal.SIGTERM, graceful_shutdown). This ensures that when the termination signal is received, the graceful_shutdown function will be executed.

By implementing this custom signal handler, you can gracefully stop your FastAPI application when using the “uvicorn” command, ensuring that any pending requests are completed and critical data is not lost during the shutdown process.

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# Configure CORS settings
origins = [
    "http://localhost",
    "http://localhost:3000",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

@app.post("/items/")
async def create_item(item: Item):
    # Access validated and deserialized item properties
    item_name = item.name
    item_price = item.price
    # ...

from fastapi import FastAPI
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")

@app.get("/items/")
async def get_items(token: str = Depends(oauth2_scheme)):
    # Validate and verify the token
    # Perform authorization checks
    # ...

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    # Perform asynchronous operations here
    # ...
    return {"item_id": item_id}

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

@app.post("/items/")
async def create_item(item: Item):
    # Process the item data here
    # ...
    return {"message": "Item created"}