Add a proper sqlite3 database to RootPythia

.env.example

@@ -31,6 +31,12 @@ API_URL=https://api.www.root-me.org/
 # [OPTIONAL] the maximum number of time a single request is retried (if relevant)
 MAX_API_ATTEMPT=5
 
+
 ### Bot Configuration
 # [OPTIONAL] in seconds, the delay between new solve checking (default is 10)
 REFRESH_DELAY=
+
+
+### Database Configuration
+# the path to the Sqlite database folder
+DB_FOLDER=data

.gitignore

@@ -8,3 +8,7 @@
 
 # Logs
 logs/**
+
+# Database folder
+data/**
+

Dockerfile.dev

@@ -5,7 +5,9 @@ COPY requirements.txt /tmp/requirements.txt
 COPY requirements-dev.txt /tmp/requirements-dev.txt
 RUN pip install -r /tmp/requirements-dev.txt
 
-RUN apt update && apt install -y vim
+RUN apt update && apt install -y vim sqlite3
 
 WORKDIR /opt/root-pythia
+RUN mkdir logs
+
 CMD ["bash"]

run.sh

@@ -10,6 +10,7 @@ run__prod () {
 	docker run --rm --interactive --tty \
 	--detach \
 	--volume $(realpath -P ${LOG_FOLDER}):/opt/${NAME}/logs \
+	--volume $(realpath -P  ${DB_FOLDER}):/opt/${NAME}/${DB_FOLDER} \
 	--env-file .env.prod \
 	--name ${NAME} \
 	${NAME}:latest

src/bot/root_pythia_bot.py

@@ -11,7 +11,7 @@
 from api.rate_limiter import RateLimiter
 from bot.custom_help_command import RootPythiaHelpCommand
 from bot.root_pythia_cogs import RootPythiaCommands
-from bot.dummy_db_manager import DummyDBManager
+from database import DatabaseManager
 
 
 CHANNEL_ID = getenv("CHANNEL_ID")
@@ -74,12 +74,15 @@ def is_my_channel(ctx):
 @BOT.event
 async def on_ready():
     # is this call secure??
-    logging.debug("channel id: %s", CHANNEL_ID)
+    BOT.logger.debug("channel id: %s", CHANNEL_ID)
 
-    # Create Rate Limiter, API Manager, and DB Manager objects
+    # Create Rate Limiter, API Manager, and Database Manager objects
     rate_limiter = RateLimiter()
+    BOT.logger.debug("Successfully created RateLimiter")
     api_manager = RootMeAPIManager(rate_limiter)
-    db_manager = DummyDBManager(api_manager)
+    BOT.logger.debug("Successfully created RootMeAPIManager")
+    db_manager = DatabaseManager(api_manager)
+    BOT.logger.debug("Successfully created DatabaseManager")
 
     # Fetch main channel and send initialization message
     BOT.channel = await BOT.fetch_channel(CHANNEL_ID)
@@ -92,8 +95,9 @@ async def on_ready():
 @BOT.event
 async def on_error(event, *args, **kwargs):
     if event == "on_ready":
-        BOT.logger.error(
-            "Event '%s' failed (probably from invalid channel ID), close connection and exit...",
+        BOT.logger.exception("Unhandled exception in 'on_ready' event:")
+        BOT.logger.critical(
+            "Event '%s' failed, please check debug logs, close connection and exit...",
             event,
         )
         await BOT.close()

src/classes/user.py

@@ -5,17 +5,23 @@ class InvalidUserData(Exception):
 class User:
     """Class for the User object"""
 
-    def __init__(self, data: dict):
+    def __init__(self, data: [dict, tuple]):
         """
         idx: int,
         username: str,
         score :int,
         rank: int,
         solves: int,
         """
-        parsed_data = User.parse_rootme_user_data(data)
+        if isinstance(data, dict):
+            parsed_data = User.parse_rootme_user_data(data)
+        elif isinstance(data, tuple):
+            parsed_data = dict(zip(User.keys(), list(data)))
 
-        self.idx = parsed_data["idx"]
+        self.as_dict = parsed_data
+        self.as_tuple = tuple(parsed_data.values())
+
+        self.idx = parsed_data["id"]
         self.username = parsed_data["username"]
         self.score = parsed_data["score"]
         self.rank = parsed_data["rank"]
@@ -25,7 +31,7 @@ def __init__(self, data: dict):
 
     @staticmethod
     def keys():
-        return ["idx", "username", "score", "rank", "nb_solves"]
+        return ["id", "username", "score", "rank", "nb_solves"]
 
     # TODO: move these static methods to rootme_api, it make more sense
     # or create a Parser object? we could then have a RootMeParser and a HTBParser
@@ -57,6 +63,12 @@ def parse_rootme_user_solves_and_yield(data):
             solve_id = int(solve["id_challenge"])
             yield solve_id
 
+    def to_dict(self):
+        return self.as_dict
+
+    def to_tuple(self):
+        return self.as_tuple
+
     def __repr__(self):
         return (
             f"User(id={self.idx}, username={self.username}, "
@@ -66,6 +78,12 @@ def __repr__(self):
     def __str__(self):
         return f"{self.username} #{self.idx}"
 
+    def __eq__(self, other) -> bool:
+        if not isinstance(other, User):
+            # don't attempt to compare against unrelated types
+            return NotImplemented
+        return self.as_tuple == other.as_tuple
+
     def update_new_solves(self, raw_user_data):
         parsed_data = User.parse_rootme_user_data(raw_user_data)
         parsed_nb_solves = parsed_data["nb_solves"]

src/database/__init__.py

@@ -0,0 +1 @@
+from database.db_manager import DatabaseManager

src/bot/dummy_db_manager.py → src/database/db_manager.py

@@ -1,9 +1,21 @@
 import logging
-
+import sqlite3
+from os import getenv, path
+
+from database.db_structure import (
+    sql_create_user_table,
+    sql_add_user,
+    sql_get_user,
+    sql_get_users,
+    sql_has_user,
+)
 from classes import User
 from classes import Challenge
 
 
+DB_FILE_NAME = "RootPythia.db"
+
+
 class InvalidUser(Exception):
     def __init__(self, idx=None, message=None):
         self.idx = idx
@@ -18,41 +30,74 @@ def __init__(self, idx=None, message=None):
         super().__init__(self.message)
 
 
-class DummyDBManager:
+class DatabaseManager:
     def __init__(self, api_manager):
-        self.users = []
+        self.logger = logging.getLogger(__name__)
+
+        self.DB_FOLDER = getenv("DB_FOLDER")
+        if self.DB_FOLDER is None or not path.isdir(self.DB_FOLDER):
+            self.logger.critical("DB_FOLDER: '%s', is not a directory", self.DB_FOLDER)
+            raise OSError(f"DB_FOLDER: '{self.DB_FOLDER}', is not a directory")
+
+        # Init Connection object allowing interaction with the database
+        db_file_path = path.join(self.DB_FOLDER, DB_FILE_NAME)
+        self.db = sqlite3.connect(db_file_path)
+        self.logger.info("Succesfully connected to database '%s'", db_file_path)
+        self._init_db()
+
         self.api_manager = api_manager
 
-        self.logger = logging.getLogger(__name__)
+    def _init_db(self):
+        """Private function that initializes the database tables (see db_strucure.py)"""
+        cur = self.db.cursor()
+        cur.execute(sql_create_user_table)
+        cur.close()
 
     async def add_user(self, idx):
         """Call the API Manager to get a user by his id then create a User object and store it"""
+        cur = self.db.cursor()
 
         # Check wether the user is already added
         if self.has_user(idx):
             return None
 
+        # Retreive information from RootMe API
         raw_user_data = await self.api_manager.get_user_by_id(idx)
-
         user = User(raw_user_data)
-        self.users.append(user)
-        self.logger.debug("add user '%s'", repr(user))
+
+        cur.execute(sql_add_user, user.to_tuple())
+        self.db.commit()
+        self.logger.debug("Add user '%s'", repr(user))
+        cur.close()
         return user
 
     def has_user(self, idx):
-        return self.get_user(idx) is not None
+        cur = self.db.cursor()
+        res = cur.execute(sql_has_user, (idx, )).fetchone()
+        cur.close()
+        return res is not None
 
     def get_user(self, idx):
         """Retrieve the user object whose id matches 'id', None if not found"""
-        return next(filter(lambda user: user.idx == idx, self.users), None)
+        cur = self.db.cursor()
+        res = cur.execute(sql_get_user, (idx, )).fetchone()
+        if res is None:
+            return None
+        user = User(res)
+        cur.close()
+        return user
 
     def get_users(self):
-        return self.users
+        cur = self.db.cursor()
+        res = cur.execute(sql_get_users).fetchall()
+        users = [User(elt) for elt in res]
+        cur.close()
+        return users
 
     async def fetch_user_new_solves(self, idx):
         user = self.get_user(idx)
         if user is None:
-            raise InvalidUser(idx, "DummyDBManager.fetch_user_new_solves: User %s not in database")
+            raise InvalidUser(idx, "DatabaseManager.fetch_user_new_solves: User %s not in database")
 
         raw_user_data = await self.api_manager.get_user_by_id(idx)
         user.update_new_solves(raw_user_data)

src/database/db_structure.py

@@ -0,0 +1,19 @@
+sql_create_user_table = """ CREATE TABLE IF NOT EXISTS users (
+                            id integer PRIMARY KEY,
+                            username text NOT NULL,
+                            score int,
+                            rank int,
+                            nb_solves int
+                        );"""
+
+
+sql_add_user = """INSERT INTO users(id,username,score,rank,nb_solves) VALUES(?,?,?,?,?);"""
+
+
+sql_get_user = """SELECT * FROM users WHERE id=?;"""
+
+
+sql_get_users = """SELECT * FROM users;"""
+
+
+sql_has_user = """SELECT * FROM users WHERE id=(?)"""

tests/conftest.py

@@ -13,7 +13,7 @@
 
 from bot.root_pythia_cogs import RootPythiaCommands
 from bot.root_pythia_cogs import NAME as COG_NAME
-from bot.dummy_db_manager import DummyDBManager
+from database import DatabaseManager
 
 # these plugins will be automatically imported by pytest
 pytest_plugins = ["pytest_mock", "pytest_asyncio"]
@@ -57,17 +57,19 @@ def mock_rootme_api_manager(mocker):
 
 
 @pytest.fixture
-def mock_dummy_db_manager(mock_rootme_api_manager):
+def mock_database_manager(mock_rootme_api_manager, monkeypatch, tmp_path):
     rootme_api_manager = mock_rootme_api_manager
+    monkeypatch.setenv("DB_FOLDER", str(tmp_path))
 
-    db = DummyDBManager(rootme_api_manager)
+    db = DatabaseManager(rootme_api_manager)
     yield db
+    db.db.close()
 
 
 # this pytest_asyncio decorator allows to automatically await async fixture before passing them
 # to tests
 @pytest_asyncio.fixture
-async def config_bot(mock_dummy_db_manager, null_logger):
+async def config_bot(mock_database_manager, null_logger):
     intents = discord.Intents.default()
     intents.members = True
     intents.message_content = True
@@ -76,7 +78,7 @@ async def config_bot(mock_dummy_db_manager, null_logger):
     _bot.logger = null_logger
 
     await _bot._async_setup_hook()
-    await _bot.add_cog(RootPythiaCommands(_bot, mock_dummy_db_manager))
+    await _bot.add_cog(RootPythiaCommands(_bot, mock_database_manager))
 
     dpytest.configure(_bot)
 

tests/test_cog.py

@@ -88,7 +88,7 @@ async def test_command_exception_handling(config_bot, mocker):
     # patching this method in particular is arbitrary: the only goal is to raise an exception
     # during a command execution
     # I chose the "getuser" command purely arbitrary and this could be changed in the future
-    mocker.patch("bot.dummy_db_manager.DummyDBManager.get_user", side_effect=Exception)
+    mocker.patch("database.DatabaseManager.get_user", side_effect=Exception)
 
     # Trigger test
     with pytest.raises(Exception):
@@ -100,8 +100,8 @@ async def test_command_exception_handling(config_bot, mocker):
 @pytest.mark.asyncio
 async def test_loop_exception_handling(config_bot, mocker):
     bot = config_bot
-    # patching "DummyDBManager.get_userS" here!
-    mocker.patch("bot.dummy_db_manager.DummyDBManager.get_users", side_effect=Exception)
+    # patching "DatabaseManager.get_userS" here!
+    mocker.patch("database.DatabaseManager.get_users", side_effect=Exception)
     # changing the check_new_solves loop delay interval to speed up the test
     cog = bot.get_cog(pytest.COG_NAME)
     cog.check_new_solves.change_interval(seconds=0.1)