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.

The architecture matters when debugging slow queries. Each aiosqlite.connect() spawns a dedicated worker thread that owns one C-level sqlite3 connection. Coroutines on the asyncio side await futures whose results are produced by that thread. A single connection therefore serializes all of its operations, even though they look concurrent in await syntax. Two await db.execute(...) calls on the same connection never actually run in parallel — they queue inside the worker thread.

The other quiet trap is the asyncio backend assumption. aiosqlite is built on asyncio.get_event_loop(). It does not natively support trio or anyio’s trio backend. If you embed an aiosqlite call inside a trio task without an anyio wrapper, the future never resolves and the call hangs forever with no error. Pick an asyncio-only stack, or use anyio.from_thread.run_sync with a vanilla sqlite3 connection.

Diagnostic Timeline

A typical database is locked debugging session.

Minute 0. Tests pass; production throws database is locked under load. First guess: “use sqlite3 instead of aiosqlite.” The async wrapper is not the problem — SQLite itself is single-writer and your default journal_mode is DELETE, which blocks readers behind writers.

Minute 5. You add PRAGMA journal_mode = WAL inside the request handler. Same error. WAL mode is a database-level setting, not per-connection — and the migration that runs at startup did not persist it because it ran on an in-memory connection. You move the pragma to a startup script that runs against the on-disk file.

Minute 13. Locks disappear under read load but still bite during bulk inserts. First guess: “use BEGIN.” Adding BEGIN does nothing because each statement was already inside an implicit transaction. The fix is BEGIN IMMEDIATE so SQLite acquires the write lock upfront instead of upgrading from a read lock mid-transaction.

Minute 22. You add a connection pool and now requests crash with Recursive use of cursor not allowed. The pool gives different requests the same connection — one is mid-cursor when another tries to reuse it. The fix is making each acquire/release strictly bracket the entire query, including fetchall().

Minute 31. Datetime columns come back as strings on Python 3.12. First guess: “missing detect_types.” You add detect_types=PARSE_DECLTYPES. Still strings. The real cause: 3.12 deprecated the built-in adapter and emits a DeprecationWarning. Register explicit register_adapter/register_converter for datetime and date.

Minute 40. Switching from asyncio to trio for new code: aiosqlite calls hang silently. The aiosqlite worker thread is waiting on an asyncio future that trio never schedules. You either keep aiosqlite on the asyncio side via anyio interop, or drop down to sqlite3 inside anyio.to_thread.run_sync.

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.
• WAL file grows unbounded in production. The auto-checkpoint runs only when a connection commits and the WAL exceeds 1000 pages. A long-lived read connection holding a snapshot blocks checkpoints. Periodically open a fresh connection and run PRAGMA wal_checkpoint(TRUNCATE).
• Database file copied via cp while the app is running is corrupt. SQLite’s WAL means the .db file alone is not consistent — you also need the -wal and -shm files, or use the online backup API via sqlite3.Connection.backup().
• uvloop plus aiosqlite races on shutdown. uvloop’s faster signal handling sometimes drops the worker thread’s join callback. Use the asyncio default loop policy in environments where you tear down connections at SIGTERM, or set loop_factory=asyncio.new_event_loop.

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.