API reference

sqlite_utils.db.Database

class sqlite_utils.db.Database(filename_or_conn: Optional[Union[str, pathlib.Path, sqlite3.Connection]] = None, memory: bool = False, recreate: bool = False, recursive_triggers: bool = True, tracer: Optional[Callable] = None, use_counts_table: bool = False)

Wrapper for a SQLite database connection that adds a variety of useful utility methods.

To create an instance:

# create data.db file, or open existing:
db = Database("data.db")
# Create an in-memory database:
dB = Database(memory=True)

• filename_or_conn - String path to a file, or a pathlib.Path object, or a sqlite3 connection

• memory - set to True to create an in-memory database

• recreate - set to True to delete and recreate a file database (dangerous)

• recursive_triggers - defaults to True, which sets PRAGMA recursive_triggers=on; - set to False to avoid setting this pragma

• tracer - set a tracer function (print works for this) which will be called with sql, parameters every time a SQL query is executed

• use_counts_table - set to True to use a cached counts table, if available. See Cached table counts using triggers.

tracer(tracer: Callable = None)

Context manager to temporarily set a tracer function - all executed SQL queries will be passed to this.

The tracer function should accept two arguments: sql and parameters

Example usage:

with db.tracer(print):
    db["creatures"].insert({"name": "Cleo"})

See Tracing queries.

__getitem__(table_name: str) → Union[sqlite_utils.db.Table, sqlite_utils.db.View]

db[table_name] returns a Table object for the table with the specified name. If the table does not exist yet it will be created the first time data is inserted into it.

register_function(fn: Optional[Callable] = None, deterministic: bool = False, replace: bool = False)

fn will be made available as a function within SQL, with the same name and number of arguments. Can be used as a decorator:

@db.register
def upper(value):
    return str(value).upper()

The decorator can take arguments:

@db.register(deterministic=True, replace=True)
def upper(value):
    return str(value).upper()

• deterministic - set True for functions that always returns the same output for a given input

• replace - set True to replace an existing function with the same name - otherwise throw an error

See Registering custom SQL functions.

register_fts4_bm25()

Register the rank_bm25(match_info) function used for calculating relevance with SQLite FTS4.

attach(alias: str, filepath: Union[str, pathlib.Path])

Attach another SQLite database file to this connection with the specified alias, equivalent to:

ATTACH DATABASE 'filepath.db' AS alias

query(sql: str, params: Optional[Union[Iterable, dict]] = None) → Generator[dict, None, None]

Execute sql and return an iterable of dictionaries representing each row.

execute(sql: str, parameters: Optional[Union[Iterable, dict]] = None) → sqlite3.Cursor

Execute SQL query and return a sqlite3.Cursor.

executescript(sql: str) → sqlite3.Cursor

Execute multiple SQL statements separated by ; and return the sqlite3.Cursor.

table(table_name: str, **kwargs) → Union[sqlite_utils.db.Table, sqlite_utils.db.View]

Return a table object, optionally configured with default options.

quote(value: str) → str

Apply SQLite string quoting to a value, including wrappping it in single quotes.

quote_fts(query: str) → str

Escape special characters in a SQLite full-text search query

table_names(fts4: bool = False, fts5: bool = False) → List[str]

A list of string table names in this database.

view_names() → List[str]

A list of string view names in this database.

property tables: List[sqlite_utils.db.Table]

A list of Table objects in this database.

property views: List[sqlite_utils.db.View]

A list of View objects in this database.

property triggers: List[sqlite_utils.db.Trigger]

A list of (name, table_name, sql) tuples representing triggers in this database.

property triggers_dict: Dict[str, str]

A {trigger_name: sql} dictionary of triggers in this database.

property schema: str

SQL schema for this database

property journal_mode: str

Current journal_mode of this database.

enable_wal()

Set journal_mode to 'wal' to enable Write-Ahead Log mode.

disable_wal()

Set journal_mode back to 'delete' to disable Write-Ahead Log mode.

enable_counts()

Enable trigger-based count caching for every table in the database, see Cached table counts using triggers.

cached_counts(tables: Optional[Iterable[str]] = None) → Dict[str, int]

Return {table_name: count} dictionary of cached counts for specified tables, or all tables if tables not provided.

