Database & Persistence

OSCAR stores persistent user data in a local SQLite database (default path: ./database/oscar.db). The database layer serves two purposes:

1. Persistence: Storing user preferences, semester plans, and feedback.
2. Caching: Storing module metadata fetched from the Nextcloud API to reduce API calls and enable SQL joins.

Database Schema

OSCAR uses SQLite (Version 3) with the following main tables:

flowchart TD
    %% Define entities with SQLite schema details
    USER[("USERS<br/>--<br/>id {PK} INTEGER<br/>name TEXT")]

    PREF[("PREFERENCES<br/>--<br/>user_id {PK,FK} INTEGER<br/>language VARCHAR2<br/>semester INTEGER<br/>major TEXT")]

    MOD[("MODULES<br/>--<br/>id {PK} INTEGER<br/>language {PK} VARCHAR2<br/>title TEXT<br/>title_en TEXT")]

    PLAN[("SEMESTER_PLANS<br/>--<br/>user_id {PK,FK} INTEGER<br/>module_id {PK,FK} INTEGER<br/>language {PK,FK} VARCHAR2<br/>UNIQUE(user_id, module_id)")]

    FEED[("FEEDBACK<br/>--<br/>id {PK} INTEGER AI<br/>user_id {FK} INTEGER<br/>timestamp INTEGER<br/>intuitiveness INT 1-4<br/>discoverability INT 1-4<br/>usefulness INT 1-4<br/>improvements TEXT<br/>wishes TEXT<br/>bugs TEXT")]

    %% Define relationships
    REL_HAS{"has"}
    REL_PLANS{"plans"}
    REL_GIVES{"gives"}
    REL_INCLUDES{"includes"}

    %% Connections with min-max notation
    %% User has Preference (0,1) to (1,1)
    USER ---|"(0,1)"| REL_HAS
    REL_HAS ---|"(1,1)"| PREF

    %% User plans Semester_Plans (0,n) to (1,1)
    USER ---|"(0,n)"| REL_PLANS
    REL_PLANS ---|"(1,1)"| PLAN

    %% Semester_Plans includes Module (0,n) to (0,m)
    PLAN ---|"(0,n)"| REL_INCLUDES
    REL_INCLUDES ---|"(0,m)"| MOD

    %% User gives Feedback (0,n) to (1,1)
    USER ---|"(0,n)"| REL_GIVES
    REL_GIVES ---|"(1,1)"| FEED

    %% Styling for EER appearance
    classDef entity fill:#e1f5ff,stroke:#01579b,stroke-width:2px;
    classDef relation fill:#fff3e0,stroke:#e65100,stroke-width:2px;

    class USER,PREF,MOD,PLAN,FEED entity;
    class REL_HAS,REL_PLANS,REL_GIVES,REL_INCLUDES relation;

Singleton Access

Do not instantiate Database directly. Use the singleton accessor to ensure thread safety and a single connection pool.

src.util.database.get_database

get_database() -> Database

Get or create the database singleton.

Source code in src/util/database.py

def get_database() -> Database:
    """Get or create the database singleton."""
    # pylint: disable=W0603 # (global-statement)
    global _db_instance
    if _db_instance is None:
        _db_instance = Database()
    return _db_instance

Data Schema (v6)

The database schema is versioned via PRAGMA user_version.

Table	Description	Key Relationships
users	Stores Discord User IDs.	Primary Key (id)
preferences	User settings (Language, Major, Semester, SPO, Winter/Summer).	FK to users
modules	Cache of module definitions (Title, Language).	PK (id, language)
semester_plans	Link table for User ↔ Module assignments.	FK to users, FK to modules
feedback	Anonymous feedback submissions.	FK to users (for rate limiting/spam protection)
study_buddies	Opt-in module study group matchmaking.	FK to users
module_ratings	Student course evaluations & difficulty ratings.	FK to users

Public API

The Database Class

The central interface for all SQL operations.

src.util.database.Database

Database(db_path: str | None = None)

Class for handling database operations.

Version: 8

Initialize database connection.

Parameters:

Name	Type	Description	Default
db_path	str | None	Path to database file. Defaults to OSCAR_DB_PATH, then ./database/oscar.db	None

