Documented testing approach does not work for SQLite in-memory instances

[posita]

First Check

• I added a very descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

• I commit to help with one of those options 👆

Example Code

# ----------------------------------------------------------------
# app.py

from __future__ import annotations

from fastapi import Depends, FastAPI
from sqlalchemy import Column, Integer, create_engine
from sqlalchemy.orm import Session, declarative_base, sessionmaker  # type: ignore

Base = declarative_base()


class DbExample(Base):  # type: ignore
    __tablename__ = "app_examples"

    example_id = Column(Integer, primary_key=True)


def create_db(url: str):
    engine = create_engine(
        url,
        connect_args={"check_same_thread": False},
        native_datetime=True,
    )
    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
    Base.metadata.create_all(bind=engine)

    return SessionLocal


SessionLocal = create_db("sqlite:///./real.db")
app = FastAPI()


def get_db():
    db = SessionLocal()

    try:
        yield db
    finally:
        db.close()


@app.get("/")
def summary(*, db: Session = Depends(get_db)):
    res = db.execute("SELECT COUNT(*) FROM app_examples").scalars()

    return {"result": next(res)}

# ----------------------------------------------------------------
# test_app.py

from __future__ import annotations

from fastapi.testclient import TestClient

from app import app, create_db, get_db

TEST_DB_URL = "sqlite://"


def test_root_with_context_manager() -> None:
    from contextlib import contextmanager

    @contextmanager
    def _session():
        TestingSessionLocal = create_db(TEST_DB_URL)
        saved_get_db = get_db in app.dependency_overrides
        old_get_db = app.dependency_overrides.get(get_db)

        def override_get_db():
            try:
                db = TestingSessionLocal()
                yield db
            finally:
                db.close()

        app.dependency_overrides[get_db] = override_get_db

        try:
            yield
        finally:
            if saved_get_db:
                app.dependency_overrides[get_db] = old_get_db

    with _session():
        client = TestClient(app)
        client.get("/")


def test_root_with_patch() -> None:
    from unittest.mock import patch

    TestingSessionLocal = create_db(TEST_DB_URL)

    with patch("app.SessionLocal", TestingSessionLocal):
        client = TestClient(app)
        client.get("/")


def test_root_by_hand() -> None:
    # Trying to follow <https://fastapi.tiangolo.com/advanced/testing-database/>
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker

    from app import Base

    engine = create_engine(
        TEST_DB_URL,
        connect_args={"check_same_thread": False},
        native_datetime=True,
    )
    TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
    Base.metadata.create_all(bind=engine)

    def override_get_db():
        try:
            db = TestingSessionLocal()
            yield db
        finally:
            db.close()

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

Description

The above is modeled after these instructions, but those do not appear to work for SQLite's in-memory databases. Running the following tests results in three errors similar to:

E       sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) no such table: app_examples
E       [SQL: SELECT COUNT(*) FROM app_examples]
E       (Background on this error at: https://sqlalche.me/e/14/e3q8)

Changing TEST_DB_URL to, e.g., sqlite:///./test.db allows those tests to pass.

Operating System

Linux, macOS

Operating System Details

No response

FastAPI Version

0.68.1

Python Version

Python 3.9.6

Additional Context

No response

[posita]

After reading up on this, I think I figured out a work-around:

# …
    def override_get_db():
        try:
            db = TestingSessionLocal()
            Base.metadata.create_all(bind=db.connection().engine)  # re-initialize the DB every time
            yield db
        finally:
            db.close()
# …

That allows the above tests to pass, but that only seems appropriate for this particular case. (More experimentation might be needed.)

UPDATE: I don't think this works. Subsequent requests using client fail with similar errors as the original post. As of right now, I have not yet identified a viable work-around. I want to start with fresh databases for each test. Right now, the only thing I can think of is use tempfile.NamedTemporaryFile and separate on-disk databases for each test.

[jnnnnn]

I got this to work following #1223 with:

from sqlalchemy.pool import StaticPool
engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)

[llamasoft]

Just a heads up that SQLAlchemy offers a SingletonThreadPool that's meant to be used for testing with :memory: databases. It maintains one connection per thread, so it also means that the {"check_same_thread": False} override shouldn't be required as each thread will use its own connection.

[chiefnoah]

Just a heads up that SQLAlchemy offers a SingletonThreadPool that's meant to be used for testing with :memory: databases. It maintains one connection per thread, so it also means that the {"check_same_thread": False} override shouldn't be required as each thread will use its own connection.

This will effectively mean you end up with n instances of your database. For in-memory sqlite, the connection is the database.

From the SQLAlchemy docs:

To use a :memory: database in a multithreaded scenario, the same connection object must be shared among threads, since the database exists only within the scope of that connection. The StaticPool implementation will maintain a single connection globally, and the check_same_thread flag can be passed to Pysqlite as False

Using SingletonThreadPool may be fine depending on your use-case, but I found it inadequate for ours due to segmented data. As the docs state, you could use StaticPool and the flag and then handle the locking yourself (yuck) or switch to using a tempfile and SingletonThreadPool. Using StaticPool with check_same_thread set to False can result in data corruption and random test failures unless you're careful.

[nathanetech]

@posita Did you find a solution to this? I'm currently running up against the same issue.

One thing I did find is that if I change SQLALCHEMY_DATABASE_URL = "sqlite://" to SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db", all other things being the same, it will successfully create a SQLite DB in my project with the tables. I'm currently not sure why that works while an in-memory DB is created without tables.

[posita]

I did not find a solution. We actually moved away from SQLAlchemy since we were fighting too much friction.

[Dentosal]

Using this solution works for me:

from sqlalchemy.pool import StaticPool
...
engine = create_engine(
    "sqlite:///:memory:", connect_args={"check_same_thread": False},
    poolclass=StaticPool
)