tests.base¶
This file contains code shared between tests, such as PrivexDBTestBase
which is the base class shared
by all unit tests.
Copyright:
+===================================================+
| © 2019 Privex Inc. |
| https://www.privex.io |
+===================================================+
| |
| Django Database Lock Manager |
| License: X11/MIT |
| |
| Core Developer(s): |
| |
| (+) Chris (@someguy123) [Privex] |
| |
+===================================================+
Classes
|
|
|
Base class for all privex-db test classes. |
|
-
class
tests.base.
PrivexDBTestBase
(methodName='runTest')[source]¶ Base class for all privex-db test classes. Includes
tearDown()
to reset database after each test.
-
class
tests.base.
SqliteWrapper
(db: str = None, isolation_level=None, **kwargs)[source]¶ Lightweight wrapper class for interacting with Sqlite3 databases.
Simple direct class usage
>>> db_path = expanduser('~/.my_app/my_db.db') >>> db = SqliteWrapper(db=db_path) >>> users = db.fetchall("SELECT * FROM users;")
Usage
Below is an example wrapper class which uses
SqliteWrapper
as it’s parent class.It overrides the class attributes
DEFAULT_DB_FOLDER
,DEFAULT_DB_NAME
, andDEFAULT_DB
- so that if no database path is passed toMyManager
, then the database file path contained inMyManager.DEFAULT_DB
will be used as a default.It also overrides
SCHEMAS
to define 2 tables (users
anditems
) which will be automatically created when the class is instantiated, unless they already exist.It adds two methods
get_items
(returns an iteratorclass MyManager(SqliteWrapper): ### # If a database path isn't specified, then the class attribute DEFAULT_DB will be used. ### DEFAULT_DB_FOLDER: str = expanduser('~/.my_app') DEFAULT_DB_NAME: str = 'my_app.db' DEFAULT_DB: str = join(DEFAULT_DB_FOLDER, DEFAULT_DB_NAME) ### # The SCHEMAS class attribute contains a list of tuples, with each tuple containing the name of a # table, as well as the SQL query required to create the table if it doesn't exist. ### SCHEMAS: List[Tuple[str, str]] = [ ('users', "CREATE TABLE users (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT);"), ('items', "CREATE TABLE items (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT);"), ] def get_items(self): # This is an example of a helper method you might want to define, which simply calls # self.fetchall with a pre-defined SQL query return self.fetchall("SELECT * FROM items;") def find_item(self, id: int): # This is an example of a helper method you might want to define, which simply calls # self.fetchone with a pre-defined SQL query, and interpolates the 'id' parameter into # the prepared statement. return self.fetchone("SELECT * FROM items WHERE id = ?;", [id]);
-
DEFAULT_DB
= '/home/docs/.privex_sqlite/privex_sqlite.db'¶ Combined
DEFAULT_DB_FOLDER
andDEFAULT_DB_NAME
used as default absolute path for the sqlite3 database
-
DEFAULT_DB_FOLDER
= '/home/docs/.privex_sqlite'¶ If an absolute path isn’t given, store the sqlite3 database file in this folder
-
DEFAULT_DB_NAME
= 'privex_sqlite.db'¶ If no database is specified to
__init__()
, then use this (appended toDEFAULT_DB_FOLDER
)
-
db
= None¶ Path to the SQLite3 database for this class instance
-
insert
(_table: str, _cursor: sqlite3.Cursor = None, **fields) → Union[privex.helpers.collections.DictObject, sqlite3.Cursor][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.
-
-
class
tests.base.
BaseQueryBuilder
(table: str, connection: privex.db.types.GenericConnection = None, **kwargs)[source]¶ This is an SQL query builder class which outputs ANSI compatible SQL queries, and can use connections/cursors to execute the queries that it builds.
This is an abstract base class (
abc.ABC
) meaning that it’s not designed to be constructed directly, instead it should be used as a parent class for a database specific query builder, for exampleSqliteQueryBuilder
orPostgresQueryBuilder
.To implement a sub-class of
BaseQueryBuilder
, you must:Implement all methods marked with
@abstractmethod
, such asbuild_query()
,all()
andfetch()
If your DBMS or it’s Python API doesn’t follow the default query configuration (see the attributes starting with
Q_
), then you should adjust theQ_
attributes in your class to match your DB / DB API.e.g. Set
Q_DEFAULT_PLACEHOLDER = "?"
if your DB API expects?
for prepared statement placeholders instead of%s
.While not required, you may wish to implement a constructor (
__init__()
), and overrideget_cursor()
to adjust it to your database API requirements
-
abstract
all
(query_mode=<QueryMode.ROW_DICT: 'dict'>) → Union[Iterable[dict], Iterable[tuple]][source]¶ Executes the current query, and returns an iterable cursor (results are loaded as you iterate the cursor)
Usage:
>>> results = BaseQueryBuilder('people').all() # Equivalent to ``SELECT * FROM people;`` >>> for r in results: >>> print(r['first_name'], r['last_name'], r['phone'])
- Return Iterable
A cursor which can be iterated using a
for
loop. Ideally, should load rows as you iterate, saving RAM.
-
abstract
build_query
() → str[source]¶ Used internally by
all()
andfetch()
- builds and returns a string SQL query using the various class attributes such aswhere_clauses
:return str query: The SQL query that will be sent to the database as a string
-
abstract
fetch
(query_mode=<QueryMode.ROW_DICT: 'dict'>) → Union[dict, tuple, None][source]¶ Executes the current query, and fetches the first result as a
dict
.If there are no results, will return None
- Return dict
The query result as a dictionary: {column: value, }
- Return None
If no results are found
-
abstract
fetch_next
(query_mode=<QueryMode.ROW_DICT: 'dict'>) → Union[dict, tuple, None][source]¶ Similar to
fetch()
, but doesn’t close the cursor after the query, so can be ran more than once to iterate row-by-row over the results.- Parameters
query_mode (QueryMode) –
- Returns
-
get_cursor
(cursor_name=None, cursor_class=None, *args, **kwargs)[source]¶ Create and return a new database cursor object.
It’s recommended to override this method if you’re inheriting from this class, as this Generic version of
get_cursor
does not make use ofcursor_name
norcursor_class
.- Parameters
- Return GenericCursor cursor
A cursor object which should implement at least the basic Python DB API Cursor functionality as specified in
GenericCursor
((PEP 249)
-
group_by
(*args)[source]¶ Add one or more columns to group by clause.
example: group_by(‘name’, ‘date’) == GROUP BY name, date
- Parameters
args – One or more columns to group by
- Returns
QueryBuilder object (for chaining)
-
limit
(limit_num, offset=None)[source]¶ Add a limit/offset. When using offset you should use an ORDER BY to avoid issues. :param limit_num: Amount of rows to limit to :param offset: Offset by this many rows (optional) :return: QueryBuilder object (for chaining)
-
order
(*args, direction='DESC')[source]¶ example: order(‘mycol’, ‘othercol’) == ORDER BY mycol, othercol DESC
- Parameters
args – One or more order columns as individual args
direction – Direction to sort
- Returns
QueryBuilder object (for chaining)
-
select
(*args)[source]¶ Add columns to select clause, specify as individual args. NOTE: no escaping!
example:
q.select(‘mycol’, ‘othercol’, ‘somecol as thiscol’)
can also chain: q.select(‘mycol’).select(‘othercol’)
- Parameters
args – columns to select as individual arguments
- Returns
QueryBuilder object (for chaining)
-
where
(col, val, compare='=', placeholder=None)[source]¶ For adding a simple col=value clause with “AND” before it (if at least 1 other clause). val is escaped properly
example: where(‘x’,’test’).where(‘y’,’thing’) produces prepared sql “WHERE x = %s AND y = %s”
- Parameters
col – the column, function etc. to query
val – the value it should be equal to. most python objects will be converted and escaped properly
compare – instead of ‘=’, compare using this comparator, e.g. ‘>’, ‘<=’ etc.
placeholder – Set the value placeholder, e.g. placeholder=’HOST(%s)’
- Returns
QueryBuilder object (for chaining)
-
where_or
(col, val, compare='=', placeholder=None)[source]¶ For adding simple col=value clause with “OR” before it (if at least 1 other clause). val is escaped properly
example: where(‘x’,’test’).where_or(‘y’,’thing’) produces prepared sql “WHERE x = %s OR y = %s”
- Parameters
col – the column, function etc. to query
val – the value it should be equal to. most python objects will be converted and escaped properly
compare – instead of ‘=’, compare using this comparator, e.g. ‘>’, ‘<=’ etc.
placeholder – Set the value placeholder, e.g. placeholder=’HOST(%s)’
- Returns
QueryBuilder object (for chaining)
-
class
tests.base.
SqliteQueryBuilder
(table: str, connection: privex.db.types.GenericConnection = None, **kwargs)[source]¶ -
all
(query_mode=<QueryMode.ROW_DICT: 'dict'>) → Union[Iterable[dict], Iterable[tuple]][source]¶ Executes the current query, and returns an iterable cursor (results are loaded as you iterate the cursor)
Usage:
>>> results = BaseQueryBuilder('people').all() # Equivalent to ``SELECT * FROM people;`` >>> for r in results: >>> print(r['first_name'], r['last_name'], r['phone'])
- Return Iterable
A cursor which can be iterated using a
for
loop. Ideally, should load rows as you iterate, saving RAM.
-
build_query
() → str[source]¶ Used internally by
all()
andfetch()
- builds and returns a string SQL query using the various class attributes such aswhere_clauses
:return str query: The SQL query that will be sent to the database as a string
-
-
class
tests.base.
QueryMode
[source]¶ A small
enum.Enum
used for thequery_mode
(whether to return rows as tuples or dicts) with Query Builder classes (seeBaseQueryBuilder
SqliteQueryBuilder
PostgresQueryBuilder
)
-
class
tests.base.
User
(*args, **kwargs)¶