api/endpoints/rand_msg.py

#!/usr/bin/env python3.12

import os
import random
from typing import LiteralString, Optional, Union
import aiosqlite as sqlite3
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from .constructors import RandMsgRequest

class RandMsg(FastAPI):
    """
    Random Message Endpoint
    """   
    def __init__(self, app: FastAPI,
                 util, constants) -> None:
        self.app: FastAPI = app
        self.util = util
        self.constants = constants
        self.endpoint_name = "randmsg"

        app.add_api_route(f"/{self.endpoint_name}", self.randmsg_handler, methods=["POST"])
    
    async def randmsg_handler(self, data: Optional[RandMsgRequest] = None) -> JSONResponse:
        """
        Get a randomly generated message
        - **short**: Optional, if True, will limit length of returned random messages to <=126 characters (Discord restriction related)
        """
        random.seed()
        short: Optional[bool] = False
        if isinstance(data, RandMsgRequest):
            short = data.short
            db_rand_selected: int = 9
        db_rand_selected = random.choice([0, 1, 3])
        title_attr: str = "Unknown"

        match db_rand_selected:
            case 0:
                randmsg_db_path: Union[str, LiteralString] = os.path.join("/usr/local/share",
                                               "sqlite_dbs", "qajoke.db")  # For qajoke db
                db_query: str = "SELECT id, ('<b>Q:</b> ' || question || '<br/><b>A:</b> ' \
                    || answer) FROM jokes ORDER BY RANDOM() LIMIT 1" # For qajoke db
                title_attr = "QA Joke DB"
            case 1 | 9:
                randmsg_db_path = os.path.join("/usr/local/share",
                                               "sqlite_dbs",
                                               "randmsg.db") # For randmsg db
                db_query = "SELECT id, msg FROM msgs WHERE \
                    LENGTH(msg) <= 180 ORDER BY RANDOM() LIMIT 1" # For randmsg db
                if db_rand_selected == 9:
                    db_query = db_query.replace("<= 180", "<= 126")
                title_attr = "Random Msg DB"
            case 2:
                randmsg_db_path = os.path.join("/usr/local/share",
                                               "sqlite_dbs",
                                               "trump.db") # For Trump Tweet DB
                db_query = "SELECT id, content FROM tweets \
                    ORDER BY RANDOM() LIMIT 1" # For Trump Tweet DB
                title_attr = "Trump Tweet DB"
            case 3:
                randmsg_db_path = os.path.join("/usr/local/share",
                                               "sqlite_dbs",
                                               "philo.db") # For Philo DB
                db_query = "SELECT id, (content || '<br> - ' || speaker) FROM quotes \
                    ORDER BY RANDOM() LIMIT 1"  # For Philo DB
                title_attr = "Philosophical Quotes DB"
            case 4:
                randmsg_db_path = os.path.join("/usr/local/share",
                                               "sqlite_dbs",
                                               "hate.db") # For Hate DB
                db_query = """SELECT id, ("<font color='#FF0000'>" || comment) FROM hate_speech \
                    WHERE length(comment) <= 180 ORDER BY RANDOM() LIMIT 1"""
                title_attr = "Hate Speech DB"
            case 5:
                randmsg_db_path = os.path.join("/usr/local/share",
                                               "sqlite_dbs",
                                               "rjokes.db") # r/jokes DB
                db_query = """SELECT id, (title || "<br>" || body) FROM jokes \
                    WHERE score >= 10000 ORDER BY RANDOM() LIMIT 1"""
                title_attr = "r/jokes DB"
                                              
        async with sqlite3.connect(database=randmsg_db_path, timeout=1) as _db:
            async with await _db.execute(db_query) as _cursor:
                result: sqlite3.Row = await _cursor.fetchone()
                (result_id, result_msg) = result
                result_msg = result_msg.strip()
                return JSONResponse(content=
                                    {
                                        "id": result_id,
                                        "msg": result_msg,
                                        "title": title_attr,
                })