Various interesting and useful bits of functionality

You need to import apsw.ext to use this module.

Pretty printing

format_query_table() makes nicely formatted query output - see the the example.

Logging and tracebacks

You can use log_sqlite() to forward SQLite log messages to the logging module.

print_augmented_traceback() prints an exception the usual way but also includes local variables, which APSW includes to make debugging quicker and easier.

Virtual Tables

Use index_info_to_dict() to get apsw.IndexInfo in an easier to print and work with format.

Use make_virtual_module() to easily turn a Python function into a virtual table source.

generate_series() and generate_series_sqlite() provide generate_series.

Accessing result rows by column name

See the example.

Use apsw.ext.DataClassRowFactory as a apsw.Connection.row_trace for an entire connection, or apsw.Cursor.row_trace for a specific cursor.

Converting types into and out of SQLite

SQLite only stores and returns 5 types:

None
int
float
str
bytes

Use TypesConverterCursorFactory as apsw.Connection.cursor_factory to adapt values going into SQLite, and convert them coming out. See the example.

To convert values going into SQLite, do either of:

Inherit from apsw.ext.SQLiteTypeAdapter and define a to_sqlite_value method on the class
Call TypesConverterCursorFactory.register_adapter() with the type and a adapter function

To adapt values coming out of SQLite:

Call TypesConverterCursorFactory.register_converter() with the exact type string in the table and a converter function

Detailed Query Information

SQLite can provide lots of information about queries. The query_info() function can gather them up for you. This includes:

readonly if the query makes no direct changes
first_query if multiple queries are provided
actions which databases, tables, columns, functions, views etc are referenced - see actions
query_plan which indices, tables scans etc are used to find the query results - see query plans
explain for the low level steps taken inside SQLite - see SQLite bytecode

See the example.

API Reference

apsw.ext.result_string(code: int) → str[source]: Turns a result or extended result code into a string. The appropriate mapping based on the value is used.

class apsw.ext.DataClassRowFactory(*, rename: bool = True, dataclass_kwargs: dict[str, Any] | None = None)[source]

Returns each row as a dataclass, accessible by column name.

To use set an instance as Connection.row_trace to affect all cursors, or on a specific cursor:

connection.row_trace = apsw.ext.DataClassRowFactory()
for row in connection.execute("SELECT title, sum(orders) AS total, ..."):
    # You can now access by name
    print (row.title, row.total)
    # you can get the underlying description
    print (row.__description__)

You can use as many instances of this class as you want, each across as many connections as you want.

Parameters:

rename – Column names could be duplicated, or not valid in Python (eg a column named continue). If rename is True, then invalid/duplicate names are replaced with _ and their position starting at zero. For example title, total, title, continue would become title, total, _2, _3. If rename is False then problem column names will result in TypeError raised by dataclasses.make_dataclass()
dataclass_kwargs – Additional parameters when creating the dataclass as described in dataclasses.dataclass(). For example you may want frozen = True to make the dataclass read-only, or slots = True to reduce memory consumption.

get_dataclass(description: tuple[tuple[str, str], ...]) → tuple[Any, tuple[str, ...]][source]

Returns dataclass and tuple of (potentially renamed) column names

The dataclass is what is returned for each row with that description

This method caches its results.

get_type(t: str | None) → Any[source]

Returns the type hint to use in the dataclass based on the type in the description

SQLite’s affinity rules are followed.

The values have no effect on how your program runs, but can be used by tools like mypy. Column information like whether null is allowed is not present, so this is just a hint.

__call__(cursor: apsw.Cursor, row: apsw.SQLiteValues) → Any[source]

What the row tracer calls

This looks up the dataclass and column names, and then returns an instance of the dataclass.

class apsw.ext.SQLiteTypeAdapter[source]

A metaclass to indicate conversion to SQLite types is supported

This is one way to indicate your type supports conversion to a value supported by SQLite. You can either inherit from this class, or call the register method:

