FastAPI Startup and Shutdown Events

FastAPI provides mechanisms to execute specific actions during the application's startup and shutdown phases. These events are crucial for initializing resources like database connections and starting background tasks, as well as for performing cleanup operations.

Why Use Startup and Shutdown Events?

Utilizing startup and shutdown events allows you to:

Initialize resources: Set up database connections, load configurations, etc.
Clean up resources: Close connections, terminate background tasks, etc.
Enhance maintainability: Centralize resource management for better code organization.

Setting Up Your Environment

Ensure you have:

Python 3.7 or higher
FastAPI and Uvicorn installed:
```
pip install fastapi uvicorn
```

Basic Example of Startup and Shutdown Events

FastAPI provides two decorators for these events:

@app.on_event("startup"): Triggered when the application starts.
@app.on_event("shutdown"): Triggered when the application stops.

Here’s an example demonstrating their usage:

from fastapi import FastAPI

app = FastAPI()

@app.on_event("startup")
async def startup_event():
    print("Application startup: Initializing resources")

@app.on_event("shutdown")
async def shutdown_event():
    print("Application shutdown: Cleaning up resources")

@app.get("/")
async def read_root():
    return {"message": "Hello, World!"}

How It Works:

When you start the app, the startup_event function executes, printing "Application startup: Initializing resources".
When you stop the app (e.g., by pressing Ctrl+C), the shutdown_event function executes, printing "Application shutdown: Cleaning up resources".

To run the app:

uvicorn main:app --reload

Managing Resources with Startup and Shutdown Events

These events are ideal for managing resources such as database connections.

Example: Database Connection Management

This example demonstrates how to establish and close a database connection using startup and shutdown events:

from fastapi import FastAPI
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine = create_async_engine(DATABASE_URL, echo=True)
async_sessionmaker = sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False)

app = FastAPI()

@app.on_event("startup")
async def startup_event():
    # Initialize the database connection
    print("Connecting to the database...")
    async with engine.begin() as conn:
        await conn.run_sync(lambda conn: print("Database connection established"))

@app.on_event("shutdown")
async def shutdown_event():
    # Dispose of the database connection
    await engine.dispose()
    print("Database connection closed.")

@app.get("/")
async def read_root():
    return {"message": "Database connection example"}

How It Works:

The startup_event function establishes the database connection when the application starts.
The shutdown_event function disposes of the connection when the application stops.

Running Background Tasks During Startup

You can use startup events to launch background tasks, such as preloading data or starting a scheduler.

Example: Preloading Data

This example demonstrates how to preload data during the application's startup:

from fastapi import FastAPI

app = FastAPI()

fake_data = {}

@app.on_event("startup")
async def preload_data():
    global fake_data
    print("Preloading data...")
    fake_data = {"key1": "value1", "key2": "value2"}
    print("Data preloaded.")

@app.get("/data/")
async def get_data():
    return fake_data

How It Works:

During startup, the preload_data function populates the fake_data dictionary.
The /data/ endpoint retrieves the preloaded data.

Example: Starting a Background Scheduler

This example demonstrates how to start a periodic task during the application's startup:

from fastapi import FastAPI
import asyncio

app = FastAPI()

async def background_scheduler():
    while True:
        print("Running background task...")
        await asyncio.sleep(10)  # Run every 10 seconds

@app.on_event("startup")
async def startup_event():
    loop = asyncio.get_event_loop()
    loop.create_task(background_scheduler())
    print("Background scheduler started.")

@app.on_event("shutdown")
async def shutdown_event():
    print("Application shutdown: Cleaning up resources")

@app.get("/")
async def read_root():
    return {"message": "Background scheduler example"}

How It Works:

The background_scheduler function runs a task every 10 seconds.
The startup_event function starts the background scheduler when the application starts.
The shutdown_event function handles cleanup when the application stops.

Wrapping Up

Managing startup and shutdown events in FastAPI is essential for initializing and cleaning up resources effectively. By leveraging these events, you can ensure that your application manages resources like database connections and background tasks efficiently.

For more advanced features and best practices, check out Craft Your Startup. It's an ideal way to accelerate your startup projects!