Source code in src/util/database.py

def __init__(self, db_path: str|None = None):
    """ Initialize database connection.

        Parameters:
            db_path: Path to database file. Defaults to `OSCAR_DB_PATH`,
                then ./database/oscar.db
    """
    if db_path is None:
        # the container points this at its /database volume, so data survives image updates
        db_path = os.environ.get("OSCAR_DB_PATH")
        if db_path:
            Path(db_path).parent.mkdir(parents=True, exist_ok=True)
    if not db_path:
        logger.debug("Missing a DB path")
        db_dir: Path = Path(__file__).parent / "database"
        logger.debug(f"Choosing DB dirctory as '{db_dir}'")
        logger.info("Ensuring the folder exists")
        db_dir.mkdir(exist_ok=True)
        logger.success(f"'{db_dir}' should now exist")
        db_path = str(db_dir / "oscar.db")
    logger.success(f"DB path has been set to '{db_path}'")

    self.db_path = str(db_path)
    self._init_db()

set_preferences

set_preferences(user_id: int, language: LanguageCode | None = None, semester: int | None = None, major: str | None = None, spo: int | None = None, winter_semester: bool = True) -> None

Update the users preferences in the database.

Parameters:

Name	Type	Description	Default
user_id	int	the users discord id.	required
language	LanguageCode | None	prefered language (e.g. ).	None
semester	int | None	integer representing the current semester.	None
major	str | None	the users course of study.	None
spo	int | None	usually the users enrollment year or the next lower spo	None
winter_semester	bool	weather the user has started to study in the winter semester	True

Source code in src/util/database.py

def set_preferences(  # pylint: disable=R0913,R0917  # (too-many-(positional-)arguments)
    self,
    user_id: int,
    language: LanguageCode | None = None,
    semester: int | None = None,
    major: str | None = None,
    spo: int | None = None,
    winter_semester: bool = True
) -> None:
    """ Update the users preferences in the database.

        Parameters:
            user_id:  the users discord id.
            language: prefered language (e.g. <LanguageCode.EN>).
            semester: integer representing the current semester.
            major:    the users course of study.
            spo:      usually the users enrollment year or the next lower spo
            winter_semester: weather the user has started to study in the winter semester
    """

    self._check_user(user_id)

    with self._get_connection() as conn:
        cursor = conn.cursor()

        # Check if preferences exist
        _ = cursor.execute(
            "SELECT user_id FROM preferences WHERE user_id = ?", (user_id,)
        )

        if cursor.fetchone():
            # Update existing preferences
            updates: list[str] = []
            params: list[str|int] = []
            if language is not None:
                updates.append("language = ?")
                params.append(language.name)
            if semester is not None:
                updates.append("semester = ?")
                params.append(semester)
            if major is not None:
                updates.append("major = ?")
                params.append(major)
            if spo is not None:
                updates.append("spo = ?")
                params.append(spo)
            updates.append("winter_semester = ?")
            params.append(winter_semester)

            if updates:
                params.append(user_id)
                query = (
                    f"UPDATE preferences SET {', '.join(updates)} WHERE user_id = ?"
                )
                _ = cursor.execute(query, params)
        else:
            # Insert new preferences
            _ = cursor.execute(
                """
                INSERT INTO preferences
                    (user_id, language, semester, major, spo, winter_semester)
                VALUES (?, ?, ?, ?, ?, ?)
                """,
                (
                    user_id,
                    language.name if language else "en",
                    semester,
                    major,
                    spo,
                    winter_semester,
                ),
            )

        conn.commit()

get_preferences

get_preferences(user_id: int) -> PrefsDict | None

Retrieve a user's preferences as a dictionary.

Parameters:

Name	Type	Description	Default
user_id	int	the users discord id.	required

Returns:

Type	Description
PrefsDict | None	dictonary corresponding to the row for the users preferences

Source code in src/util/database.py

