Created
May 22, 2020 08:16
-
-
Save NuarkNoir/ec5689ed6063253f48c09129842d98e1 to your computer and use it in GitHub Desktop.
Simple sqlite3 wrapper for python 3+
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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