Created
March 24, 2021 14:26
-
-
Save DavidKatz-il/03e6d9ef9938fa1e1a4a0000bc2adf5f to your computer and use it in GitHub Desktop.
DataBase class is a wrapper around `sqlite` database.
This file contains 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 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