def get_preferences(self, user_id: int) -> PrefsDict | None:
    """ Retrieve a user's preferences as a dictionary.

        Parameters:
            user_id:  the users discord id.

        Returns:
            dictonary corresponding to the row for the users preferences
    """

    with self._get_connection() as conn:
        cursor = conn.cursor()
        _ = cursor.execute("SELECT * FROM preferences WHERE user_id = ?", (user_id,))
        row: sqlite3.Row | None = cursor.fetchone()  # pyright: ignore[reportAny]

        if not row:
            return None

        prefs: PrefsDict = PrefsDict(
            user_id = int(row["user_id"]),  # pyright: ignore[reportAny]
            # pylint: disable=C0301 # (line-too-long)
            language = LanguageCode.from_language_code(str(row["language"])),  # pyright: ignore[reportAny]
            # semester = int(row["semester"]),
        )

        try:
            major: StudyCourse = StudyCourse.from_str(
                str(row["major"])  # pyright: ignore[reportAny]
            )
            prefs["major"] = major
        except ValueError:
            # no user id here: a log line outlives the row, and `/my_data` cannot delete it
            logger.warning(f"A user has the invalid major '{row['major']}'")

        if row["semester"] is not None:
            prefs["semester"] = int(row["semester"])  # pyright: ignore[reportAny]

        if row["spo"] is not None:
            prefs["spo"] = row["spo"] # pyright: ignore[reportAny]

        prefs["winter_semester"] = bool(row["winter_semester"]) # pyright: ignore[reportAny]

        return prefs

get_users

get_users() -> list[int] | None

Fetches all users

Returns:

Type	Description
list[int] | None	List of user (discord) id's.

Source code in src/util/database.py

def get_users(self) -> list[int] | None:
    """Fetches all users

        Returns:
            List of user (discord) id's.
    """

    with self._get_connection() as conn:
        cursor = conn.cursor()
        _ = cursor.execute("SELECT id FROM users")
        rows: list[sqlite3.Row] = cursor.fetchall()
        return [row["id"] for row in rows] if rows else None

get_module

get_module(module_id: int, language: LanguageCode) -> Module | None

Fetches a module from the database.

Parameters:

Name	Type	Description	Default
module_id	int	the modules id.	required
language	LanguageCode	the modules language.	required

Source code in src/util/database.py

def get_module(self, module_id: int, language: LanguageCode) -> Module | None:
    """Fetches a module from the database.

        Parameters:
            module_id:  the modules id.
            language:   the modules language.
    """

    with self._get_connection() as conn:
        cursor = conn.cursor()
        _ = cursor.execute("SELECT * FROM modules WHERE id = ?", (module_id,))
        rows: list[sqlite3.Row] = cursor.fetchall()

        if not rows:
            return None

        # Try to find the module with the requested language
        for row in rows:
            if row["language"] == language:
                return Module(
                    id_ = row["id"],            # pyright: ignore[reportAny]
                    language = row["language"], # pyright: ignore[reportAny]
                    title = row["title"],       # pyright: ignore[reportAny]
                    title_en = row["title_en"], # pyright: ignore[reportAny]
                )
        # Fallback: return the first module found (any language)
        row = rows[0]
        return Module(
            id_ = row["id"],             # pyright: ignore[reportAny]
            language = row["language"], # pyright: ignore[reportAny]
            title = row["title"],       # pyright: ignore[reportAny]
            title_en = row["title_en"], # pyright: ignore[reportAny]
        )

add_to_semesterplan

add_to_semesterplan(user_id: int, module_id: int, language: LanguageCode = LanguageCode.DE) -> None

Adds a module to the users semester plan.

Parameters:

Name	Type	Description	Default
user_id	int	the users discord id.	required
module_id	int	the modules id.	required
language	LanguageCode	the modules language.	DE

Source code in src/util/database.py

def add_to_semesterplan(
    self,
    user_id: int,
    module_id: int,
    language: LanguageCode = LanguageCode.DE
)-> None:
    """Adds a module to the users semester plan.

        Parameters:
            user_id:   the users discord id.
            module_id: the modules id.
            language:  the modules language.
    """
    self._check_user(user_id)
    checked_module: tuple[int, str] | None  = self._check_module(module_id, language)

    if not checked_module:
        # asserts are stripped with -O and the button callback only shows a generic error
        raise KeyError(f"can't find information on the module with the id: {module_id}")

    with self._get_connection() as conn:
        cursor = conn.cursor()
        _ = cursor.execute("""
            INSERT OR REPLACE INTO semester_plans (user_id, module_id, language)
            VALUES (?, ?, ?)
        """, (user_id, checked_module[0], checked_module[1]))
        conn.commit()

