Asynchronous SQLite client with AnyIO

0.2.8 · active · verified Thu Apr 16

sqlite-anyio is an asynchronous client for SQLite databases, built on top of the AnyIO library. It provides an `async`/`await` interface for interacting with SQLite, enabling non-blocking database operations within asynchronous Python applications. The current version is 0.2.8, and it maintains a frequent release cadence, often introducing minor features or bug fixes.

Common errors

```
RuntimeError: You must use `await` with an async function
```
cause Attempting to call an asynchronous method (like `conn.execute` or `cur.fetchone`) without `await` inside an `async` function, or calling an `async` function (like `main`) directly from a synchronous context without an async event loop runner (e.g., `anyio.run`).

fix Ensure all calls to `sqlite-anyio` methods are preceded by `await` within an `async` function. The top-level `async` function must be executed using `anyio.run()` (or `asyncio.run()` if using asyncio backend).
```
sqlite3.OperationalError: database is locked
```
cause SQLite is a file-based database that has limitations on concurrent write access. This error typically occurs when multiple processes, threads, or even multiple `sqlite-anyio` connections within the same application try to write to the database simultaneously, or if a previous connection was not properly closed.

fix Design your application to minimize concurrent writes or use a queuing mechanism for write operations. Ensure that `Connection` and `Cursor` objects are always properly closed using `async with` context managers. Consider using WAL (Write-Ahead Logging) journal mode for better concurrency if the underlying SQLite setup supports it, though `sqlite-anyio` abstracts this.
```
NameError: name 'Connection' is not defined
```
cause The `Connection` class was not correctly imported from the `sqlite_anyio` library.

fix Add the correct import statement: `from sqlite_anyio import Connection` at the top of your Python file.

Warnings

breaking The `Connection.execute()` method in version 0.2.8 and later now returns `self` (the Connection object) instead of a new Cursor object. Prior versions might have returned a Cursor or similar.
Fix: If you previously relied on `Connection.execute()` to return a cursor for chaining operations, refactor your code to explicitly use `async with conn.cursor() as cur:` to obtain a cursor and then call `await cur.execute(...)`.
gotcha Both `Connection` and `Cursor` objects are designed to be used as asynchronous context managers (`async with`). Failing to use them as such can lead to unclosed database connections or cursors, potentially causing resource leaks or database locking issues.
Fix: Always wrap `Connection` and `Cursor` instantiation in `async with` statements to ensure proper resource management, e.g., `async with Connection("db.db") as conn:` and `async with conn.cursor() as cur:`.

Install

pip install sqlite-anyio Install stable version

Imports

Connection
```
from sqlite_anyio import Connection
```

Quickstart

This quickstart demonstrates opening an asynchronous connection to an SQLite database, creating a table, inserting data, and querying it using both `Connection.execute` and `Cursor` objects, all within an AnyIO runtime. It also shows proper resource management using async context managers.

import anyio
from sqlite_anyio import Connection
import os

async def main():
    db_path = "test.db"
    async with Connection(db_path) as conn:
        await conn.execute("CREATE TABLE IF NOT EXISTS users (id INTEGER, name TEXT)")
        await conn.execute("INSERT INTO users VALUES (?, ?)", 1, "Alice")
        await conn.execute("INSERT INTO users VALUES (?, ?)", 2, "Bob")

        async with conn.cursor() as cur:
            await cur.execute("SELECT id, name FROM users")
            rows = await cur.fetchall()
            print(f"All users: {rows}")

            await cur.execute("SELECT id, name FROM users WHERE id = ?", 1)
            row = await cur.fetchone()
            print(f"User with ID 1: {row}")

    # Clean up the database file
    if os.path.exists(db_path):
        os.remove(db_path)

if __name__ == "__main__":
    anyio.run(main)

view raw JSON →