GenericDBWrapper¶

class privex.db.base.GenericDBWrapper(db=None, connector_func: callable = None, **kwargs)[source]¶

This is a generic database wrapper class, which implements various methods such as:

Querying methods such as query(), fetch(), fetchone(), fetchall()
Table management functions such as create_schemas(), drop_schemas() and drop_table()
Connection and cursor methods / properties: conn, get_cursor()

While this class is intended to be subclassed by DBMS-specific wrapper classes, all methods follow the Python DB API (PEP 249) and the ANSI SQL standard, meaning very little modification is actually required to adapt this wrapper to most database systems.

See PostgresWrapper and SqliteWrapper for implementation examples.

__init__(db=None, connector_func: callable = None, **kwargs)[source]

Initialise the database wrapper class.

This constructor sets _conn to None, and sets up various instance variables such as connector_func and query_mode.

While you can set various instance variables such as query_mode via this constructor, if you’re inheriting this class, it’s recommended that you override the DEFAULT_ static class attributes to your preference.

Parameters

db (str) – The name / path of the database to connect to
connector_func (callable) – A function / method / lambda which returns a database connection object which acts like GenericConnection

Key bool auto_create_schema

(Default: True) If True, call create_schemas() during constructor.

Key list connector_args

A list of arguments to be passed as positional args to connector_func

Key dict connector_kwargs

A dict of arguments to passed as keyword args to connector_func

Key str query_mode

Either flat (return tuples) or dict (return dicts of column:value) Controls how results are returned from query functions, e.g. query() and fetch()

Key str table_query

The query used to check for existence of a table. The query should take one prepared statement argument (the table name to check for), and the first column returned must be named table_count - an integer containing how many tables were found under the given name (usually just 0 if not found, 1 if found).

Key str table_list_query

The query used to obtain a list of tables in the database. The query should take no arguments, and return rows containing one column each, name - the name of the table.

AUTO_ZIP_COLS = True¶

If your database API doesn’t support returning rows as dicts or a dict-like structure (e.g. sqlite3), then when this setting is enabled, _zip_cols() will be called to zip each row with the result column names into a dictionary.

If your database API supports returning rows as a dictionary either by default, or via a cursor/connection class (e.g. PostgreSQL with psycopg2) then you should set this to False and use the cursor/connection class instead.

DEFAULT_PLACEHOLDER = '?'¶: The placeholder used by the database API for prepared statements in .execute()

DEFAULT_QUERY_MODE = 'dict'¶: This attribute can be overridden on your inheriting wrapper class to change the default query mode used if one isn’t specified in the constructor.

SCHEMAS = []¶

This should be set as a class attribute to a list of two value tuples, each containing the name of a table, and the SQL query to create the table if it doesn’t exist.

Example:

SCHEMAS = [
    (
        'my_table',
        "CREATE TABLE my_table (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT);"
    ),
    (
        'other_table',
        "CREATE TABLE other_table (id INTEGER PRIMARY KEY AUTOINCREMENT, example TEXT);"
    ),
]

action(sql: str, *params, **kwparams) → int[source]¶

Use action() as a simple alias method for running “action” queries which don’t return results, only affected row counts.

For example INSERT, UPDATE, CREATE etc. queries.

This method calls query() with fetch='no', saves the row count into a variable, closes the cursor, then returns the affected row count as an integer.

>>> db = GenericDBWrapper('SomeDB')
>>> rows_affected = db.action("DELETE FROM users WHERE first_name LIKE 'John%';")
>>> rows_affected
4

Parameters

sql (str) – An SQL query to execute on the current DB, as a string.
params – Extra arguments will be passed through to cursor.execute(sql, *params, **kwparams)
kwparams – Extra arguments will be passed through to cursor.execute(sql, *params, **kwparams)

Return int row_count

Number of rows affected

property conn¶: Get or create a database connection

create_schema(table: str, schema: str = None)[source]¶

Create the individual table table, either uses the create statement schema, or if schema is empty, then checks for a pre-existing CREATE statement for table in SCHEMAS.

>>> db = GenericDBWrapper('SomeDBName')
>>> db.create_schema('users', 'CREATE TABLE users (id INT PRIMARY KEY, name VARCHAR(200));')