reset_counts()

Re-calculate cached counts for tables.

create_table_sql(name: str, columns: Dict[str, Any], pk: Optional[Any] = None, foreign_keys: Optional[Union[Iterable[str], Iterable[sqlite_utils.db.ForeignKey], Iterable[Tuple[str, str]], Iterable[Tuple[str, str, str]], Iterable[Tuple[str, str, str, str]]]] = None, column_order: Optional[List[str]] = None, not_null: Optional[Iterable[str]] = None, defaults: Optional[Dict[str, Any]] = None, hash_id: Optional[Any] = None, extracts: Optional[Union[Dict[str, str], List[str]]] = None) → str

Returns the SQL CREATE TABLE statement for creating the specified table.

create_table(name: str, columns: Dict[str, Any], pk: Optional[Any] = None, foreign_keys: Optional[Union[Iterable[str], Iterable[sqlite_utils.db.ForeignKey], Iterable[Tuple[str, str]], Iterable[Tuple[str, str, str]], Iterable[Tuple[str, str, str, str]]]] = None, column_order: Optional[List[str]] = None, not_null: Optional[Iterable[str]] = None, defaults: Optional[Dict[str, Any]] = None, hash_id: Optional[Any] = None, extracts: Optional[Union[Dict[str, str], List[str]]] = None) → sqlite_utils.db.Table

Create a table with the specified name and the specified {column_name: type} columns.

See Explicitly creating a table.

create_view(name: str, sql: str, ignore: bool = False, replace: bool = False)

Create a new SQL view with the specified name - sql should start with SELECT ....

• ignore - set to True to do nothing if a view with this name already exists

• replace - set to True to replace the view if one with this name already exists

m2m_table_candidates(table: str, other_table: str) → List[str]

Given two table names returns the name of tables that could define a many-to-many relationship between those two tables, based on having foreign keys to both of the provided tables.

add_foreign_keys(foreign_keys: Iterable[Tuple[str, str, str, str]])

See Adding multiple foreign key constraints at once.

foreign_keys should be a list of (table, column, other_table, other_column) tuples, see Adding multiple foreign key constraints at once.

index_foreign_keys()

Create indexes for every foreign key column on every table in the database.

vacuum()

Run a SQLite VACUUM against the database.

sqlite_utils.db.Queryable

Table and View are both subclasses of Queryable, providing access to the following methods:

class sqlite_utils.db.Queryable(db, name)

exists() → bool

Does this table or view exist yet?

count_where(where: Optional[str] = None, where_args: Optional[Union[Iterable, dict]] = None) → int

Executes SELECT count(*) FROM table WHERE ... and returns a count.

property count: int

A count of the rows in this table or view.

property rows: Generator[dict, None, None]

Iterate over every dictionaries for each row in this table or view.

rows_where(where: Optional[str] = None, where_args: Optional[Union[Iterable, dict]] = None, order_by: Optional[str] = None, select: str = '*', limit: Optional[int] = None, offset: Optional[int] = None) → Generator[dict, None, None]

Iterate over every row in this table or view that matches the specified where clause.

• where - a SQL fragment to use as a WHERE clause, for example age > ? or age > :age.

• where_args - a list of arguments (if using ?) or a dictionary (if using :age).

• order_by - optional column or fragment of SQL to order by.

• select - optional comma-separated list of columns to select.

• limit - optional integer number of rows to limit to.

• offset - optional integer for SQL offset.

Returns each row as a dictionary. See Listing rows for more details.

pks_and_rows_where(where: Optional[str] = None, where_args: Optional[Union[Iterable, dict]] = None, order_by: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None) → Generator[Tuple[Any, Dict], None, None]

Like .rows_where() but returns (pk, row) pairs - pk can be a single value or tuple.

property columns: List[sqlite_utils.db.Column]

List of Columns representing the columns in this table or view.

property columns_dict: Dict[str, Any]

{column_name: python-type} dictionary representing columns in this table or view.

property schema: str

SQL schema for this table or view.

sqlite_utils.db.Table

