Fix: aiosqlite Not Working — Single Writer, WAL Mode, Row Factory, and Connection Patterns

The Error

You hit database is locked under any concurrent write:

async with aiosqlite.connect("app.db") as db:
    await db.execute("INSERT INTO posts (title) VALUES (?)", ("hi",))
    await db.commit()
# Two requests in flight → second one hangs or errors.

Or rows come back as tuples instead of dicts:

async with aiosqlite.connect("app.db") as db:
    async with db.execute("SELECT id, name FROM users") as cursor:
        async for row in cursor:
            print(row[0], row[1])  # No row["name"]

Or foreign keys are silently not enforced:

await db.execute(
    "INSERT INTO posts (user_id) VALUES (?)",
    (99999,),  # nonexistent user
)
await db.commit()  # Succeeds! Foreign key not checked.

Or datetime fields return as strings:

async with db.execute("SELECT created_at FROM posts LIMIT 1") as cur:
    row = await cur.fetchone()
    print(type(row[0]))  # <class 'str'>, not datetime

Why This Happens

aiosqlite is an asyncio wrapper around Python’s sqlite3. Underneath, it runs SQLite in a background thread per connection and uses callbacks to bridge to asyncio. SQLite itself has limits:

• Single-writer. SQLite allows multiple readers but only one writer at a time. With journal_mode=DELETE (default), writers also block readers. WAL mode fixes most of this.
• No automatic foreign key checks. PRAGMA foreign_keys is off by default per connection.
• No native datetime. SQLite stores dates as text/int. Python’s sqlite3 can auto-convert with detect_types.
• Connections are stateful. PRAGMA settings, transactions, prepared statements — all per-connection. Reusing connections across requests requires care.

aiosqlite adds asyncio compatibility but doesn’t fix SQLite’s underlying limitations.

Fix 1: Enable WAL Mode

WAL (Write-Ahead Logging) lets readers proceed during writes:

import aiosqlite

async def setup_db():
    async with aiosqlite.connect("app.db") as db:
        await db.execute("PRAGMA journal_mode = WAL")
        await db.execute("PRAGMA synchronous = NORMAL")
        await db.commit()

PRAGMA journal_mode = WAL sticks — it’s a database-file property, not per-connection. Run once during setup; all future connections use WAL.

PRAGMA synchronous = NORMAL trades a tiny crash-safety margin for big throughput gains. For most apps, the trade is worth it.

Other useful pragmas:

await db.execute("PRAGMA foreign_keys = ON")          # Per-connection
await db.execute("PRAGMA busy_timeout = 5000")        # 5 seconds before locked
await db.execute("PRAGMA temp_store = MEMORY")        # Faster temp tables
await db.execute("PRAGMA cache_size = -64000")        # 64 MB cache

busy_timeout is critical for “database is locked” — instead of failing immediately, SQLite waits this long for the lock.

Pro Tip: Use a single setup function called once at startup that opens the DB, sets pragmas, runs migrations. Subsequent connections inherit WAL (it’s persistent) but still need per-connection foreign_keys.

Fix 2: Connection Patterns — Per-Request vs Shared

For asyncio web apps (FastAPI/Starlette), there are three patterns:

Pattern A — connection per request (simple, safe):

from fastapi import FastAPI, Depends
import aiosqlite

app = FastAPI()

async def get_db():
    async with aiosqlite.connect("app.db") as db:
        await db.execute("PRAGMA foreign_keys = ON")
        db.row_factory = aiosqlite.Row
        yield db

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: aiosqlite.Connection = Depends(get_db)):
    async with db.execute("SELECT * FROM users WHERE id = ?", (user_id,)) as cur:
        row = await cur.fetchone()
        return dict(row) if row else None

Pros: simple, per-request transaction isolation. Cons: connection setup overhead per request (~1-5 ms).

Pattern B — shared connection (FAST but care needed):

import aiosqlite
from contextlib import asynccontextmanager
from fastapi import FastAPI

db: aiosqlite.Connection | None = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global db
    db = await aiosqlite.connect("app.db")
    await db.execute("PRAGMA journal_mode = WAL")
    await db.execute("PRAGMA foreign_keys = ON")
    db.row_factory = aiosqlite.Row
    yield
    await db.close()

app = FastAPI(lifespan=lifespan)

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    async with db.execute("SELECT * FROM users WHERE id = ?", (user_id,)) as cur:
        row = await cur.fetchone()
        return dict(row) if row else None

Pros: zero connection overhead, fastest. Cons: single connection serializes all queries — you lose concurrency.

Pattern C — pool (best for high concurrency):

# aiosqlite doesn't include a built-in pool; use a small one yourself:
import asyncio
import aiosqlite

