Async Support Demos

These demos show how to use SQLiter's async API via AsyncSqliterDB. The async interface mirrors the sync API almost exactly — the main differences are the await keyword before every database operation and async with for context managers.

Installation

Async support requires the aiosqlite package. Install it with:

pip install sqliter-py[async]
# or, for everything:
pip install sqliter-py[full]

Async Connection

Create an async database connection. The constructor parameters are identical to SqliterDB — only the method calls need await.

import asyncio
from sqliter.asyncio import AsyncSqliterDB

async def main():
    db = AsyncSqliterDB(memory=True)
    print(f"Created database: {db}")
    print(f"Is memory: {db.is_memory}")
    print(f"Filename: {db.filename}")

    await db.connect()
    print(f"Connected: {db.is_connected}")

    await db.close()
    print(f"After close: {db.is_connected}")

asyncio.run(main())

What Happens

AsyncSqliterDB creates the database object but does not open a connection until await db.connect() is called (or an async with block is entered). All properties — is_memory, filename, is_connected — behave the same as the sync version.

When to Use

Use AsyncSqliterDB whenever your application already runs an async event loop (FastAPI, aiohttp, asyncio-based CLI tools). Mixing sync SQLiter inside an async context can block the event loop.

Async Context Manager

The async with db: block manages transaction scope automatically — identical semantics to the sync with db:, but you must use the async form.

import asyncio
from sqliter.asyncio import AsyncSqliterDB
from sqliter.model import BaseDBModel

class Task(BaseDBModel):
    title: str
    done: bool = False

async def main():
    db = AsyncSqliterDB(memory=True)

    async with db:
        await db.create_table(Task)
        task = await db.insert(Task(title="Learn async SQLiter"))
        print(f"Inserted: {task.title} (pk={task.pk})")
        print("Transaction auto-commits on exit")

    print(f"\nAfter context: connected={db.is_connected}")
    await db.close()

asyncio.run(main())

What Happens

On entry, async with db: opens the connection and begins a transaction. On clean exit, the transaction is committed. If an exception escapes the block, all changes are rolled back automatically.

When to Use

Prefer async with db: when you need atomicity. Starting with 0.21.0, the context manager no longer closes the connection automatically; call await db.close() explicitly when the async database instance is no longer needed.

Must use async with, not with

Using the plain with db: on an AsyncSqliterDB instance will raise a TypeError. Always use async with db:.

Async CRUD Operations

All data operations — including create_table — are coroutines that must be awaited. The method signatures are otherwise identical to the sync API.

import asyncio
from sqliter.asyncio import AsyncSqliterDB
from sqliter.model import BaseDBModel

class Product(BaseDBModel):
    name: str
    price: float

async def main():
    db = AsyncSqliterDB(memory=True)
    await db.create_table(Product)

    # Insert
    widget = await db.insert(Product(name="Widget", price=9.99))
    print(f"Inserted: {widget.name} pk={widget.pk}")

    # Get by primary key
    fetched = await db.get(Product, widget.pk)
    if fetched is not None:
        print(f"Fetched: {fetched.name}")

    # Update
    widget.price = 12.99
    await db.update(widget)
    updated = await db.get(Product, widget.pk)
    if updated is not None:
        print(f"Updated price: {updated.price}")

    # Delete
    await db.delete(Product, widget.pk)
    gone = await db.get(Product, widget.pk)
    print(f"After delete: {gone}")

    await db.close()

asyncio.run(main())

What Happens

Each database call suspends the coroutine while the I/O completes, letting the event loop run other tasks. The returned objects are the same Pydantic model instances as the sync API.

Async Bulk Insert

Insert multiple records efficiently in a single async transaction.

import asyncio
from sqliter.asyncio import AsyncSqliterDB
from sqliter.model import BaseDBModel

class Tag(BaseDBModel):
    label: str

async def main():
    db = AsyncSqliterDB(memory=True)
    await db.create_table(Tag)

    tags = await db.bulk_insert([
        Tag(label="python"),
        Tag(label="async"),
        Tag(label="sqlite"),
    ])

    print(f"Inserted {len(tags)} tags:")
    for tag in tags:
        print(f"  pk={tag.pk}  {tag.label}")

    count = await db.select(Tag).count()
    print(f"\nTotal in DB: {count}")

    await db.close()