class sqlite_utils.db.Table(db: sqlite_utils.db.Database, name: str, pk: Optional[Any] = None, foreign_keys: Optional[Union[Iterable[str], Iterable[sqlite_utils.db.ForeignKey], Iterable[Tuple[str, str]], Iterable[Tuple[str, str, str]], Iterable[Tuple[str, str, str, str]]]] = None, column_order: Optional[List[str]] = None, not_null: Optional[Iterable[str]] = None, defaults: Optional[Dict[str, Any]] = None, batch_size: int = 100, hash_id: Optional[Any] = None, alter: bool = False, ignore: bool = False, replace: bool = False, extracts: Optional[Union[Dict[str, str], List[str]]] = None, conversions: Optional[dict] = None, columns: Optional[Dict[str, Any]] = None)

Bases: sqlite_utils.db.Queryable

Tables should usually be initialized using the db.table(table_name) or db[table_name] methods.

last_rowid: Optional[int] = None

The rowid of the last inserted, updated or selected row.

last_pk: Optional[Any] = None

The primary key of the last inserted, updated or selected row.

property count: int

Count of the rows in this table - optionally from the table count cache, if configured.

exists() → bool

Does this table or view exist yet?

property pks: List[str]

Primary key columns for this table.

property use_rowid: bool

Does this table use rowid for its primary key (no other primary keys are specified)?

get(pk_values: Union[list, tuple, str, int]) → dict

Return row (as dictionary) for the specified primary key.

Primary key can be a single value, or a tuple for tables with a compound primary key.

Raises NotFoundError if a matching row cannot be found.

property foreign_keys: List[sqlite_utils.db.ForeignKey]

List of foreign keys defined on this table.

property virtual_table_using: Optional[str]

Type of virtual table, or None if this is not a virtual table.

property indexes: List[sqlite_utils.db.Index]

List of indexes defined on this table.

property xindexes: List[sqlite_utils.db.XIndex]

List of indexes defined on this table using the more detailed XIndex format.

property triggers: List[sqlite_utils.db.Trigger]

List of triggers defined on this table.

property triggers_dict: Dict[str, str]

{trigger_name: sql} dictionary of triggers defined on this table.

create(columns: Dict[str, Any], pk: Optional[Any] = None, foreign_keys: Optional[Union[Iterable[str], Iterable[sqlite_utils.db.ForeignKey], Iterable[Tuple[str, str]], Iterable[Tuple[str, str, str]], Iterable[Tuple[str, str, str, str]]]] = None, column_order: Optional[List[str]] = None, not_null: Optional[Iterable[str]] = None, defaults: Optional[Dict[str, Any]] = None, hash_id: Optional[Any] = None, extracts: Optional[Union[Dict[str, str], List[str]]] = None) → sqlite_utils.db.Table

Create a table with the specified columns.

See Explicitly creating a table for full details.

transform(*, types=None, rename=None, drop=None, pk=<sqlite_utils.db.Default object>, not_null=None, defaults=None, drop_foreign_keys=None, column_order=None) → sqlite_utils.db.Table

Apply an advanced alter table, including operations that are not supported by ALTER TABLE in SQLite itself.

See Transforming a table for full details.

transform_sql(*, types=None, rename=None, drop=None, pk=<sqlite_utils.db.Default object>, not_null=None, defaults=None, drop_foreign_keys=None, column_order=None, tmp_suffix=None) → List[str]

Returns a list of SQL statements that would be executed in order to apply this transformation.

extract(columns: Union[str, Iterable[str]], table: Optional[str] = None, fk_column: Optional[str] = None, rename: Optional[Dict[str, str]] = None) → sqlite_utils.db.Table

Extract specified columns into a separate table.

See Extracting columns into a separate table for details.

create_index(columns: Iterable[Union[str, sqlite_utils.db.DescIndex]], index_name: Optional[str] = None, unique: bool = False, if_not_exists: bool = False)

Create an index on this table.

• columns - a single columns or list of columns to index. These can be strings or, to create an index using the column in descending order, db.DescIndex(column_name) objects.

• index_name - the name to use for the new index. Defaults to the column names joined on _.

• unique - should the index be marked as unique, forcing unique values?

• if_not_exists - only create the index if one with that name does not already exist.

See Creating indexes.

add_column(col_name: str, col_type=None, fk=None, fk_col=None, not_null_default=None)

Add a column to this table. See Adding columns.

drop(ignore: bool = False)

