Migration Guide: aiosqlite to rapsqlite

This guide helps you migrate from aiosqlite to rapsqlite for true async SQLite operations with GIL-independent performance.

Quick Start

The simplest migration is a one-line change:

# Before (aiosqlite)
import aiosqlite

# After (rapsqlite)
import rapsqlite as aiosqlite

For most applications, this is all you need! rapsqlite is designed to be a drop-in replacement for aiosqlite.

Why Migrate?

True async: All database operations execute outside the Python GIL
Better performance: No fake async, no event loop stalls
Same API: Drop-in replacement with minimal code changes
Verified: Passes Fake Async Detector benchmarks

Migration Steps

1. Install rapsqlite

pip install rapsqlite

2. Update Imports

Option A: Simple alias (recommended)

import rapsqlite as aiosqlite

Option B: Direct import

from rapsqlite import connect, Connection

3. Verify Compatibility

Run your existing tests. Most code should work without changes.

API Compatibility

Core API (100% Compatible)

All core aiosqlite APIs are supported:

✅ connect() - Connection factory
✅ Connection - Connection class
✅ Cursor - Cursor class
✅ execute(), executemany() - Query execution
✅ fetchone(), fetchall(), fetchmany() - Result fetching
✅ begin(), commit(), rollback() - Transactions
✅ transaction() - Transaction context manager
✅ Exception types: Error, OperationalError, ProgrammingError, IntegrityError
✅ Parameterized queries (named and positional)
✅ Row factories (dict, tuple, callable)
✅ PRAGMA settings
✅ Connection string URIs

Enhanced APIs

rapsqlite includes additional methods not in aiosqlite:

fetch_all() - Fetch all rows (returns list)
fetch_one() - Fetch single row (raises if not found)
fetch_optional() - Fetch single row or None
last_insert_rowid() - Get last insert ID
changes() - Get number of affected rows
init_hook - Database initialization hook (rapsqlite-specific)
Schema introspection methods (get_tables(), get_table_info(), etc.)

Code Examples

Basic Connection

# Works the same in both libraries
import rapsqlite as aiosqlite

async with aiosqlite.connect("example.db") as db:
    await db.execute("CREATE TABLE test (id INTEGER PRIMARY KEY, value TEXT)")
    await db.execute("INSERT INTO test (value) VALUES ('hello')")
    rows = await db.fetch_all("SELECT * FROM test")
    print(rows)

Output: [[1, 'hello']]

Parameterized Queries

# Named parameters
await db.execute(
    "INSERT INTO users (name, email) VALUES (:name, :email)",
    {"name": "Alice", "email": "alice@example.com"}
)

# Positional parameters
await db.execute(
    "INSERT INTO users (name, email) VALUES (?, ?)",
    ["Bob", "bob@example.com"]
)

Transactions

# Explicit transaction
await db.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.commit()
except Exception:
    await db.rollback()

# Transaction context manager
async with db.transaction():
    await db.execute("INSERT INTO test (value) VALUES (1)")
    await db.execute("INSERT INTO test (value) VALUES (2)")

Differences and Limitations

For a detailed compatibility analysis based on running the aiosqlite test suite, see aiosqlite Compatibility Analysis. For per-test results when running the aiosqlite suite against rapsqlite, see docs/AIOSQLITE_TEST_RESULTS.md.

If you see test failures when running the aiosqlite test suite against rapsqlite (e.g. via scripts/run_aiosqlite_tests.py), see docs/AIOSQLITE_TEST_RESULTS.md for a per-test breakdown and failure categories (fix / document / environment). Known intentional differences are listed there and in aiosqlite Compatibility Analysis.

Migrating from aiosqlite: Common Patterns

Connection lifecycle: Always use async with connect(...) or await conn.close(). Abandoning a connection without closing can cause issues during garbage collection.
total_changes and in_transaction: These are now sync properties in rapsqlite, matching aiosqlite behavior. Use db.total_changes and db.in_transaction (no await needed).
Backup API: rapsqlite supports backup to both rapsqlite and sqlite3 connections. Backup to sqlite3 requires a file-backed database (not :memory:).
Transaction queue: aiosqlite queues transaction operations on a background thread. rapsqlite does not use a transaction queue; operations run directly on the connection. For most applications this is transparent.

Known Differences