get_semesterplan

get_semesterplan(user_id: int) -> list[Module]

Fetches the semester plan for a user.

Parameters:

Name	Type	Description	Default
user_id	int	the users discord id.	required

Returns:

Type	Description
list[Module]	List of modules in the users semester plan.

Source code in src/util/database.py

def get_semesterplan(self, user_id:int)-> list[Module]:
    """Fetches the semester plan for a user.

        Parameters:
            user_id: the users discord id.

        Returns:
            List of modules in the users semester plan.
    """

    with self._get_connection() as conn:
        cursor = conn.cursor()
        _ = cursor.execute("""
            SELECT m.id, m.language, m.title, m.title_en
            FROM semester_plans sp
            JOIN modules m ON sp.module_id = m.id AND sp.language = m.language
            WHERE sp.user_id = ?
        """, (user_id,))
        rows: list[sqlite3.Row] = cursor.fetchall()

        result: list[Module] = []
        for row in rows:
            result.append(
                Module(
                    id_ = row["id"],            # pyright: ignore[reportAny]
                    language = row["language"], # pyright: ignore[reportAny]
                    title = row["title"],       # pyright: ignore[reportAny]
                    title_en = row["title_en"], # pyright: ignore[reportAny]
                )
            )
        return result

send_feedback_dict

send_feedback_dict(feedback: FeedbackDict)

Sends user feedback to the database.

Source code in src/util/database.py

def send_feedback_dict(self, feedback: FeedbackDict):
    """ Sends user feedback to the database."""
    self.send_feedback(
        user_id = feedback["user_id"],
        intuitiveness = feedback["intuitiveness"],
        discoverability = feedback["discoverability"],
        usefulness = feedback["usefulness"],
        improvements = feedback.get("improvements", ""),
        wishes = feedback.get("wishes", ""),
        bugs = feedback.get("bugs", "")
    )

send_feedback

send_feedback(user_id: int, intuitiveness: int, discoverability: int, usefulness: int, improvements: str = '', wishes: str = '', bugs: str = '') -> None

Sends user feedback to the database.

Source code in src/util/database.py

def send_feedback(self,
    user_id: int,
    intuitiveness: int,
    discoverability: int,
    usefulness: int,
    improvements: str = "",
    wishes: str = "",
    bugs: str = "")-> None:
    """ Sends user feedback to the database."""

    # check that user exists
    self._check_user(user_id)

    # get user language
    language: str = "en"
    prefs: PrefsDict | None = self.get_preferences(user_id)
    if prefs:
        language = prefs["language"].name

    timestamp: int = int(time.time())
    # as Unix Time, the number of seconds since 1970-01-01 00:00:00 UTC.

    with self._get_connection() as conn:
        cursor = conn.cursor()
        _ = cursor.execute("""
            INSERT INTO feedback (user_id, language, time,
                                  intuitiveness, discoverability, usefulness,
                                  improvements, wishes, bugs)
            VALUES (?, ?, ?,
                    ?, ?, ?,
                    ?, ?, ?)
            """,
            (user_id, language, timestamp,
             intuitiveness, discoverability, usefulness,
             improvements, wishes, bugs)
        )
        conn.commit()

get_feedback

get_feedback() -> list[FeedbackReviewDict]

Fetches all feedback from the database, sorted by ccreation date.

Source code in src/util/database.py