apsw.ext.SQLiteTypeAdapter.register(YourClassHere)

Doing either is entirely sufficient and there is no need to register with TypesConverterCursorFactory

abstract to_sqlite_value() → apsw.SQLiteValue[source]: Return a SQLite compatible value for this object

class apsw.ext.TypesConverterCursorFactory(abstract_base_class: ABCMeta = SQLiteTypeAdapter)[source]

Provides cursors that can convert objects into one of the types supported by SQLite, or back from SQLite

Parameters:: abstract_base_class – Which metaclass to consider as conversion capable

register_adapter(klass: type, callable: Callable[[Any], apsw.SQLiteValue]) → None[source]: Registers a callable that converts from klass to one of the supported SQLite types

register_converter(name: str, callable: Callable[[apsw.SQLiteValue], Any]) → None[source]: Registers a callable that converts from a SQLite value

__call__(connection: apsw.Connection) → TypeConverterCursor[source]: Returns a new convertor cursor for the connection

adapt_value(value: Any) → apsw.SQLiteValue[source]: Returns SQLite representation of value

convert_value(schematype: str, value: apsw.SQLiteValue) → Any[source]: Returns Python object from schema type and SQLite value

wrap_bindings(bindings: apsw.Bindings | None) → apsw.Bindings | None[source]: Wraps bindings that are supplied to underlying execute

wrap_sequence_bindings(sequenceofbindings: Iterable[apsw.Bindings]) → Generator[apsw.Bindings, None, None][source]: Wraps a sequence of bindings that are supplied to the underlying executemany

class DictAdapter(factory: TypesConverterCursorFactory, data: collections.abc.Mapping[str, apsw.SQLiteValue])[source]: Used to wrap dictionaries supplied as bindings

class TypeConverterCursor(connection: Connection, factory: TypesConverterCursorFactory)[source]

Cursor used to do conversions

execute(statements: str, bindings: apsw.Bindings | None = None, *, can_cache: bool = True, prepare_flags: int = 0, explain: int = -1) → apsw.Cursor[source]

Executes the statements doing conversions on supplied and returned values

See apsw.Cursor.execute() for parameter details

executemany(statements: str, sequenceofbindings: Iterable[apsw.Bindings], *, can_cache: bool = True, prepare_flags: int = 0, explain: int = -1) → apsw.Cursor[source]

Executes the statements against each item in sequenceofbindings, doing conversions on supplied and returned values

See apsw.Cursor.executemany() for parameter details

apsw.ext.log_sqlite(*, level: int = logging.ERROR, logger: Logger | None = None) → None[source]

Send SQLite log messages to logging

Parameters:

level – level to log at
logger – Use the specific logger

apsw.ext.print_augmented_traceback(exc_type: type[BaseException], exc_value: BaseException, exc_traceback: TracebackType, *, file: TextIO | None = None) → None[source]

Prints a standard exception, but also includes the value of variables in each stack frame which APSW adds to help diagnostics and debugging.

Parameters:

exc_type – The exception type
exc_value – The exception value
exc_traceback – Traceback for the exception
file – (default sys.stderr) Where the print goes

try:
    ....
except Exception as exc:
    apsw.ext.print_augmented_traceback(*sys.exc_info())

apsw.ext.index_info_to_dict(o: IndexInfo, *, column_names: list[str] | None = None, rowid_name: str = '__ROWID__') → dict[str, Any][source]

Returns a apsw.IndexInfo as a dictionary.

If column_names is supplied then additional keys with column names are present, using rowid_name for the rowid.

Here is an example response:

query = '''
    SELECT orderid, note FROM sales WHERE
            price > 74.99
            AND quantity<=?
            AND customer in ('Acme Widgets', 'Acme Industries')
    ORDER BY date desc
    LIMIT 10'''

