SQLAlchemy Sessions

Source module: fastapi_restful.sessions

Note

To use please install with: pip install fastapi-restful[session] or pip install fastapi-restful[all]

---

One of the most commonly used ways to power database functionality with FastAPI is SQLAlchemy’s ORM.

FastAPI has great documentation about how to integrate ORM into your application.

However, the recommended approach for using SQLAlchemy’s ORM with FastAPI has evolved over time to reflect both insights from the community and the addition of new features to FastAPI.

The fastapi_restful.session module contains an implementation making use of the most up-to-date best practices for managing SQLAlchemy sessions with FastAPI.

---

FastAPISessionMaker

The fastapi_restful.session.FastAPISessionMaker class conveniently wraps session-making functionality for use with FastAPI. This section contains an example showing how to use this class.

Let’s begin with some infrastructure. The first thing we’ll do is make sure we have an ORM table to query:

from functools import lru_cache
from typing import Iterator
from uuid import UUID

import sqlalchemy as sa
from fastapi import Depends, FastAPI
from pydantic import BaseSettings
from sqlalchemy.orm import Session, declarative_base

from fastapi_restful.guid_type import GUID, GUID_DEFAULT_SQLITE
from fastapi_restful.session import FastAPISessionMaker

Base = declarative_base()


class User(Base):
    __tablename__ = "user"
    id = sa.Column(GUID, primary_key=True, default=GUID_DEFAULT_SQLITE)
    name = sa.Column(sa.String, nullable=False)


class DBSettings(BaseSettings):
    """Parses variables from environment on instantiation"""

    database_uri: str  # could break up into scheme, username, password, host, db


def get_db() -> Iterator[Session]:
    """FastAPI dependency that provides a sqlalchemy session"""
    yield from _get_fastapi_sessionmaker().get_db()


@lru_cache()
def _get_fastapi_sessionmaker() -> FastAPISessionMaker:
    """This function could be replaced with a global variable if preferred"""
    database_uri = DBSettings().database_uri
    return FastAPISessionMaker(database_uri)


app = FastAPI()


@app.get("/{user_id}")
def get_user_name(db: Session = Depends(get_db), *, user_id: UUID) -> str:
    user = db.get(User, user_id)
    username = user.name
    return username

Next, we set up infrastructure for loading the database uri from the environment:

from functools import lru_cache
from typing import Iterator
from uuid import UUID

import sqlalchemy as sa
from fastapi import Depends, FastAPI
from pydantic import BaseSettings
from sqlalchemy.orm import Session, declarative_base

from fastapi_restful.guid_type import GUID, GUID_DEFAULT_SQLITE
from fastapi_restful.session import FastAPISessionMaker

Base = declarative_base()


class User(Base):
    __tablename__ = "user"
    id = sa.Column(GUID, primary_key=True, default=GUID_DEFAULT_SQLITE)
    name = sa.Column(sa.String, nullable=False)


class DBSettings(BaseSettings):
    """Parses variables from environment on instantiation"""

    database_uri: str  # could break up into scheme, username, password, host, db


def get_db() -> Iterator[Session]:
    """FastAPI dependency that provides a sqlalchemy session"""
    yield from _get_fastapi_sessionmaker().get_db()


@lru_cache()
def _get_fastapi_sessionmaker() -> FastAPISessionMaker:
    """This function could be replaced with a global variable if preferred"""
    database_uri = DBSettings().database_uri
    return FastAPISessionMaker(database_uri)


app = FastAPI()


@app.get("/{user_id}")
def get_user_name(db: Session = Depends(get_db), *, user_id: UUID) -> str:
    user = db.get(User, user_id)
    username = user.name
    return username

We use the pydantic.BaseSettings to load variables from the environment. There is documentation for this class in the pydantic docs, but the basic idea is that if a model inherits from this class, any fields not specified during initialization are read from the environment if possible.

Info

Since database_uri is not an optional field, a ValidationError will be raised if the DATABASE_URI environment variable is not set.

Info

For finer grained control, you could remove the database_uri field, and replace it with separate fields for scheme, username, password, host, and db. You could then give the model a @property called database_uri that builds the uri from these components.

