42 lines
1.3 KiB
Python
42 lines
1.3 KiB
Python
import logging
|
|
import tomllib
|
|
import aiomysql
|
|
from typing import Optional
|
|
|
|
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._pool: Optional[aiomysql.Pool] = None
|
|
|
|
async def init_db_pool(self):
|
|
self._pool = await aiomysql.create_pool(
|
|
host=self.db_config["db_host"],
|
|
port=self.db_config["db_port"],
|
|
user=self.db_config["db_user"],
|
|
password=self.db_config["db_password"],
|
|
db=self.db_config["db_name"],
|
|
minsize=1,
|
|
maxsize=40,
|
|
autocommit=True
|
|
)
|
|
logger.info("Connected to database.")
|
|
|
|
async def get_user_badges(self) -> list:
|
|
async with self._pool.acquire() as conn:
|
|
async with conn.cursor(aiomysql.DictCursor) as cursor:
|
|
await 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;"""
|
|
)
|
|
return await cursor.fetchall()
|