Python API¶

sqlty¶

from sqlty import SQL, UNTYPED, SQLTemplate, row_form, sqlty

SQL[T]¶

class SQL[T](str): ...

A str subclass parameterized by the row type T. It adds no methods and works anywhere a string works. sqlty rewrite creates instances:

query = SQL[tuple[int, str]]("--sql\nSELECT id::INTEGER, name::TEXT FROM users")

T is tuple[...] as inferred, a row type you write, or UNTYPED. String operations return plain str.

SQLTemplate[T]¶

class SQLTemplate[T]:
    template: string.templatelib.Template
    def __init__(self, template: Template, /) -> None: ...

A t-string query whose rows are T. Template cannot be subclassed, so the template is kept in template: pass that to a driver that runs templates (psycopg 3.3), or the whole query to execute and scalar. sqlty rewrite creates instances:

>>> from sqlty import SQLTemplate, row_form
>>> user_id = 1
>>> query = SQLTemplate[tuple[int]](t"--sql\nSELECT id::INTEGER FROM users WHERE id = {user_id}")
>>> query.template.values
(1,)
>>> row_form(query)
tuple[int]

UNTYPED¶

A PEP 661 sentinel used as the row type of a query sqlty could not type: SQL[UNTYPED]. Type checkers reject using its rows as a concrete type. See Untyped queries.

row_form(query[, default])¶

def row_form[T](query: SQL[T] | SQLTemplate[T], /) -> TypeForm[T]: ...
def row_form[T, D](query: SQL[T] | SQLTemplate[T], default: D, /) -> TypeForm[T] | D: ...

Returns the T an instance was built with, read from __orig_class__:

>>> from sqlty import SQL, row_form
>>> row_form(SQL[tuple[int, str]]("--sql\nSELECT 1::INTEGER, 'a'::TEXT"))
tuple[int, str]
>>> print(row_form(SQL("--sql\nSELECT 1::INTEGER"), None))
None

Plain strings, SQL("..."), annotation-only q: SQL[T] = "..." and stripped sources carry no runtime type: row_form raises TypeError for them unless default is given.

sqlty.drivers.psycopg¶

Requires sqlty[psycopg]. See psycopg.

from sqlty.drivers.psycopg import execute, row_factory, scalar

execute(conn, query, params=None)¶

def execute[T](conn: Connection[Any], query: SQL[T], params: Params | None = None) -> Cursor[T]: ...

Runs query on a new cursor whose row factory is row_factory(row_form(query)), or tuple_row when the query has no runtime row type.

row_factory(form)¶

def row_factory[T](form: TypeForm[T]) -> BaseRowFactory[T]: ...

Raises TypeError for anything else.

scalar(conn, query, [form,] params=None)¶

def scalar[T](conn: Connection[Any], query: SQL[tuple[T]], /, params: Params | None = None) -> T: ...
def scalar[T](conn: Connection[Any], query: LiteralString | SQL[Any], form: TypeForm[T], /, params: Params | None = None) -> T: ...

Returns the only column of the first row, typed from SQL[tuple[T]] or from form. form accepts any type expression, including unions and Literal, and is not checked at runtime. Raises LookupError when the query returns no rows and ValueError when the row has more than one column.

sqlty.core¶

from sqlty.core import (
    analyze_params,
    analyze_select,
    check_source,
    expand_star,
    explain_select,
    fix_source,
    infer_type,
    load_schema,
    returns_rows,
    rewrite_source,
    strip_source,
)

The functions behind the CLI, implemented by the Rust engine. Use them to build your own tooling:

>>> from sqlty.core import infer_type, rewrite_source, strip_source
>>> infer_type("SELECT id::INTEGER, name::TEXT FROM users")
'tuple[int, str]'
>>> infer_type("SELECT id FROM users", schema_sql="CREATE TABLE users (id INTEGER NOT NULL);")
'tuple[int]'
>>> source = rewrite_source('q = "--sql\\nSELECT 1::INTEGER"')
>>> print(source)
from sqlty import SQL
q = SQL[tuple[int]]("--sql\nSELECT 1::INTEGER")
>>> print(strip_source(source))
q = "--sql\nSELECT 1::INTEGER"

analyze_params types each placeholder (%s, %(name)s, $1, ?, :name) from its context. nullable is False for a parameter stored in a NOT NULL column, so None is not a valid argument:

>>> from sqlty.core import analyze_params
>>> schema = "CREATE TABLE users (id INTEGER NOT NULL, name TEXT NOT NULL, age INTEGER)"
>>> for p in analyze_params("INSERT INTO users (id, name, age) VALUES (%s, %s, %s)", schema_sql=schema):
...     print(p.number, p.python_type, p.nullable)
1 int False
2 str False
3 int True
>>> analyze_params("SELECT name FROM users WHERE id = ANY(%(ids)s)", schema_sql=schema)
[ParamInfo(number=1, name='ids', python_type='list[int]', sql_type='ARRAY<INT>', nullable=None, cast=False, untyped_sql_type=None)]

dialect takes postgres (the default), mysql, sqlite or another sqlparser dialect, in any case. An unknown name raises ValueError rather than reading the query as generic SQL:

>>> infer_type("SELECT 1::INTEGER", dialect="PostgreSQL")
'tuple[int]'
>>> infer_type("SELECT 1", dialect="postgress")  # doctest: +ELLIPSIS
Traceback (most recent call last):
ValueError: unknown dialect "postgress"; expected one of: postgres, mysql, sqlite, ...

schema_sql takes the schema's SQL text, or a Schema from load_schema. Text is parsed once and cached, but each call still finds it by comparing the whole text. When you check many queries against one schema, load it once and pass the handle:

>>> from sqlty.core import load_schema
>>> schema = load_schema("CREATE TABLE users (id INTEGER NOT NULL, name TEXT);")
>>> schema
Schema(dialect="postgres")
>>> infer_type("SELECT id, name FROM users", schema_sql=schema)
'tuple[int, str | None]'