Drop this table. ignore=True means errors will be ignored.

guess_foreign_table(column: str) → str

For a given column, suggest another table that might be referenced by this column should it be used as a foreign key.

For example, a column called tag_id or tag or tags might suggest a tag table, if one exists.

If no candidates can be found, raises a NoObviousTable exception.

add_foreign_key(column: str, other_table: Optional[str] = None, other_column: Optional[str] = None, ignore: bool = False)

Alter the schema to mark the specified column as a foreign key to another table.

• column - the column to mark as a foreign key.

• other_table - the table it refers to - if omitted, will be guessed based on the column name.

• other_column - the column on the other table it - if omitted, will be guessed.

• ignore - set this to True to ignore an existing foreign key - otherwise a AlterError will be raised.

enable_counts()

Set up triggers to update a cache of the count of rows in this table.

See Cached table counts using triggers for details.

property has_counts_triggers: bool

Does this table have triggers setup to update cached counts?

enable_fts(columns: Iterable[str], fts_version: str = 'FTS5', create_triggers: bool = False, tokenize: Optional[str] = None, replace: bool = False)

Enable SQLite full-text search against the specified columns.

• columns - list of column names to include in the search index.

• fts_version - FTS version to use - defaults to FTS5 but you may want FTS4 for older SQLite versions.

• create_triggers - should triggers be created to keep the search index up-to-date? Defaults to False.

• tokenize - custom SQLite tokenizer to use, for example "porter" to enable Porter stemming.

• replace - should any existing FTS index for this table be replaced by the new one?

See Full-text search for more details.

populate_fts(columns: Iterable[str]) → sqlite_utils.db.Table

Update the associated SQLite full-text search index with the latest data from the table for the specified columns.

disable_fts() → sqlite_utils.db.Table

Remove any full-text search index and related triggers configured for this table.

rebuild_fts()

Run the rebuild operation against the associated full-text search index table.

detect_fts() → Optional[str]

Detect if table has a corresponding FTS virtual table and return it

optimize() → sqlite_utils.db.Table

Run the optimize operation against the associated full-text search index table.

search_sql(columns: Optional[Iterable[str]] = None, order_by: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None) → str

Return SQL string that can be used to execute searches against this table.

search(q: str, order_by: Optional[str] = None, columns: Optional[Iterable[str]] = None, limit: Optional[int] = None, offset: Optional[int] = None, quote: bool = False) → Generator[dict, None, None]

Execute a search against this table using SQLite full-text search, returning a sequence of dictionaries for each row.

• q - terms to search for

• order_by - defaults to order by rank, or specify a column here.

• columns - list of columns to return, defaults to all columns.

• limit - optional integer limit for returned rows.

• offset - optional integer SQL offset.

• quote - apply quoting to disable any special characters in the search query

See Searching with table.search().

delete(pk_values: Union[list, tuple, str, int, float]) → sqlite_utils.db.Table

Delete row matching the specified primary key.

delete_where(where: Optional[str] = None, where_args: Optional[Union[Iterable, dict]] = None) → sqlite_utils.db.Table

Delete rows matching specified where clause, or delete all rows in the table.

update(pk_values: Union[list, tuple, str, int, float], updates: Optional[dict] = None, alter: bool = False, conversions: Optional[dict] = None) → sqlite_utils.db.Table

Execute a SQL UPDATE against the specified row.

• pk_values - the primary key of an individual record - can be a tuple if the table has a compound primary key.

• updates - a dictionary mapping columns to their updated values.

• alter - set to True to add any missing columns.

• conversions - optional dictionary of SQL functions to apply during the update, for example {"mycolumn": "upper(?)"}.

See Updating a specific record.

convert(columns: Union[str, List[str]], fn: Callable, output: Optional[str] = None, output_type: Optional[Any] = None, drop: bool = False, multi: bool = False, where: Optional[str] = None, where_args: Optional[Union[Iterable, dict]] = None, show_progress: bool = False)

Apply conversion function fn to every value in the specified columns.

• columns - a single column or list of string column names to convert.

• fn - a callable that takes a single argument, value, and returns it converted.

• output - optional string column name to write the results to (defaults to the input column).

• output_type - if the output column needs to be created, this is the type that will be used for the new column.

