privex.db.sqlite

This module holds SqliteWrapper - a somewhat higher level class for interacting with SQLite3 databases.

Copyright:

    +===================================================+
    |                 © 2019 Privex Inc.                |
    |               https://www.privex.io               |
    +===================================================+
    |                                                   |
    |        Privex's Python Database Library           |
    |        License: X11 / MIT                         |
    |                                                   |
    |        Originally Developed by Privex Inc.        |
    |        Core Developer(s):                         |
    |                                                   |
    |          (+)  Chris (@someguy123) [Privex]        |
    |                                                   |
    +===================================================+

Copyright (c) 2019     Privex Inc.   ( https://www.privex.io )

Classes

SqliteWrapper([db, isolation_level])

Lightweight wrapper class for interacting with Sqlite3 databases.

class privex.db.sqlite.SqliteAsyncWrapper(db: str = None, isolation_level=None, **kwargs)

Usage

Creating an instance:

>>> from privex.db import SqliteAsyncWrapper
>>> db = SqliteAsyncWrapper('my_app.db')

Inserting rows:

>>> db.insert('users', first_name='John', last_name='Doe')
>>> db.insert('users', first_name='Dave', last_name='Johnson')

Running raw queries:

>>> # fetchone() allows you to run a raw query, and a dict is returned with the first row result
>>> row = await db.fetchone("SELECT * FROM users WHERE first_name = ?;", ['John'])
>>> row['first_name']
John
>>> row['last_name']
Doe

>>> # fetchall() runs a query and returns an iterator of the returned rows
>>> rows = await db.fetchall("SELECT * FROM users;")
>>> for user in rows:
...     print(f"First Name: {row['first_name']}   ||    Last Name: {row['last_name']}")
...
First Name: John   ||    Last Name: Doe
First Name: Dave   ||    Last Name: Johnson

>>> # action() is for running queries where you don't want to fetch any results. It simply returns the
>>> # affected row count as an integer.
>>> row_count = await db.action('UPDATE users SET first_name = ? WHERE id = ?;', ['David', 2])
>>> print(row_count)
1

Creating tables if they don’t already exist:

>>> # If the table 'users' doesn't exist, the CREATE TABLE query will be executed.
>>> await db.create_schema(
...    'users',
...    "CREATE TABLE users (id INTEGER PRIMARY KEY AUTOINCREMENT, first_name TEXT, last_name TEXT);"
... )
>>>

Using the query builder:

>>> # You can either use it directly
>>> q = db.builder('users')
>>> q.select('first_name', 'last_name').where('first_name', 'John').where_or('last_name', 'Doe')
>>> results = q.all()
>>> async for row in results:
...     print(f"First Name: {row['first_name']}   ||    Last Name: {row['last_name']}")
...
First Name: John   ||    Last Name: Doe

>>> # Or, you can use it in a ``with`` statement to maintain a singular connection, which means you
>>> # can use .fetch_next to fetch a singular row at a time (you can still use .all() and .fetch())
>>> async with db.builder('users') as q:
...     q.select('first_name', 'last_name')
...     row = q.fetch_next()
...     print('Name:', row['first_name'], row['last_name'])   # John Doe
...     row = q.fetch_next()
...     print('Name:', row['first_name'], row['last_name'])   # Dave Johnson
...
Name: John Doe
Name: Dave Johnson

Creating a wrapper sub-class of SqliteAsyncWrapper:

class MyManager(SqliteAsyncWrapper):
    ###
    # 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);"),
    ]
    
    async 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 await self.fetchall("SELECT * FROM items;")
    
    async 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 await 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)

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

async get_cursor(cursor_name=None, cursor_class=None, *args, **kwargs) → aiosqlite.core.Cursor

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)

async insert(_table: str, _cursor: aiosqlite.core.Cursor = None, **fields) → Union[privex.helpers.collections.DictObject, aiosqlite.core.Cursor]

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 privex.db.sqlite.SqliteWrapper(db: str = None, isolation_level=None, **kwargs)

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]

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.