{'aConstraint': [{'collation': 'BINARY',
                'iColumn': 5,
                'iColumn_name': 'price',
                'op': 4,
                'op_str': 'SQLITE_INDEX_CONSTRAINT_GT',
                'rhs': 74.99,
                'usable': True},
                {'collation': 'BINARY',
                'iColumn': 7,
                'iColumn_name': 'quantity',
                'op': 8,
                'op_str': 'SQLITE_INDEX_CONSTRAINT_LE',
                'rhs': None,
                'usable': True},
                {'collation': 'BINARY',
                'iColumn': 8,
                'iColumn_name': 'customer',
                'op': 2,
                'op_str': 'SQLITE_INDEX_CONSTRAINT_EQ',
                'rhs': None,
                'usable': True},
                {'collation': 'BINARY',
                'op': 73,
                'op_str': 'SQLITE_INDEX_CONSTRAINT_LIMIT',
                'rhs': 10,
                'usable': True}],
'aConstraintUsage': [{'argvIndex': 0, 'in': False, 'omit': False},
                    {'argvIndex': 0, 'in': False, 'omit': False},
                    {'argvIndex': 0, 'in': True, 'omit': False},
                    {'argvIndex': 0, 'in': False, 'omit': False}],
'aOrderBy': [{'desc': True, 'iColumn': 9, 'iColumn_name': 'date'}],
'colUsed': {0, 3, 5, 7, 8, 9},
'colUsed_names': {'date', 'note', 'customer', 'quantity', 'orderid', 'price'},
'distinct': 0,
'estimatedCost': 5e+98,
'estimatedRows': 25,
'idxFlags': 0,
'idxFlags_set': set(),
'idxNum': 0,
'idxStr': None,
'nConstraint': 4,
'nOrderBy': 1,
'orderByConsumed': False}

apsw.ext.dbinfo(db: Connection, schema: str = 'main') → tuple[DatabaseFileInfo | None, JournalFileInfo | WALFileInfo | None][source]

Extracts fields from the database, journal, and wal files

Based on the file format description. The headers are read using apsw.Connection.read() so you see inside encrypted, compressed, zip etc formats, not necessarily the actual on disk file.

Memory databases return None for both.

apsw.ext.format_query_table(db: apsw.Connection, query: str, bindings: apsw.Bindings | None = None, *, colour: bool = False, quote: bool = False, string_sanitize: Callable[[str], str] | Literal[0] | Literal[1] | Literal[2] = 1, binary: Callable[[bytes], str] = lambda x: ..., null: str = '(null)', truncate: int = 4096, truncate_val: str = ' ...', text_width: int = 80, use_unicode: bool = True, word_wrap: bool = True) → str[source]

Produces query output in an attractive text table

See the example.

Parameters:

db – Connection to run the query on
query – Query to run
bindings – Bindings for query (if needed)
colour – If True then ANSI colours are used to outline the header, and show the type of each value.
quote – If True then apsw.format_sql_value() is used to get a textual representation of a value

string_sanitize –

If this is a callable then each string is passed to it for cleaning up. Bigger numbers give more sanitization to the string. Using an example source string of:

'''hello \\ \t\f\0日本語 world'''

param	example output	description
0	hello \\ 0日本語 world	Various whitespace (eg tabs, vertical form feed) are replaced. backslashes are escaped, embedded nulls become \0
1	hello \\ \0{CJK UNIFIED IDEOGRAPH-65E5}{CJK UNIFIED IDEOGRAPH-672C}{CJK UNIFIED IDEOGRAPH-8A9E} world	After step 0, all non-ascii characters are replaced with their `unicodedata.name()` or \x and hex value
2	hello.\……..world	All non-ascii characters and whitespace are replaced by a dot

binary – Called to convert bytes to string
null – How to represent the null value
truncate – How many characters to truncate long strings at (after sanitization)
truncate_val – Appended to truncated strings to show it was truncated
text_width – Maximum output width to generate
use_unicode – If True then unicode line drawing characters are used. If False then +—+ and | are used.
word_wrap – If True then textwrap is used to break wide text to fit column width