def get_feedback(self)-> list[FeedbackReviewDict]:
    """ Fetches all feedback from the database, sorted by ccreation date."""
    result: list[FeedbackReviewDict] = []

    with self._get_connection() as conn:
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()
        _ = cursor.execute("SELECT * FROM feedback")
        rows: list[sqlite3.Row] = cursor.fetchall()

        if rows:
            for row in rows:
                result.append(
                    FeedbackReviewDict(
                        id = row["id"],                             # pyright: ignore[reportAny]
                        user_id = row["user_id"],                   # pyright: ignore[reportAny]
                        language = row["language"],                 # pyright: ignore[reportAny]
                        timestamp = row["time"],                    # pyright: ignore[reportAny]
                        intuitiveness = row["intuitiveness"],       # pyright: ignore[reportAny]
                        discoverability = row["discoverability"],   # pyright: ignore[reportAny]
                        usefulness = row["usefulness"],             # pyright: ignore[reportAny]
                        improvements = row["improvements"],         # pyright: ignore[reportAny]
                        wishes = row["wishes"],                     # pyright: ignore[reportAny]
                        bugs = row["bugs"],                         # pyright: ignore[reportAny]
                    )
                )

    result.sort(key=lambda x: x["timestamp"])

    return result

delete_feedback

delete_feedback(feedback_id: int) -> None

Deletes feedback from the database by its id.

Source code in src/util/database.py

def delete_feedback(self, feedback_id: int) -> None:
    """ Deletes feedback from the database by its id."""

    with self._get_connection() as conn:
        cursor = conn.cursor()
        _ = cursor.execute(
            "DELETE FROM feedback WHERE id = ?", (feedback_id,)
        )
        conn.commit()

rate_module

rate_module(user_id: int, module_id: int, rating: int, difficulty: int, comment: str | None = None) -> None

Stores or updates a student's rating and difficulty evaluation for a module.

Parameters:

Name	Type	Description	Default
user_id	int	The student's discord id.	required
module_id	int	The module identifier.	required
rating	int	Overall rating (1-5 stars).	required
difficulty	int	Subjective difficulty (1-5, 1=easy, 5=hard).	required
comment	str | None	Optional review or advice for future students.	None

Source code in src/util/database.py

def rate_module(
    self,
    user_id: int,
    module_id: int,
    rating: int,
    difficulty: int,
    comment: str | None = None,
) -> None:
    """ Stores or updates a student's rating and difficulty evaluation for a module.

        Parameters:
            user_id: The student's discord id.
            module_id: The module identifier.
            rating: Overall rating (1-5 stars).
            difficulty: Subjective difficulty (1-5, 1=easy, 5=hard).
            comment: Optional review or advice for future students.
    """
    if not 1 <= rating <= 5:
        raise ValueError("Rating must be between 1 and 5.")
    if not 1 <= difficulty <= 5:
        raise ValueError("Difficulty must be between 1 and 5.")

    self._check_user(user_id)
    now = int(time.time())
    clean_comment = comment.strip() if comment else None

    with self._get_connection() as conn:
        cursor = conn.cursor()
        _ = cursor.execute("""
            INSERT INTO module_ratings (user_id, module_id, rating, difficulty, comment, timestamp)
            VALUES (?, ?, ?, ?, ?, ?)
            ON CONFLICT(user_id, module_id) DO UPDATE SET
                rating = excluded.rating,
                difficulty = excluded.difficulty,
                comment = excluded.comment,
                timestamp = excluded.timestamp
        """, (user_id, module_id, rating, difficulty, clean_comment, now))
        conn.commit()

get_module_ratings

get_module_ratings(module_id: int) -> ModuleRatingDict

Returns aggregated ratings and difficulty statistics for a module.

Parameters:

Name	Type	Description	Default
module_id	int	The module identifier.	required

Returns:

Type	Description
ModuleRatingDict	Dictionary with count, avg_rating, and avg_difficulty.

Source code in src/util/database.py

def get_module_ratings(self, module_id: int) -> ModuleRatingDict:
    """ Returns aggregated ratings and difficulty statistics for a module.

        Parameters:
            module_id: The module identifier.

        Returns:
            Dictionary with count, avg_rating, and avg_difficulty.
    """
    with self._get_connection() as conn:
        cursor = conn.cursor()
        _ = cursor.execute("""
            SELECT COUNT(*) as count, AVG(rating) as avg_rating, AVG(difficulty) as avg_difficulty
            FROM module_ratings
            WHERE module_id = ?
        """, (module_id,))
        row = cursor.fetchone()
        if row and row["count"] > 0:
            return {
                "count": int(row["count"]),
                "avg_rating": round(float(row["avg_rating"]), 1),
                "avg_difficulty": round(float(row["avg_difficulty"]), 1),
            }
        return {
            "count": 0,
            "avg_rating": None,
            "avg_difficulty": None,
        }

