55 lines
2.2 KiB
Python
55 lines
2.2 KiB
Python
import logging
|
|
import tomllib
|
|
import pymysql
|
|
from typing import Optional, List, Dict
|
|
|
|
logger = logging.getLogger(__name__.split(".")[-1])
|
|
|
|
|
|
class DatabaseService:
|
|
def __init__(self, config: str):
|
|
with open(config, "rb") as f:
|
|
config = tomllib.load(f)
|
|
|
|
self.db_config = config["database"]
|
|
self._conn: Optional[pymysql.Connection] = None
|
|
|
|
def init_db(self):
|
|
self._conn = pymysql.connect(
|
|
host=self.db_config["db_host"],
|
|
port=self.db_config["db_port"],
|
|
user=self.db_config["db_user"],
|
|
password=self.db_config["db_password"],
|
|
database=self.db_config["db_name"],
|
|
cursorclass=pymysql.cursors.DictCursor)
|
|
logger.info("Connected to database.")
|
|
|
|
def get_user_badges(self, user_id: Optional[int] = None) -> List[Dict]:
|
|
if user_id is None:
|
|
with self._conn.cursor() as cursor:
|
|
cursor.execute(
|
|
"""SELECT u.user_id, u.user_name, s.seat_id, upp.picture, tic.ticket_id
|
|
FROM users AS u
|
|
LEFT JOIN seats AS s
|
|
ON u.user_id = s.`user`
|
|
LEFT JOIN user_profile_picture AS upp
|
|
ON u.user_id = upp.user_id
|
|
INNER JOIN tickets AS tic
|
|
ON u.user_id = tic.user;"""
|
|
)
|
|
user_badges = cursor.fetchall()
|
|
else:
|
|
with self._conn.cursor() as cursor:
|
|
cursor.execute(
|
|
"""SELECT u.user_id, u.user_name, s.seat_id, upp.picture
|
|
FROM users AS u
|
|
LEFT JOIN seats AS s
|
|
ON u.user_id = s.`user`
|
|
LEFT JOIN user_profile_picture AS upp
|
|
ON u.user_id = upp.user_id
|
|
WHERE u.user_id = %s;""", (user_id,)
|
|
)
|
|
user_badges = cursor.fetchall()
|
|
logger.info(f"Got {len(user_badges)} user badges from database.")
|
|
return user_badges
|