Compare commits

...
4 Commits
Author SHA1 Message Date
imcompiling 157b8d4f3f Begin updating utils. 2025-03-03 15:54:28 -05:00
imcompiling 1c3fb231b7 Add more cspell words. 2025-03-03 15:54:20 -05:00
imcompiling b72dd35e04 Add init file to the database package. 2025-03-03 15:28:00 -05:00
imcompiling 0b2a1cd293 Organize import in db.py. 2025-03-03 15:19:46 -05:00
4 changed files with 35 additions and 33 deletions
+2 -5
View File
@@ -1,11 +1,8 @@
{
"cSpell.words": [
"highscore",
"highscores"
],
"cSpell.words": ["gmtime", "highscore", "highscores", "imcompiling", "nextcord"],
"python.analysis.diagnosticSeverityOverrides": {
"reportOptionalMemberAccess": false,
"reportArgumentType": "information",
"reportAttributeAccessIssue": "warning"
}
}
}
+1
View File
@@ -0,0 +1 @@
"""Package for database interaction."""
+1
View File
@@ -1,6 +1,7 @@
"""Simple file to set up the sqlmodel connection to database."""
from sqlmodel import SQLModel, create_engine
from utilities import config
from . import models # noqa: F401 Need this import for creating the models.
+31 -28
View File
@@ -8,7 +8,8 @@ from sqlalchemy import distinct, select
from sqlalchemy.orm import Session, aliased
# pylint: disable=import-error
from database.db import Category, CategoryGroup, Guild, Score, engine
from database.db import engine
from database.models import Category, CategoryGroup, Guild, Score
MAX_LINE_LENGTH = 142
@@ -31,13 +32,13 @@ def autofill_categories(guild_id: int, group_id: int, category_begin: str) -> "d
# search database for all category names
subq = (
select(CategoryGroup)
.where(CategoryGroup.GuildId == guild_id)
.where(CategoryGroup.GroupId == group_id)
.where(CategoryGroup.guild_id == guild_id)
.where(CategoryGroup.id == group_id)
.subquery()
)
cat_group_alias = aliased(CategoryGroup, subq)
stmt = select(Category).where(Category.GroupId == cat_group_alias.GroupId)
categories_dict: dict[str, str] = {cat.CategoryName: str(cat.CategoryId) for cat in session.scalars(stmt)}
stmt = select(Category).where(Category.group_id == cat_group_alias.id)
categories_dict: dict[str, str] = {cat.name: str(cat.id) for cat in session.scalars(stmt)}
# if string empty, return all category names
if not category_begin:
@@ -64,10 +65,10 @@ def autofill_groups(guild_id: int, group_begin: str) -> "dict[str, str]":
"""
with Session(engine) as session:
# Select all groups in guild.
stmt = select(CategoryGroup).where(CategoryGroup.GuildId == guild_id)
stmt = select(CategoryGroup).where(CategoryGroup.guild_id == guild_id)
# dict has to have key values of same type for discord.
group_dict: dict[str, str] = {group.GroupName: str(group.GroupId) for group in session.scalars(stmt)}
group_dict: dict[str, str] = {group.name: str(group.id) for group in session.scalars(stmt)}
# if string empty, return all group names
if not group_begin:
@@ -83,8 +84,10 @@ def autofill_groups(guild_id: int, group_begin: str) -> "dict[str, str]":
def autofill_users_in_group_category(guild_id: int, group_id: int, category_id: int, user_begin: str) -> "list[str]":
"""
Returns list of users that match user_begin. Either all users in group or just users in category if category is
not None. Might be expensive due to unioning all users in group.
Returns list of users that match user_begin.
Either all users in group or just users in category if category is not None. Might be expensive due to union-ing all
users in group.
Args:
guild_id (int): Guild id corresponding to group and category.
@@ -99,27 +102,27 @@ def autofill_users_in_group_category(guild_id: int, group_id: int, category_id:
# get all categories in the group
subq = (
select(CategoryGroup)
.where(CategoryGroup.GuildId == guild_id)
.where(CategoryGroup.GroupId == group_id)
.where(CategoryGroup.guild_id == guild_id)
.where(CategoryGroup.id == group_id)
.subquery()
)
category_group = aliased(CategoryGroup, subq)
# If no category id, get all users in group.
if category_id == 0:
subq = select(Category.CategoryId).where(Category.GroupId == category_group.GroupId)
stmt = select(distinct(Score.UserName)).where(Score.CategoryId.in_(subq))
subq = select(Category.id).where(Category.group_id == category_group.id)
stmt = select(distinct(Score.username)).where(Score.category_id.in_(subq))
else:
# get all users in category_id
subq = (
select(Category)
.where(Category.GroupId == category_group.GroupId)
.where(Category.CategoryId == category_id)
.where(Category.group_id == category_group.id)
.where(Category.id == category_id)
.subquery()
)
category = aliased(Category, subq)
stmt = select(distinct(Score.UserName)).where(Score.CategoryId == category.CategoryId)
stmt = select(distinct(Score.username)).where(Score.category == category.id)
users_list: list[str] = list(session.scalars(stmt))
@@ -145,9 +148,9 @@ def autofill_users_in_guild(guild_id: int, user_begin: str) -> "list[str]":
list[str]: All users with submissions in the guild.
"""
with Session(engine) as session:
group_ids = select(CategoryGroup.GroupId).where(CategoryGroup.GuildId == guild_id)
category_ids = select(Category.CategoryId).where(Category.GroupId.in_(group_ids))
stmt = select(distinct(Score.UserName)).where(Score.CategoryId.in_(category_ids))
group_ids = select(CategoryGroup.id).where(CategoryGroup.guild_id == guild_id)
category_ids = select(Category.id).where(Category.group_id.in_(group_ids))
stmt = select(distinct(Score.username)).where(Score.category_id.in_(category_ids))
users_list: list[str] = list(session.scalars(stmt))
# if string empty, return all users
@@ -176,34 +179,34 @@ def format_highscore_message(session: Session, group: CategoryGroup) -> str:
Returns:
str: Formatted string for discord message
"""
ret_string = f"**{group.GroupName}**\n```"
ret_string = f"**{group.name}**\n```"
stmt = select(Guild).where(Guild.GuildId == group.GuildId)
stmt = select(Guild).where(Guild.id == group.guild_id)
# Get all categories in the group
subq = (
select(CategoryGroup)
.where(CategoryGroup.GuildId == group.GuildId)
.where(CategoryGroup.GroupName == group.GroupName)
.where(CategoryGroup.guild_id == group.guild_id)
.where(CategoryGroup.name == group.name)
.subquery()
)
cat_group_alias = aliased(CategoryGroup, subq)
stmt = select(Category).where(Category.GroupId == cat_group_alias.GroupId)
stmt = select(Category).where(Category.group_id == cat_group_alias.id)
# set up a map for the categories
# also grab the max category size
categories = []
category_ids = []
max_cat_size: int = 0
pad_length = group.PadLength
pad_length = group.pad_length
for cat in session.scalars(stmt):
max_cat_size = max(cat.Size, max_cat_size)
max_cat_size = max(cat.size, max_cat_size)
categories.append(cat)
category_ids.append(cat.CategoryId)
category_ids.append(cat.id)
# get all scores in category group, load into data frame.
d_f = pd.read_sql_query(
sql=session.query(Score).where(Score.CategoryId.in_(category_ids)).statement,
sql=session.query(Score).where(Score.category_id.in_(category_ids)).statement,
con=engine,
)