Parameters

table –
schema –

Returns

create_schemas(*tables) → dict[source]¶

Create all tables listed in SCHEMAS if they don’t already exist.

Parameters: tables (str) – If one or more table names are specified, then only these tables will be affected.
Return dict result: dict_keys(['executions', 'created', 'skipped', 'tables_created', 'tables_skipped'])

drop_schemas(*tables) → dict[source]¶

Drop all tables listed in SCHEMAS if they exist.

Parameters: tables (str) – If one or more table names are specified, then only these tables will be affected.
Return dict result: dict_keys(['executions', 'created', 'skipped', 'tables_dropped', 'tables_skipped'])

drop_table(table: str) → bool[source]¶

Drop the table table if it exists. If the table exists, it will be dropped and True will be returned.

If the table doesn’t exist (thus can’t be dropped), False will be returned.

drop_tables(*tables) → List[Tuple[str, bool]][source]¶

Drop one or more tables as positional arguments.

Returns a list of tuples containing (table_name:str, was_dropped:bool,) :param str tables: One or more table names to drop as positional args :return list drop_results: List of tuples containing (table_name:str, was_dropped:bool,)

fetch(sql: str, *params, fetch='all', **kwparams) → Union[dict, tuple, List[dict], List[tuple], None][source]¶

Similar to query() but only returns the fetch results, not the execution object nor cursor.

Example Usage (default query mode)::

>>> s = GenericDBWrapper()
>>> user = s.fetch("SELECT * FROM users WHERE id = ?;", [123], fetch='one')
>>> user
(123, 'john', 'doe',)

Example Usage (dict query mode):

>>> s.query_mode = 'dict'    # Or s = SqliteWrapper(query_mode='dict')
>>> res = s.fetch("SELECT * FROM users WHERE id = ?;", [123], fetch='one')
>>> res
{'id': 123, 'first_name': 'john', 'last_name': 'doe'}

Parameters

fetch (str) – Either 'all' or 'one' - controls whether the result is fetched with GenericCursor.fetchall() or GenericCursor.fetchone()
sql (str) – An SQL query to execute on the current DB, as a string.
params – Extra arguments will be passed through to cursor.execute(sql, *params, **kwparams)
kwparams – Extra arguments will be passed through to cursor.execute(sql, *params, **kwparams)

Returns

fetchall(sql: str, *params, **kwparams) → Union[List[dict], List[tuple], None][source]¶: Alias for fetch() with fetch='all'

fetchone(sql: str, *params, **kwparams) → Union[dict, tuple, None][source]¶: Alias for fetch() with fetch='one'

get_cursor(cursor_name=None, cursor_class=None, *args, **kwargs) → Union[privex.db.base.CursorManager, privex.db.types.GenericCursor][source]¶

Create and return a new database cursor object, by default the cursor will be wrapped with CursorManager to ensure context management (with statements) works regardless of whether the database API supports context managing cursors (e.g. sqlite does not support cursor contexts).

For sub-classes, you should override _get_cursor(), which returns an actual native DB cursor.

Parameters

cursor_name (str) – (If DB API supports it) The name for this cursor
cursor_class (type) – (If DB API supports it) The cursor class to use

Key bool cursor_mgr

(Default: True) If True, wrap the returned cursor with CursorManager

Key callable close_callback

(Default: None) Passed onto CursorManager

Return GenericCursor cursor

