import sqlite3
class Database:
def __init__(self, path_to_db="Sqlite/database.db"):
self.path_to_db = path_to_db
self.connection
@property
def connection(self):
return sqlite3.connect(self.path_to_db)
def execute(self, sql: str, parameters: tuple = None, fetchone = False,
fetchall = False, commit = False):
if not parameters:
parameters = tuple()
connection = self.connection
connection.set_trace_callback(logger)
cursor = connection.cursor()
data = None
cursor.execute(sql, parameters)
if commit:
connection.commit()
if fetchone:
data = cursor.fetchone()
if fetchall:
data = cursor.fetchall()
connection.close()
return data
def create_table_users(self):
sql = """
CREATE TABLE Users (
id int NOT NULL,
name varchar(255) NOT NULL,
treshka int,
kasan int,
presents int,
vstrecha int,
podklich int
PRIMARY KEY (id)
);
"""
self.execute(sql, commit=True)
def add_user(self, id:int, name:str, treshka:int = None, kasan:int = None, presents:int = None, vstrecha:int = None, podklich:int = None):
sql = "INSERT INTO Users(id, name, treshka, kasan, presents, vstrecha, podklich) VALUES(?, ?, ?, ?, ?, ?, ?)"
parameters = (id, name, treshka, kasan, presents, vstrecha, podklich)
self.execute(sql, parameters=parameters, commit=True)
def select_all_users(self):
sql = "SELECT * FROM Users"
return self.execute(sql, fetchall=True)
@staticmethod
def format_args(sql, parameters=dict):
sql += " AND ".join([
f"{item} = ?" for item in parameters
])
return sql, tuple(parameters.values())
def select_user(self, **kwargs):
sql = "SELECT * FROM Users WHERE"
sql, parameters = self.format_args(sql, kwargs)
return self.execute(sql, parameters, fetchone=True)
def count_users(self):
return self.execute("SELECT COUNT(*) from Users", fetchone=True)
#def update_email(self, email, id):
# sql = "UPDATE Users SET email =? WHERE id=?"
# return self.execute(sql, parameters=(email, id), commit=True)
def update_treshka(self, treshka, id):
sql = "UPDATE Users SET treshka =? WHERE id=?"
return self.execute(sql, parameters=(treshka, id), commit=True)
def delete_user(self):
self.execute("DELETE FROM Users WHERE True")
def logger(statement):
print(f"""
_________________________________________
Executing:
{statement}
_________________________________________
""")Traceback (most recent call last):
File "d:\Programming\Python\C8\bot.py", line 31, in <module>
test()
File "d:\Programming\Python\C8\bot.py", line 20, in test
db.create_table_user()
File "d:\Programming\Python\C8\sqlite.py", line 40, in create_table_user
self.execute(sql, commit=True)
File "d:\Programming\Python\C8\sqlite.py", line 21, in execute
cursor.execute(sql, parameters)
sqlite3.OperationalError: table Users already existimport sqlite3
class Database:
def __init__(self, path_to_db="database.db"):
self.path_to_db = path_to_db
self.connection
@property
def connection(self):
return sqlite3.connect(self.path_to_db)
def execute(self, sql: str, parameters: tuple = None, fetchone = False,
fetchall = False, commit = False):
if not parameters:
parameters = tuple()
connection = self.connection
connection.set_trace_callback(logger)
cursor = connection.cursor()
data = None
cursor.execute(sql, parameters)
if commit:
connection.commit()
if fetchone:
data = cursor.fetchone()
if fetchall:
data = cursor.fetchall()
connection.close()
return data
def create_table_user(self):
sql = """
CREATE TABLE Users (
id int NOT NULL,
Name varchar(255) NOT NULL,
email varcahr(255),
PRIMARY KEY (id)
);
"""
self.execute(sql, commit=True)
def add_user(self, id:int, name:str, email:str=None):
sql = "INSERT INTO Users(id, name, email) VALUES(?, ?, ?)"
parameters = (id, name, email)
self.execute(sql, parameters=parameters, commit=True)
@staticmethod
def format_args(sql, parameters=dict):
sql += " AND ".join([
f"{item} = ?" for item in parameters
])
return sql, tuple(parameters.values())
def select_user(self, **kwargs):
sql = "SELECT * FROM Users WHERE"
sql, parameters = self.format_args(sql, **kwargs)
self.execute(sql, parameters, fetchone=True)
def count_user(self):
return self.execute("SELECT COUNT(*) from Users", fetchone=True)
def update_email(self, email, id):
sql = "UPDATE Users SET email =? WHERE id=?"
return self.execute(sql, parameters=(email, id), commit=True)
def delete_user(self):
self.execute("DELETE FROM Users WHERE True")
def logger(statement):
print(f"""
_________________________________________
Executing:
{statement}
_________________________________________
""")