get_user_module_rating

get_user_module_rating(user_id: int, module_id: int) -> dict[str, Any] | None

Returns a student's own rating for a module if already submitted.

Parameters:

Name	Type	Description	Default
user_id	int	The student's discord id.	required
module_id	int	The module identifier.	required

Returns:

Type	Description
dict[str, Any] | None	Rating dict or None if unrated.

Source code in src/util/database.py

def get_user_module_rating(self, user_id: int, module_id: int) -> dict[str, Any] | None:
    """ Returns a student's own rating for a module if already submitted.

        Parameters:
            user_id: The student's discord id.
            module_id: The module identifier.

        Returns:
            Rating dict or None if unrated.
    """
    with self._get_connection() as conn:
        cursor = conn.cursor()
        _ = cursor.execute("""
            SELECT rating, difficulty, comment, timestamp
            FROM module_ratings
            WHERE user_id = ? AND module_id = ?
        """, (user_id, module_id))
        row = cursor.fetchone()
        if row:
            return {
                "rating": int(row["rating"]),
                "difficulty": int(row["difficulty"]),
                "comment": row["comment"],
                "timestamp": int(row["timestamp"]),
            }
        return None

get_module_comments

get_module_comments(module_id: int, limit: int = 5) -> list[dict[str, Any]]

Returns recent student reviews/comments for a module.

Parameters:

Name	Type	Description	Default
module_id	int	The module identifier.	required
limit	int	Maximum number of comments to return.	5

Returns:

Type	Description
list[dict[str, Any]]	List of dicts with rating, difficulty, comment, timestamp.

Source code in src/util/database.py

def get_module_comments(self, module_id: int, limit: int = 5) -> list[dict[str, Any]]:
    """ Returns recent student reviews/comments for a module.

        Parameters:
            module_id: The module identifier.
            limit: Maximum number of comments to return.

        Returns:
            List of dicts with rating, difficulty, comment, timestamp.
    """
    with self._get_connection() as conn:
        cursor = conn.cursor()
        _ = cursor.execute("""
            SELECT rating, difficulty, comment, timestamp
            FROM module_ratings
            WHERE module_id = ? AND comment IS NOT NULL AND TRIM(comment) != ''
            ORDER BY timestamp DESC
            LIMIT ?
        """, (module_id, limit))
        return [
            {
                "rating": int(r["rating"]),
                "difficulty": int(r["difficulty"]),
                "comment": str(r["comment"]),
                "timestamp": int(r["timestamp"]),
            }
            for r in cursor.fetchall()
        ]

Manual Documentation overrides

add_module

Hybrid Caching Logic: This method is polymorphic. If called with just an int (ID), it attempts to fetch the module details from the Nextcloud Tables API (util.tables) and cache them locally before insertion.

Adds a module to the database.

• Overloads:
  • add_module(module: Module)
  • add_module(module_id: int, language: LanguageCode = DE): Fetches data from API.
  • add_module(module_id: int, language: LanguageCode, title: str, title_en: str)
• Parameters:
  • *args: Variable arguments based on usage above.

remove_from_semesterplan

Note: Documented manually due to docstring mismatch in source.

Removes a module from the user's semester plan.

• Parameters:
  • user_id (int): The user's Discord ID.
  • module_id (int): The module's unique ID.

Helper Functions

get_user_language

A helper function used extensively in UI Views to determine the correct language string (DE or EN) for a user. Defaults to EN if no preference is saved.

src.util.database.get_user_language

get_user_language(user_id: int) -> LanguageCode

Liest die Sprache eines Benutzers aus der Datenbank.

Source code in src/util/database.py

def get_user_language(user_id: int) -> LanguageCode:
    """Liest die Sprache eines Benutzers aus der Datenbank."""
    _db: Database = get_database()
    prefs: PrefsDict | None = _db.get_preferences(user_id)

    if not prefs:
        logger.debug("A user has no prefered language set, defaulting to EN")
        return LanguageCode.EN
    return prefs["language"]