• drop - boolean, should the original column be dropped once the conversion is complete?

• multi - boolean, if True the return value of fn(value) will be expected to be a dictionary, and new columns will be created for each key of that dictionary.

• where - a SQL fragment to use as a WHERE clause to limit the rows to which the conversion is applied, for example age > ? or age > :age.

• where_args - a list of arguments (if using ?) or a dictionary (if using :age).

• show_progress - boolean, should a progress bar be displayed?

See Converting data in columns.

insert(record: Dict[str, Any], pk=<sqlite_utils.db.Default object>, foreign_keys=<sqlite_utils.db.Default object>, column_order: Optional[Union[List[str], sqlite_utils.db.Default]] = <sqlite_utils.db.Default object>, not_null: Optional[Union[Set[str], sqlite_utils.db.Default]] = <sqlite_utils.db.Default object>, defaults: Optional[Union[Dict[str, Any], sqlite_utils.db.Default]] = <sqlite_utils.db.Default object>, hash_id: Optional[Union[str, sqlite_utils.db.Default]] = <sqlite_utils.db.Default object>, alter: Optional[Union[bool, sqlite_utils.db.Default]] = <sqlite_utils.db.Default object>, ignore: Optional[Union[bool, sqlite_utils.db.Default]] = <sqlite_utils.db.Default object>, replace: Optional[Union[bool, sqlite_utils.db.Default]] = <sqlite_utils.db.Default object>, extracts: Optional[Union[Dict[str, str], List[str], sqlite_utils.db.Default]] = <sqlite_utils.db.Default object>, conversions: Optional[Union[Dict[str, str], sqlite_utils.db.Default]] = <sqlite_utils.db.Default object>, columns: Optional[Union[Dict[str, Any], sqlite_utils.db.Default]] = <sqlite_utils.db.Default object>) → sqlite_utils.db.Table

Insert a single record into the table. The table will be created with a schema that matches the inserted record if it does not already exist, see Creating tables.

• record - required: a dictionary representing the record to be inserted.

The other parameters are optional, and mostly influence how the new table will be created if that table does not exist yet.

Each of them defaults to DEFAULT, which indicates that the default setting for the current Table object (specified in the table constructor) should be used.

• pk - if creating the table, which column should be the primary key.

• foreign_keys - see Specifying foreign keys.

• column_order - optional list of strings specifying a full or partial column order to use when creating the table.

• not_null - optional set of strings specifying columns that should be NOT NULL.

• defaults - optional dictionary specifying default values for specific columns.

• hash_id - optional name of a column to create and use as a primary key, where the value of thet primary key will be derived as a SHA1 hash of the other column values in the record. hash_id="id" is a common column name used for this.

• alter - boolean, should any missing columns be added automatically?

• ignore - boolean, if a record already exists with this primary key, ignore this insert.

• replace - boolean, if a record already exists with this primary key, replace it with this new record.

• extracts - a list of columns to extract to other tables, or a dictionary that maps {column_name: other_table_name}. See Populating lookup tables automatically during insert/upsert.

• conversions - dictionary specifying SQL conversion functions to be applied to the data while it is being inserted, for example {"name": "upper(?)"}. See Converting column values using SQL functions.

• columns - dictionary over-riding the detected types used for the columns, for example {"age": int, "weight": float}.

insert_all(records, pk=<sqlite_utils.db.Default object>, foreign_keys=<sqlite_utils.db.Default object>, column_order=<sqlite_utils.db.Default object>, not_null=<sqlite_utils.db.Default object>, defaults=<sqlite_utils.db.Default object>, batch_size=<sqlite_utils.db.Default object>, hash_id=<sqlite_utils.db.Default object>, alter=<sqlite_utils.db.Default object>, ignore=<sqlite_utils.db.Default object>, replace=<sqlite_utils.db.Default object>, truncate=False, extracts=<sqlite_utils.db.Default object>, conversions=<sqlite_utils.db.Default object>, columns=<sqlite_utils.db.Default object>, upsert=False) → sqlite_utils.db.Table

Like .insert() but takes a list of records and ensures that the table that it creates (if table does not exist) has columns for ALL of that data.