Row Format: rapsqlite returns rows as lists [[1, 'a']] by default; aiosqlite/sqlite3 return tuples [(1, 'a')]. For drop-in compatibility (e.g. import rapsqlite as aiosqlite) use aiosqlite_compat=True so rows are tuples by default. Otherwise set row_factory to the string "tuple":

# Option A: Tuple rows by default (aiosqlite-compatible)
async with connect("example.db", aiosqlite_compat=True) as db:
    rows = await db.fetch_all("SELECT id, name FROM users")
    # rows = [(1, 'Alice'), (2, 'Bob')]

# Option B: rapsqlite default (lists)
rows = await db.fetch_all("SELECT id, name FROM users")
# rows = [[1, 'Alice'], [2, 'Bob']]

# Option C: Set row_factory on an existing connection
db.row_factory = "tuple"
rows = await db.fetch_all("SELECT id, name FROM users")
# rows = [(1, 'Alice'), (2, 'Bob')]

``set_progress_handler``: rapsqlite accepts both (n, callback) and (callback, n) for sqlite3/aiosqlite compatibility.
Tuple parameters: rapsqlite accepts tuple as a single parameter (converted to text representation for one placeholder).

``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]

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

register_adapter and register_converter

rapsqlite implements per-connection register_adapter(type, adapter) and register_converter(typename, converter) (sqlite3-style). Use conn.register_adapter(MyType, lambda x: ...) for parameter binding and conn.register_converter("TYPENAME", lambda b: ...) for result column decoding. See Type Conversion Strategy for details and examples.

Advanced Features

Database Initialization Hooks

rapsqlite supports automatic database initialization:

async def init_hook(conn):
    """Initialize database schema and data."""
    await conn.execute("""
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY,
            name TEXT NOT NULL,
            email TEXT UNIQUE
        )
    """)
    await conn.set_pragma("foreign_keys", True)

async with Connection("example.db", init_hook=init_hook) as conn:
    # Tables are already created and initialized
    users = await conn.fetch_all("SELECT * FROM users")

Schema Introspection

rapsqlite provides comprehensive schema introspection:

# Get all tables
tables = await conn.get_tables()

# Get table information
columns = await conn.get_table_info("users")

# Get indexes
indexes = await conn.get_indexes("users")

# Get foreign keys
foreign_keys = await conn.get_foreign_keys("posts")

# Get comprehensive schema
schema = await conn.get_schema("users")

Troubleshooting

Import Errors

Problem: ImportError: Could not import _rapsqlite

Solution: Make sure rapsqlite is built. If installing from source:

pip install maturin
maturin develop

Performance Issues

Problem: Queries seem slower than expected

Solution:

Ensure you’re using parameterized queries (not string formatting)
Use connection pooling for concurrent operations
Consider using execute_many() for batch inserts

Transaction Issues

Problem: “Transaction connection not available” error

Solution: Make sure you call begin() before using transaction methods, or use the transaction() context manager.

Testing Your Migration

Run existing tests: Your aiosqlite tests should work with minimal changes
Use compatibility tests: See tests/test_dropin_replacement.py for examples
Verify performance: Use benchmarks to ensure performance meets expectations

For best practices and common anti-patterns when using rapsqlite, see Advanced Usage Guide.

Example: Complete Migration

Here’s a complete example of migrating an application:

Before (aiosqlite):

import aiosqlite

async def main():
    async with aiosqlite.connect("app.db") as db:
        await db.execute("CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT)")
        await db.execute("INSERT INTO users (name) VALUES (?)", ["Alice"])
        async with db.execute("SELECT * FROM users") as cursor:
            rows = await cursor.fetchall()
            print(rows)

asyncio.run(main())

After (rapsqlite):

import rapsqlite as aiosqlite

async def main():
    async with aiosqlite.connect("app.db") as db:
        await db.execute("CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT)")
        await db.execute("INSERT INTO users (name) VALUES (?)", ["Alice"])
        # Option 1: Use async with pattern (same as aiosqlite)
        async with db.execute("SELECT * FROM users") as cursor:
            rows = await cursor.fetchall()
            print(rows)
        # Option 2: Use fetch_all (rapsqlite enhancement)
        rows = await db.fetch_all("SELECT * FROM users")
        print(rows)

asyncio.run(main())

Output (for one row: id=1, name=”Alice”): [[1, 'Alice']]

The migration is complete! Your code now uses true async SQLite operations.