class SqlitePool:
    def __init__(self, db_path: str, size: int = 5):
        self.db_path = db_path
        self.size = size
        self._pool = asyncio.Queue(maxsize=size)
    
    async def init(self):
        for _ in range(self.size):
            conn = await aiosqlite.connect(self.db_path)
            await conn.execute("PRAGMA foreign_keys = ON")
            conn.row_factory = aiosqlite.Row
            await self._pool.put(conn)
    
    async def acquire(self):
        return await self._pool.get()
    
    async def release(self, conn):
        await self._pool.put(conn)
    
    async def close(self):
        while not self._pool.empty():
            conn = await self._pool.get()
            await conn.close()

pool = SqlitePool("app.db", size=5)
await pool.init()

# Per request:
conn = await pool.acquire()
try:
    async with conn.execute("SELECT * FROM users") as cur:
        rows = await cur.fetchall()
finally:
    await pool.release(conn)

A pool lets N concurrent queries run, capped at N connections to SQLite.

Pro Tip: For most apps, Pattern A is fine — connection overhead is small compared to query time. Switch to Pattern B/C only if profiling shows connection setup as a bottleneck.

Fix 3: Row Factory for Dict-Like Access

async with aiosqlite.connect("app.db") as db:
    db.row_factory = aiosqlite.Row
    
    async with db.execute("SELECT id, name FROM users WHERE id = ?", (1,)) as cur:
        row = await cur.fetchone()
        if row:
            print(row["name"])    # Dict-like access
            print(row[0])          # Index access still works
            print(dict(row))       # Convert to plain dict

aiosqlite.Row (same as sqlite3.Row) is a tuple-like object with named-field access. Setting row_factory is per-connection.

For Pydantic models:

from pydantic import BaseModel

class User(BaseModel):
    id: int
    name: str

async with db.execute("SELECT id, name FROM users WHERE id = ?", (1,)) as cur:
    row = await cur.fetchone()
    if row:
        user = User(**dict(row))

For mass conversion:

async with db.execute("SELECT id, name FROM users") as cur:
    rows = await cur.fetchall()
    users = [User(**dict(row)) for row in rows]

Common Mistake: Forgetting row_factory = aiosqlite.Row per connection. The default returns plain tuples. Set it in your connection setup function.

Fix 4: Datetime Handling

SQLite has no native datetime — values stored as text/int. To convert in Python:

import aiosqlite
import sqlite3

# Detect types from column declarations or queries:
db = await aiosqlite.connect(
    "app.db",
    detect_types=sqlite3.PARSE_DECLTYPES | sqlite3.PARSE_COLNAMES,
)

await db.execute("""
    CREATE TABLE posts (
        id INTEGER PRIMARY KEY,
        title TEXT,
        created_at TIMESTAMP
    )
""")

With PARSE_DECLTYPES, the TIMESTAMP column type triggers automatic datetime conversion.

For column aliases:

async with db.execute('SELECT created_at AS "ts [timestamp]" FROM posts') as cur:
    row = await cur.fetchone()
    print(type(row[0]))  # datetime

The [timestamp] suffix on the alias is the explicit type hint for PARSE_COLNAMES.

For explicit conversion without detect_types:

from datetime import datetime

async def fetch_posts():
    async with db.execute("SELECT id, created_at FROM posts") as cur:
        rows = await cur.fetchall()
        return [
            {"id": r[0], "created_at": datetime.fromisoformat(r[1])}
            for r in rows
        ]

Pro Tip: Recent Python (3.12+) deprecated the default datetime adapter. Use register_adapter + register_converter for explicit control:

import sqlite3
from datetime import datetime, date

sqlite3.register_adapter(datetime, lambda d: d.isoformat())
sqlite3.register_adapter(date, lambda d: d.isoformat())
sqlite3.register_converter("timestamp", lambda b: datetime.fromisoformat(b.decode()))
sqlite3.register_converter("date", lambda b: date.fromisoformat(b.decode()))

These run for all sqlite3 (and aiosqlite) connections.

Fix 5: Foreign Keys

Enable on every connection:

await db.execute("PRAGMA foreign_keys = ON")

This is per-connection. Doing it once at app startup is not enough — every new connection needs it.

To verify:

async with db.execute("PRAGMA foreign_keys") as cur:
    row = await cur.fetchone()
    print(row[0])  # 1 = on, 0 = off

For schemas with foreign key constraints to be enforced, this must be ON. Schema definition itself is correct without the PRAGMA — it just doesn’t enforce.

CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT);
CREATE TABLE posts (
    id INTEGER PRIMARY KEY,
    user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    title TEXT
);

