Query Stack Example

This example builds an immutable StatementStack and executes it against both the synchronous SQLite adapter and the asynchronous AioSQLite adapter. Each stack:

  1. Inserts an audit log row

  2. Updates the user’s last action

  3. Fetches the user’s roles

docs/examples/patterns/stacks/query_stack_example.py
  1"""Demonstrate StatementStack usage across sync and async SQLite adapters."""
  2
  3import asyncio
  4from typing import Any
  5
  6from sqlspec import SQLSpec, StatementStack
  7from sqlspec.adapters.aiosqlite import AiosqliteConfig
  8from sqlspec.adapters.sqlite import SqliteConfig
  9
 10__all__ = ("build_stack", "main", "run_async_example", "run_sync_example")
 11
 12SCHEMA_SCRIPT = """
 13CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, last_action TEXT);
 14CREATE TABLE IF NOT EXISTS audit_log (
 15    id INTEGER PRIMARY KEY AUTOINCREMENT,
 16    user_id INTEGER NOT NULL,
 17    action TEXT NOT NULL
 18);
 19CREATE TABLE IF NOT EXISTS user_roles (
 20    user_id INTEGER NOT NULL,
 21    role TEXT NOT NULL
 22);
 23"""
 24
 25
 26def build_stack(user_id: int, action: str) -> "StatementStack":
 27    """Add audit, update, and select operations to the stack."""
 28    return (
 29        StatementStack()
 30        .push_execute(
 31            "INSERT INTO audit_log (user_id, action) VALUES (:user_id, :action)", {"user_id": user_id, "action": action}
 32        )
 33        .push_execute(
 34            "UPDATE users SET last_action = :action WHERE id = :user_id", {"action": action, "user_id": user_id}
 35        )
 36        .push_execute("SELECT role FROM user_roles WHERE user_id = :user_id ORDER BY role", {"user_id": user_id})
 37    )
 38
 39
 40def _seed_sync_tables(session: "Any", user_id: int, roles: "tuple[str, ...]") -> None:
 41    """Create tables and seed sync demo data."""
 42    session.execute_script(SCHEMA_SCRIPT)
 43    session.execute(
 44        "INSERT INTO users (id, last_action) VALUES (:user_id, :action)", {"user_id": user_id, "action": "start"}
 45    )
 46    session.execute_many(
 47        "INSERT INTO user_roles (user_id, role) VALUES (:user_id, :role)",
 48        [{"user_id": user_id, "role": role} for role in roles],
 49    )
 50
 51
 52async def _seed_async_tables(session: "Any", user_id: int, roles: "tuple[str, ...]") -> None:
 53    """Create tables and seed async demo data."""
 54    await session.execute_script(SCHEMA_SCRIPT)
 55    await session.execute(
 56        "INSERT INTO users (id, last_action) VALUES (:user_id, :action)", {"user_id": user_id, "action": "start"}
 57    )
 58    await session.execute_many(
 59        "INSERT INTO user_roles (user_id, role) VALUES (:user_id, :role)",
 60        [{"user_id": user_id, "role": role} for role in roles],
 61    )
 62
 63
 64def run_sync_example() -> None:
 65    """Execute the stack with the synchronous SQLite adapter."""
 66    registry = SQLSpec()
 67    config = registry.add_config(SqliteConfig(pool_config={"database": ":memory:"}))
 68    with registry.provide_session(config) as session:
 69        _seed_sync_tables(session, 1, ("admin", "editor"))
 70        results = session.execute_stack(build_stack(user_id=1, action="sync-login"))
 71        audit_insert, user_update, role_select = results
 72        print("[sync] rows inserted:", audit_insert.rows_affected)
 73        print("[sync] rows updated:", user_update.rows_affected)
 74        if role_select.result is not None:
 75            roles = [row["role"] for row in role_select.result.data]
 76            print("[sync] roles:", roles)
 77
 78
 79def run_async_example() -> None:
 80    """Execute the stack with the asynchronous AioSQLite adapter."""
 81
 82    async def _inner() -> None:
 83        registry = SQLSpec()
 84        config = registry.add_config(AiosqliteConfig(pool_config={"database": ":memory:"}))
 85        async with registry.provide_session(config) as session:
 86            await _seed_async_tables(session, 2, ("viewer",))
 87            results = await session.execute_stack(build_stack(user_id=2, action="async-login"))
 88            audit_insert, user_update, role_select = results
 89            print("[async] rows inserted:", audit_insert.rows_affected)
 90            print("[async] rows updated:", user_update.rows_affected)
 91            if role_select.result is not None:
 92                roles = [row["role"] for row in role_select.result.data]
 93                print("[async] roles:", roles)
 94
 95    asyncio.run(_inner())
 96
 97
 98def main() -> None:
 99    """Run both sync and async StatementStack demonstrations."""
100    run_sync_example()
101    run_async_example()
102
103
104if __name__ == "__main__":
105    main()

Run the script:

uv run python docs/examples/patterns/stacks/query_stack_example.py

Expected output shows inserted/updated row counts plus the projected role list for each adapter.