FastAPI Event Handlers

In any web application, there are times when you need to perform certain actions when your application starts up or shuts down. FastAPI provides event handlers specifically designed for these scenarios, allowing you to execute code at critical moments in your application's lifecycle.

Introduction to Event Handlers

Event handlers in FastAPI are functions that get executed when specific events occur during the application lifecycle. The two primary events you can hook into are:

1. Startup: Code that runs before the application starts accepting requests
2. Shutdown: Code that runs when the application is shutting down

These event handlers are particularly useful for tasks like:

• Establishing database connections
• Loading machine learning models into memory
• Initializing resources
• Creating initial data
• Cleaning up resources before shutdown
• Closing connections properly

Basic Event Handler Syntax

To create event handlers in FastAPI, you use the @app.on_event() decorator:

python

from fastapi import FastAPI

app = FastAPI()

@app.on_event("startup")
async def startup_event():
    # Code to run on application startup
    print("Application startup!")

@app.on_event("shutdown")
async def shutdown_event():
    # Code to run on application shutdown
    print("Application shutdown!")

The event handlers can be both synchronous or asynchronous functions, though asynchronous is recommended for potentially blocking operations.

Practical Example: Database Connection

Let's see how event handlers can be used to manage database connections:

python

from fastapi import FastAPI
import databases
import sqlalchemy

# Database URL
DATABASE_URL = "sqlite:///./test.db"

# Database instance
database = databases.Database(DATABASE_URL)

# SQLAlchemy setup
metadata = sqlalchemy.MetaData()

# Define a table
notes = sqlalchemy.Table(
    "notes",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("text", sqlalchemy.String),
    sqlalchemy.Column("completed", sqlalchemy.Boolean),
)

# Create the database
engine = sqlalchemy.create_engine(DATABASE_URL)
metadata.create_all(engine)

app = FastAPI()

@app.on_event("startup")
async def startup_db_client():
    await database.connect()
    print("Database connection established")

@app.on_event("shutdown")
async def shutdown_db_client():
    await database.disconnect()
    print("Database connection closed")

@app.get("/notes/")
async def get_notes():
    query = notes.select()
    return await database.fetch_all(query)

In this example:

1. On application startup, we establish a connection to the database
2. During the application's runtime, we can use this connection to query the database
3. On shutdown, we properly close the database connection

Multiple Event Handlers

You can define multiple handlers for the same event. They will be executed in the order they were defined:

python

from fastapi import FastAPI

app = FastAPI()

@app.on_event("startup")
async def startup_event_1():
    print("Startup event 1")

@app.on_event("startup")
async def startup_event_2():
    print("Startup event 2")

# Output when the application starts:
# Startup event 1
# Startup event 2

Real-World Example: Caching System

Let's implement a simple caching system that gets initialized at startup and cleared at shutdown:

python

from fastapi import FastAPI, Depends
from pydantic import BaseModel
import time

app = FastAPI()

# A simple in-memory cache
cache = {}

class CacheInfo(BaseModel):
    size: int
    uptime: float

startup_time = None

@app.on_event("startup")
async def initialize_cache():
    global startup_time
    cache.clear()  # Ensure the cache is empty
    startup_time = time.time()
    print("Cache initialized")
    
    # Populate cache with initial data
    cache["greeting"] = "Hello, World!"
    cache["counter"] = 0

@app.on_event("shutdown")
async def cleanup_cache():
    cache.clear()
    print("Cache cleared")

@app.get("/cache/{key}")
async def get_cache_item(key: str):
    if key in cache:
        return {key: cache[key]}
    return {"error": "Key not found"}

@app.post("/cache/{key}/{value}")
async def set_cache_item(key: str, value: str):
    cache[key] = value
    return {"status": "success"}

@app.get("/cache-info", response_model=CacheInfo)
async def get_cache_info():
    return CacheInfo(
        size=len(cache),
        uptime=round(time.time() - startup_time, 2)
    )

In this example:

1. We initialize a cache when the application starts
2. We provide API endpoints to interact with the cache
3. We clean up the cache when the application shuts down

Advanced Example: Background Tasks with Event Handlers

You might want to start a background task when your application starts. Here's how you can do it:

python

from fastapi import FastAPI
import asyncio
import datetime

app = FastAPI()

# Global variable to control the background task
should_continue = True

async def periodic_task():
    while should_continue:
        now = datetime.datetime.now()
        print(f"Background task running at: {now}")
        # Simulate some work
        await asyncio.sleep(10)

@app.on_event("startup")
async def start_background_tasks():
    global should_continue
    should_continue = True
    # Start the background task
    asyncio.create_task(periodic_task())
    print("Background task started")

@app.on_event("shutdown")
async def stop_background_tasks():
    global should_continue
    should_continue = False
    print("Background task stopped")

@app.get("/")
async def read_root():
    return {"status": "Application is running with background tasks"}

This code starts a background task that runs every 10 seconds when the application starts and stops it gracefully when the application shuts down.

Best Practices for Event Handlers

1. Keep them focused: Each event handler should have a single responsibility.
2. Handle exceptions: Add proper exception handling to prevent startup failures.
3. Use async when appropriate: For I/O-bound operations, use async functions.
4. Log important information: Log the status of initialization and cleanup tasks.
5. Avoid long-running blocking operations: These can delay your application startup.

Here's an example with improved error handling:

python

import logging
from fastapi import FastAPI

app = FastAPI()
logger = logging.getLogger("uvicorn")

@app.on_event("startup")
async def startup_event():
    try:
        # Initialize resources
        logger.info("Application resources initialized successfully")
    except Exception as e:
        logger.error(f"Failed to initialize application resources: {str(e)}")
        # You might want to exit the application if critical initialization fails
        # import sys
        # sys.exit(1)

@app.on_event("shutdown")
async def shutdown_event():
    try:
        # Clean up resources
        logger.info("Application resources cleaned up successfully")
    except Exception as e:
        logger.error(f"Error during resource cleanup: {str(e)}")

Summary

FastAPI event handlers provide a convenient way to execute code at specific points in your application's lifecycle:

• startup events run before the application starts accepting requests
• shutdown events run when the application is shutting down
• Event handlers are useful for initializing and cleaning up resources
• Multiple handlers can be defined for the same event
• They help manage connections, caches, and background tasks effectively

By effectively using event handlers, you can ensure your FastAPI application properly initializes resources it needs and cleans up when it's done, leading to more robust and resource-efficient applications.

Additional Resources

• FastAPI Official Documentation on Events
• Starlette Events Documentation (FastAPI is built on Starlette)

Exercises

1. Create a FastAPI application that counts the total number of requests processed between startup and shutdown.
2. Implement a simple file-based logging system that opens a log file at startup and properly closes it at shutdown.
3. Create a FastAPI application that loads a JSON configuration file at startup and uses it throughout the application.
4. Implement a background task that periodically checks a website's availability and logs the results.

Happy coding with FastAPI event handlers!