40 lines
1.3 KiB
Python
40 lines
1.3 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) -> List[Dict]:
|
|
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;"""
|
|
)
|
|
user_badges = cursor.fetchall()
|
|
logger.info(f"Got {len(user_badges)} user badges from database.")
|
|
return user_badges
|