Reference / SQL API for Python
SQL API for Python
Every public class, function and type of the SQL engine in tinystore, generated from its source. The SQL guide explains how to use them, and the Bun and Node, Go and Rust pages list the same API.
Sql
@dataclass(frozen=True, slots=True)
class Sql:
text: str
values: tuple[SqlArg, ...] = ()
bare: bool = False # whether it stands beside others in an and or an or without parentheses: one column's comparisonA piece of SQL and the values of its ?s, which are always values and never SQL.
sql makes one; a call, a batch and a table's query take it.
sql
sql = SqlMaker()Done
@dataclass(frozen=True, slots=True)
class Done:
changes: int
last_insert_rowid: intWhat a write changed: the rows, and SQLite's rowid of the row it inserted, 0 when none.
Database
class Database:
name: strAn application's database: await store.database("app", migrations=...).
The verb says what comes back: all the rows, one or none, one value, or
what a write changed. A write with returning goes through a read's
verb and runs on the writer, its rows given once it is durable.
Database.describe
@property
def describe() -> str: ...Database.handle
async def handle(connection: Connection) -> int: ...Database.keep
def keep(
method: int,
body: bytes,
make: Callable[[Connection], Awaitable[bytes]],
) -> Opener: ...Database.via
def via(what: str) -> Via: ...Database.refuse_around
def refuse_around(what: str) -> None: ...Refuses a call made around the database's transaction from inside its function.
It would wait for the writer the transaction holds until the store rolled the transaction back, and then commit alone.
Database.bucket
@overload
def bucket[T](
name: str,
of: type[T],
*,
ttl: Duration | None = None,
idle: Duration | None = None,
encrypted: bool = False,
) -> Bucket[T]: ...
@overload
def bucket(
name: str,
of: None = None,
*,
ttl: Duration | None = None,
idle: Duration | None = None,
encrypted: bool = False,
) -> Bucket[Any]: ...
def bucket(
name: str,
of: Any = None,
*,
ttl: Duration | None = None,
idle: Duration | None = None,
encrypted: bool = False,
) -> Bucket[Any]: ...A bucket of values by key kept in this database's file, beside its tables.
The same calls as store.bucket's, and its writes commit with the
rows in db.tx, through tx.with_(bucket):
sessions = db.bucket("sessions", Session, idle="30d")Database.search_index
@overload
def search_index[T](
name: str,
of: type[T],
*,
table: str,
key: str | None = None,
fields: Fields,
filters: Sequence[str] = (),
typo_tolerance: bool | None = None,
vectors: Vectors | None = None,
hybrid: Hybrid | None = None,
rebuild: Literal['automatic', 'manual'] | None = None,
) -> TableIndex[T]: ...
@overload
def search_index(
name: str,
of: None = None,
*,
table: str,
key: str | None = None,
fields: Fields,
filters: Sequence[str] = (),
typo_tolerance: bool | None = None,
vectors: Vectors | None = None,
hybrid: Hybrid | None = None,
rebuild: Literal['automatic', 'manual'] | None = None,
) -> TableIndex[dict[str, Any]]: ...
def search_index(
name: str,
of: Any = None,
*,
table: str,
key: str | None = None,
fields: Fields,
filters: Sequence[str] = (),
typo_tolerance: bool | None = None,
vectors: Vectors | None = None,
hybrid: Hybrid | None = None,
rebuild: Literal['automatic', 'manual'] | None = None,
) -> TableIndex[Any]: ...An index of a table of this database, kept in its file: its rows found again by their words or vectors.
Each hit's value is its row as all reads it. The table owns the
rows and the index follows them: every change of a row, by any
statement, trigger or migration, reaches it in the commit that makes
it, and one by a process beside the store within a second:
articles = db.search_index("articles_search", Article, table="articles", fields={"title": 2, "body": 1})Database.drop_search_index
async def drop_search_index(name: str) -> bool: ...Drops a table's index this database keeps, its triggers off the table; says whether there was one.
Database.queue
@overload
def queue[T](
name: str,
of: type[T],
*,
attempts: int | None = None,
backoff: Backoff | None = None,
timeout: Duration | None = None,
concurrency: int | Concurrency | None = None,
rate: str | None = None,
dedupe: Duration | None = None,
keep: Duration | None = None,
max_waiting: int | None = None,
) -> Queue[T]: ...
@overload
def queue(
name: str,
of: None = None,
*,
attempts: int | None = None,
backoff: Backoff | None = None,
timeout: Duration | None = None,
concurrency: int | Concurrency | None = None,
rate: str | None = None,
dedupe: Duration | None = None,
keep: Duration | None = None,
max_waiting: int | None = None,
) -> Queue[Any]: ...
def queue(
name: str,
of: Any = None,
*,
attempts: int | None = None,
backoff: Backoff | None = None,
timeout: Duration | None = None,
concurrency: int | Concurrency | None = None,
rate: str | None = None,
dedupe: Duration | None = None,
keep: Duration | None = None,
max_waiting: int | None = None,
) -> Queue[Any]: ...A queue of jobs kept in this database's file, as store.queue keeps one in jobs.db.
The same calls and workers, and its jobs commit with the rows in
db.tx, through tx.with_(queue). It shares the database's
writer, which is the point and its cost.
Database.all
@overload
async def all[T](of: type[T], statement: Statement, /, *values: Any) -> list[T]: ...
@overload
async def all(statement: Statement, /, *values: Any) -> list[dict[str, Any]]: ...
async def all(first: Any, /, *rest: Any) -> list[Any]: ...Every row a statement gives: await db.all(Note, "select * from notes where author_id = ?", user_id).
Database.one
@overload
async def one[T](of: type[T], statement: Statement, /, *values: Any) -> T | None: ...
@overload
async def one(statement: Statement, /, *values: Any) -> dict[str, Any] | None: ...
async def one(first: Any, /, *rest: Any) -> Any: ...The one row a statement gives, or None; two are InvalidError, and a write that gave them rolls back.
Database.scalar
@overload
async def scalar[T](of: type[T], statement: Statement, /, *values: Any) -> T: ...
@overload
async def scalar(statement: Statement, /, *values: Any) -> Any: ...
async def scalar(first: Any, /, *rest: Any) -> Any: ...The one value of the one row a statement gives, as count(*) does; None as SQL answers it.
Database.columns
async def columns(statement: Statement, /, *values: Any) -> dict[str, list[Any]]: ...What all gives, by its columns: a list a column, by the column's name.
For rows by the thousand, a chart's points: they come faster, and nobody turns rows into columns after.
Database.each
@overload
def each[T](of: type[T], statement: Statement, /, *values: Any) -> AsyncIterator[T]: ...
@overload
def each(statement: Statement, /, *values: Any) -> AsyncIterator[dict[str, Any]]: ...
def each(first: Any, /, *rest: Any) -> AsyncIterator[Any]: ...The rows a statement gives, one at a time as their parts arrive, from one snapshot.
Database.table
@overload
def table[T](name: str, of: type[T]) -> Table[T]: ...
@overload
def table(name: str, of: None = None) -> Table[dict[str, Any]]: ...
def table(name: str, of: Any = None) -> Table[Any]: ...A table of the database, typed where it opens: its rows' inserts and upserts, and queries.
It costs nothing until its first call; open each once, in the module
that owns it: notes = db.table("notes", Note).
Database.exec
async def exec(statement: Statement, /, *values: Any) -> Done: ...Runs a write once it is durable; writes from many callers share a commit, each failing alone.
Database.batch
async def batch(statements: Sequence[Sql]) -> list[Done]: ...Statements known before they run, written as one: all of them or none, in a shared commit.
Database.tx
async def tx[R](fn: Callable[[SqlTx], Awaitable[R]]) -> R: ...Runs fn in a transaction that holds the writer, for writes that depend on what its reads found.
Returning commits, an exception rolls back. It runs once, five seconds at most, past which the store rolls it back; inside it, call nothing that waits on the world.
Database.rows
async def rows(statement: Sql, want: Want) -> Answered: ...Database.exec_piece
async def exec_piece(statement: Sql) -> Done: ...Database.each_piece
def each_piece(statement: Sql) -> AsyncIterator[dict[str, SqlValue]] | None: ...TxQueue
class TxQueue[T]:
...A queue's calls inside a database's transaction, from tx.with_(queue).
A job it adds is there only if the transaction commits, and the queue's workers find it once it has. A worker, a watch and a page work outside.
TxQueue.add
async def add(value: T, **options: Any) -> bool: ...Adds a job unless its id is taken, as queue.add does, with the transaction.
TxQueue.set
async def set(id: str, value: T, **options: Any) -> None: ...TxQueue.update
async def update(id: str, value: T, **options: Any) -> bool: ...TxQueue.cancel
async def cancel(id: str) -> bool: ...TxQueue.get
async def get(id: str) -> Job[T] | None: ...The job under id as the transaction sees it, its own writes among it.
SqlTx
class SqlTx:
...A database's transaction, from db.tx.
Its reads see its writes, and a call that fails leaves it as it was before the call. Its calls go one at a time, in order.
SqlTx.describe
@property
def describe() -> str: ...SqlTx.with_
@overload
def with_[T](handle: Bucket[T]) -> Bucket[T]: ...
@overload
def with_[T](handle: Queue[T]) -> TxQueue[T]: ...
@overload
def with_[T](handle: Channel[T]) -> TxChannel[T]: ...
def with_(
handle: Bucket[Any] | Queue[Any] | Channel[Any],
) -> Bucket[Any] | TxQueue[Any] | TxChannel[Any]: ...A bucket or a queue opened from this database, or a channel, inside the transaction.
What it writes commits with the rows or not at all, and what it reads sees what the transaction wrote; a message is handed over at the commit.
SqlTx.all
@overload
async def all[T](of: type[T], statement: Statement, /, *values: Any) -> list[T]: ...
@overload
async def all(statement: Statement, /, *values: Any) -> list[dict[str, Any]]: ...
async def all(first: Any, /, *rest: Any) -> list[Any]: ...SqlTx.one
@overload
async def one[T](of: type[T], statement: Statement, /, *values: Any) -> T | None: ...
@overload
async def one(statement: Statement, /, *values: Any) -> dict[str, Any] | None: ...
async def one(first: Any, /, *rest: Any) -> Any: ...SqlTx.scalar
@overload
async def scalar[T](of: type[T], statement: Statement, /, *values: Any) -> T: ...
@overload
async def scalar(statement: Statement, /, *values: Any) -> Any: ...
async def scalar(first: Any, /, *rest: Any) -> Any: ...SqlTx.exec
async def exec(statement: Statement, /, *values: Any) -> Done: ...SqlTx.table
@overload
def table[T](name: str, of: type[T]) -> Table[T]: ...
@overload
def table(name: str, of: None = None) -> Table[dict[str, Any]]: ...
def table(name: str, of: Any = None) -> Table[Any]: ...A table inside the transaction: its calls are the transaction's.
SqlTx.rows
async def rows(statement: Sql, want: Want) -> Answered: ...SqlTx.exec_piece
async def exec_piece(statement: Sql) -> Done: ...SqlTx.each_piece
def each_piece(statement: Sql) -> AsyncIterator[dict[str, SqlValue]] | None: ...SqlTx.end
async def end(commit: bool) -> None: ...Sends the last call, commit or roll back, and waits for the store's word.
SqlTx.nested
async def nested(method: int, body: bytes) -> bytes: ...A call of kv or jobs in the transaction: its method and its request, and the method's answer.
RowPage
@dataclass(frozen=True, slots=True)
class RowPage[T]:
rows: list[T]
next: str | NoneA page of a query's rows, and the cursor that reads the next; None after the last.
Query
class Query[T]:
...A query on a table: a value, each call returning a new one, so that the first stays as it was.
all, one, scalar, count, each and list read;
update and delete write.
Query.where
def where(
condition: Condition | str | None = None,
/,
*values: Any,
**equal: Any,
) -> Query[T]: ...Adds a condition with and.
Equal values, as keywords or a dict, None as is null and a list as in
(...); a piece of SQL from sql; or text and the values of its ?s.
Query.select
@overload
def select[R](of: type[R], columns: str | Sql, /) -> Query[R]: ...
@overload
def select(columns: str | Sql, /) -> Query[dict[str, Any]]: ...
def select(first: Any, /, *rest: Any) -> Query[Any]: ...The columns a row holds, as SQL text, and the type of the row they make.
Query.include
def include(name: str, query: Query[Any]) -> Query[T]: ...Gives each row the rows of another query as a list under name, from the same statement.
The query names the row it belongs to in its where, by this table's alias, names its columns with select, and has a limit, which counts for each row. Every value comes back as it is kept.
Query.join
def join(table: str, on: str | Sql) -> Query[T]: ...Joins a table, "users as u", on a condition as SQL text or a piece of SQL.
Query.left_join
def left_join(table: str, on: str | Sql) -> Query[T]: ...Joins a table whose columns may be None where no row of it meets the condition.
Query.group_by
def group_by(*columns: str) -> Query[T]: ...Groups the rows by columns, quoted as names.
Query.having
def having(condition: Condition | str, /, *values: Any) -> Query[T]: ...A condition on the groups, as where takes one.
Query.order_by
def order_by(column: str | Sql, direction: Direction = 'asc') -> Query[T]: ...Orders the rows by a column, quoted as a name, or by a piece of SQL: order_by(sql("lower(title)")).
Query.limit
def limit(rows: int) -> Query[T]: ...Query.offset
def offset(rows: int) -> Query[T]: ...Query.to_sql
def to_sql() -> tuple[str, list[SqlArg]]: ...The text and the values the query sends, to read or to log.
Query.embedded
def embedded() -> Sql: ...The query inside another piece of SQL: a subquery, in parentheses.
Query.parts
@property
def parts() -> _Parts: ...Query.of
@property
def of() -> Any: ...Query.all
async def all() -> list[T]: ...Query.one
async def one() -> T | None: ...Query.scalar
async def scalar(of: Any = None) -> Any: ...Query.count
async def count() -> int: ...Every row the conditions match, without the order and the limit.
Query.each
async def each() -> AsyncIterator[T]: ...The rows one at a time, as their parts arrive.
Query.list
async def list(*, limit: int = 100, after: str | None = None) -> RowPage[T]: ...A page after the last row of the one before, by the query's order.
The order ends in a unique column so that no row is read twice or skipped; next reads the next page, and only of this query.
Query.explain
async def explain() -> list[str]: ...SQLite's plan for the query, a line a step.
Query.update
async def update(
values: Mapping[str, Any] | None = None,
/,
**columns: Any,
) -> Done: ...Sets columns of the rows the conditions match, and says how many changed.
Query.delete
async def delete() -> Done: ...Deletes the rows the conditions match, and says how many it deleted.
Table
class Table[T](Query[T]):
...A table: its queries, and its rows' inserts and upserts.
notes = db.table("notes", Note), a dataclass, a TypedDict or a
pydantic model, whose fields are the table's columns.
Table.insert
@overload
async def insert(row: T | Mapping[str, Any], /) -> T: ...
@overload
async def insert(rows: Sequence[T | Mapping[str, Any]], /) -> list[T]: ...
async def insert(given: Any, /) -> Any: ...Inserts a row and gives it back as the file keeps it; a list of rows is one write, all or none.
Table.upsert
async def upsert(
row: T | Mapping[str, Any],
*,
conflict: str | Sequence[str],
update: Sequence[str],
where: Condition | None = None,
) -> T | None: ...Inserts a row, or, when its key is taken, sets the columns update names on the row that holds it.
It does so when that row meets where, and gives back the row as the file keeps it, or None when where kept it.