Skip to content

Instantly share code, notes, and snippets.

@NuarkNoir
Created May 22, 2020 08:16
Show Gist options
  • Save NuarkNoir/ec5689ed6063253f48c09129842d98e1 to your computer and use it in GitHub Desktop.
Save NuarkNoir/ec5689ed6063253f48c09129842d98e1 to your computer and use it in GitHub Desktop.
Simple sqlite3 wrapper for python 3+
import warnings
import sqlite3 as sqlite
class Database(object):
__connected = False
__connection = None
__autocommit = True
def __init__(self, path=":memory:", autocommit=True):
self.__path = path
self.__autocommit = autocommit
self.open_connection()
def open_connection(self):
if not self.__connected:
try:
self.__connection = sqlite.connect(database=self.__path)
self.__connected = True
except Exception as e:
raise e
else:
warnings.warn(f"Database connection with `{self.__path}` already established")
def close_connection(self):
if self.__connected:
self.__connection.close()
self.__connection = None
self.__connected = False
else:
warnings.warn(f"Database connection with `{self.__path}` already closed")
def session(self):
return ConnectionProxy(self.__connection, self.__autocommit)
def list_tables(self):
if not self.__connected:
raise Exception(f"Database connection with `{self.__path}` closed")
with self.session() as sess:
query_str = "SELECT name FROM sqlite_master WHERE type ='table';"
sess.execute(query_str)
return sess.fetchall()
class ConnectionProxy(object):
def __init__(self, connection: sqlite.Connection, autocommit=True):
self.__connection = connection
self.__autocommit = autocommit
def __enter__(self):
self.cursor = self.__connection.cursor()
return self
def __getattr__(self, attr):
return self.cursor.__getattribute__(attr)
def __exit__(self, exc_type, exc_value, exc_traceback):
if self.__autocommit:
self.__connection.commit()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment