Enabling SQLite foreign keys and WAL mode with aiosqlite

SQLite disables foreign keys on every new connection, so add a connect event listener that issues PRAGMA foreign_keys = ON — then set journal_mode = WAL once per file and a busy_timeout per connection, and use StaticPool for :memory: databases so the schema survives. This guide belongs to selecting async drivers for SQLite, MySQL and Postgres.

Quick Answer

A test suite on SQLite with default pragmas does not enforce foreign keys, which means constraint bugs pass every test and appear in production.

Before — defaults, so nothing is enforced:

from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine("sqlite+aiosqlite:///./test.db")

# An order referencing a customer that does not exist is accepted.
# A cascade delete removes nothing, because the foreign key is not enforced.
# Two concurrent writes: sqlite3.OperationalError: database is locked

After — pragmas applied on every connection:

from sqlalchemy import event, text
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

engine = create_async_engine(
    "sqlite+aiosqlite:///./test.db",
    connect_args={"timeout": 15},        # seconds the driver waits for a lock
)


@event.listens_for(engine.sync_engine, "connect")
def _sqlite_pragmas(dbapi_connection, connection_record) -> None:
    cursor = dbapi_connection.cursor()
    try:
        cursor.execute("PRAGMA foreign_keys = ON")     # per connection, resets to OFF
        cursor.execute("PRAGMA busy_timeout = 5000")   # per connection, milliseconds
    finally:
        cursor.close()


async def set_wal_mode() -> None:
    """Persistent in the database file: set it once, not per connection."""
    async with engine.begin() as conn:
        await conn.execute(text("PRAGMA journal_mode = WAL"))


Session = async_sessionmaker(engine, expire_on_commit=False)

The listener is registered on engine.sync_engine, because pool and connection events live on the synchronous engine the async engine wraps. Inside it, the cursor calls work through SQLAlchemy's greenlet bridge, so ordinary synchronous code is correct there.

Execution Context & Async Workflow Integration

SQLite's pragmas divide into two kinds, and the division decides where each one belongs.

Per-connection pragmas apply to one connection and reset when it closes. foreign_keys is the important one: SQLite has enforced foreign keys since 3.6.19, and defaults to off for backwards compatibility, on every connection. A pooled engine opens several connections over its life, so setting it once after create_async_engine covers one connection and misses the rest — which is why the connect event is the only reliable place. busy_timeout and synchronous are also per connection.

Persistent pragmas are stored in the database file. journal_mode = WAL is the notable one: set it once and every future connection to that file uses WAL, including from other processes. Setting it on every connect is harmless and unnecessary.

WAL mode changes the concurrency story in a way worth understanding, because it is often oversold. In the default rollback-journal mode, a writer blocks readers. In WAL mode, readers continue while a writer is active, which is a genuine improvement. What does not change is that only one writer can hold the database at a time — SQLite serialises writers with a lock, and a second writer either waits (up to busy_timeout) or fails with database is locked.

That single-writer limit is what makes SQLite unable to stand in for PostgreSQL in any test that exercises write concurrency: SELECT ... FOR UPDATE SKIP LOCKED, the job-queue pattern in building a job queue with SELECT FOR UPDATE SKIP LOCKED, optimistic-locking races, and deadlock handling all need a real database.

aiosqlite itself is a thread wrapper around the standard library's sqlite3, so the async API does not make SQLite concurrent — it makes it awaitable. For a test suite that is exactly right: the calls integrate with pytest-asyncio and the same application code runs unchanged, which is the whole reason to use it, as using aiosqlite for async tests and local development describes.

One async-specific detail: because the pragmas are applied in a connect listener, they also apply to connections opened by Alembic during a test-suite migration run — which matters, since a migration that relies on foreign keys being enforced would otherwise behave differently from the application.

Resolving Warnings, Errors & Common Mistakes

The in-memory case is the one that produces the most confusing error, because the schema appears to be created and then is not there:

from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.pool import StaticPool

engine = create_async_engine(
    "sqlite+aiosqlite:///:memory:",
    poolclass=StaticPool,                       # one connection, shared
    connect_args={"check_same_thread": False},  # aiosqlite uses a worker thread
)

:memory: means "a fresh, private database for this connection". Without StaticPool, the connection that ran create_all() is returned to the pool, a second connection is opened for the first query, and that connection's database is empty. StaticPool keeps exactly one connection for the engine's lifetime, which is what makes an in-memory database usable at all — and also means no concurrency, which is fine for a test.

check_same_thread=False is needed because aiosqlite runs the underlying sqlite3 connection in a worker thread, and sqlite3 refuses cross-thread use by default.

Batch mode is the other thing a SQLite-backed test suite needs from Alembic, because migrations written for PostgreSQL routinely alter columns:

from alembic import op
import sqlalchemy as sa


