voldemort/CDRM-Project
1
0
Fork 0
forked from tpd94/CDRM-Project
Code Pull Requests Activity

Refactor database connection handling in user_db.py to utilize os.path.join for improved path management and consistency across user-related functions.

Browse Source
This commit is contained in:
voldemort 2025-07-23 16:36:10 +07:00
parent 6890c6b464
commit bbeeffcd9d
1 changed files with 29 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
custom_functions/database/user_db.py
View File

@ -7,9 +7,11 @@ import bcrypt
def create_user_database(): def create_user_database():
"""Create the user database.""" """Create the user database."""
os.makedirs(f"{os.getcwd()}/databases/sql", exist_ok=True) os.makedirs(os.path.join(os.getcwd(), "databases", "sql"), exist_ok=True)
with sqlite3.connect(f"{os.getcwd()}/databases/sql/users.db") as conn: with sqlite3.connect(
os.path.join(os.getcwd(), "databases", "sql", "users.db")
) as conn:
cursor = conn.cursor() cursor = conn.cursor()
cursor.execute( cursor.execute(
""" """
@ -27,7 +29,9 @@ def add_user(username, password, api_key):
"""Add a user to the database.""" """Add a user to the database."""
hashed_pw = bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()) hashed_pw = bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt())
with sqlite3.connect(f"{os.getcwd()}/databases/sql/users.db") as conn: with sqlite3.connect(
os.path.join(os.getcwd(), "databases", "sql", "users.db")
) as conn:
cursor = conn.cursor() cursor = conn.cursor()
try: try:
cursor.execute( cursor.execute(
@ -42,7 +46,9 @@ def add_user(username, password, api_key):
def verify_user(username, password): def verify_user(username, password):
"""Verify a user.""" """Verify a user."""
with sqlite3.connect(f"{os.getcwd()}/databases/sql/users.db") as conn: with sqlite3.connect(
os.path.join(os.getcwd(), "databases", "sql", "users.db")
) as conn:
cursor = conn.cursor() cursor = conn.cursor()
cursor.execute( cursor.execute(
"SELECT Password FROM user_info WHERE Username = ?", (username.lower(),) "SELECT Password FROM user_info WHERE Username = ?", (username.lower(),)
@ -55,13 +61,14 @@ def verify_user(username, password):
if isinstance(stored_hash, str): if isinstance(stored_hash, str):
stored_hash = stored_hash.encode("utf-8") stored_hash = stored_hash.encode("utf-8")
return bcrypt.checkpw(password.encode("utf-8"), stored_hash) return bcrypt.checkpw(password.encode("utf-8"), stored_hash)
else: return False
return False
def fetch_api_key(username): def fetch_api_key(username):
"""Fetch the API key for a user.""" """Fetch the API key for a user."""
with sqlite3.connect(f"{os.getcwd()}/databases/sql/users.db") as conn: with sqlite3.connect(
os.path.join(os.getcwd(), "databases", "sql", "users.db")
) as conn:
cursor = conn.cursor() cursor = conn.cursor()
cursor.execute( cursor.execute(
"SELECT API_Key FROM user_info WHERE Username = ?", (username.lower(),) "SELECT API_Key FROM user_info WHERE Username = ?", (username.lower(),)
@ -70,8 +77,7 @@ def fetch_api_key(username):
if result: if result:
return result[0] return result[0]
else: return None
return None
def change_password(username, new_password): def change_password(username, new_password):
@ -80,7 +86,9 @@ def change_password(username, new_password):
new_hashed_pw = bcrypt.hashpw(new_password.encode("utf-8"), bcrypt.gensalt()) new_hashed_pw = bcrypt.hashpw(new_password.encode("utf-8"), bcrypt.gensalt())
# Update the password in the database # Update the password in the database
with sqlite3.connect(f"{os.getcwd()}/databases/sql/users.db") as conn: with sqlite3.connect(
os.path.join(os.getcwd(), "databases", "sql", "users.db")
) as conn:
cursor = conn.cursor() cursor = conn.cursor()
cursor.execute( cursor.execute(
"UPDATE user_info SET Password = ? WHERE Username = ?", "UPDATE user_info SET Password = ? WHERE Username = ?",
@ -93,7 +101,9 @@ def change_password(username, new_password):
def change_api_key(username, new_api_key): def change_api_key(username, new_api_key):
"""Change the API key for a user.""" """Change the API key for a user."""
# Update the API key in the database # Update the API key in the database
with sqlite3.connect(f"{os.getcwd()}/databases/sql/users.db") as conn: with sqlite3.connect(
os.path.join(os.getcwd(), "databases", "sql", "users.db")
) as conn:
cursor = conn.cursor() cursor = conn.cursor()
cursor.execute( cursor.execute(
"UPDATE user_info SET API_Key = ? WHERE Username = ?", "UPDATE user_info SET API_Key = ? WHERE Username = ?",
@ -105,7 +115,9 @@ def change_api_key(username, new_api_key):
def fetch_styled_username(username): def fetch_styled_username(username):
"""Fetch the styled username for a user.""" """Fetch the styled username for a user."""
with sqlite3.connect(f"{os.getcwd()}/databases/sql/users.db") as conn: with sqlite3.connect(
os.path.join(os.getcwd(), "databases", "sql", "users.db")
) as conn:
cursor = conn.cursor() cursor = conn.cursor()
cursor.execute( cursor.execute(
"SELECT Styled_Username FROM user_info WHERE Username = ?", "SELECT Styled_Username FROM user_info WHERE Username = ?",
@ -115,18 +127,18 @@ def fetch_styled_username(username):
if result: if result:
return result[0] return result[0]
else: return None
return None
def fetch_username_by_api_key(api_key): def fetch_username_by_api_key(api_key):
"""Fetch the username for a user by API key.""" """Fetch the username for a user by API key."""
with sqlite3.connect(f"{os.getcwd()}/databases/sql/users.db") as conn: with sqlite3.connect(
os.path.join(os.getcwd(), "databases", "sql", "users.db")
) as conn:
cursor = conn.cursor() cursor = conn.cursor()
cursor.execute("SELECT Username FROM user_info WHERE API_Key = ?", (api_key,)) cursor.execute("SELECT Username FROM user_info WHERE API_Key = ?", (api_key,))
result = cursor.fetchone() result = cursor.fetchone()
if result: if result:
return result[0] # Return the username return result[0] # Return the username
else: return None # If no user is found for the API key
return None # If no user is found for the API key