We need to make sure the connection and cursor is properly closed even if an exception is thrown.
Idea:
# ... other imports ...
from contextlib import closing
def query(db_name, sql):
"""Performs a database query and returns the result.
It will uses contextlib to safely wrap a sqlite3 connection and
a cursor to ensure the connection safely commits and closes (same
as the cursor) in both happy and sad paths.
Args:
db_name (str): The name of the sqlite3 database to open.
sql (str): The SQL query to perform.
Returns:
list: A list of tuples that gets returned by the performed query,
as with calling ``cursor.fetchall()``. This will return an empty
list in case nothing gets returned (such as for an UPDATE).
"""
# Auto-closes
with closing(sqlite3.connect(db_name)) as conn:
# Auto-commits
with conn:
# Auto-closes
with contextlib.closing(conn.cursor()) as cur:
cur.execute(sql)
return cur.fetchall()
You can then use wrap this in the try...except block:
try:
# This will create a tuple that can be passed to the cursor
query_tuple = "insert into Events values (?,?,?,?,?,?)", # ....
query(DB_NAME, query_tuple)
except sqlite3.IntegrityError as e:
logging.error(e)
We need to make sure the connection and cursor is properly closed even if an exception is thrown.
Idea:
You can then use wrap this in the
try...except
block:This needs to be unit-tested as well.