Patterns and Practices for using SQLAlchemy 2.0 with FastAPI

While Django and Flask remain the first choices for many Python engineers, has already been recognized as an undeniably reliable pick. It is a highly flexible, well-optimized, structured framework that gives the developer endless possibilities for building backend applications. FastAPI Working with databases is an essential aspect of most backend applications. As a result, the ORM plays a critical role in the backend code. However, unlike Django, FastAPI does not have an ORM built-in. It is entirely the developer's responsibility to select a suitable library and integrate it into the codebase. Python engineers widely consider to be the most popular ORM available. It's a legendary library that's been in use since 2006 and has been adopted by thousands of projects. In 2023, it received a major update to version 2.0. Similar to FastAPI, SQLAlchemy provides developers with powerful features and utilities without forcing them to use them in a specific way. Essentially, it's a versatile toolkit that empowers developers to use it however they see fit. SQLAlchemy FastAPI and SQLAlchemy are a match made in heaven. They are both reliable, performant, and modern technologies, which enable the creation of powerful and unique applications. This article explores creating a FastAPI backend application that utilizes SQLAlchemy 2.0 as the ORM. The content covers: building models using and Mapped mapped_column defining an abstract model handling database session using the ORM creating a common repository class for all models preparing a test setup and adding tests Afterward, you will be able to combine the FastAPI application with SQLAlchemy ORM easily. Additionally, you will get familiar with best practices and patterns for creating well-structured, robust, and performant applications. Prerequisites Code examples included in the article come from the project, which is a basic API for creating and reading ingredient and potion objects. The main focus of the article is to explore the combination of FastAPI and SQLAlchemy. It doesn't cover other topics, such as: alchemist configuring Docker setup starting the uvicorn server setting up linting If you are interested in these topics, you can explore them on your own by examining the codebase. To access the code repository of the alchemist project, please follow this link . Additionally, you can find the file structure of the project below: here alchemist ├─ alchemist │ ├─ api │ │ ├─ v1 │ │ │ ├─ __init__.py │ │ │ └─ routes.py │ │ ├─ v2 │ │ │ ├─ __init__.py │ │ │ ├─ dependencies.py │ │ │ └─ routes.py │ │ ├─ __init__.py │ │ └─ models.py │ ├─ database │ │ ├─ __init__.py │ │ ├─ models.py │ │ ├─ repository.py │ │ └─ session.py │ ├─ __init__.py │ ├─ app.py │ └─ config.py ├─ requirements │ ├─ base.txt │ └─ dev.txt ├─ scripts │ ├─ create_test_db.sh │ ├─ migrate.py │ └─ run.sh ├─ tests │ ├─ conftest.py │ └─ test_api.py ├─ .env ├─ .gitignore ├─ .pre-commit-config.yaml ├─ Dockerfile ├─ Makefile ├─ README.md ├─ docker-compose.yaml ├─ example.env └─ pyproject.toml Although the tree may seem large, some of the contents are not relevant to the main point of this article. Additionally, the code may appear simpler than what is necessary in certain areas. For instance, the project lacks: production stage in the Dockerfile alembic setup for migrations subdirectories for tests This was done intentionally to reduce complexity and avoid unnecessary overhead. However, it is important to keep these factors in mind if dealing with a more production-ready project. API Requirements When beginning to develop an app, it's critical to consider the models that your app will use. These models will represent the and that your app will work with and will be exposed in the API. In the case of the alchemist app, there are two entities: ingredients and potions. The API should allow for creating, and retrieving these entities. The file contains the models that will be used in the API: objects entities alchemist/api/models.py import uuid from pydantic import BaseModel, Field class Ingredient(BaseModel): """Ingredient model.""" pk: uuid.UUID name: str class Config: orm_mode = True class IngredientPayload(BaseModel): """Ingredient payload model.""" name: str = Field(min_length=1, max_length=127) class Potion(BaseModel): """Potion model.""" pk: uuid.UUID name: str ingredients: list[Ingredient] class Config: orm_mode = True class PotionPayload(BaseModel): """Potion payload model.""" name: str = Field(min_length=1, max_length=127) ingredients: list[uuid.UUID] = Field(min_items=1) The API will be returning and models. Setting to in the configuration will make it easier to work with the SQLAlchemy objects in the future. models will be utilized for creating new objects. Ingredient Potion orm_mode True Payload The use of pydantic makes the classes more detailed and clear in their roles and functions. Now, it's time to create the database models. Declaring Models A model is essentially a of something. In the context of APIs, models represent what the backend expects in the request body and what it will return in the response data. Database models, on the other hand, are more complex and represent the stored in the database and the types between them. representation data structures relationship The file contains models for ingredient and potion objects: alchemist/database/models.py import uuid from sqlalchemy import Column, ForeignKey, Table, orm from sqlalchemy.dialects.postgresql import UUID class Base(orm.DeclarativeBase): """Base database model.""" pk: orm.Mapped[uuid.UUID] = orm.mapped_column( primary_key=True, default=uuid.uuid4, ) potion_ingredient_association = Table( "potion_ingredient", Base.metadata, Column("potion_id", UUID(as_uuid=True), ForeignKey("potion.pk")), Column("ingredient_id", UUID(as_uuid=True), ForeignKey("ingredient.pk")), ) class Ingredient(Base): """Ingredient database model.""" __tablename__ = "ingredient" name: orm.Mapped[str] class Potion(Base): """Potion database model.""" __tablename__ = "potion" name: orm.Mapped[str] ingredients: orm.Mapped[list["Ingredient"]] = orm.relationship( secondary=potion_ingredient_association, backref="potions", lazy="selectin", ) Every model in SQLAlchemy starts with the class. Inheriting from it allows building database models that are compatible with Python type checkers. DeclarativeBase It's also a good practice to create an model— class in this case—that includes fields required in all models. These fields include the primary key, which is a unique identifier of every object. The abstract model often also stores the creation and update dates of an object, which are set automatically, when an object is created or updated. However, the model will be kept simple. abstract Base Base Moving on to the model, the attribute specifies the name of the database table, while the field uses the new SQLAlchemy syntax, allowing model fields to be declared with type annotations. This concise and modern approach is both powerful and advantageous for type checkers and IDEs, as it recognizes the field as a string. Ingredient __tablename__ name name Things get more complex in the model. It also includes and attributes, but on top of that, it stores the relationship to ingredients. The use of indicates that the potion can contain multiple ingredients, and in this case, the relationship is many-to-many (M2M). This means that a single ingredient can be assigned to multiple potions. Potion __tablename__ name Mapped[list["Ingredient"]] M2M requires additional configuration, usually involving the creation of an association table that stores the connections between the two entities. In this case, the object stores only the identifiers of the ingredient and potion, but it could also include extra attributes, such as the amount of a specific ingredient needed for the potion. potion_ingredient_association The function configures the relationship between the potion and its ingredients. The argument specifies how related items should be loaded. In other words: what should SQLAlchemy do with related ingredients when you are fetching a potion. Setting it to means that ingredients will be loaded with the potion, eliminating the need for additional queries in the code. relationship lazy selectin Building well-designed models is crucial when working with an ORM. Once this is done, the next step is establishing the connection with the database. Session Handler When working with a database, particularly when using SQLAlchemy, it is essential to understand the following concepts: dialect engine connection connection pool session Out of all these terms, the most important one is the . According to the SQLAlchemy documentation, the engine object is responsible for connecting the and to facilitate database connectivity and behavior. In simpler terms, the engine object is the source of the database connection, while the provides high-level functionalities like executing SQL statements, managing transactions, and retrieving results from the database. engine Pool Dialect connection A is a unit of work that groups related operations within a single transaction. It is an abstraction over the underlying database connections and efficiently manages connections and transactional behavior. session is a component that provides support for a specific database backend. It acts as an intermediary between SQLAlchemy and the database, handling the details of communication. The alchemist project uses Postgres as the database, so the dialect must be compatible with this specific database type. Dialect The final question mark is the . In the context of SQLAlchemy, a connection pool is a mechanism that manages a collection of database connections. It is designed to improve the performance and efficiency of database operations by reusing existing connections rather than creating new ones for each request. By reusing connections, the connection pool reduces the overhead of establishing new connections and tearing them down, resulting in improved performance. connection pool With that knowledge covered, you can now take a look at file, which contains a function that will be used as a dependency for connecting to the database: alchemist/database/session.py from collections.abc import AsyncGenerator from sqlalchemy import exc from sqlalchemy.ext.asyncio import ( AsyncSession, async_sessionmaker, create_async_engine, ) from alchemist.config import settings async def get_db_session() -> AsyncGenerator[AsyncSession, None]: engine = create_async_engine(settings.DATABASE_URL) factory = async_sessionmaker(engine) async with factory() as session: try: yield session await session.commit() except exc.SQLAlchemyError as error: await session.rollback() raise The first important detail to notice is that the function is a generator function. This is because the FastAPI dependency system supports generators. As a result, this function can handle both successful and failed scenarios. get_db_session The first two lines of the function create a database engine and a session. However, the session object can also be used as a context manager. This gives you more control over potential exceptions and successful outcomes. get_db_session Although SQLAlchemy handles the closing of connections, it is a good practice to explicitly declare how to handle the connection after it's done. In the function, the session is committed if everything goes well and rolled back if an exception is raised. get_db_session It's important to note that this code is built around the asyncio extension. This feature of SQLAlchemy allows the app to interact with the database asynchronously. It means that requests to the database won't block other API requests, making the app way more efficient. Once the models and connection are set up, the next step is to ensure that the models are added to the database. Quick Migrations SQLAlchemy models represent the structures of a database. However, simply creating them does not result in immediate changes to the database. To make changes, you must first them. This is typically done using a migration library such as alembic, which tracks every model and updates the database accordingly. apply Since no further changes to the models are planned in this scenario, a basic migration script will suffice. Below is an example code from the file. scripts/migrate.py import asyncio import logging from sqlalchemy.ext.asyncio import create_async_engine from alchemist.config import settings from alchemist.database.models import Base logger = logging.getLogger() async def migrate_tables() -> None: logger.info("Starting to migrate") engine = create_async_engine(settings.DATABASE_URL) async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) logger.info("Done migrating") if __name__ == "__main__": asyncio.run(migrate_tables()) To put it simply, the function reads the structure of models and recreates it in the database using the SQLAlchemy engine. To run this script, use the command. migrate_tables python scripts/migrate.py The models are now present both in the code and in the database and can facilitate interactions with the database. You can now begin working on the API logic. get_db_session API with the ORM As mentioned previously, the API for ingredients and potions is meant to support three operations: creating objects listing objects retrieving objects by ID Thanks to the prior preparations, all these features can already be implemented with SQLAlchemy as the ORM and FastAPI as the web framework. To begin, review the ingredients API located in the file: alchemist/api/v1/routes.py import uuid from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from alchemist.api import models from alchemist.database import models as db_models from alchemist.database.session import get_db_session router = APIRouter(prefix="/v1", tags=["v1"]) @router.post("/ingredients", status_code=status.HTTP_201_CREATED) async def create_ingredient( data: models.IngredientPayload, session: AsyncSession = Depends(get_db_session), ) -> models.Ingredient: ingredient = db_models.Ingredient(**data.dict()) session.add(ingredient) await session.commit() await session.refresh(ingredient) return models.Ingredient.from_orm(ingredient) @router.get("/ingredients", status_code=status.HTTP_200_OK) async def get_ingredients( session: AsyncSession = Depends(get_db_session), ) -> list[models.Ingredient]: ingredients = await session.scalars(select(db_models.Ingredient)) return [models.Ingredient.from_orm(ingredient) for ingredient in ingredients] @router.get("/ingredients/{pk}", status_code=status.HTTP_200_OK) async def get_ingredient( pk: uuid.UUID, session: AsyncSession = Depends(get_db_session), ) -> models.Ingredient: ingredient = await session.get(db_models.Ingredient, pk) if ingredient is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Ingredient does not exist", ) return models.Ingredient.from_orm(ingredient) Under the API, there are three routes available. The POST endpoint takes an ingredient payload as an object from a previously created model and a database session. The generator function initializes the session and enables database interactions. /ingredients get_db_session In the actual function body, there are five steps taking place: An ingredient object is created from the incoming payload. The method of the session object adds the ingredient object to the session tracking system and marks it as pending for insertion into the database. add The session is committed. The ingredient object is refreshed to ensure its attributes match the database state. The database ingredient instance is converted to the API model instance using the method. from_orm For a quick test, a simple curl can be executed against the running app: curl -X 'POST' \ 'http://localhost:8000/api/v1/ingredients' \ -H 'accept: application/json' \ -H 'Content-Type: application/json' \ -d '{"name": "Salty water"}' In the response, there should be an ingredient object that has an ID coming from the database: { "pk":"2eb255e9-2172-4c75-9b29-615090e3250d", "name":"Salty water" } Although SQLAlchemy's multiple layers of abstraction may seem unnecessary for a simple API, they keep the ORM details separated and contribute to SQLAlchemy's efficiency and scalability. When combined with asyncio, the ORM features perform exceptionally well in the API. The remaining two endpoints are less complex and share similarities. One part that deserves a deeper explanation is the use of the method inside the function. While querying the database using SQLAlchemy, the method is often used with a query as the argument. While method returns row-like tuples, returns ORM entities directly, making the endpoint cleaner. scalars get_ingredients execute execute scalars Now, consider the potions API, in the same file: @router.post("/potions", status_code=status.HTTP_201_CREATED) async def create_potion( data: models.PotionPayload, session: AsyncSession = Depends(get_db_session), ) -> models.Potion: data_dict = data.dict() ingredients = await session.scalars( select(db_models.Ingredient).where( db_models.Ingredient.pk.in_(data_dict.pop("ingredients")) ) ) potion = db_models.Potion(**data_dict, ingredients=list(ingredients)) session.add(potion) await session.commit() await session.refresh(potion) return models.Potion.from_orm(potion) @router.get("/potions", status_code=status.HTTP_200_OK) async def get_potions( session: AsyncSession = Depends(get_db_session), ) -> list[models.Potion]: potions = await session.scalars(select(db_models.Potion)) return [models.Potion.from_orm(potion) for potion in potions] @router.get("/potions/{pk}", status_code=status.HTTP_200_OK) async def get_potion( pk: uuid.UUID, session: AsyncSession = Depends(get_db_session), ) -> models.Potion: potion = await session.get(db_models.Potion, pk) if potion is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Potion does not exist", ) return models.Potion.from_orm(potion) The GET endpoints for potions are identical to those for ingredients. However, the POST function requires additional code. This is because creating potions involves including at least one ingredient ID, which means that the ingredients must be fetched and linked to the newly created potion. To achieve this, the method is used again, but this time with a query that specifies the IDs of the fetched ingredients. The remaining part of the potion creation process is identical to that of the ingredients. scalars To test the endpoint, again a curl command can be executed. curl -X 'POST' \ 'http://localhost:8000/api/v1/potions' \ -H 'accept: application/json' \ -H 'Content-Type: application/json' \ -d '{"name": "Salty soup", "ingredients": ["0b4f1de5-e780-418d-a74d-927afe8ac954"}' It results in the following response: { "pk": "d4929197-3998-4234-a5f7-917dc4bba421", "name": "Salty soup", "ingredients": [ { "pk": "0b4f1de5-e780-418d-a74d-927afe8ac954", "name": "Salty water" } ] } It's important to notice that each ingredient is represented as a complete object within the potion, thanks to the argument specified in the relationship. lazy="selectin" The APIs are functional, but there is a major issue with the code. While SQLAlchemy gives you the freedom to interact with the database as you please, it does not offer any high-level "manager" utility similar to Django's . As a result, you will need to create it yourself, which is essentially the logic used in the ingredient and potion APIs. However, if you keep this logic directly in the endpoints without extracting it into a separate space, you will end up with a lot of duplicated code. Additionally, making changes to the queries or models will become increasingly difficult to manage. Model.objects The upcoming chapter introduces the repository pattern: an elegant solution for extracting ORM code. Repository The pattern allows to abstract away the details of working with the database. In the case of using SQLAlchemy, such as in the example of the alchemist, the repository class would be responsible for managing multiple models and interacting with the database session. repository Take a look at the following code from the file: alchemist/database/repository.py import uuid from typing import Generic, TypeVar from sqlalchemy import BinaryExpression, select from sqlalchemy.ext.asyncio import AsyncSession from alchemist.database import models Model = TypeVar("Model", bound=models.Base) class DatabaseRepository(Generic[Model]): """Repository for performing database queries.""" def __init__(self, model: type[Model], session: AsyncSession) -> None: self.model = model self.session = session async def create(self, data: dict) -> Model: instance = self.model(**data) self.session.add(instance) await self.session.commit() await self.session.refresh(instance) return instance async def get(self, pk: uuid.UUID) -> Model | None: return await self.session.get(self.model, pk) async def filter( self, *expressions: BinaryExpression, ) -> list[Model]: query = select(self.model) if expressions: query = query.where(*expressions) return list(await self.session.scalars(query)) The class holds all the logic that was previously included in the endpoints. The difference is that it allows for the specific model class to be passed in the method, making it possible to reuse the code for all models instead of duplicating it in each endpoint. DatabaseRepository __init__ Furthermore, the uses Python generics, with the generic type bounded to the abstract database model. This allows for the repository class to benefit more from static type checking. When used with a specific model, the return types of the repository methods will reflect this specific model. DatabaseRepository Model Because the repository needs to use the database session, it must be initialized along with the dependency. Consider the new dependency in the file: get_db_session alchemist/api/v2/dependencies.py from collections.abc import Callable from fastapi import Depends from sqlalchemy.ext.asyncio import AsyncSession from alchemist.database import models, repository, session def get_repository( model: type[models.Base], ) -> Callable[[AsyncSession], repository.DatabaseRepository]: def func(session: AsyncSession = Depends(session.get_db_session)): return repository.DatabaseRepository(model, session) return func Simply put, the function is a dependency factory. It first takes the database model that you will use the repository with. Then, it returns the dependency, which will be used to receive the database session and initialize the repository object. To gain a better understanding, check out the new API from the file. It shows only the POST endpoints, but it should be enough to give you a clearer idea of how the code gets improved: get_repository alchemist/api/v2/routes.py from typing import Annotated from fastapi import APIRouter, Depends, status from alchemist.api import models from alchemist.api.v2.dependencies import get_repository from alchemist.database import models as db_models from alchemist.database.repository import DatabaseRepository router = APIRouter(prefix="/v2", tags=["v2"]) IngredientRepository = Annotated[ DatabaseRepository[db_models.Ingredient], Depends(get_repository(db_models.Ingredient)), ] PotionRepository = Annotated[ DatabaseRepository[db_models.Potion], Depends(get_repository(db_models.Potion)), ] @router.post("/ingredients", status_code=status.HTTP_201_CREATED) async def create_ingredient( data: models.IngredientPayload, repository: IngredientRepository, ) -> models.Ingredient: ingredient = await repository.create(data.dict()) return models.Ingredient.from_orm(ingredient) @router.post("/potions", status_code=status.HTTP_201_CREATED) async def create_potion( data: models.PotionPayload, ingredient_repository: IngredientRepository, potion_repository: PotionRepository, ) -> models.Potion: data_dict = data.dict() ingredients = await ingredient_repository.filter( db_models.Ingredient.pk.in_(data_dict.pop("ingredients")) ) potion = await potion_repository.create({**data_dict, "ingredients": ingredients}) return models.Potion.from_orm(potion) The first important feature to note is the use of , a new way of working with FastAPI dependencies. By specifying the dependency's return type as and declaring its usage with you can end up using simple type annotations in the endpoint: . Annotated DatabaseRepository[db_models.Ingredient] Depends(get_repository(db_models.Ingredient)) repository: IngredientRepository Thanks to the repository, the endpoints don't have to store all the ORM-related burden. Even in the more complicated potion case, all you need to do is to use two repositories at the same time. You may wonder if initializing two repositories will initialize the session twice. The answer is no. FastAPI dependency system caches the same dependency calls in a single request. This means that session initialization gets cached and both repositories use the exact same session object. Yet another great feature of the combination of SQLAlchemy and FastAPI. The API is fully functional and has a reusable, high-performing data-access layer. The next step is to make sure the requirements are met by writing some end-to-end tests. Testing Tests play a crucial role in software development. Projects can contain unit, integration, and end-to-end (E2E) tests. While it is usually best to have a high number of meaningful unit tests, it is also good to write at least a few E2E tests to ensure the entire workflow is functioning correctly. To create some E2E tests for the alchemist app, two additional libraries are required: pytest to actually create and run the tests httpx to make async requests inside the tests Once these are installed, the next step is to have a separate, test database in place. You don't want your default database to be polluted or dropped. Since alchemist includes a Docker setup, only a simple script is needed to create a second database. Take a look at the code from the file: scripts/create_test_db.sh #!/bin/bash psql -U postgres psql -c "CREATE DATABASE test" In order for the script to be executed, it must be added as a volume to the Postgres container. This can be achieved by including it in the section of the file. volumes docker-compose.yaml The final step of preparation is to create pytest fixtures within the file: tests/conftest.py from collections.abc import AsyncGenerator import pytest import pytest_asyncio from fastapi import FastAPI from httpx import AsyncClient from sqlalchemy.ext.asyncio import ( AsyncSession, async_sessionmaker, create_async_engine, ) from alchemist.app import app from alchemist.config import settings from alchemist.database.models import Base from alchemist.database.session import get_db_session @pytest_asyncio.fixture() async def db_session() -> AsyncGenerator[AsyncSession, None]: """Start a test database session.""" db_name = settings.DATABASE_URL.split("/")[-1] db_url = settings.DATABASE_URL.replace(f"/{db_name}", "/test") engine = create_async_engine(db_url) async with engine.begin() as conn: await conn.run_sync(Base.metadata.drop_all) await conn.run_sync(Base.metadata.create_all) session = async_sessionmaker(engine)() yield session await session.close() @pytest.fixture() def test_app(db_session: AsyncSession) -> FastAPI: """Create a test app with overridden dependencies.""" app.dependency_overrides[get_db_session] = lambda: db_session return app @pytest_asyncio.fixture() async def client(test_app: FastAPI) -> AsyncGenerator[AsyncClient, None]: """Create an http client.""" async with AsyncClient(app=test_app, base_url="http://test") as client: yield client One thing that is essential to be changed in the tests, is how the app interacts with the database. This includes not only changing the database URL but also ensuring that each test is isolated by starting with an empty database. The fixture accomplishes both of these goals. Its body takes the following steps: db_session Create an engine with a modified database URL. Delete all existing tables to make sure that the test has a clean database. Create all the tables inside the database (same code as in the migration script). Create and yield a session object. Manually close the session when the test is done. Although the last step could also be implemented as a context manager, manual closing works just fine in this case. The two remaining fixtures should be quite self-explanatory: is the FastAPI instance from the file, with the dependency replaced with the fixture test_app alchemist/app.py get_db_session db_session is the httpx that will make API requests against the client AsyncClient test_app With all this being set up, finally the actual tests can be written. For conciseness, the example below from the file shows only a test for creating an ingredient: tests/test_api.py from fastapi import status class TestIngredientsAPI: """Test cases for the ingredients API.""" async def test_create_ingredient(self, client): response = await client.post("/api/v2/ingredients", json={"name": "Carrot"}) assert response.status_code == status.HTTP_201_CREATED pk = response.json().get("pk") assert pk is not None response = await client.get("/api/v2/ingredients") assert response.status_code == status.HTTP_200_OK assert len(response.json()) == 1 assert response.json()[0]["pk"] == pk The test uses a client object created in a fixture, that makes requests to the FastAPI instance with overridden dependency. As a result, the test is able to interact with a separate database that will be cleared after the test is done. The structure of the remaining test suite for both APIs is pretty much the same. Summary FastAPI and SQLAlchemy are excellent technologies for creating modern and powerful backend applications. The freedom, simplicity, and flexibility they offer make them one of the best options for Python-based projects. If developers follow best practices and patterns, they can create performant, robust, and well-structured applications that handle database operations and API logic with ease. This article aimed to provide you with a good understanding of how to set up and maintain this amazing combination. Sources The source code for the alchemist project can be found here: . link Also published . here