SQLite

from pyxpo import Button, Text, View, component, use_async, use_state
from pyxpo_sqlite import SQLiteProvider, use_sqlite_context


async def migrate(db):
    row = await db.get_first_async("PRAGMA user_version")
    version = row["user_version"]
    if version < 1:
        await db.exec_async("CREATE TABLE todo (id INTEGER PRIMARY KEY, title TEXT NOT NULL, done INTEGER DEFAULT 0)")
        await db.exec_async("PRAGMA user_version = 1")


@component
def TodoList():
    db = use_sqlite_context()
    n, set_n = use_state(0)
    rows, set_rows = use_state([])

    async def refresh():
        if n > 0:
            await db.run_async("INSERT INTO todo (title) VALUES (?)", "Item %d" % n)
        set_rows(await db.get_all_async("SELECT * FROM todo ORDER BY id"))

    use_async(refresh, [n])
    return View(
        *[Text(("[x] " if r["done"] else "[ ] ") + r["title"]) for r in rows],
        Button("Add", lambda: set_n(n + 1)),
    )


def App():
    return SQLiteProvider(TodoList(), database_name="todos.db", on_init=migrate, fallback=Text("Opening..."))

async def transfer(db):
    async def move():
        await db.run_async("UPDATE account SET cents = cents - 500 WHERE id = 1")
        await db.run_async("UPDATE account SET cents = cents + 500 WHERE id = 2")

    await db.with_transaction_async(move)          # rolls back if move raises


async def add_many(db, titles):
    stmt = await db.prepare_async("INSERT INTO todo (title) VALUES ($title)")
    try:
        for t in titles:
            await stmt.execute_async({"$title": t})
    finally:
        await stmt.finalize_async()

import asyncio
import sqlite3

import pytest
import pyxpo_sqlite as SQLite


def test_insert_and_query():
    async def go():
        db = await SQLite.open_database_async(":memory:")
        await db.exec_async("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT NOT NULL)")
        res = await db.run_async("INSERT INTO t (name) VALUES (?)", "a")
        rows = await db.get_all_async("SELECT * FROM t WHERE name = $n", {"$n": "a"})
        return res, rows

    res, rows = asyncio.run(go())
    assert (res.last_insert_row_id, res.changes) == (1, 1)
    assert rows == [{"id": 1, "name": "a"}]


def test_constraint():
    db = SQLite.open_database_sync(":memory:")
    db.exec_sync("CREATE TABLE t (x TEXT NOT NULL)")
    with pytest.raises(sqlite3.IntegrityError):
        db.run_sync("INSERT INTO t VALUES (NULL)")