Source code for dbutils
"""Module that defines common database functions.
See Also
--------
genutils : module that defines many general and useful functions.
logutils : module that defines common logging functions.
"""
import logging
import os
import sqlite3
import time
from logging import NullHandler
# Custom modules
from pyutils.exceptions.sql import SQLSanityCheckError
logging.getLogger(__name__).addHandler(NullHandler())
logger = logging.getLogger(__name__)
[docs]def connect_db(db_path, autocommit=False):
"""Open a database connection to a SQLite database.
Parameters
----------
db_path : str
File path to the database file.
autocommit : bool, optional
In autocommit mode, all changes to the database are committed as soon
as all operations associated with the current database connection
complete [1]_ (the default value is False, which implies that statements
that modify the database don't take effect immediately [2]_. You have to
call :meth:`~sqlite3.Connection.commit` to close the transaction.).
Raises
------
sqlite3.Error
Raised if any SQLite-related errors occur, e.g. :exc:`IntegrityError` or
:exc:`OperationalError`, since :exc:`sqlite3.Error` is the class for all
exceptions of the module.
Returns
-------
sqlite3.Connection
Connection object that represents the SQLite database.
.. [1] `7.0 Transaction Control At The SQL Level
<https://www.sqlite.org/lockingv3.html>`_.
.. [2] `Controlling Transactions
<https://docs.python.org/3/library/sqlite3.html#controlling-transactions>`_.
"""
# TODO: add reference
try:
if autocommit:
# If isolation_level is None, it will leave the underlying sqlite3
# library operating in autocommit mode
# Ref.: https://bit.ly/2mg5Hie
conn = sqlite3.connect(db_path, isolation_level=None)
else:
conn = sqlite3.connect(db_path)
except sqlite3.Error as e:
raise sqlite3.Error(e)
else:
return conn
[docs]def create_db(overwrite_db, db_filepath, schema_filepath, **kwargs):
"""Create a SQLite database.
A schema file is needed for creating the database. If an existing SQLite
database will be overwritten, the user is given 5 seconds to stop the script
before the database is overwritten.
Parameters
----------
overwrite_db : bool
Whether the database will be overwritten. The user is given some time
to stop the script before the database is overwritten.
db_filepath : str
Path to the SQLite database.
schema_filepath : str
Path to the schema file.
**kwargs
TODO
Raises
------
IOError
Raised if there is any IOError when opening the schema file, e.g. the
schema file doesn't exist (OSError).
"""
# TODO: add verbose option
db_filepath = os.path.expanduser(db_filepath)
schema_filepath = os.path.expanduser(schema_filepath)
db_exists = os.path.exists(db_filepath)
if overwrite_db and db_exists:
logger.warning("{} will be overwritten ...".format(db_filepath))
# Exit program before delay expires or the database is overwritten
time.sleep(5)
os.remove(db_filepath)
if not db_exists or overwrite_db:
logger.info("Creating database '{}'".format(db_filepath))
with sqlite3.connect(db_filepath) as conn:
try:
with open(schema_filepath, 'rt') as f:
schema = f.read()
conn.executescript(schema)
except IOError as e:
raise IOError(e)
else:
logger.info("Database created!")
else:
logger.warning("Database '{}' already exists!".format(db_filepath))
[docs]def sql_sanity_check(sql, values):
"""Perform sanity checks on an SQL query.
Only SQL queries that have values to be added are checked, i.e. `INSERT`
queries.
These are the checks performed:
- Whether the values are of :obj:`tuple` type
- Whether the SQL expression contains the right number of values
Parameters
----------
sql : str
SQL query to be executed.
values : tuple of str
The values to be inserted in the database.
Raises
------
SQLSanityCheckError
Raised when the sanity check on a SQL query fails: e.g. the query's
values are not of :obj:`tuple` type or wrong number of values in the SQL
query.
"""
if type(values) is not tuple:
raise SQLSanityCheckError(
"[TypeError] The values for the SQL expression are not of "
"tuple type")
if len(values) != sql.count('?'):
raise SQLSanityCheckError(
"[AssertionError] Wrong number of values ({}) in the SQL "
"expression '{}'".format(len(values), sql))