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]")