upsert(record, pk=<sqlite_utils.db.Default object>, foreign_keys=<sqlite_utils.db.Default object>, column_order=<sqlite_utils.db.Default object>, not_null=<sqlite_utils.db.Default object>, defaults=<sqlite_utils.db.Default object>, hash_id=<sqlite_utils.db.Default object>, alter=<sqlite_utils.db.Default object>, extracts=<sqlite_utils.db.Default object>, conversions=<sqlite_utils.db.Default object>, columns=<sqlite_utils.db.Default object>) → sqlite_utils.db.Table

Like .insert() but performs an UPSERT, where records are inserted if they do not exist and updated if they DO exist, based on matching against their primary key.

See Upserting data.

upsert_all(records, pk=<sqlite_utils.db.Default object>, foreign_keys=<sqlite_utils.db.Default object>, column_order=<sqlite_utils.db.Default object>, not_null=<sqlite_utils.db.Default object>, defaults=<sqlite_utils.db.Default object>, batch_size=<sqlite_utils.db.Default object>, hash_id=<sqlite_utils.db.Default object>, alter=<sqlite_utils.db.Default object>, extracts=<sqlite_utils.db.Default object>, conversions=<sqlite_utils.db.Default object>, columns=<sqlite_utils.db.Default object>) → sqlite_utils.db.Table

Like .upsert() but can be applied to a list of records.

lookup(column_values: Dict[str, Any])

Create or populate a lookup table with the specified values.

db["Species"].lookup({"name": "Palm"}) will create a table called Species (if one does not already exist) with two columns: id and name. It will set up a unique constraint on the name column to guarantee it will not contain duplicate rows.

It well then inserts a new row with the name set to Palm and return the new integer primary key value.

See Working with lookup tables for more details.

m2m(other_table: Union[str, sqlite_utils.db.Table], record_or_iterable: Optional[Union[Iterable[Dict[str, Any]], Dict[str, Any]]] = None, pk: Optional[Union[Any, sqlite_utils.db.Default]] = <sqlite_utils.db.Default object>, lookup: Optional[Dict[str, Any]] = None, m2m_table: Optional[str] = None, alter: bool = False)

After inserting a record in a table, create one or more records in some other table and then create many-to-many records linking the original record and the newly created records together.

For example:

db["dogs"].insert({"id": 1, "name": "Cleo"}, pk="id").m2m(
    "humans", {"id": 1, "name": "Natalie"}, pk="id"
)

See Working with many-to-many relationships for details.

• other_table - the name of the table to insert the new records into.

• record_or_iterable - a single dictionary record to insert, or a list of records.

• pk - the primary key to use if creating other_table.

• lookup - same dictionary as for .lookup(), to create a many-to-many lookup table.

• m2m_table - the string name to use for the many-to-many table, defaults to creating this automatically based on the names of the two tables.

• alter - set to True to add any missing columns on other_table if that table already exists.

analyze_column(column: str, common_limit: int = 10, value_truncate=None, total_rows=None) → sqlite_utils.db.ColumnDetails

Return statistics about the specified column.

See Analyzing a column.

sqlite_utils.db.View

class sqlite_utils.db.View(db, name)

Bases: sqlite_utils.db.Queryable

exists()

Does this table or view exist yet?

drop(ignore=False)

enable_fts(*args, **kwargs)

Other

sqlite_utils.db.Column

class sqlite_utils.db.Column(cid, name, type, notnull, default_value, is_pk)

Describes a SQLite column returned by the Table.columns property.

cid

Column index

name

Column name

type

Column type

notnull

Does the column have a not null constraint

default_value

Default value for this column

is_pk

Is this column part of the primary key

sqlite_utils.db.ColumnDetails

class sqlite_utils.db.ColumnDetails(table, column, total_rows, num_null, num_blank, num_distinct, most_common, least_common)

Summary information about a column, see Analyzing a column.

table

The name of the table

column

The name of the column

total_rows

The total number of rows in the table

num_null

The number of rows for which this column is null

num_blank

The number of rows for which this column is blank (the empty string)

num_distinct

The number of distinct values in this column

most_common

The N most common values as a list of (value, count) tuples, or None if the table consists entirely of distinct values

least_common

The N least common values as a list of (value, count) tuples, or None if the table is entirely distinct or if the number of distinct values is less than N (since they will already have been returned in most_common)