User Database

teamprojekt_competition_server/server/user_database.py

@@ -0,0 +1,175 @@
+"""User-Database-Module"""
+
+import sqlite3
+import logging as log
+
+USER_DB_NAME = "user"
+
+
+# Connect to the database:
+connection = sqlite3.connect("teamprojekt_competition_server/server/COMP_database.db")
+cursor = connection.cursor()
+cursor.execute(
+    f"""
+    CREATE TABLE IF NOT EXISTS {USER_DB_NAME} (
+        user_id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
+        name TEXT NOT NULL,
+        token INTEGER NOT NULL,
+        mu FLOAT NOT NULL,
+        sigma FLOAT NOT NULL
+    )"""
+)
+
+
+def _is_token_taken(new_token: int) -> bool:
+    """tests whether a token has already been assigned to another user
+
+    Args:
+        new_token (int): the token that should be tested
+
+    Returns:
+        bool:
+        returns true if token has already been assigned to another user
+        returns false if the token does not yet exist
+    """
+    cursor.execute(
+        f"""
+        SELECT COUNT(*) FROM {USER_DB_NAME} WHERE token=?
+    """,
+        (new_token,),
+    )
+    count = cursor.fetchone()[0]
+    return count > 0
+
+
+def add_user(
+    user_name: str, user_token: int, user_mu=25.000, user_sigma=8.333
+) -> int | None:
+    """adds a new user to the database
+
+    Args:
+        user_name (str): name of the user
+        user_token (int): token for the user (must be unique for every user)
+        user_mu (float, optional): needed for Matchmaking. Defaults to 25.000.
+        user_sigma (float, optional): needed for Matchmaking. Defaults to 8.333.
+
+    Returns:
+        int | None: returns the user_id
+    """
+    assert not _is_token_taken(user_token)
+    cursor.execute(
+        f"""
+        INSERT INTO {USER_DB_NAME}(name, token, mu, sigma) VALUES (?,?,?,?)""",
+        (user_name, user_token, user_mu, user_sigma),
+    )
+    connection.commit()
+    id = cursor.lastrowid
+    log.debug(
+        (
+            "inserted user("
+            f"user_id={id}, name={user_name}, token={user_token}, "
+            f"mu={user_mu}, sigma={user_sigma})"
+        )
+    )
+    return id
+
+
+def get_user(id: int) -> tuple:
+    """returns the database entry for the user with this id
+
+    Args:
+        id (int): the id of the user
+
+    Returns:
+        tuple(int, string, int, float, float): database entry
+        (user_id, name, token, mu, sigma)
+    """
+    cursor.execute(
+        f"""
+        SELECT * FROM {USER_DB_NAME} WHERE user_id = ?
+    """,
+        (id,),
+    )
+    user = cursor.fetchone()
+    return user
+
+
+def verify_user(user_token: int) -> int:
+    """returns the corresponding user_id for a token
+
+    Args:
+        user_token (int): token for which the user should be found
+
+    Returns:
+        int: user_id
+    """
+    res = cursor.execute(
+        f"""
+        SELECT user_id FROM {USER_DB_NAME} WHERE token = ?
+    """,
+        (user_token,),
+    )
+    (id,) = res.fetchone()
+    return id
+
+
+def get_all_users() -> list[tuple]:
+    """returns the database entries for all users
+
+    Returns:
+        list[tuple(int, string, int, float, float)]: database entries of all users
+        list[(user_id, name, token, mu, sigma)]
+    """
+    cursor.execute(
+        f"""
+        SELECT * FROM {USER_DB_NAME}
+    """
+    )
+    users = cursor.fetchall()
+    return users
+
+
+def update_matchmaking_parameters(id: int, new_mu: float, new_sigma: float):
+    """updates the mu and sigma entries required for Matchmaking for one user
+
+    Args:
+        id (int): user id
+        new_mu (float): new mu value
+        new_sigma (float): new sigma value
+    """
+    cursor.execute(
+        f"""
+        UPDATE {USER_DB_NAME} SET mu=?, sigma=? WHERE user_id=?
+    """,
+        (new_mu, new_sigma, id),
+    )
+    connection.commit()
+
+
+def get_matchmaking_parameters(id: int) -> tuple[float, float]:
+    """gets the mu and sigma entries required for Matchmaking of the user
+
+    Args:
+        id (int): user_id
+
+    Returns:
+        (float, float): (mu, sigma)
+    """
+    res = cursor.execute(
+        f"""
+        SELECT mu, sigma FROM {USER_DB_NAME} WHERE user_id = ?
+    """,
+        (id,),
+    )
+    parameters = res.fetchone()
+    return parameters
+
+
+def delete_user(id: int) -> None:
+    """deletes a user from the database
+
+    Args:
+        id (int): user id
+    """
+    cursor.execute(f""" DELETE FROM {USER_DB_NAME} WHERE user_id = ? """, (id,))
+    connection.commit()

tests/user_database_test.py

@@ -0,0 +1,41 @@
+import teamprojekt_competition_server.server.user_database as user_db
+import logging
+
+# run with python -m tests.user_database_test
+
+logging.basicConfig(level=logging.DEBUG)
+
+
+def user_database_tests():
+    userID1 = user_db.add_user(user_name="player_1", user_token=123)
+    userID2 = user_db.add_user(user_name="player_2", user_token=456)
+    userID3 = user_db.add_user(user_name="player_3", user_token=789)
+    userID4 = user_db.add_user(user_name="player_4", user_token=444)
+    print(userID1, userID2, userID3, userID4)
+
+    assert (
+        userID1 % 3 == 1 and userID2 % 3 == 2 and userID3 % 3 == 0 and userID4 % 3 == 1
+    )
+    all_users = user_db.get_all_users()
+    print(all_users)
+    assert len(all_users) == 4
+    assert user_db.verify_user(123) == userID1
+
+    user = user_db.get_user(id=userID1)
+    print(
+        f"User ID: {user[0]}, Name: {user[1]}, \
+          Token: {user[2]}, Mu: {user[3]}, Sigma: {user[4]}"
+    )
+
+    user_db.update_matchmaking_parameters(id=userID3, new_mu=24.000, new_sigma=9.333)
+    (mu, sigma) = user_db.get_matchmaking_parameters(id=userID3)
+    assert mu == 24.000 and sigma == 9.333
+
+    user_db.delete_user(userID4)
+    all_users = user_db.get_all_users()
+    assert len(all_users) == 3
+
+
+if __name__ == "__main__":
+    # user_database_tests()  # only enable for manual testing
+    pass