class apsw.ext.VTColumnAccess(value, names=_not_given, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

How the column value is accessed from a row, for make_virtual_module()

By_Index = 1: By number like with tuples and lists - eg row[3]

By_Name = 2: By name like with dicts - eg row['quantity']

By_Attr = 3: By attribute like with dataclasses - eg row.quantity

apsw.ext.get_column_names(row: Any) → tuple[Sequence[str], VTColumnAccess][source]

Works out column names and access given an example row

row can be an instance of a row, or the class used to make one (eg a dataclass)

Type	Access	Column names From
`dataclasses.is_dataclass()`	`VTColumnAccess.By_Attr`	`dataclasses.fields()`
`isinstance`(`tuple`) and `hasattr`(`"_fields"`) - eg `namedtuple()`	`VTColumnAccess.By_Index`	`row._fields`
`hasattr`(`"__match_args__"`)	`VTColumnAccess.By_Attr`	`row.__match_args__` (if not empty)
`isinstance`(`dict`)	`VTColumnAccess.By_Name`	`dict.keys()`
`isinstance`(`tuple`)	`VTColumnAccess.By_Index`	`columnX` where X is zero up to `len`(`row`)

Example usage:

def method(arg1, arg2):
    yield {"fruit": "orange", "price": 17, "quantity": 2}

example_row = next(method(0, 10))
method.columns, method.column_access = apsw.ext.get_column_names(example_row)

apsw.ext.make_virtual_module(db: Connection, name: str, callable: Callable, *, eponymous: bool = True, eponymous_only: bool = False, repr_invalid: bool = False) → None[source]

Registers a read-only virtual table module with db based on callable. The callable must have an attribute named columns with a list of column names, and an attribute named column_access with a VTColumnAccess saying how to access columns from a row. See get_column_names() for easily figuring that out.

The goal is to make it very easy to turn a Python function into a virtual table. For example the following Python function:

def gendata(start_id, end_id=1000, include_system=False):
    yield (10, "2020-10-21", "readme.txt)
    yield (11, "2019-05-12", "john.txt)

gendata.columns = ("user_id", "start_date", "file_name")
gendata.column_access = VTColumnAccess.By_Index

Will generate a table declared like this, using HIDDEN for parameters:

CREATE TABLE table_name(user_id,
                        start_date,
                        file_name,
                        start_id HIDDEN,
                        end_id HIDDEN,
                        include_system HIDDEN);

inspect.signature() is used to discover parameter names.

Positional parameters to callable come from the table definition.

SELECT * from table_name(1, 100, 1);

Keyword arguments come from WHERE clauses.

SELECT * from table_name(1) WHERE
      include_system=1;

iter() is called on callable with each iteration expected to return the next row. That means callable can return its data all at once (eg a list of rows), or yield them one row at a time. The number of columns must always be the same, no matter what the parameter values.

Parameters:

eponymous – Lets you use the name as a table name without having to create a virtual table
eponymous_only – Can only reference as a table name
repr_invalid – If True then values that are not valid apsw.SQLiteValue will be converted to a string using repr()

See the example

Advanced

The callable may also have an attribute named primary_key. By default the id() of each row is used as the primary key. If present then it must be a column number to use as the primary key. The contents of that column must be unique for every row.

If you specify a parameter to the table and in WHERE, or have non-equality for WHERE clauses of parameters then the query will fail with apsw.SQLError and a message from SQLite of “no query solution”

apsw.ext.generate_series_sqlite(start=None, stop=0xffffffff, step=1)[source]

Behaves like SQLite’s generate_series

Only integers are supported. If step is negative then values are generated from stop to start

To use:

apsw.ext.make_virtual_module(db,
                             "generate_series",
                             apsw.ext.generate_series_sqlite)


db.execute("SELECT value FROM generate_series(1, 10))