#240: Asynchronous SQLAlchemy With FastAPI

Last week we got pytest to run asynchronous test methods. That was the preparation step for this post where we switch to asynchronous SQLAlchemy for our to-do application. As it turns out, switching to asynchronous methods for SQLAlchemy takes a lot of work. Let us get through the different changes we need to make.

Install the asynchronous SQLite driver

The default driver for SQLite only works with synchronous commands. To access SQLite with the asynchronous engine, we need to install this package:

pip install aiosqlite

Update the data store tests

We start our change by turning our tests for the data store into asynchronous methods. As we learned last week, this consists of these 4 main steps:

Add the async keyword in front of our test methods.
Add the async decorator to our test methods.
Add an await in front of all method calls to our data store.
Add the async fixture decorator to our fixture.

In our fixture we call the function create_async_session_factory() to get the asynchronous session factory. While we could work with the session in our data store in the synchronous world, we cannot do the same with the asynchronous engine. If we try, we get an endless list of errors about closed transactions and other problems. Therefore, we will put the factory inside our data store.

import os

import pytest_asyncio

from ..data.database import create_async_session_factory, create_session_factory
from ..data.datastore_db import DataStoreDb
from ..models.todo import TaskInput
from datetime import date, datetime, timedelta
import pytest


@pytest_asyncio.fixture()
async def with_db():
    db_file = os.path.join(
        os.path.dirname(__file__),
        '..',
        'db',
        'test_db.sqlite')

    factory = await create_async_session_factory(db_file)
    yield factory


@pytest.mark.asyncio
async def test_can_add_entry(with_db):
    current_time = datetime.now()
    entry = TaskInput(name="a simple task", 
                      priority=1, 
                      due_date=date.today(), 
                      done=False)
    store = DataStoreDb(with_db)

    data = await store.add(entry)

    assert data.name == "a simple task"
    assert data.priority == 1
    assert data.due_date == date.today()
    assert data.done == False
    assert data.created_at == date.today()
    assert data.id >= 1


@pytest.mark.asyncio
async def test_can_add_multiple_entries(with_db):
    entry_a = TaskInput(name="a simple task", 
                        priority=1, 
                        due_date=date.today(), 
                        done=False)
    entry_b = TaskInput(name="b simple task", 
                        priority=2, 
                        due_date=date.today(), 
                        done=False)
    store = DataStoreDb(with_db)

    data_a = await store.add(entry_a)
    data_b = await store.add(entry_b)

    assert data_a.id < data_b.id


@pytest.mark.asyncio
async def test_can_get_specific_entry_back(with_db):
    entry_a = TaskInput(name="Find a specific task", 
                        priority=1, 
                        due_date=date.today(), 
                        done=False)
    store = DataStoreDb(with_db)
    saved = await store.add(entry_a)

    entry = await store.get(saved.id)

    assert saved == entry


@pytest.mark.asyncio
async def test_missing_entry_gets_None_back(with_db):
    store = DataStoreDb(with_db)

    entry = await store.get(-1)

    assert entry == None


@pytest.mark.asyncio
async def test_can_get_all_entrries_back(with_db):
    entry_a = TaskInput(name="a simple task", 
                        priority=1, 
                        due_date=date.today(), 
                        done=False)
    entry_b = TaskInput(name="b simple task", 
                        priority=2, 
                        due_date=date.today(), 
                        done=False)
    entry_c = TaskInput(name="b simple task", 
                        priority=2, 
                        due_date=date.today(), 
                        done=False)
    store = DataStoreDb(with_db)
    await store.add(entry_a)
    await store.add(entry_b)
    await store.add(entry_c)

    entries = await store.all()

    assert len(entries) >= 3


@pytest.mark.asyncio
async def test_can_delete_entry(with_db):
    entry_a = TaskInput(name="a simple task", 
                        priority=1, 
                        due_date=date.today(), 
                        done=False)
    store = DataStoreDb(with_db)
    saved = await store.add(entry_a)
    id = saved.id
    print(id)
    await store.delete(id)

    result = await store.get(id)
    assert result == None


@pytest.mark.asyncio
async def test_can_update_entry(with_db):
    old = TaskInput(name="a simple task", 
                    priority=1, 
                    due_date=date.today(), 
                    done=False)
    store = DataStoreDb(with_db)
    old_saved = await store.add(old)

    new = TaskInput(name="b simple task", 
                    priority=2, 
                    due_date=date.today() + timedelta(days=2), 
                    done=True)
    await store.update(old_saved.id, new)

    entry = await store.get(old_saved.id)
    assert entry.name == "b simple task"
    assert entry.priority == 2
    assert entry.due_date == date.today() + timedelta(days=2)
    assert entry.done == True


@pytest.mark.asyncio
async def test_non_existing_entry_cannot_be_updated(with_db):
    store = DataStoreDb(with_db)

    new = TaskInput(name="b simple task", 
                    priority=2, 
                    due_date=date.today() + timedelta(days=2), 
                    done=True)
    with pytest.raises(ValueError) as e_info:
        await store.update(-123, new)
    assert str(e_info.value) == "no taks known with id '-123'"


@pytest.mark.asyncio
async def test_fetches_statistics(with_db):
    store = DataStoreDb(with_db)
    await store.add(TaskInput(name="counter", 
                              priority=1, 
                              due_date=date.today(), 
                              done=False))
    stats = await store.get_statistics()

    assert stats.total_tasks == stats.total_open + stats.total_done
    assert stats.total_tasks >= 1

Create the asynchronous session factory

In our data/database.py file, we can add the new method to create an asynchronous session factory:

from typing import AsyncIterator
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine


def create_session_factory(db_file: str) -> sessionmaker:
    engine = create_engine(
        'sqlite:///' + db_file, connect_args={"check_same_thread": False}, echo=False
    )

    factory = sessionmaker(autocommit=False, autoflush=False, bind=engine)
    return factory


async def create_async_session_factory(db_file: str) -> AsyncIterator[AsyncSession]:
    engine = create_async_engine(
        'sqlite+aiosqlite:///' + db_file, connect_args={"check_same_thread": False}, echo=False
    )

    factory = async_sessionmaker(engine, autocommit=False, autoflush=False, expire_on_commit=False, class_=AsyncSession)
    return factory

We now use our new driver to access the SQLite database, but otherwise the method works as before.

Change the DataStoreDb

For our DataStoreDb class, we need to make these adjustments:

Use AsyncIterator[AsyncSession] instead of the session in the constructor.
Add the async keyword in front of all public methods.
Wrap an async with block around everything that uses the database.
Rewrite the queries to run asynchronously.
In the statistics method, we turn the result cursor into a list to prevent a cursor error.

With all those changes, our DataStoreDb class now looks like this:

from typing import AsyncIterator
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import func, select
from datetime import date, datetime

from ..models.statistics import StatisticOverview
from ..models.todo import TaskInput, TaskOutput
from .entities import Task

class DataStoreDb:
    def __init__(self, db: AsyncIterator[AsyncSession]):
        self.db = db


    async def add(self, entry: TaskInput) -> TaskOutput:
        async with self.db() as session:
            task = Task(id=None, created_at=datetime.now(), **dict(entry))        
            session.add(task)
            await session.commit()

            return self.__to_output(task) 


    async def get(self, id: int) -> TaskOutput:
        async with self.db() as session:
            query = select(Task).where(Task.id == id)
            result = await session.scalar(query)

        if result:
            return self.__to_output(result)
        else:
            return None


    async def all(self):
        async with self.db() as session:
            query = select(Task)
            entries = await session.scalars(query)

            results = []

            for entry in entries:
                results.append(self.__to_output(entry))

            return results


    async def delete(self, id: int) -> None:
        async with self.db() as session:
            query = select(Task).where(Task.id == id)
            entry = await session.scalar(query)

            if entry:
                await session.delete(entry)
                await session.commit()


    async def update(self, id: int, update: TaskInput) -> TaskOutput:
        async with self.db() as session:
            query = select(Task).where(Task.id == id)
            entry = await session.scalar(query)

            if entry: 
                entry.name = update.name
                entry.priority = update.priority
                entry.due_date = update.due_date
                entry.done = update.done
                await session.commit()

                return self.__to_output(entry)
            else: 
                raise ValueError(f"no taks known with id '{id}'")


    async def get_statistics(self) -> StatisticOverview:
        async with self.db() as session:
            query = (
                select(
                    func.count("*").label("total"),
                    func.count("*").filter(Task.done==True).label("done"),
                    func.count("*").filter(Task.done==False).label("open") ,
                )
            )
            result_db = await session.execute(query)

            #https://stackoverflow.com/questions/36515882/command-cursor-object-is-not-subscriptable
            result = list(result_db)[0]

            return StatisticOverview(total_tasks=result[0], total_done=result[1], total_open=result[2])


    def __to_output(self, entity: Task) -> TaskOutput:
        return TaskOutput(id=entity.id, 
                          name=entity.name, 
                          priority=entity.priority, 
                          due_date=entity.due_date,
                          done=entity.done, 
                          created_at=date.today())