REFERENCES users(id) ON DELETE CASCADE — with foreign_keys = ON, deleting a user deletes their posts. Without it, the user deletes but posts orphan.

Fix 6: Transactions and BEGIN

aiosqlite (like sqlite3) starts implicit transactions on writes. To control explicitly:

async with aiosqlite.connect("app.db") as db:
    await db.execute("BEGIN")
    try:
        await db.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
        await db.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
        await db.execute("COMMIT")
    except Exception:
        await db.execute("ROLLBACK")
        raise

Or use isolation_level=None for autocommit:

async with aiosqlite.connect("app.db", isolation_level=None) as db:
    await db.execute("INSERT INTO logs VALUES (?)", ("event",))
    # Auto-committed.

For repeated writes that need batching:

async with aiosqlite.connect("app.db") as db:
    await db.execute("BEGIN IMMEDIATE")  # Acquire write lock now
    for record in many_records:
        await db.execute("INSERT INTO ...", record)
    await db.commit()

BEGIN IMMEDIATE reserves the write lock right away — preventing “database is locked” mid-batch.

Common Mistake: Long-running transactions block all other writers. SQLite is single-writer; a transaction held for 30 seconds blocks every other write for 30 seconds. Keep transactions short.

Fix 7: Concurrent Reads

WAL mode lets unlimited readers run concurrently with one writer:

async def read_many(ids):
    async with aiosqlite.connect("app.db") as db:
        db.row_factory = aiosqlite.Row
        cur = await db.execute(
            f"SELECT * FROM users WHERE id IN ({','.join('?' for _ in ids)})",
            ids,
        )
        return [dict(r) for r in await cur.fetchall()]

# Many concurrent reads:
results = await asyncio.gather(
    read_many([1, 2, 3]),
    read_many([4, 5, 6]),
    read_many([7, 8, 9]),
)

In WAL mode, all three reads proceed in parallel against snapshots — no blocking.

For sustained read throughput, use a connection pool (Fix 2 Pattern C).

Pro Tip: Periodic PRAGMA wal_checkpoint(TRUNCATE) keeps the WAL file from growing without bound. SQLite checkpoints automatically but you can force it during quiet periods.

Fix 8: Migrations

aiosqlite has no migration framework. For simple cases, use Alembic with SQLAlchemy + aiosqlite driver:

pip install alembic sqlalchemy[asyncio] aiosqlite

alembic.ini:

sqlalchemy.url = sqlite+aiosqlite:///app.db

For just running a versioned schema:

async def migrate(db: aiosqlite.Connection):
    await db.execute("CREATE TABLE IF NOT EXISTS schema_version (version INTEGER)")
    
    async with db.execute("SELECT version FROM schema_version") as cur:
        row = await cur.fetchone()
        current = row[0] if row else 0
    
    migrations = [
        (1, "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)"),
        (2, "ALTER TABLE users ADD COLUMN email TEXT"),
        (3, "CREATE INDEX idx_users_email ON users(email)"),
    ]
    
    for version, sql in migrations:
        if version > current:
            await db.execute(sql)
            if current == 0:
                await db.execute("INSERT INTO schema_version VALUES (?)", (version,))
            else:
                await db.execute("UPDATE schema_version SET version = ?", (version,))
            await db.commit()

For more complex schema changes (table rename, type change), SQLite often requires the “create new table, copy data, drop old, rename” pattern. Alembic handles this automatically.

Still Not Working?

A few less-obvious failures:

• database is locked despite WAL. Another process opened the file in journal mode and broke WAL. Reset: PRAGMA journal_mode = WAL.
• Slow writes after enabling foreign keys. Foreign key checks add overhead. For bulk loads, temporarily disable: PRAGMA foreign_keys = OFF; ...; PRAGMA foreign_keys = ON; and verify integrity afterward.
• Connection closed mid-request. A timeout killed the connection. Increase aiosqlite.connect(timeout=10) — default is 5 seconds for getting a lock.
• Recursive use of cursor not allowed. Two queries on the same cursor. Use separate async with db.execute(...) blocks.
• High memory with big result sets. fetchall() loads everything into memory. Iterate instead: async for row in cur: reads one at a time.
• Pickle errors with custom types. SQLite stores limited types. Convert custom objects to JSON/string before insert.
• Memory database (:memory:) shared across connections. Not by default — each :memory: connection is its own DB. Use file::memory:?cache=shared URI for shared in-memory.
• asyncio event loop closed errors at shutdown. aiosqlite uses a background thread. Always await db.close() before exiting; in FastAPI, do so in the lifespan shutdown.

For related Python async DB and SQLite issues, see SQLite database is locked, SQLAlchemy not working, Python asyncio not running, and FastAPI dependency injection error.