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:

To adapt values coming out of SQLite:

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 – highest 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))
apsw.ext.generate_series(start, stop, step=None)[source]

Behaves like Postgres and SQL Server

Postgres doc SQL server doc

Operates on floating point as well as integer. If step is not specified then it is 1 if stop is greater than start and -1 if stop is less than start.

To use:

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

db.execute("SELECT value FROM generate_series(1, 10))
apsw.ext.query_info(db: apsw.Connection, query: str, bindings: apsw.Bindings | None = None, *, prepare_flags: int = 0, actions: bool = False, expanded_sql: bool = False, explain: bool = False, explain_query_plan: bool = False) QueryDetails[source]

Returns information about the query, without running it.

bindings can be None if you want to find out what the bindings for the query are.

Set the various parameters to True if you also want the actions, expanded_sql, explain, query_plan etc filled in.

class apsw.ext.QueryDetails(query: str, bindings: apsw.Bindings | None, first_query: str, query_remaining: str | None, is_explain: int, is_readonly: bool, has_vdbe: bool, bindings_count: int, bindings_names: tuple[str | None], description: tuple[tuple[str, str], ...], description_full: tuple[tuple[str, str, str, str, str], ...] | None, expanded_sql: str | None, actions: list[QueryAction] | None, explain: list[VDBEInstruction] | None, query_plan: QueryPlan | None)[source]

A dataclass that provides detailed information about a query, returned by query_info()

query: str

Original query provided

bindings: apsw.Bindings | None

Bindings provided

first_query: str

The first statement present in query

query_remaining: str | None

Query text after the first one if multiple were in query, else None

is_explain: int

Cursor.is_explain

is_readonly: bool

Cursor.is_readonly

has_vdbe: bool

Cursor.has_vdbe

bindings_count: int

How many bindings are in the query

bindings_names: tuple[str | None]

The names. The leading marker (?:@$) is omitted

description: tuple[tuple[str, str], ...]

Cursor.get_description

description_full: tuple[tuple[str, str, str, str, str], ...] | None

Cursor.description_full

expanded_sql: str | None

Cursor.expanded_sql

actions: list[QueryAction] | None

A list of the actions taken by the query, as discovered via Connection.authorizer

explain: list[VDBEInstruction] | None

A list of instructions of the internal code used by SQLite to execute the query

query_plan: QueryPlan | None

The steps taken against tables and indices described here

class apsw.ext.QueryAction(action: int, action_name: str, column_name: str | None = None, database_name: str | None = None, file_name: str | None = None, function_name: str | None = None, module_name: str | None = None, operation: str | None = None, pragma_name: str | None = None, pragma_value: str | None = None, table_name: str | None = None, trigger_name: str | None = None, trigger_or_view: str | None = None, view_name: str | None = None)[source]

A dataclass that provides information about one action taken by a query

Depending on the action, only a subset of the fields will have non-None values

action: int

Authorizer code (also present in apsw.mapping_authorizer_function)

action_name: str

The string corresponding to the action. For example action could be 21 in which case action_name will be SQLITE_SELECT

column_name: str | None = None
database_name: str | None = None

main, temp, the name in ATTACH

file_name: str | None = None
function_name: str | None = None
module_name: str | None = None
operation: str | None = None
pragma_name: str | None = None
pragma_value: str | None = None
table_name: str | None = None
trigger_name: str | None = None
trigger_or_view: str | None = None

This action is happening due to a trigger or view, and not directly expressed in the query itself

view_name: str | None = None
class apsw.ext.QueryPlan(detail: str, sub: list[QueryPlan] | None = None)[source]

A dataclass for one step of a query plan

detail: str

Description of this step

sub: list[QueryPlan] | None = None

Steps that run inside this one

class apsw.ext.VDBEInstruction(addr: int, opcode: str, comment: str | None = None, p1: int | None = None, p2: int | None = None, p3: int | None = None, p4: int | None = None, p5: int | None = None)[source]

A dataclass representing one instruction and its parameters

addr: int

Address of this opcode. It will be the target of goto, loops etc

opcode: str

The instruction

comment: str | None = None

Additional human readable information

p1: int | None = None

First opcode parameter

p2: int | None = None

Second opcode parameter

p3: int | None = None

Third opcode parameter

p4: int | None = None

Fourth opcode parameter

p5: int | None = None

Fifth opcode parameter

class apsw.ext.DatabaseFileInfo(filename: str, header: bytes, page_size: int, write_format: int, read_format: int, reserved_bytes: int, file_change_counter: int, page_count: int, freelist_pages: int, schema_cookie: int, schema_format: int, default_cache_size: int, autovacuum_top_root: int, text_encoding: str, user_version: int, incremental_vacuum: bool, application_id: int, version_valid_for: int, sqlite_version: int)[source]

Information about the main database file returned by dbinfo()

See file format description

filename: str

database filena name

header: bytes

Header string

page_size: int

The database page size in bytes

write_format: int

File format write version. 1 for legacy; 2 for WAL

read_format: int

File format read version. 1 for legacy; 2 for WAL.

reserved_bytes: int

Bytes of unused “reserved” space at the end of each page. Usually 0

file_change_counter: int

File change counter

page_count: int

Size of the database file in pages

freelist_pages: int

Total number of freelist pages

The schema cookie

schema_format: int

The schema format number. Supported schema formats are 1, 2, 3, and 4

default_cache_size: int

The schema format number. Supported schema formats are 1, 2, 3, and 4

autovacuum_top_root: int

The page number of the largest root b-tree page when in auto-vacuum or incremental-vacuum modes, or zero otherwise

text_encoding: str

The database text encoding

user_version: int

The “user version” as read and set by the user_version pragma.

incremental_vacuum: bool

True (non-zero) for incremental-vacuum mode. False (zero) otherwise.

application_id: int

The “Application ID” set by PRAGMA application_id

version_valid_for: int

The version-valid-for number.

sqlite_version: int

SQLite version that lost wrote

class apsw.ext.JournalFileInfo(filename: str, header: bytes, header_valid: bool, page_count: int, random_nonce: int, initial_pages: int, sector_size: int, page_size: int)[source]

Information about the rollback journal returned by dbinfo()

See the file format description

filename: str

journal file name

header: bytes

Header string

header_valid: bool

If the header is the expected bytes

page_count: int

The “Page Count” - The number of pages in the next segment of the journal, or -1 to mean all content to the end of the file

random_nonce: int

A random nonce for the checksum

initial_pages: int

Initial size of the database in pages

sector_size: int

Size of a disk sector assumed by the process that wrote this journal

page_size: int

Size of pages in this journal

class apsw.ext.WALFileInfo(filename: str, magic_number: int, format_version: int, page_size: int, checkpoint_sequence_number: int, salt_1: int, salt_2: int, checksum_1: int, checksum_2: int)[source]

Information about the rollback journal returned by dbinfo()

See the file format description

filename: str

WAL file name

magic_number: int

Magic number

format_version: int

File format version. Currently 3007000

page_size: int

Database page size

checkpoint_sequence_number: int

Checkpoint sequence number

salt_1: int

Salt-1: random integer incremented with each checkpoint

salt_2: int

Salt-2: a different random number for each checkpoint

checksum_1: int

Checksum-1: First part of a checksum on the first 24 bytes of header

checksum_2: int

Checksum-2: Second part of the checksum on the first 24 bytes of header