Jocaru/wallamanta
1
0
Fork 0
Code Issues 15 Packages Projects Releases Wiki Activity

Changed all sqlite queries to params

Browse Source
This commit is contained in:
Joan
2023-03-11 18:25:41 +01:00
parent f4446ed6fb
commit 23fa1205b3
1 changed files with 40 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
wallamanta/walladb.py
View File

@@ -28,7 +28,8 @@ def setup_db():
def is_user_valid(telegram_user_id):
con = sqlite3.connect(constants.DB)
cur = con.cursor()
res = cur.execute(f"SELECT * FROM users WHERE telegram_user_id={telegram_user_id} AND active=True")
params = (telegram_user_id,)
res = cur.execute(f"SELECT * FROM users WHERE telegram_user_id=? AND active=True", params)
ret = res.fetchone() != None
con.close()
return ret
@@ -36,7 +37,8 @@ def is_user_valid(telegram_user_id):
def is_user_expired(telegram_user_id):
con = sqlite3.connect(constants.DB)
cur = con.cursor()
res = cur.execute(f"SELECT until FROM users WHERE telegram_user_id={telegram_user_id}")
params = (telegram_user_id,)
res = cur.execute(f"SELECT until FROM users WHERE telegram_user_id=?", params)
q_res = res.fetchone()
ret = True
if q_res != None:
@@ -48,7 +50,8 @@ def is_user_expired(telegram_user_id):
def is_user_premium(telegram_user_id):
con = sqlite3.connect(constants.DB)
cur = con.cursor()
res = cur.execute(f"SELECT * FROM users WHERE telegram_user_id={telegram_user_id} AND active=True AND type='premium'")
params = (telegram_user_id,)
res = cur.execute(f"SELECT * FROM users WHERE telegram_user_id=? AND active=True AND type='premium'", params)
ret = res.fetchone() != None
con.close()
return ret
@@ -56,7 +59,8 @@ def is_user_premium(telegram_user_id):
def is_user_testing(telegram_user_id):
con = sqlite3.connect(constants.DB)
cur = con.cursor()
res = cur.execute(f"SELECT * FROM users WHERE telegram_user_id={telegram_user_id} AND active=True AND type='testing'")
params = (telegram_user_id,)
res = cur.execute(f"SELECT * FROM users WHERE telegram_user_id=? AND active=True AND type='testing'", params)
ret = res.fetchone() != None
con.close()
return ret
@@ -65,13 +69,15 @@ def add_premium_user(telegram_user_id, telegram_name, until):
found = False
con = sqlite3.connect(constants.DB)
cur = con.cursor()
res = cur.execute(f"SELECT * FROM users WHERE telegram_user_id={telegram_user_id}")
params = (telegram_name,)
res = cur.execute(f"SELECT * FROM users WHERE telegram_user_id=?", params)
if res.fetchone() is None:
params = (telegram_user_id, True, 'premium', until, telegram_name.first_name)
cur.execute("INSERT INTO users VALUES (?, ?, ?, ?, ?)", params)
con.commit()
else:
cur.execute(f"UPDATE users SET active = True, type = 'premium', until = '{until}' WHERE telegram_user_id={telegram_user_id}")
params = (until, telegram_user_id)
cur.execute(f"UPDATE users SET active = True, type = 'premium', until = ? WHERE telegram_user_id=?", params)
con.commit()
found = True
con.close()
@@ -82,7 +88,8 @@ def add_test_user(telegram_user_id, telegram_name, until):
found = False
con = sqlite3.connect(constants.DB)
cur = con.cursor()
res = cur.execute(f"SELECT * FROM users WHERE telegram_user_id={telegram_user_id}")
params = (telegram_name,)
res = cur.execute(f"SELECT * FROM users WHERE telegram_user_id=?", params)
if res.fetchone() is None:
params = (telegram_user_id, True, 'testing', until, telegram_name.first_name)
cur.execute("INSERT INTO users VALUES (?, ?, ?, ?, ?)", params)
@@ -96,9 +103,10 @@ def add_test_user(telegram_user_id, telegram_name, until):
def remove_valid_user(telegram_user_id):
con = sqlite3.connect(constants.DB)
cur = con.cursor()
res = cur.execute(f"SELECT * FROM users WHERE telegram_user_id={telegram_user_id}")
params = (telegram_user_id,)
res = cur.execute(f"SELECT * FROM users WHERE telegram_user_id=?", params)
if res.fetchone() != None:
cur.execute(f"UPDATE users SET active = False WHERE telegram_user_id={telegram_user_id}")
cur.execute(f"UPDATE users SET active = False WHERE telegram_user_id=?", params)
con.commit()
con.close()
logging.info(f"De-activated user {telegram_user_id}")
@@ -114,7 +122,8 @@ def get_user_list():
def get_user_type(telegram_user_id):
con = sqlite3.connect(constants.DB)
cur = con.cursor()
res = cur.execute(f"SELECT type FROM users WHERE telegram_user_id={telegram_user_id}")
params = (telegram_user_id,)
res = cur.execute(f"SELECT type FROM users WHERE telegram_user_id=?", params)
ret = res.fetchone()
con.close()
return ret[0]
@@ -122,7 +131,8 @@ def get_user_type(telegram_user_id):
def get_user_until(telegram_user_id):
con = sqlite3.connect(constants.DB)
cur = con.cursor()
res = cur.execute(f"SELECT until FROM users WHERE telegram_user_id={telegram_user_id}")
params = (telegram_user_id,)
res = cur.execute(f"SELECT until FROM users WHERE telegram_user_id=?", params)
ret = res.fetchone()
con.close()
return ret[0]
@@ -133,8 +143,9 @@ def get_product(product):
con = sqlite3.connect(constants.DB)
con.row_factory = dict_factory
cur = con.cursor()
res = cur.execute(f"SELECT * FROM products WHERE telegram_user_id={telegram_user_id} \
AND product_name='{product_name}'")
params = (telegram_user_id, product_name)
res = cur.execute(f"SELECT * FROM products WHERE telegram_user_id=? \
AND product_name=?", params)
ret = res.fetchone()
con.close()
return ret
@@ -143,7 +154,8 @@ def get_products_from_user(telegram_user_id):
con = sqlite3.connect(constants.DB)
con.row_factory = dict_factory
cur = con.cursor()
res = cur.execute(f"SELECT * FROM products WHERE telegram_user_id={telegram_user_id}")
params = (telegram_user_id,)
res = cur.execute(f"SELECT * FROM products WHERE telegram_user_id=?", params)
ret = res.fetchall()
con.close()
return ret
@@ -171,27 +183,30 @@ def add_product(product):
if category == '0':
category = ''
telegram_user_id = product.get('telegram_user_id')
logging.info(f"Trying to add: {product_name}, {telegram_user_id}")
con = sqlite3.connect(constants.DB)
cur = con.cursor()
params = (telegram_user_id, product_name)
res = cur.execute(f"SELECT * FROM products WHERE telegram_user_id=? \
AND product_name=?", params)
if res.fetchone() is None:
params = (product_name, \
distance, latitude, longitude, condition, min_price, \
max_price, category, title_exclude, title_description_exclude, telegram_user_id)
logging.info(f"Trying to add: {product_name}, {title_exclude}, {title_description_exclude}, {telegram_user_id}")
con = sqlite3.connect(constants.DB)
cur = con.cursor()
res = cur.execute(f"SELECT * FROM products WHERE telegram_user_id={telegram_user_id} \
AND product_name='{product_name}'")
if res.fetchone() is None:
cur.execute("INSERT INTO products VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", params)
con.commit()
con.close()
def remove_product(product):
telegram_user_id = product.get('telegram_user_id')
product_name = product.get('product_name').lower()
removed = False
if get_product(product):
con = sqlite3.connect(constants.DB)
cur = con.cursor()
res = cur.execute(f"DELETE FROM products WHERE telegram_user_id={product.get('telegram_user_id')} \
AND product_name='{product.get('product_name').lower()}'")
params = (telegram_user_id, product_name)
res = cur.execute(f"DELETE FROM products WHERE telegram_user_id=? \
AND product_name=?", params)
con.commit()
con.close()
logging.info(f"Removed product {product['product_name']}")
@@ -201,7 +216,8 @@ def remove_product(product):
def count_user_products(telegram_user_id):
con = sqlite3.connect(constants.DB)
cur = con.cursor()
res = cur.execute(f"SELECT Count() FROM products WHERE telegram_user_id={telegram_user_id}")
params = (telegram_user_id,)
res = cur.execute(f"SELECT Count() FROM products WHERE telegram_user_id=?", params)
ret = res.fetchone()[0]
con.close()
return ret