A cursor object which should implement at least the basic Python DB API Cursor functionality as specified in GenericCursor ((PEP 249)

insert(_table: str, _cursor: privex.db.types.GenericCursor = None, **fields) → Union[privex.helpers.collections.DictObject, privex.db.types.GenericCursor][source]¶

Builds and executes an insert query into the table _table using the keyword arguments for column names and values.

>>> db = GenericDBWrapper(db='SomeDB')
>>> cur = db.insert('users', first_name='John', last_name='Doe', phone='+1-800-123-4567')
>>> cur.lastrowid
15

Parameters

_table (str) – The table to insert into
_cursor (GenericCursor) – Optionally, specify a cursor to use, instead of the default cursor
fields – Keyword args mapping column names to values

Return DictObject cur

If no custom cursor was specified, the cursor used to execute the query is converted into a DictObject before closing it, then the dict is returned.

Return GenericCursor cur

If a custom cursor (_cursor) was specified, then that cursor will NOT be auto-closed, and the original cursor instance will be returned.

list_tables() → List[str][source]¶

Get a list of tables present in the current database.

Example:

>>> GenericDBWrapper().list_tables()
['sqlite_sequence', 'nodes', 'node_api', 'node_failures']

Return List[str] tables: A list of tables in the database

make_connection(*args, **kwargs) → privex.db.types.GenericConnection[source]¶

Creates a database connection using connector_func, passing all arguments/kwargs directly to the connector function.

Return GenericConnection conn: A database connection object, which should implement at least the basic connection object methods as specified in the Python DB API (PEP 249), and in the Protocol type class GenericConnection

query(sql: str, *params, fetch='all', **kwparams) → Tuple[Optional[iter], Any, privex.db.types.GenericCursor][source]¶

Create an instance of your database wrapper:

>>> db = GenericDBWrapper()

Querying with prepared SQL queries and returning a single row:

>>> res, res_exec, cur = db.query("SELECT * FROM users WHERE first_name = ?;", ['John'], fetch='one')
>>> res
(12, 'John', 'Doe', '123 Example Road',)
>>> cur.close()

Querying with plain SQL queries, using query_mode, handling an iterator result, and using the cursor

If your database API returns rows as tuple``s or ``list``s, you can use ``query_mode='dict' (or set query_mode in the constructor) to convert any row results into dictionaries which map each column to their values.

>>> res, _, cur = db.query("SELECT * FROM users;", fetch='all', query_mode='dict')

When querying with fetch='all', depending on your database API, res may be an iterator, and cannot be accessed via an index like res[0].

You should make sure to iterate the rows using a for loop:

>>> for row in res:
...     print(row['first_name'], ':', row)
John : {'first_name': 'John', 'last_name': 'Doe', 'id': 12}
Dave : {'first_name': 'Dave', 'last_name': 'Johnson', 'id': 13}
Aaron : {'first_name': 'Aaron', 'last_name': 'Swartz', 'id': 14}

Or, if the result object is a generator, then you can auto-iterate the results into a list using x = list(res):

>>> rows = list(res)
>>> rows[0]
{'first_name': 'John', 'last_name': 'Doe', 'id': 12}

Using the returned cursor (third return item), we can access various metadata about our query. Note that cursor objects vary between database APIs, and not all methods/attributes may be available, or may return different data than shown below:

>>> cur.description  # cursor.description often contains the column names matching the query columns
(('id', None, None, None, None, None, None), ('first_name', None, None, None, None, None, None),
 ('last_name', None, None, None, None, None, None))

>>> _, _, cur = db.query("INSERT INTO users (first_name, last_name) VALUES ('a', 'b')", fetch='no')
>>> cur.rowcount   # cursor.rowcount tells us how many rows were affected by a query
1
>>> cur.lastrowid  # cursor.lastrowid tells us the ID of the last row we inserted with this cursor
3

Parameters

sql (str) – An SQL query to execute
params – Any positional arguments other than sql will be passed to cursor.execute.
fetch (str) – Fetch mode. Either all (return cursor.fetchall() as first return arg), one (return cursor.fetchone()), or no (do not fetch. first return arg is None).
kwparams – Any keyword arguments that aren’t specified as parameters / keyword args for this method will be forwarded to cursor.execute

Key GenericCursor cursor

Use this specific cursor instead of automatically obtaining one

Key cursor_name

If your database API supports named cursors (e.g. PostgreSQL), then you may specify cursor_name as a keyword argument to use a named cursor for this query.

Key query_mode

Either flat (fetch results as they were originally returned from the DB), or dict (use _zip_cols() to convert tuple/list rows into dicts mapping col:value).

Return iter results

(tuple item 1) An iterable such as a generator, or storage type e.g. list or dict. NOTE: If you’ve set fetch='all', depending on your database adapter, this may be a generator or other form of iterator that cannot be directly accessed via index (i.e. res[123]). Instead you must iterate it with a for loop, or cast it into a list/tuple to automatically iterate it into an indexed object, e.g. ``list(res)`

Return Any res_exec

(tuple item 2) The object returned from running cur.execute(sql, *params, **kwparams). This may be a cursor, but may also vary based on database API.

Return GenericCursor cur

(tuple item 3) The cursor that was used to execute and fetch your query. To allow for use with server side cursors, the cursor is NOT closed automatically. To avoid stale cursors, it’s best to run cur.close() when you’re done with handling the returned results.

query_mode = None¶

Per-instance attribute, either:

'flat' (query() returns rows as tuples)
'dict' (query() returns rows as dicts mapping column names to values)

recreate_schemas(*tables) → dict[source]¶

Drop all tables then re-create them.

Shortcut for running drop_schemas() followed by create_schemas().

Parameters: tables (str) – If one or more table names are specified, then only these tables will be affected.
Return dict result: dict_keys(['tables_created', 'skipped_create', 'skipped_drop', 'tables_dropped'])

table_exists(table: str) → bool[source]¶

Returns True if the table table exists in the database, otherwise False.

>>> GenericDBWrapper().table_exists('some_table')
True
>>> GenericDBWrapper().table_exists('other_table')
False

Parameters: table (str) – The table to check for existence.
Return bool exists: True if the table table exists in the database, otherwise False.

tables_created = {}¶: This is a static class attribute which tracks which tables have already been created, avoiding create_schemas() having to make as many queries every time the class is constructed.

Methods¶

Methods

`__init__`([db, connector_func])	Initialise the database wrapper class.
`action`(sql, params, *kwparams)	Use `action()` as a simple alias method for running “action” queries which don’t return results, only affected row counts.
`close_cursor`()
`create_schema`(table[, schema])	Create the individual table `table`, either uses the create statement `schema`, or if `schema` is empty, then checks for a pre-existing CREATE statement for `table` in `SCHEMAS`.
`create_schemas`(*tables)	Create all tables listed in `SCHEMAS` if they don’t already exist.
`drop_schemas`(*tables)	Drop all tables listed in `SCHEMAS` if they exist.
`drop_table`(table)	Drop the table `table` if it exists.
`drop_tables`(*tables)	Drop one or more tables as positional arguments.
`fetch`(sql, *params[, fetch])	Similar to `query()` but only returns the fetch results, not the execution object nor cursor.
`fetchall`(sql, params, *kwparams)	Alias for `fetch()` with `fetch='all'`
`fetchone`(sql, params, *kwparams)	Alias for `fetch()` with `fetch='one'`
`get_cursor`([cursor_name, cursor_class])	Create and return a new database cursor object, by default the cursor will be wrapped with `CursorManager` to ensure context management (`with` statements) works regardless of whether the database API supports context managing cursors (e.g.
`list_tables`()	Get a list of tables present in the current database.
`make_connection`(args, *kwargs)	Creates a database connection using `connector_func`, passing all arguments/kwargs directly to the connector function.
`query`(sql, *params[, fetch])	Create an instance of your database wrapper:
`recreate_schemas`(*tables)	Drop all tables then re-create them.
`table_exists`(table)	Returns `True` if the table `table` exists in the database, otherwise `False`.

Attributes¶

Attributes

`AUTO_ZIP_COLS`	If your database API doesn’t support returning rows as dicts or a dict-like structure (e.g.
`DEFAULT_ENABLE_EXECUTION_LOG`
`DEFAULT_QUERY_MODE`	This attribute can be overridden on your inheriting wrapper class to change the default query mode used if one isn’t specified in the constructor.
`DEFAULT_TABLE_LIST_QUERY`
`DEFAULT_TABLE_QUERY`
`SCHEMAS`	This should be set as a class attribute to a list of two value tuples, each containing the name of a table, and the SQL query to create the table if it doesn’t exist.
`conn`	Get or create a database connection
`cursor`
`tables_created`	This is a static class attribute which tracks which tables have already been created, avoiding `create_schemas()` having to make as many queries every time the class is constructed.