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()
anddrop_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
andSqliteWrapper
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 asconnector_func
andquery_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 theDEFAULT_
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, callcreate_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) ordict
(return dicts of column:value) Controls how results are returned from query functions, e.g.query()
andfetch()
- 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()
withfetch='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 statementschema
, or ifschema
is empty, then checks for a pre-existing CREATE statement fortable
inSCHEMAS
.>>> 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 andTrue
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 withGenericCursor.fetchall()
orGenericCursor.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()
withfetch='all'
-
fetchone
(sql: str, *params, **kwparams) → Union[dict, tuple, None][source]¶ Alias for
fetch()
withfetch='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
- Key bool cursor_mgr
(Default:
True
) If True, wrap the returned cursor withCursorManager
- Key callable close_callback
(Default:
None
) Passed ontoCursorManager
- 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 setquery_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 likeres[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 tocursor.execute
.fetch (str) – Fetch mode. Either
all
(returncursor.fetchall()
as first return arg),one
(returncursor.fetchone()
), orno
(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), ordict
(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
ordict
. NOTE: If you’ve setfetch='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 afor
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:
-
recreate_schemas
(*tables) → dict[source]¶ Drop all tables then re-create them.
Shortcut for running
drop_schemas()
followed bycreate_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 tabletable
exists in the database, otherwiseFalse
.>>> 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 tabletable
exists in the database, otherwiseFalse
.
-
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
|
Initialise the database wrapper class. |
|
Use |
|
Create the individual table |
|
Create all tables listed in |
|
Drop all tables listed in |
|
Drop the table |
|
Drop one or more tables as positional arguments. |
|
Similar to |
|
Alias for |
|
Alias for |
|
Create and return a new database cursor object, by default the cursor will be wrapped with |
Get a list of tables present in the current database. |
|
|
Creates a database connection using |
|
Create an instance of your database wrapper: |
|
Drop all tables then re-create them. |
|
Returns |
Attributes¶
Attributes
If your database API doesn’t support returning rows as dicts or a dict-like structure (e.g. |
|
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. |
|
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. |
|
Get or create a database connection |
|
This is a static class attribute which tracks which tables have already been created, avoiding |