77 lines
2 KiB
Python
77 lines
2 KiB
Python
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
|