maple/util/db.py

import sqlite3
from sqlite3 import Error, Connection

from util.config import DB_PATH


def setup_database() -> bool:
    """
    First-time DB setup (creating tables etc.)
    :return: True if all went smoothly, else False
    """
    conn = create_connection()
    if conn is None:
        return False
    try:
        c = conn.cursor()
        c.execute("""CREATE TABLE IF NOT EXISTS connections (
                    id integer PRIMARY KEY,
                    uid text NOT NULL,
                    animelist text NOT NULL,
                    service text NOT NULL
                    );""")
    except Error as e:
        print("Error while creating table 'connections':")
        print(e)
        return False
    conn.close()
    return True


def get_connection(uid: str) -> tuple | None:
    conn = create_connection()
    try:
        c = conn.cursor()
        c.execute(f"""SELECT * FROM connections WHERE uid='{uid}';""")
        o = c.fetchall()
        conn.close()
        if len(o) == 0:
            return None
        return o[0]
    except Error as e:
        print("Error finding user connection:")
        print(e)
        return None


def add_user_connection(uid: str, listname: str, service: str) -> bool:
    conn = create_connection()
    if conn is None:
        print("No connection")
        return False
    try:
        c = conn.cursor()
        c.execute(f"""INSERT INTO connections(uid,animelist,service) VALUES('{uid}','{listname}','{service}')""")
        conn.commit()
    except Error as e:
        print("Error while adding a new connection:")
        print(e)
        return False
    conn.close()
    return True


def create_connection() -> Connection | None:
    """
    Creates a connection to an SQLite database
    :return: Connection if connected successfully, else None
    """
    conn = None
    try:
        conn = sqlite3.connect(DB_PATH)
        print(f"Connected, SQLite vetsion {sqlite3.version}")
        return conn
    except Error as e:
        print("Error while connecting to DB:")
        print(e)
    return conn