CNLearn FastAPI - Changing Startup/Shutdown Events to Lifespan

  • Fri, Apr 21, 2023
  • 2-minute read

This post will make some changes to the way startup and shutdown events are handled in our CNLearn backend application. This was prompted by the fact that the application “startup” and “shutdown” events are deprecated and that instead a lifespan handler should be used. The changes that will be presented below are a mixture of the information from FastAPI’s documentation on the subject as well as of the underlying Starlette documentation.

App state

We add a new file, app/state.py, which will contain the state that can be used in our app (for database connections and other common things).

import contextlib
from typing import AsyncGenerator, TypedDict

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

from app.settings.base import settings


class AppState(TypedDict):
    _db: async_sessionmaker[AsyncSession]


@contextlib.asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[AppState, None]:
    if settings.SQLALCHEMY_POSTGRES_URI is None:
        return
    ASYNC_URI: str = settings.SQLALCHEMY_POSTGRES_URI
    engine = create_async_engine(ASYNC_URI, echo=False)
    async_session = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
    yield AppState(_db=async_session)
    close_all_sessions()

What is our AppState? It’s just a typed dictionary containing the various things we’ll have shared across the app. For now, it’s just our _db, which is a SQLAlchemy async session maker. We then do what we used to do in:

  • app/db/connections.py
  • app/tasks/shutdown.py
  • app/tasks/startup.py

all in one place. Please note that the lifespan handler is an AsyncGenerator that will yield the AppState on startup, and do the rest when the app exits (i.e. close_all_sessions for now).

Lifespan handler usage

Where do we use the lifespan handler? We use it on our app configuration when creating it:

    app = FastAPI(
        title=settings.APP_NAME,
        version=settings.VERSION,
        lifespan=lifespan,
    )

One more important thing to note is that before, we saved the db on the request.app.state. When you use it in a handler, you can access it via the request state.

Instead of async_session_maker = request.app.state._db, we now have async_session_maker = request.state._db in our get_async_session dependency that we use in requests.

There are some changes I needed to make in the tests, but otherwise that’s it for today. Super short maintenance post. The commit for the post is here.

pythonfastapi fastapipython