We can now run our tests for the datastore, and they all should pass. However, the tests for FastAPI application will fail. Let us fix that.

Fix the endpoint tests

The tests for our FastAPI endpoints need an adjustment for the override_get_db() function. All the methods need the await keyword and the marker, and we need to await the call to the prepare_task() function:

from datetime import date, timedelta
import os
import re
from bs4 import BeautifulSoup
from fastapi.testclient import TestClient
import pytest

from ..dependencies import get_db
from ..data.datastore_db import DataStoreDb
from ..data.database import create_async_session_factory
from ..main import app

import logging
logging.getLogger("httpx").setLevel(logging.WARNING)

async def override_get_db():
    db_file = os.path.join(
        os.path.dirname(__file__),
        '..',
        'db',
        'test_db.sqlite')
    factory = await create_async_session_factory(db_file)
    db = DataStoreDb(factory)
    yield db


client = TestClient(app)
app.dependency_overrides[get_db] = override_get_db

@pytest.mark.asyncio
async def test_create_task():
    data = {
        "name": "A first task",
        "priority": 5,
        "due_date": str(date.today() + timedelta(days=1)),
        "done": False
    }

    response = client.post("/api/todo/", json=data)
    assert response.status_code == 201
    result = response.json()
    assert result['id'] > 0
    assert result['done'] == False
    assert result['created_at'] == str(date.today())
    assert result['name'] == data['name']
    assert result['priority'] == data['priority']
    assert result['due_date'] == data['due_date']
    assert f"http://testserver/api/todo/{result['id']}" == response.headers['location']


@pytest.mark.asyncio
async def prepare_task(name, priority=4, due_date=None, done=False):
    if due_date == None:
        due_date = date.today() + timedelta(days=1)

    data = {
        "name": name,
        "priority": priority,
        "due_date": str(due_date),
        "done": done
    }

    prepare_response = client.post("/api/todo/", json=data)
    assert prepare_response.status_code == 201
    return prepare_response.json()['id']


@pytest.mark.asyncio
async def test_show_task():
    name = "A second task"
    id = await prepare_task(name)

    response = client.get(f"/api/todo/{id}")
    assert response.status_code == 200
    details = response.json()
    assert details['name'] == name


@pytest.mark.asyncio
async def test_show_task_where_task_is_unknown():
    response = client.get(f"/api/todo/-1")
    assert response.status_code == 404
    assert response.json()['detail'] == "Task not found"


@pytest.mark.asyncio
async def test_update_task():
    id = await prepare_task("original")

    update = {
        "name": "An updated task",
        "priority": 5,
        "due_date": str(date.today() + timedelta(days=2)),
        "done": False
    }

    response = client.put(f"/api/todo/{id}", json=update)
    assert response.status_code == 200
    assert response.json()['name'] == "An updated task"

    check = client.get(f"/api/todo/{id}")
    assert check.json()['name'] == "An updated task"


@pytest.mark.asyncio
async def test_delete_task():
    id = await prepare_task("to delete")
    response = client.delete(f"/api/todo/{id}")
    assert response.status_code == 204

    check = client.get(f"/api/todo/{id}")
    assert check.status_code == 404


@pytest.mark.asyncio
async def test_main_page_shows_info_message():
    response = client.get("/")

    assert response.status_code == 200
    assert response.json()['message'] == "The minimalistic ToDo API"


@pytest.mark.asyncio
async def test_show_all_tasks():
    await prepare_task("a first task")
    await prepare_task("a second task")
    await prepare_task("a third task")

    response = client.get("/api/todo")

    assert response.status_code == 200
    tasks = response.json()
    assert len(tasks) >= 3


@pytest.mark.asyncio
async def test_show_all_tasks_that_are_not_done():
    await prepare_task("a finished task", done=True)
    await prepare_task("an open task", done=False)

    response = client.get("/api/todo?include_done=false")

    assert response.status_code == 200
    tasks = response.json()
    done = [task for task in tasks if task['done'] == True]
    assert len(done) == 0


@pytest.mark.asyncio
async def test_show_all_tasks_that_are_due_within_five_days():
    await prepare_task("in 10 days", due_date=date.today() + timedelta(days=10))

    response = client.get(f"/api/todo?due_before={date.today() + timedelta(days=5)}")

    assert response.status_code == 200
    tasks = response.json()
    done = [task for task in tasks if date.fromisoformat(task['due_date']) > date.today() + timedelta(days=5)]
    assert len(done) == 0


@pytest.mark.asyncio
async def test_docs_endpoint_works():
    response = client.get("/openapi.json")
    # No exception -> test passes


