discord-havoc/util/lovehate_db.py

import os
import logging
from typing import LiteralString, Optional, Union
import aiosqlite as sqlite3
from constructors import LoveHateException


class DB:
    """LoveHate DB Utility Class"""

    def __init__(self, bot) -> None:
        self.db_path: str | LiteralString = os.path.join(
            "/usr/local/share", "sqlite_dbs", "lovehate.db"
        )

    async def get_wholovehates(
        self, thing: str, loves: bool = False, hates: bool = False
    ) -> Union[list[tuple], bool]:
        """
        Get a list of users who have professed their love OR hatred for <thing>

        Args:
            thing (str): The <thing> to check
            loves (bool): Are we looking for loves?
            hates (bool): ...or are we looking for hates?
        Returns:
            Union[list[tuple], bool]
        """

        query: str = "SELECT display_name FROM lovehate WHERE thing LIKE ? AND flag = ?"
        params: tuple = tuple()
        flag: Optional[int] = None

        if hates and loves:
            raise LoveHateException("Both hates and loves may not be True")
        elif hates:
            flag = -1
        elif loves:
            flag = 1
        elif not hates and not loves:
            raise LoveHateException("Neither loves nor hates were requested")

        params = (
            thing,
            flag,
        )
        async with sqlite3.connect(self.db_path, timeout=2) as db_conn:
            async with await db_conn.execute(query, params) as db_cursor:
                result: list[tuple] = await db_cursor.fetchall()
                logging.debug("Result: %s", result)
                if not result:
                    return False
                return result

    async def get_lovehates(
        self,
        loves: bool = False,
        hates: bool = False,
        user: Optional[str] = None,
        thing: Optional[str] = None,
    ) -> Union[list[tuple], bool]:
        """
        Get a list of either 1) what {user} loves/hates, or who loves/hates {thing}, depending on bools loves, hates

        Args:
            loves (bool): Are we looking for loves?
            hates (bool): ...OR are we looking for hates?
            user (Optional[str]): the user to query against
            thing (Optional[str]): ... OR the thing to query against

        Returns:
            Union[list[tuple], bool]
        """

        query: str = ""
        params: tuple = tuple()

        if not user and not thing:
            raise LoveHateException(
                "Neither a <user> or <thing> was specified to query against"
            )

        flag: Optional[int] = None

        if hates and loves:
            raise LoveHateException("Both hates and loves may not be True")
        elif hates:
            flag = -1
        elif loves:
            flag = 1
        elif not hates and not loves:
            raise LoveHateException("Neither loves nor hates were requested")

        if user:
            query = "SELECT thing FROM lovehate WHERE display_name LIKE ? AND flag == ?"
            params = (
                user,
                flag,
            )
        elif thing:
            query = "SELECT display_name FROM lovehate WHERE thing LIKE ? AND flag == ?"
            params = (
                thing,
                flag,
            )

        async with sqlite3.connect(self.db_path, timeout=2) as db_conn:
            async with await db_conn.execute(query, params) as db_cursor:
                result = await db_cursor.fetchall()
                if not result:
                    return False
                return result

    async def check_existence(self, user: str, thing: str) -> Optional[int]:
        """
        Determine whether a user is opinionated on a <thing>

        Args:
            user (str): The user to check
            thing (str): The thing to check if the user has an opinion on
        Returns:
            Optional[int]
        """

        params = (
            user,
            thing,
        )

        async with sqlite3.connect(self.db_path, timeout=2) as db_conn:
            async with await db_conn.execute(
                "SELECT id, flag FROM lovehate WHERE display_name LIKE ? AND thing LIKE ?",
                params,
            ) as db_cursor:
                result = await db_cursor.fetchone()
                if not result:
                    return None
                (_, flag) = result
                return flag

    async def update(self, user: str, thing: str, flag: int) -> str:
        """
        Updates the lovehate database, and returns an appropriate response

        Args:
            user (str): The user to update
            thing (str): The thing the user loves/hates/doesn't care about anymore
            flag (int): int representation of love (1), hate (-1), and dontcare (0)
        Returns:
            str
        """
        if not flag in range(-1, 2):
            raise LoveHateException(
                f"Invalid flag {flag} specified, is this love (1), hate (-1), or dontcare? (0)"
            )

        db_query: str = ""
        params: tuple = (
            user,
            thing,
        )

        already_opinionated: Optional[int] = await self.check_existence(user, thing)
        if already_opinionated:
            if flag == 0:
                db_query = (
                    "DELETE FROM lovehate WHERE display_name LIKE ? AND thing LIKE ?"
                )
            else:
                loves_or_hates: str = "loves"
                if already_opinionated == -1:
                    loves_or_hates = "hates"
                raise LoveHateException(
                    f"But {user} already {loves_or_hates} {thing}..."
                )
        else:
            match flag:
                case -1:
                    db_query = "INSERT INTO lovehate(display_name, flag, thing) VALUES(?, -1, ?)"
                case 1:
                    db_query = "INSERT INTO lovehate(display_name, flag, thing) VALUES(?, 1, ?)"
                case _:
                    raise LoveHateException("Unknown error, default case matched")

        async with sqlite3.connect(self.db_path, timeout=2) as db_conn:
            async with await db_conn.execute(db_query, params) as db_cursor:
                await db_conn.commit()
                if db_cursor.rowcount != 1:
                    raise LoveHateException(
                        f"DB Error - RowCount: {db_cursor.rowcount} for INSERT query"
                    )
                match flag:
                    case -1:
                        return f"We're done here, {user} hates {thing}."
                    case 0:
                        return f"We're done here, {user} no longer cares one way or the other about {thing}."
                    case 1:
                        return f"We're done here, {user} loves {thing}."
                    case _:
                        raise LoveHateException(
                            "Unknown error, default case matched [2]"
                        )