aiosqlite Compatibility Analysis

This guide covers aiosqlite compatibility, SQLAlchemy async (sqlite+rapsqlite dialect), and Alembic migrations. It analyzes compatibility between rapsqlite and aiosqlite based on running the aiosqlite test suite.

Test Execution

Date: 2026-01-30 rapsqlite Version: 0.3.1 aiosqlite Test Suite: Latest from https://github.com/omnilib/aiosqlite

Test baseline: Run scripts/run_aiosqlite_tests.py to execute the aiosqlite test suite against rapsqlite (patched imports). Results are written to docs/AIOSQLITE_TEST_RESULTS.md with a per-test breakdown. Summary: some tests pass in perf.py and smoke.py; failures are often due to document (intentional differences: async methods for total_changes/in_transaction, no transaction queue, backup API), environment (temp dir, package layout when run from script), or compatibility gaps. See docs/AIOSQLITE_TEST_RESULTS.md for the per-test breakdown.

Known API Differences

1. `async with db.execute(...)` Pattern

Status: ✅ NOW SUPPORTED - async with db.execute(...) pattern is fully implemented via ExecuteContextManager. Both async with and direct await patterns work.

Impact: High - Many aiosqlite tests use this pattern. Now fully compatible.

2. Cursor as Async Context Manager / Async Iteration

Status: ✅ NOW SUPPORTED - Cursors support async iteration via __aiter__ and __anext__ methods.

Impact: Medium - Some tests use async iteration. Now fully compatible.

3. Connection Properties

Status: ✅ ALL PROPERTIES NOW SUPPORTED - All connection properties are implemented. total_changes and in_transaction are synchronous properties in rapsqlite (no await), matching aiosqlite behavior.

4. Row Format (lists vs tuples)

Status: ✅ CONFIGURABLE - rapsqlite default is list rows; aiosqlite/sqlite3 use tuples. Use connect(..., aiosqlite_compat=True) for tuple rows by default (drop-in import rapsqlite as aiosqlite), or set conn.row_factory = "tuple" on an existing connection. See Migration Guide: aiosqlite to rapsqlite for examples.

5. Row Factory: `aiosqlite.Row`

Status: ✅ NOW SUPPORTED - rapsqlite.Row class is implemented with dict-like access (row["column"], row[0], keys(), values(), items()).

6. `executescript()` Method

Status: ✅ NOW IMPLEMENTED

7. `load_extension()` Method

Status: ✅ NOW IMPLEMENTED

8. Backup to sqlite3.Connection

rapsqlite.Connection.backup() now supports both rapsqlite.Connection and sqlite3.Connection targets. For sqlite3 targets, it uses Python’s sqlite3 backup API over the on-disk database file (file-backed databases only; :memory: and non-file URIs are not supported).

9. `iterdump()` Return Type

rapsqlite supports both async iteration and await-to-list:

# aiosqlite and rapsqlite (async iterator)
async for line in db.iterdump():
    ...

# rapsqlite (backwards compatible)
lines = await db.iterdump()  # Returns List[str]

10. `init_hook` parameter

This is a rapsqlite-specific enhancement for automatic database initialization. It’s not available in aiosqlite.

True Async DBAPI (SQLAlchemy-style)

rapsqlite implements a True Async DBAPI 2.0 interface in rapsqlite.dbapi, suitable for SQLAlchemy and other consumers that expect an async driver with no thread offloading. See docs/true_async_dbapi_spec.md for the contract.

apilevel = "2.0", threadsafety = 0, paramstyle = "qmark"
Full exception hierarchy (Error, InterfaceError, DataError, etc.)
async def connect(database, **kwargs) -> AsyncConnection
AsyncConnection: async cursor(), async execute(), executemany, commit, rollback, close
Cursor: execute, executemany, fetchone, fetchmany, fetchall, close; description, rowcount, lastrowid, arraysize

Use rapsqlite.dbapi when integrating with async frameworks that require a DBAPI-compliant module.

SQLAlchemy

rapsqlite provides a sqlite+rapsqlite dialect for SQLAlchemy 2.0+. Full ORM support is implemented: AsyncSession, session.add/add_all, session.commit, session.get, transaction rollback, and INSERT...RETURNING identity fetch all work correctly.

Requirements: SQLAlchemy 2.0 or newer. Install with pip install sqlalchemy (or pip install rapsqlite[sqlalchemy] to install rapsqlite and SQLAlchemy in one step).

Core usage:

import rapsqlite.sqlalchemy  # register dialect
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine("sqlite+rapsqlite:///path.db")
# or sqlite+rapsqlite:///:memory:

async with engine.connect() as conn:
    result = await conn.execute(text("SELECT 1"))
    print(result.fetchone())

ORM usage (AsyncSession):

import rapsqlite.sqlalchemy
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, select

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(50))

engine = create_async_engine("sqlite+rapsqlite:///app.db")
async_session = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)

async with engine.begin() as conn:
    await conn.run_sync(Base.metadata.create_all)