asyncio.run(main())

What Happens

All inserts are wrapped in a single transaction for performance. The method returns the list of inserted instances, each with its pk populated. Passing an empty list is a no-op that returns [].

Async Queries

Query builder chaining methods (filter, limit, offset, order) are synchronous and return the builder. Only the terminal methods that actually touch the database are coroutines.

import asyncio
from sqliter.asyncio import AsyncSqliterDB
from sqliter.model import BaseDBModel

class Item(BaseDBModel):
    name: str
    qty: int

async def main():
    db = AsyncSqliterDB(memory=True)
    await db.create_table(Item)
    await db.bulk_insert([
        Item(name="Apple", qty=5),
        Item(name="Banana", qty=12),
        Item(name="Cherry", qty=3),
        Item(name="Date", qty=8),
    ])

    # Terminal methods require await
    all_items = await db.select(Item).fetch_all()
    print(f"All items: {len(all_items)}")

    in_stock = await db.select(Item).filter(qty__gt=4).fetch_all()
    print(f"qty > 4: {[i.name for i in in_stock]}")

    first = await db.select(Item).order("qty", reverse=True).fetch_first()
    if first:
        print(f"Most stock: {first.name} ({first.qty})")

    count = await db.select(Item).filter(qty__lt=6).count()
    print(f"Low-stock count: {count}")

    has_apple = await db.select(Item).filter(name__eq="Apple").exists()
    print(f"Apple exists: {has_apple}")

    await db.close()

asyncio.run(main())

Async Terminal Methods

Method	Returns	Description
`await .fetch_all()`	`list[T]`	All matching records
`await .fetch_one()`	`T \\| None`	First match or None
`await .fetch_first()`	`T \\| None`	First by ordering
`await .fetch_last()`	`T \\| None`	Last by ordering
`await .fetch_dicts()`	`list[dict]`	Results as dicts (field selection)
`await .count()`	`int`	Number of matching records
`await .exists()`	`bool`	True if any match found
`await .delete()`	`int`	Deletes matches, returns count
`await .update(values)`	`int`	Bulk-updates matches, returns count

FK Lazy Loading

This is the most important difference between sync and async mode. In sync mode, accessing book.author triggers an automatic lazy load. In async mode, FK fields return an AsyncLazyLoader object — you must explicitly call await loader.fetch() to retrieve the related object.

import asyncio
from typing import cast
from sqliter.asyncio import AsyncSqliterDB
from sqliter.asyncio.orm import AsyncBaseDBModel, AsyncForeignKey, AsyncLazyLoader

class Author(AsyncBaseDBModel):
    name: str

class Book(AsyncBaseDBModel):
    title: str
    author: AsyncForeignKey[Author] = AsyncForeignKey(Author)

async def main():
    db = AsyncSqliterDB(memory=True)
    await db.create_table(Author)
    await db.create_table(Book)

    tolkien = await db.insert(Author(name="J.R.R. Tolkien"))
    book = await db.insert(Book(title="The Hobbit", author=tolkien))

    # Re-fetch to simulate loading from DB (no in-memory state)
    fresh = await db.get(Book, book.pk)
    if fresh is None:
        raise ValueError("Book not found")

    # WRONG — raises AttributeError in async mode:
    # fresh.author.name
    # AttributeError: Async foreign key 'name' is not loaded.
    #   Use `await relation.fetch()` first.

    # CORRECT — get the loader, then await it.
    # mypy: AsyncForeignKey is typed as returning T (Author) to keep
    # eager-loaded access ergonomic. For lazy loading, a cast is needed
    # under --strict. See the async guide for the full explanation.
    loader = cast(AsyncLazyLoader[Author], fresh.author)
    author = await loader.fetch()
    if author is None:
        raise ValueError("Author not found")
    print(f"Book: {fresh.title}")
    print(f"Author: {author.name}")

    await db.close()

asyncio.run(main())

Async FK access is always explicit