Now that we have a way to load the database uri, we can create the FastAPI dependency we’ll use to obtain the sqlalchemy session:

from functools import lru_cache
from typing import Iterator
from uuid import UUID

import sqlalchemy as sa
from fastapi import Depends, FastAPI
from pydantic import BaseSettings
from sqlalchemy.orm import Session, declarative_base

from fastapi_restful.guid_type import GUID, GUID_DEFAULT_SQLITE
from fastapi_restful.session import FastAPISessionMaker

Base = declarative_base()


class User(Base):
    __tablename__ = "user"
    id = sa.Column(GUID, primary_key=True, default=GUID_DEFAULT_SQLITE)
    name = sa.Column(sa.String, nullable=False)


class DBSettings(BaseSettings):
    """Parses variables from environment on instantiation"""

    database_uri: str  # could break up into scheme, username, password, host, db


def get_db() -> Iterator[Session]:
    """FastAPI dependency that provides a sqlalchemy session"""
    yield from _get_fastapi_sessionmaker().get_db()


@lru_cache()
def _get_fastapi_sessionmaker() -> FastAPISessionMaker:
    """This function could be replaced with a global variable if preferred"""
    database_uri = DBSettings().database_uri
    return FastAPISessionMaker(database_uri)


app = FastAPI()


@app.get("/{user_id}")
def get_user_name(db: Session = Depends(get_db), *, user_id: UUID) -> str:
    user = db.get(User, user_id)
    username = user.name
    return username

Info

The get_db dependency makes use of a context-manager dependency, rather than a middleware-based approach. This means that any endpoints that don’t make use of a sqlalchemy session will not be exposed to any session-related overhead.

This is in contrast with middleware-based approaches, where the handling of every request would result in a session being created and closed, even if the endpoint would not make use of it.

Warning

The get_db dependency will not finalize your ORM session until after a response is returned to the user.

This has minor response-latency benefits, but also means that if you have any uncommitted database writes that will raise errors, you may return a success response to the user (status code 200), but still raise an error afterward during request clean-up.

To deal with this, for any request where you expect a database write to potentially fail, you should manually perform a commit inside your endpoint logic and appropriately handle any resulting errors.

---

Note that while middleware-based approaches can automatically ensure database errors are visible to users, the result would be a generic 500 internal server error, which you should strive to avoid sending to clients under normal circumstances.

You can still log any database errors raised during cleanup by appropriately modifying the get_db function with a try: except: block.

The get_db function can be used as a FastAPI dependency that will inject a sqlalchemy ORM session where used:

from functools import lru_cache
from typing import Iterator
from uuid import UUID

import sqlalchemy as sa
from fastapi import Depends, FastAPI
from pydantic import BaseSettings
from sqlalchemy.orm import Session, declarative_base

from fastapi_restful.guid_type import GUID, GUID_DEFAULT_SQLITE
from fastapi_restful.session import FastAPISessionMaker

Base = declarative_base()


class User(Base):
    __tablename__ = "user"
    id = sa.Column(GUID, primary_key=True, default=GUID_DEFAULT_SQLITE)
    name = sa.Column(sa.String, nullable=False)


class DBSettings(BaseSettings):
    """Parses variables from environment on instantiation"""

    database_uri: str  # could break up into scheme, username, password, host, db


def get_db() -> Iterator[Session]:
    """FastAPI dependency that provides a sqlalchemy session"""
    yield from _get_fastapi_sessionmaker().get_db()


@lru_cache()
def _get_fastapi_sessionmaker() -> FastAPISessionMaker:
    """This function could be replaced with a global variable if preferred"""
    database_uri = DBSettings().database_uri
    return FastAPISessionMaker(database_uri)


app = FastAPI()


@app.get("/{user_id}")
def get_user_name(db: Session = Depends(get_db), *, user_id: UUID) -> str:
    user = db.get(User, user_id)
    username = user.name
    return username

Info

We make use of @lru_cache on _get_fastapi_sessionmaker to ensure the same FastAPISessionMaker instance is reused across requests. This reduces the per-request overhead while still ensuring the instance is created lazily, making it possible to have the database_uri reflect modifications to the environment performed after importing the relevant source file.

This can be especially useful during testing if you want to override environment variables programmatically using your testing framework.