gws.lib.sqlitex

Convenience wrapper for the SQLite driver.

The Object class accepts a database path and optionally an “init” DDL script. It executes queries given in a text form, with named parameters (:name).

Each query runs on its own connection in autocommit mode, which is closed immediately afterwards.

If a query fails with “no such table”, the wrapper runs the “init” script and repeats the query once. The script can contain multiple statements.

A query that fails with a recoverable error (e.g. the database is locked) is repeated on a new connection, up to MAX_ATTEMPTS times. Other errors are raised as Error.

Example:

import gws.lib.sqlitex

db = gws.lib.sqlitex.Object(
    '/data/jobs.sqlite',
    init_ddl='CREATE TABLE jobs (uid TEXT PRIMARY KEY, state TEXT)',
)
db.insert('jobs', {'uid': 'a1', 'state': 'open'})
db.update('jobs', {'state': 'done'}, 'a1')
rows = db.select('SELECT * FROM jobs WHERE state=:state', state='done')

Source code: gws.lib.sqlitex

Package Contents

gws.lib.sqlitex.BUSY_TIMEOUT = 5.0

Time in seconds to wait for a lock before giving up.

class gws.lib.sqlitex.Error

Bases: gws.Error

Raised when a query fails.

gws.lib.sqlitex.MAX_ATTEMPTS = 3

How many times to repeat a query that failed with a recoverable error.

class gws.lib.sqlitex.Object(db_path: str, init_ddl: str = '', uid_column: str = 'uid')

SQLite database wrapper.

dbPath
initDDL = ''
uidName = 'uid'
delete(table_name: str, uid)

Delete a record from a table.

Parameters:

  • table_name – Table name.

  • uid – Value of the primary key column of the record.

Raises:

Error – If the statement fails.

execute(stmt: str, **params)

Execute a text statement that does not return rows.

Parameters:

  • stmt – SQL statement.

  • **params – Values for the named parameters of the statement.

Raises:

Error – If the statement fails.

insert(table_name: str, rec: dict, on_conflict: str = '')

Insert a record into a table.

Parameters:

  • table_name – Table name.

  • rec – Record as a dict of column names and values.

  • on_conflict – What to do if the record conflicts with an existing primary or unique key: empty to fail, ignore to skip the record, update to update the existing record with the given values (the conflict is detected on the primary key column).

Raises:

Error – If the statement fails, or on_conflict is invalid.

select(stmt: str, **params) → list[dict]

Execute a text select statement.

Parameters:

  • stmt – SQL statement.

  • **params – Values for the named parameters of the statement.

Returns:

Result rows as dicts.

Raises:

Error – If the statement fails.

update(table_name: str, rec: dict, uid)

Update a record in a table.

Parameters:

  • table_name – Table name.

  • rec – Columns to update, as a dict of column names and values.

  • uid – Value of the primary key column of the record.

Raises:

Error – If the statement fails.

gws.lib.sqlitex.SLEEP_TIME = 0.1

Time in seconds to wait between the attempts.