Unlike sync mode where book.author.name triggers an automatic database query, in async mode you must call await book.author.fetch() first. Accessing any attribute on an unloaded AsyncLazyLoader raises:

AttributeError: Async foreign key 'name' is not loaded.
  Use `await relation.fetch()` first.

This is intentional — hidden I/O inside property access is not safe in async code.

mypy strict mode

AsyncForeignKey is typed to return the model type (T) so that eager-loaded access (book.author.name after select_related) type-checks without any extra annotation. At runtime the lazy-loaded value is AsyncLazyLoader[T], so strict mypy requires the cast shown above. See mypy and Static Type Checking for the full explanation and both workarounds.

Sync vs Async FK Access

	Sync	Async
Model base	`BaseDBModel`	`AsyncBaseDBModel`
FK field	`ForeignKey[T]`	`AsyncForeignKey[T]`
Access pattern	`book.author.name`	`author = await book.author.fetch()` then `author.name`
Already loaded?	Cached automatically	Cached after first `fetch()`

FK Eager Loading

Use select_related() to load the FK in the same query via a JOIN. The related object is immediately available without any extra fetch() call — the closest async equivalent to sync lazy loading ergonomics.

import asyncio
from sqliter.asyncio import AsyncSqliterDB
from sqliter.asyncio.orm import AsyncBaseDBModel, AsyncForeignKey

class Author(AsyncBaseDBModel):
    name: str

class Book(AsyncBaseDBModel):
    title: str
    author: AsyncForeignKey[Author] = AsyncForeignKey(Author)

async def main():
    db = AsyncSqliterDB(memory=True)
    await db.create_table(Author)
    await db.create_table(Book)

    austen = await db.insert(Author(name="Jane Austen"))
    await db.insert(Book(title="Pride and Prejudice", author=austen))
    await db.insert(Book(title="Emma", author=austen))

    # select_related performs a JOIN — author is loaded immediately
    books = await db.select(Book).select_related("author").fetch_all()
    for b in books:
        # No await needed — already loaded via JOIN
        print(f"  {b.title} by {b.author.name}")

    print(f"\nLoaded {len(books)} books with eager FK")

    await db.close()

asyncio.run(main())

When to Use Each Approach

Approach	When to use
Lazy (`await loader.fetch()`)	You only sometimes need the related object, or are fetching a large list where not every item needs its FK
Eager (`select_related()`)	You know you will always access the related object — avoids N+1 queries

Reverse Relationships

When an AsyncForeignKey has related_name set, the related model gains a reverse accessor. In async mode this returns an AsyncReverseQuery — call await accessor.fetch_all() (or any other terminal method) to get results.

import asyncio
from typing import cast
from sqliter.asyncio import AsyncSqliterDB
from sqliter.asyncio.orm import AsyncBaseDBModel, AsyncForeignKey, AsyncReverseQuery

class Author(AsyncBaseDBModel):
    name: str

class Book(AsyncBaseDBModel):
    title: str
    author: AsyncForeignKey[Author] = AsyncForeignKey(
        Author, related_name="books"
    )

async def main():
    db = AsyncSqliterDB(memory=True)
    await db.create_table(Author)
    await db.create_table(Book)

    dickens = await db.insert(Author(name="Charles Dickens"))
    await db.insert(Book(title="Oliver Twist", author=dickens))
    await db.insert(Book(title="Great Expectations", author=dickens))

    author = await db.get(Author, dickens.pk)
    if author is None:
        raise ValueError("Author not found")

    # mypy: reverse accessors are set dynamically via setattr, so
    # __getattribute__ returns `object`. Cast to AsyncReverseQuery for
    # strict mypy. fetch_all() returns list[BaseDBModel], cast to the
    # concrete type. Without --strict these casts are not needed.
    books_query = cast(AsyncReverseQuery, author.books)
    books = cast(list[Book], await books_query.fetch_all())
    print(f"Author: {author.name}")
    print(f"Books ({len(books)}):")
    for b in books:
        print(f"  - {b.title}")

    count = await books_query.count()
    print(f"Total via .count(): {count}")

    await db.close()

asyncio.run(main())