@pytest.mark.asyncio
async def test_about_page():
    response = client.get("/about")
    assert response.status_code == 200

    soup = BeautifulSoup(response.text, 'html.parser')
    assert soup.title.text == "About To-Do Task API"
    assert soup.body.h1.text == "About"


@pytest.mark.asyncio
async def test_dashboard():
    response = client.get("/dashboard")
    assert response.status_code == 200

    soup = BeautifulSoup(response.text, 'html.parser')
    assert soup.title.text == "Dashboard To-Do Task API"
    assert soup.body.h1.text == "Dashboard"
    numbers = re.findall(r"\d+", soup.body.p.text)
    assert  int(numbers[0]) == int(numbers[1]) + int(numbers[2])

Fix the dependencies.py code

We need to switch to the create_async_session_factory() function and change how we initialise our data store:

import os

from .data.database import create_async_session_factory
from .data.datastore_db import DataStoreDb


async def get_db():
    """
    Creates the datastore 
    """
    db_file = os.path.join(
        os.path.dirname(__file__),
        '.',
        'db',
        'todo_api.sqlite')

    factory = await create_async_session_factory(db_file)
    db = DataStoreDb(factory)

    yield db

Fix the FastAPI endpoints

In all our endpoints we need to await the call to the db.* methods, the rest of the files can stay the same.

Our main.py file has one place to await the call to the db.get_statistics() method:

from fastapi import Depends, FastAPI, Request
from fastapi.encoders import jsonable_encoder
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates

from .data.datastore_db import DataStoreDb
from .dependencies import get_db
from .routers import todo
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent

app = FastAPI()
app.include_router(todo.router, prefix="/api/todo")

app.mount("/static", StaticFiles(directory=str(Path(BASE_DIR, 'static'))), name="static")

templates = Jinja2Templates(directory=str(Path(BASE_DIR, 'templates')))

@app.get("/about", response_class=HTMLResponse)
async def about(request: Request):
    return templates.TemplateResponse(
        request=request, name="about.html"
    )

@app.get("/", include_in_schema=False)
async def main():
    return {'message':'The minimalistic ToDo API'}


@app.get("/dashboard", include_in_schema=False)
async def dashboard(request: Request, db: DataStoreDb = Depends(get_db)):
    stats = await db.get_statistics()
    return templates.TemplateResponse(
        request=request, name="dashboard.html", context=jsonable_encoder(stats)
    )

In the routers/todo.py file, we need an await in every method:

from datetime import date, timedelta
from typing import Annotated, List

from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse

from ..dependencies import get_db

from ..models.todo import TaskOutput, TaskInput
from ..data.datastore_db import DataStoreDb

router = APIRouter()


async def filter_parameters(q: str | None = None, 
                            include_done: bool = True, 
                            due_before: date = date.today() + timedelta(days=365)):
    return {"q": q, "include_done": include_done, "due_before": due_before }


@router.get("/")
async def show_all_tasks(filter: Annotated[dict, Depends(filter_parameters)], 
                         db: DataStoreDb = Depends(get_db)) -> List[TaskOutput]:
    result = await db.all()

    if not filter["include_done"]:
        result = [item for item in result if item.done == False ]

    result = [item for item in result if item.due_date <= filter["due_before"] ]

    return result


@router.post("/", status_code=status.HTTP_201_CREATED)
async def create_task(task: TaskInput, 
                      request: Request, 
                      db: DataStoreDb = Depends(get_db)) -> TaskOutput:
    result = await db.add(task)
    headers = {"Location": f"{request.base_url}api/todo/{result.id}"}
    return JSONResponse(content=jsonable_encoder(result), 
                        status_code=status.HTTP_201_CREATED,
                        headers=headers)


@router.get("/{id}")
async def show_task(id: int, 
                    db: DataStoreDb = Depends(get_db)) -> TaskOutput:
    result = await db.get(id)

    if result:
        return result
    else:
        raise HTTPException(status_code=404, detail="Task not found")


@router.put("/{id}")
async def update_task(id: int, 
                      task: TaskInput, 
                      db: DataStoreDb = Depends(get_db)) -> TaskOutput:
    try:
        result = await db.update(id, task)
        return result
    except ValueError:
        raise HTTPException(status_code=404, detail="Task not found")


@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_task(id: int, 
                      db: DataStoreDb = Depends(get_db)) -> None:
    await db.delete(id)
    return Response(status_code=status.HTTP_204_NO_CONTENT)

We can now run all tests and they should pass. Without those tests we would need to spend an awful amount of time to run the application by hand. Therefore, make sure that you have a good test coverage for your code before you start such a massive change.