def upgrade() -> None:
    with op.batch_alter_table("orders") as batch:
        batch.alter_column("total_cents", existing_type=sa.Integer(), type_=sa.BigInteger())

Batch mode recreates the table with the new definition and copies the rows, which SQLite can do. It also needs constraints to be named, which a metadata naming convention provides.

Advanced: A Fixture That Behaves Like the Real Thing

When SQLite is used for tests, the goal is for a test that passes to mean something. That takes three things beyond the pragmas: the same schema-creation path, per-test isolation, and an explicit marker for tests that cannot run there.

import pytest
import pytest_asyncio
from sqlalchemy import event, text
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.pool import StaticPool

from shop.models import Base


def _install_pragmas(engine) -> None:
    @event.listens_for(engine.sync_engine, "connect")
    def _pragmas(dbapi_connection, connection_record) -> None:
        cursor = dbapi_connection.cursor()
        try:
            cursor.execute("PRAGMA foreign_keys = ON")
            cursor.execute("PRAGMA busy_timeout = 5000")
        finally:
            cursor.close()


@pytest_asyncio.fixture
async def sqlite_engine():
    engine = create_async_engine(
        "sqlite+aiosqlite:///:memory:",
        poolclass=StaticPool,
        connect_args={"check_same_thread": False},
    )
    _install_pragmas(engine)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    await engine.dispose()


@pytest_asyncio.fixture
async def sqlite_session(sqlite_engine):
    factory = async_sessionmaker(sqlite_engine, expire_on_commit=False)
    async with factory() as session:
        yield session


@pytest.mark.asyncio
async def test_foreign_keys_are_enforced(sqlite_session):
    from sqlalchemy.exc import IntegrityError

    from shop.models import Order

    sqlite_session.add(Order(customer_id=999_999, total_cents=100))
    with pytest.raises(IntegrityError):
        await sqlite_session.commit()

That last test is worth keeping permanently. It asserts that the pragma listener is installed and working, so a refactor that moves the fixture cannot silently turn foreign-key enforcement off again — which is the failure this whole guide exists to prevent, and one that produces no error when it happens.

A per-test in-memory database gives isolation for free: each test gets a new engine, a new database and a new schema. The cost is create_all() per test, which for a large model is tens of milliseconds. Where that matters, the alternative is a file database created once per session with per-test transaction rollback — the pattern in rolling back database state between async tests.

Note that create_all() builds the schema the models describe, which is not necessarily the schema the migrations produce. For a suite that uses SQLite, that gap is usually accepted deliberately: the migration path is tested against PostgreSQL, where it will actually run.

Deciding What SQLite May Test

The useful question is not "SQLite or PostgreSQL" but "which tests may run on SQLite". Drawing that line explicitly, and enforcing it with markers, is what keeps a fast suite from becoming a misleading one.

SQLite is suitable for tests of code that uses only portable constructs: business logic over simple selects and inserts, validation, mapping behaviour, cascades (once foreign keys are on), and anything where the database is incidental to the assertion.

SQLite is unsuitable for anything this site's PostgreSQL-specific guidance covers: JSONB and array operators, full-text search, ON CONFLICT upsert semantics, row-level security, SKIP LOCKED, window frames, recursive-CTE cycle handling, timezone-aware timestamp behaviour, and every error code an application branches on.

Making the distinction mechanical means a marker and a fixture per backend:

# conftest.py
import pytest


def pytest_configure(config) -> None:
    config.addinivalue_line("markers", "postgres: requires a real PostgreSQL instance")


# In a test module:
@pytest.mark.postgres
@pytest.mark.asyncio
async def test_jsonb_containment_filter(session, product_factory):
    ...

Running pytest -m "not postgres" then gives the fast suite, and CI runs everything. The marker names the reason, so a reader knows why the test is restricted rather than guessing.

One trap worth naming: a test that happens to pass on SQLite because the feature degraded silently. A JSONB column becomes JSON text on SQLite, so a containment filter may run and return nothing rather than raising, and the test asserting "no results" passes for the wrong reason. Tests should assert the positive case — that the matching row was found — which fails loudly on a backend that cannot do the query.

Finally, be honest about what the fast suite is for. It gives quick feedback on logic during development; it does not tell you the application works. The deployment gate should be the PostgreSQL run, using the container fixture in running tests against a Postgres testcontainer — and if maintaining two backends in the suite starts costing more than the speed is worth, running everything against a container and keeping a warm one locally is a perfectly good answer.

Frequently Asked Questions

Related

• Selecting Async Drivers for SQLite, MySQL and Postgres — The parent guide: driver options per database.
• Using aiosqlite for async tests and local development — When SQLite is the right test backend.
• Running tests against a Postgres testcontainer — The fixture for everything SQLite cannot test.
• Using async drivers for Oracle and SQL Server — The other thread-backed async drivers.