What Happens

author.books returns an AsyncReverseQuery — a lazy query builder, not a list. Only when you call a terminal method (fetch_all(), fetch_one(), count(), exists()) is the database queried. You can also call .filter() on it before fetching to narrow the results.

mypy strict mode

Reverse accessors are installed dynamically so __getattribute__ returns object for them. Under --strict mypy, cast to AsyncReverseQuery and cast the result of fetch_all() to list[YourModel] as shown above. Without strict type checking these casts are unnecessary.

Async Transactions

The async with db: context manager provides the same atomic transaction semantics as the sync version: commits on clean exit, rolls back on any unhandled exception.

import asyncio
import tempfile
from pathlib import Path
from sqliter.asyncio import AsyncSqliterDB
from sqliter.model import BaseDBModel

class Account(BaseDBModel):
    name: str
    balance: float

async def main():
    with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
        db_path = f.name

    db = AsyncSqliterDB(db_filename=db_path)
    await db.create_table(Account)

    alice = await db.insert(Account(name="Alice", balance=100.0))
    print(f"Initial: Alice=${alice.balance}")

    try:
        async with db:
            alice.balance -= 50.0
            await db.update(alice)
            print("Inside txn: deducted $50")
            raise RuntimeError("Simulated payment failure")
    except RuntimeError:
        print("Error — transaction rolled back")

    # Verify with a fresh connection
    db2 = AsyncSqliterDB(db_filename=db_path)
    restored = await db2.get(Account, alice.pk)
    if restored is not None:
        print(f"Restored: Alice=${restored.balance}")
        if restored.balance == 100.0:
            print("Rollback confirmed")
    await db2.close()
    await db.close()
    Path(db_path).unlink(missing_ok=True)

asyncio.run(main())

What Happens

The async with db: block begins an implicit transaction. When RuntimeError is raised, the context manager catches it, rolls back all changes, then re-raises. The second connection confirms the original balance was preserved.

Warning

Breaking change in 0.21.0: async with db: no longer closes the connection automatically. Call await db.close() explicitly when the async database instance is no longer needed.

When to Use

Use async with db: whenever you need atomicity; either all operations in the block succeed, or none are persisted.

Sync vs Async Quick Reference

Feature	Sync	Async
Import	`from sqliter import SqliterDB`	`from sqliter.asyncio import AsyncSqliterDB`
Model base (FK/M2M)	`BaseDBModel`	`AsyncBaseDBModel`
FK field	`ForeignKey[T]`	`AsyncForeignKey[T]`
Context manager	`with db:`	`async with db:`
Connect	`db.connect()`	`await db.connect()`
Create table	`db.create_table(M)`	`await db.create_table(M)`
Insert	`db.insert(obj)`	`await db.insert(obj)`
Get	`db.get(M, pk)`	`await db.get(M, pk)`
Update	`db.update(obj)`	`await db.update(obj)`
Delete	`db.delete(M, pk)`	`await db.delete(M, pk)`
Bulk insert	`db.bulk_insert([...])`	`await db.bulk_insert([...])`
Query terminal	`.fetch_all()`	`await .fetch_all()`
FK access	`book.author.name`	`author = await book.author.fetch()`
Eager FK	`.select_related("x")`	`await .select_related("x").fetch_all()`
Reverse FK	`author.books.fetch_all()`	`await author.books.fetch_all()`

Async Guide - Full async usage guide
AsyncSqliterDB API - API reference
AsyncQueryBuilder API - Query builder reference
Async ORM API - FK, reverse, M2M async API
Connection Demos - Sync connection patterns
ORM Features - Sync ORM and relationship demos
Transactions - Sync transaction demos

Async Support Demos

Async Connection

What Happens

When to Use

Async Context Manager

What Happens

When to Use

Async CRUD Operations

What Happens

Async Bulk Insert

What Happens

Async Queries

Async Terminal Methods

FK Lazy Loading

Sync vs Async FK Access

FK Eager Loading

When to Use Each Approach

Reverse Relationships

What Happens

Async Transactions

What Happens

When to Use

Sync vs Async Quick Reference

Related Documentation