aiosqlite
aiosqlite is an async wrapper around Python's builtin sqlite3 module. Every connection spawns one background thread, and every query you await gets pushed onto that thread's queue, runs there, and sends the result back to the event loop. The API is a near-copy of sqlite3: connect, cursor, execute, executemany, fetchone, fetchall, commit, rollback, row_factory, create_function, backup, iterdump. The difference is that the methods are coroutines and the connection and cursor objects double as async context managers and async iterators. It exists so that a query against a local SQLite file does not freeze every other coroutine in your process while the disk reads happen. It is a thread-offload layer, not a new database driver.
The obvious choice for keeping SQLite off your event loop, and the API match with sqlite3 makes the port almost mechanical. Understand what it does not buy: one connection is still one serialized worker thread, so this fixes responsiveness, not throughput.
Use it if
- You have an asyncio web service or bot backed by a SQLite file and a slow query currently blocks every other request on the loop
- You are converting existing sqlite3 code to async and want a mechanical port: the connect signature, the cursor methods, and the exception classes are all re-exported unchanged
- You want zero dependencies and no build step, which matters for slim containers and for anything that ships as a single wheel
- You need the less common sqlite3 features asynchronously too, including create_function, load_extension, set_progress_handler, set_authorizer, backup, and iterdump
- You expect concurrency inside one connection: all work for a connection runs on that connection's single worker thread in queue order, so ten awaited queries against one Connection execute one after another, exactly as they would synchronously
- You are write-heavy: SQLite takes a database-level write lock, and moving the call off the loop does nothing about SQLITE_BUSY. You still need WAL mode, a busy_timeout, and a design that funnels writes through one place
- Your queries are small and fast: a primary-key lookup on a warm page cache takes microseconds, and the queue hop plus thread wakeup costs more than the query, so a plain sqlite3 call is faster
- You want a pool or an ORM: there is no connection pooling, no query builder, and no migrations here. SQLAlchemy's async engine already drives aiosqlite underneath and gives you all three
- You are on Python 3.12 or newer and only have a couple of call sites: asyncio.to_thread around your existing sqlite3 code gets you the same non-blocking behaviour with no new dependency
- You upgraded from 0.21: as of 0.22.0 Connection no longer subclasses threading.Thread, and a connection that is garbage collected without close() or stop() now emits a ResourceWarning and leaves a thread parked until it is stopped
Setup reality
pip install aiosqlite installs in a second with no dependencies and no compiler, and it needs Python 3.9 or newer. The friction is behavioural rather than installational. connect() returns an object that is both awaitable and an async context manager, so a bare aiosqlite.connect(path) with neither await nor async with silently does nothing. Since 0.22.0 the connection is no longer a Thread subclass and it owns a worker thread you are responsible for shutting down: use async with, or call await close(), or call the synchronous stop() when there is no running loop, otherwise you get a ResourceWarning and a lingering thread. Each connection is one thread, so opening a connection per request in a busy service creates a thread per request. Cursors returned by execute should also be closed, which async with on the execute call does for you. PRAGMA settings such as journal_mode and busy_timeout are per connection, so they have to be re-applied every time you open one.
Patterns
Open a connection and run a queryconnect-and-query
import aiosqlite
async def get_user(user_id: int):
async with aiosqlite.connect('app.db') as db:
async with db.execute(
'SELECT name, email FROM users WHERE id = ?', (user_id,)
) as cursor:
return await cursor.fetchone()Both async with blocks matter. The outer one closes the connection and stops its worker thread; the inner one closes the cursor. aiosqlite.connect(path) on its own returns an unstarted object and touches no disk.
Get rows as mappings instead of tuplesrow-factory
import aiosqlite
async def list_users():
async with aiosqlite.connect('app.db') as db:
db.row_factory = aiosqlite.Row
async with db.execute('SELECT id, name FROM users') as cursor:
return [dict(row) async for row in cursor]row_factory is a property on the connection and must be set before you execute, since cursors inherit it at creation. aiosqlite.Row is re-exported from sqlite3, so it indexes by column name and by position.
Write rows and commit explicitlyinsert-and-commit
import aiosqlite
async def add_users(rows):
async with aiosqlite.connect('app.db') as db:
await db.executemany(
'INSERT INTO users (name, email) VALUES (?, ?)', rows
)
await db.commit()Nothing is durable until you await commit(). Closing the connection without committing rolls the transaction back, and because the failure is silent this is the most common bug people hit on the way over from sqlite3.
Get the id of the row you just insertedlast-insert-rowid
import aiosqlite
async def create_order(customer_id: int) -> int:
async with aiosqlite.connect('app.db') as db:
row = await db.execute_insert(
'INSERT INTO orders (customer_id) VALUES (?)', (customer_id,)
)
await db.commit()
return row[0]execute_insert runs the insert and last_insert_rowid() in one trip to the worker thread, which is why it beats reading cursor.lastrowid afterwards. It returns a row, so take element 0, and it can return None if the statement inserted nothing.
Iterate a large result set without loading it allstream-large-result
import aiosqlite
async def export_events(handle):
async with aiosqlite.connect('app.db') as db:
async with db.execute('SELECT id, payload FROM events') as cursor:
async for row in cursor:
await handle(row)async for pulls rows in batches through the worker thread rather than materialising the whole result. If handle() awaits anything slow, the cursor and its read transaction stay open the whole time, which blocks writers in rollback-journal mode.
Run queries in parallel with separate connectionsconnection-per-task
import asyncio
import aiosqlite
async def count(table: str) -> int:
async with aiosqlite.connect('app.db') as db:
async with db.execute(f'SELECT count(*) FROM {table}') as cur:
return (await cur.fetchone())[0]
async def totals():
return await asyncio.gather(count('users'), count('orders'))One connection means one worker thread and one queue, so gathering queries on a shared connection just serializes them. Separate connections give real parallel reads, at the cost of one OS thread each, so cap how many you open.
Set the pragmas that make concurrent access workwal-and-busy-timeout
import aiosqlite
async def open_db(path='app.db'):
db = await aiosqlite.connect(path)
await db.execute('PRAGMA journal_mode=WAL')
await db.execute('PRAGMA busy_timeout=5000')
await db.execute('PRAGMA foreign_keys=ON')
return dbjournal_mode is stored in the database file and survives, but busy_timeout and foreign_keys are per connection and reset every time you open one. Without a busy_timeout, a concurrent writer raises OperationalError: database is locked immediately instead of waiting.
Hold a long-lived connection and close it properlymanual-lifecycle
import aiosqlite
class Store:
async def start(self):
self.db = await aiosqlite.connect('app.db')
async def stop(self):
await self.db.close()
# no running event loop, for example in an atexit hook:
# store.db.stop()Since 0.22.0 a connection dropped without close() emits a ResourceWarning and its worker thread stays parked. close() is the async path; the synchronous stop() exists for teardown paths where the loop is already gone.
Register a Python function callable from SQLcustom-sql-function
import aiosqlite
async def search(term: str):
async with aiosqlite.connect('app.db') as db:
await db.create_function('slugify', 1, lambda s: s.lower().replace(' ', '-'))
async with db.execute(
'SELECT id FROM posts WHERE slugify(title) = ?', (term,)
) as cur:
return await cur.fetchall()The callback runs synchronously on the worker thread, once per row, so it cannot await anything and it blocks the whole connection while it runs. Registration is per connection and has to be repeated on every new one.
Roll back a batch when one statement failstransaction-rollback
import aiosqlite
async def transfer(src: int, dst: int, amount: int):
async with aiosqlite.connect('app.db') as db:
try:
await db.execute('UPDATE accounts SET bal = bal - ? WHERE id = ?', (amount, src))
await db.execute('UPDATE accounts SET bal = bal + ? WHERE id = ?', (amount, dst))
await db.commit()
except aiosqlite.Error:
await db.rollback()
raiseThe exception classes are re-exported straight from sqlite3, so catching sqlite3.Error works identically. Check db.in_transaction if you need to know whether an implicit transaction is actually open before rolling back.
Copy a live database to another filebackup-database
import aiosqlite
async def snapshot(src_path='app.db', dst_path='backup.db'):
async with aiosqlite.connect(src_path) as src, aiosqlite.connect(dst_path) as dst:
await src.backup(dst, pages=100, sleep=0.05)This wraps sqlite3's online backup API, so it is safe while writers are active, unlike copying the file. pages=0 copies everything in one blocking call on the worker thread; a small page count plus a sleep lets other queries on that connection interleave.
Dump the whole database as SQL textdump-to-sql
import aiosqlite
async def dump(path='app.db', out='dump.sql'):
async with aiosqlite.connect(path) as db:
with open(out, 'w') as fh:
async for line in db.iterdump():
fh.write(line + '\n')iterdump runs the dump on the worker thread and hands lines back through a queue that the coroutine polls with a 10 ms sleep, so on a small database the polling interval, not the dump, dominates the wall time.
Alternatives
| Package | Registry | Pick it when |
|---|---|---|
| sqlalchemy | PyPI | You want an async engine with pooling, a session layer, and migrations, and are happy for aiosqlite to sit underneath as the driver |
| databases | PyPI | You want a thin async query interface with pooling that can point at SQLite in development and Postgres in production without changing call sites |
| apsw | PyPI | You need the full SQLite C API surface, including VFS hooks, virtual tables, and precise busy handling, and can run it synchronously or behind your own thread offload |