Skip to content

Instantly share code, notes, and snippets.

@DavidKatz-il
Created March 24, 2021 14:26
Show Gist options
  • Save DavidKatz-il/03e6d9ef9938fa1e1a4a0000bc2adf5f to your computer and use it in GitHub Desktop.
Save DavidKatz-il/03e6d9ef9938fa1e1a4a0000bc2adf5f to your computer and use it in GitHub Desktop.
DataBase class is a wrapper around `sqlite` database.
import sqlite3
from typing import List, Union
class DataBase:
"""
DataBase class is a wrapper around sqlite database.
"""
def __init__(self, db_path: str):
"""
Initialize DataBase.
If db_path is not already exist it will be created.
:param db_path: The path of a database file, or ':memory:' for open a database in the RAM.
"""
self.conn = None
self.cur = None
self.db_path = db_path
self.open()
def open(self):
"""
Connecting to the database.
"""
self.conn = sqlite3.connect(self.db_path)
self.cur = self.conn.cursor()
def close(self):
"""
Closing the connection to the database.
"""
self.conn.commit()
self.conn.close()
def __exit__(self):
self.close()
def create_table(self, table: str, columns: Union[str, List[str]]):
"""
Creates a table into the database.
:param table: Name of the new table.
:param columns: Columns to create in the table (e.g. 'id integer PRIMARY KEY').
"""
columns = self.tolist_of_strings(columns)
sql_create_table = f"""
CREATE TABLE IF NOT EXISTS {table} (
{','.join(columns)}
); """
self.cur.execute(sql_create_table)
def insert(self, table: str, columns: str, values: Union[str, List[str]]):
"""
Insert values into a table.
:param table: Name of the table.
:param columns: Columns where to store the values, separated with a comma (e.g. 'col1,col2').
:param values: Values to store into the table, (e.g. 'val1,val2' or ['val1', 'val2']).
"""
data = self.tolist_of_strings(values)
sql_insert = f"""
INSERT INTO {table} ({columns}) VALUES (
{','.join('?'*len(data))}
); """
self.cur.execute(sql_insert, data)
self.conn.commit()
def exist(self, table: str, column: str, value: str) -> bool:
"""
Checking if a value exist.
:param table: Name of the table.
:param column: Name of the column.
:param value: Value to search.
:return: True if exist otherwise returns False.
"""
if (
self.cur.execute(
f"SELECT 1 FROM {table} WHERE {column} = {value};"
).fetchone()
is None
):
return False
return True
def get_tables(self) -> list:
"""
Returns a list of the database tables.
:return: list of table names.
"""
return list(
self.cur.execute(
"SELECT name FROM sqlite_master WHERE type='table';"
).fetchall()
)
def get_columns(self, table: str) -> list:
"""
Returns columns of a given table name.
:param table: Name of a table.
:return: list of columns names.
"""
return list(self.cur.execute(f"SELECT * FROM {table} LIMIT 1;").fetchall())
def query(self, query: str):
"""
Execute a query in the database.
:param query: query string.
"""
self.cur.execute(query)
self.conn.commit()
@staticmethod
def tolist_of_strings(data: Union[str, int, List[str], List[int]]):
"""
Convert all values inside the list to string,
if data is not a list and one value is given it will be insert into a list.
:param data: int or string or list of objects or integers.
:return: list of strings
"""
if not isinstance(data, list):
data = [data]
return list(map(str, data))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment