aiosqlite: Async SQLite in Python (Tested Examples)

Python’s built-in sqlite3 module blocks, so a slow query inside an async app holds up every other coroutine on the event loop. aiosqlite wraps the same API so you can await it instead. In this article I’ll walk you through installing aiosqlite and using it, with a test that shows the difference.

TLDR

  • Install it with pip install aiosqlite. A ModuleNotFoundError: No module named 'aiosqlite' means it isn’t installed in the Python you’re running.
  • Open a database with async with aiosqlite.connect("file.db") as db: and put await in front of every database call.
  • Each connection runs its queries on its own background thread, so the event loop keeps running while SQLite works.
  • SQLAlchemy’s async engine uses it through the sqlite+aiosqlite:/// URL, which needs pip install "sqlalchemy[asyncio]".

What is aiosqlite, and what will we build?

aiosqlite is an “asyncio bridge to the standard sqlite3 module” (PyPI). Its README says it “replicates the standard sqlite3 module, but with async versions of all the standard connection and cursor methods”. It does that “by using a single, shared thread per connection” (aiosqlite README).

sqlite3aiosqlite
Where queries runThe calling thread (in async code, the event loop)A background thread per connection
Calling stylecon.execute(...)await db.execute(...)
Closingwith closes transactions, not the connectionasync with closes the connection
InstallPart of the standard librarypip install aiosqlite

To show you how it works, I will build a small events log with one table.

Prerequisites

RequirementVersion used here
Python3.9 or newer, per PyPI. Tested on 3.13
aiosqlite0.22.1
SQLiteWhatever your Python’s sqlite3 was built with
Python 3.13.14 | aiosqlite 0.22.1 | SQLite 3.53.4

Building an events log with aiosqlite

Each step below is a separate script that works on the same events.db file.

Step 1: Install aiosqlite

Before installing, the import fails:

$ python -c "import aiosqlite"
ModuleNotFoundError: No module named 'aiosqlite'

Install it in your project’s virtual environment:

$ pip install aiosqlite
Successfully installed aiosqlite-0.22.1

Still getting ModuleNotFoundError after installing? Then pip installed into a different Python. Run python -m pip install aiosqlite so both are the same Python.

Step 2: Connect, create a table and insert a row

async with aiosqlite.connect() opens the database, creating the file if it’s missing, and closes it at the end of the block. Writes need await db.commit(), the same as in sqlite3:

import asyncio
import aiosqlite


async def main():
    async with aiosqlite.connect("events.db") as db:
        await db.execute(
            "CREATE TABLE IF NOT EXISTS events ("
            "id INTEGER PRIMARY KEY, user_id TEXT NOT NULL, name TEXT NOT NULL)"
        )
        await db.execute(
            "INSERT INTO events (user_id, name) VALUES (?, ?)", ("u1", "signup")
        )
        await db.commit()

        async with db.execute("SELECT id, user_id, name FROM events") as cursor:
            async for row in cursor:
                print(row)


asyncio.run(main())
$ python quickstart.py
(1, 'u1', 'signup')

async with db.execute(...) as cursor closes the cursor for you, and async for reads the rows one at a time.

Step 3: Insert many rows and read them by column name

import asyncio
import aiosqlite

EVENTS = [("u1", "search"), ("u2", "signup"), ("u1", "purchase")]


async def main():
    async with aiosqlite.connect("events.db") as db:
        await db.executemany("INSERT INTO events (user_id, name) VALUES (?, ?)", EVENTS)
        cursor = await db.execute_insert(
            "INSERT INTO events (user_id, name) VALUES (?, ?)", ("u3", "signup")
        )
        print("new row id:", cursor[0])
        await db.commit()

        db.row_factory = aiosqlite.Row
        async with db.execute("SELECT * FROM events WHERE user_id = ?", ("u1",)) as cur:
            row = await cur.fetchone()
            print("first u1 event:", row["id"], row["name"])

        rows = await db.execute_fetchall("SELECT name, count(*) AS n FROM events GROUP BY name ORDER BY name")
        for r in rows:
            print(r["name"], r["n"])


asyncio.run(main())
$ python rows.py
new row id: 5
first u1 event: 1 signup
purchase 1
search 1
signup 3

What to notice:

  • executemany() inserts a list of rows with one call.
  • execute_insert() is an aiosqlite helper that returns the new row’s ID.
  • db.row_factory = aiosqlite.Row lets you read columns by name, like row["name"].
  • execute_fetchall() runs a query and returns all its rows in one await. There’s no execute_fetchone(). Use execute() and then await cursor.fetchone().

Step 4: Open and close a connection without a context manager

Sometimes a connection has to outlive one function. Then you keep the object and close it yourself. close() is a coroutine, so it needs await too:

import asyncio
import aiosqlite


async def main():
    db = await aiosqlite.connect("events.db")
    cursor = await db.execute("SELECT count(*) FROM events")
    print("events:", (await cursor.fetchone())[0])
    await cursor.close()
    await db.close()
    print("closed")


asyncio.run(main())
$ python manual_close.py
events: 5
closed

Step 5: Use aiosqlite through SQLAlchemy

SQLAlchemy’s async engine talks to SQLite through aiosqlite. The URL scheme is sqlite+aiosqlite:///:

import asyncio
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine


async def main():
    engine = create_async_engine("sqlite+aiosqlite:///events.db")
    async with engine.connect() as conn:
        result = await conn.execute(text("SELECT count(*) FROM events"))
        print("events via SQLAlchemy:", result.scalar_one())
    await engine.dispose()

asyncio.run(main())

A plain pip install sqlalchemy isn’t enough for the async engine:

$ pip install sqlalchemy
Successfully installed sqlalchemy-2.1.3 typing-extensions-4.16.0
$ python sqlalchemy_async.py
ImportError: The SQLAlchemy asyncio module requires that the Python 'greenlet' library is installed.  In order to ensure this dependency is available, use the 'sqlalchemy[asyncio]' install target:  'pip install sqlalchemy[asyncio]'

Install the asyncio extra, as the error says, and the script runs:

$ pip install "sqlalchemy[asyncio]"
Successfully installed greenlet-3.5.6
$ python sqlalchemy_async.py
events via SQLAlchemy: 5

For the full SQLAlchemy setup, see SQLAlchemy SQLite.

Running it: does aiosqlite keep the event loop free?

To check the claim, I ran the same slow query both ways while a ticker coroutine tried to wake up every 0.1 seconds for 0.6 seconds:

# A ticker coroutine counts how often it wakes up while a slow query runs.
# With sqlite3 the event loop stalls; with aiosqlite the ticker keeps going.
import asyncio
import sqlite3
import time
import aiosqlite

SLOW = ("WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < 10000000) "
        "SELECT count(*) FROM n")


async def ticker(label):
    ticks = 0
    start = time.perf_counter()
    while time.perf_counter() - start < 0.6:
        await asyncio.sleep(0.1)
        ticks += 1
    return ticks


async def with_sqlite3():
    con = sqlite3.connect(":memory:")
    t = asyncio.create_task(ticker("sqlite3"))
    await asyncio.sleep(0)
    con.execute(SLOW).fetchone()  # runs on the event loop thread
    return await t


async def with_aiosqlite():
    async with aiosqlite.connect(":memory:") as db:
        t = asyncio.create_task(ticker("aiosqlite"))
        await asyncio.sleep(0)
        await (await db.execute(SLOW)).fetchone()  # runs on aiosqlite's thread
        return await t


async def main():
    print("ticks during sqlite3 query:  ", await with_sqlite3())
    print("ticks during aiosqlite query:", await with_aiosqlite())

asyncio.run(main())
$ python blocking.py
ticks during sqlite3 query:   1
ticks during aiosqlite query: 6

With sqlite3, the ticker only woke up once, because the query held the event loop. With aiosqlite, it woke up on schedule while the query ran on the connection’s own thread. The query itself isn’t faster. aiosqlite helps when other coroutines need to run while SQLite works, such as requests in a web server.

Common problems

Each problem starts with its error message. The first two came up while testing.

ModuleNotFoundError: No module named ‘aiosqlite’

aiosqlite isn’t installed in the interpreter that runs your code. Install it with python -m pip install aiosqlite, using the same python you run the script with.

RuntimeWarning: coroutine ‘Connection.execute’ was never awaited

Every database call is a coroutine. Leave out await, and you get an object that isn’t a cursor yet:

import asyncio
import aiosqlite


async def main():
    async with aiosqlite.connect("events.db") as db:
        cursor = db.execute("SELECT count(*) FROM events")  # missing await
        print(await cursor.fetchone())


asyncio.run(main())
$ python forgot_await.py
AttributeError: 'Result' object has no attribute 'fetchone'
<sys>:0: RuntimeWarning: coroutine 'Connection.execute' was never awaited

Put await in front of the call, or use it as async with db.execute(...).

database is locked

aiosqlite doesn’t change SQLite’s locking rules. Only one connection can write at a time. If two connections write at once, see database is locked.

Conclusion

Use aiosqlite when SQLite runs inside an asyncio program, such as an async web app or bot. It keeps the sqlite3 API you already know. You add await, and the queries run off the event loop. In a plain script with no event loop, the standard sqlite3 module does the same job with less code.

Further reading

Tested with aiosqlite 0.22.1 on Python 3.13.14 (SQLite 3.53.4), 7 October 2026.