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

ExampleWrapper(*args, **kwargs)

PrivexDBTestBase([methodName])

Base class for all privex-db test classes.

User(*args, **kwargs)

class tests.base.PrivexDBTestBase(methodName='runTest')[source]

Base class for all privex-db test classes. Includes tearDown() to reset database after each test.

setUp() → None[source]

Hook method for setting up the test fixture before exercising it.

tearDown() → None[source]

Hook method for deconstructing the test fixture after testing it.

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, and DEFAULT_DB - so that if no database path is passed to MyManager, then the database file path contained in MyManager.DEFAULT_DB will be used as a default.

It also overrides SCHEMAS to define 2 tables (users and items) which will be automatically created when the class is instantiated, unless they already exist.

It adds two methods get_items (returns an iterator

class 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 and DEFAULT_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 to DEFAULT_DB_FOLDER)

property conn

Get or create an SQLite3 connection using DB file db and return it

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 example SqliteQueryBuilder or PostgresQueryBuilder.

To implement a sub-class of BaseQueryBuilder, you must:

  • Implement all methods marked with @abstractmethod, such as build_query(), all() and fetch()

  • 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 the Q_ 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 override get_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() and fetch() - builds and returns a string SQL query using the various class attributes such as where_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 of cursor_name nor cursor_class.

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

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)

order_by(*args, **kwargs)[source]

Alias of order()

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() and fetch() - builds and returns a string SQL query using the various class attributes such as where_clauses :return str query: The SQL query that will be sent to the database as a string

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

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

class tests.base.QueryMode[source]

A small enum.Enum used for the query_mode (whether to return rows as tuples or dicts) with Query Builder classes (see BaseQueryBuilder SqliteQueryBuilder PostgresQueryBuilder)

class tests.base.ExampleWrapper(*args, **kwargs)[source]
class tests.base.User(*args, **kwargs)
class tests.base.ExampleAsyncWrapper(*args, **kwargs)[source]