async with async_session() as session:
    async with session.begin():
        session.add_all([User(name="alice"), User(name="bob")])

async with async_session() as session:
    result = await session.execute(select(User).order_by(User.id))
    users = result.scalars().all()
    # users has 2 User instances

Transaction rollback with on_connect UDFs (e.g. regexp, floor) works correctly: conn.begin(), inserts, and conn.rollback() behave as expected.

Tests: The suite tests/test_sqlalchemy_rapsqlite.py validates Core and ORM patterns against both sqlite+rapsqlite and sqlite+aiosqlite.

See docs/true_async_dbapi_spec.md for the full async DBAPI contract.

Alembic with rapsqlite

Alembic is fully supported with the sqlite+rapsqlite dialect. Use the async template (alembic init -t async) and point your engine URL to sqlite+rapsqlite:///... for true async migrations. Upgrade and downgrade (including downgrade base) work correctly.

You can run Alembic migrations using the sqlite+rapsqlite dialect as follows.

Install dependencies: pip install rapsqlite sqlalchemy alembic (or pip install rapsqlite[sqlalchemy] alembic to install rapsqlite and SQLAlchemy together).
Initialize Alembic with async support: alembic init -t async alembic
In alembic/env.py, ensure the dialect is registered and the URL uses sqlite+rapsqlite:

import rapsqlite.sqlalchemy  # register sqlite+rapsqlite dialect
from sqlalchemy.ext.asyncio import create_async_engine
# ...
# In run_migrations_online(), use:
connectable = create_async_engine(
    config.get_main_option("sqlalchemy.url").replace("sqlite://", "sqlite+rapsqlite://")
)

Set sqlalchemy.url in alembic.ini (e.g. sqlite:///app.db). The env.py replacement above switches it to sqlite+rapsqlite:///app.db for async runs.
Run migrations: alembic upgrade head (runs in the async context provided by the template).

For file-backed databases, use a path or sqlite+rapsqlite:///path/to/db.db. In-memory (:memory:) is supported for the engine but each connection gets its own database; use a file path for migrations so the same database is used across runs.

FastAPI

Use a connection dependency with rapsqlite.connect() and lifespan for setup. See examples/fastapi_db.py:

from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends
from rapsqlite import connect

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with connect("app.db") as conn:
        await conn.execute("CREATE TABLE IF NOT EXISTS ...")
    yield

async def get_db():
    async with connect("app.db") as conn:
        yield conn

app = FastAPI(lifespan=lifespan)

@app.get("/")
async def root(db=Depends(get_db)):
    rows = await db.fetch_all("SELECT ...")
    return {"data": rows}

Run with uvicorn examples.fastapi_db:app --reload. The test tests/test_fastapi_example.py validates this pattern.

Starlette

Starlette uses the same ASGI base as FastAPI. Use lifespan for setup and create a connection per request. See examples/starlette_db.py:

from contextlib import asynccontextmanager
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route
from rapsqlite import connect

@asynccontextmanager
async def lifespan(app):
    async with connect("app.db") as conn:
        await conn.execute("CREATE TABLE IF NOT EXISTS items ...")
    yield

async def homepage(request):
    async with connect("app.db") as conn:
        rows = await conn.fetch_all("SELECT * FROM items")
        return JSONResponse({"items": [list(r) for r in rows]})

app = Starlette(routes=[Route("/", homepage)], lifespan=lifespan)

Run with uvicorn examples.starlette_db:app --reload. The test tests/test_starlette_example.py validates this pattern.

aiohttp

Use on_startup for schema setup and create a connection per request. See examples/aiohttp_db.py:

from aiohttp import web
from rapsqlite import connect

async def init_db(app):
    async with connect("app.db") as conn:
        await conn.execute("CREATE TABLE IF NOT EXISTS items ...")
async def homepage(request):
    async with connect("app.db") as conn:
        rows = await conn.fetch_all("SELECT * FROM items")
        return web.json_response({"items": [list(r) for r in rows]})

app = web.Application()
app.on_startup.append(init_db)
app.router.add_get("/", homepage)

Run with python examples/aiohttp_db.py or aiohttp run examples/aiohttp_db:app. The test tests/test_aiohttp_example.py validates this pattern.

Compatibility Summary

Core API Compatibility: ~95%

✅ All core APIs supported
✅ All high-priority compatibility features implemented
✅ Drop-in replacement for most use cases
⚠️ Minor differences in property vs method access for total_changes and in_transaction

Performance Characteristics

Connection pooling: rapsqlite uses connection pooling internally. The default pool size is 1, but can be configured.
Prepared statements: sqlx (the underlying library) caches prepared statements per connection automatically.
True async: All operations execute outside the GIL, providing better concurrency under load.

For type conversion and custom types (including register_adapter and register_converter), see Type Conversion Strategy.

For the aiosqlite test suite baseline and per-test failure categories, see docs/AIOSQLITE_TEST_RESULTS.md.

For more details, see Migration Guide: aiosqlite to rapsqlite.