Feat: Swap Election to use new models.

Address many pylance errors. Rewrite a few functions, will require testing to ensure
functionality remains the same.

.vscode/settings.json

@@ -8,7 +8,9 @@
         "CLANLEADER",
         "Dakiny",
         "DISCORDMANAGER",
+        "gitea",
         "highcouncil",
+        "imcompiling",
         "nextcord",
         "ondelete"
     ]

dakiny_bot/classes/__init__.py

@@ -0,0 +1 @@
+"""Empty init file for classes module."""

dakiny_bot/classes/election.py

@@ -5,17 +5,17 @@ from datetime import UTC, datetime, timedelta
 from pathlib import Path
 
 import nextcord
+from sqlalchemy import select, update
+from sqlalchemy.orm import Session
+from sqlmodel import col
+
 from dakiny_bot.classes.nominate_highcouncil_views import NominateHighCouncilView
 from dakiny_bot.classes.nominate_views import NominateView
 from dakiny_bot.classes.voting_highcouncil_views import VoteHighcouncilView
 from dakiny_bot.classes.voting_views import VoteView
 from dakiny_bot.dakiny_settings import settings
-from dakiny_bot.database.sql_data_classes import (
+from dakiny_bot.database.db import engine
-    CouncilMembers,
+from dakiny_bot.database.models import CouncilMember, ElectionInformation, Vote
-    ElectionInformation,
-    Votes,
-    engine,
-)
 from dakiny_bot.logger import dakiny_logger
 from dakiny_bot.utils import (
     TEMPLATES_DIR,
@@ -26,8 +26,6 @@ from dakiny_bot.utils import (
     get_datetime,
     get_election_name,
 )
-from sqlalchemy import select, true, update
-from sqlalchemy.orm import Session
 
 REQUIRED_VOTES_TO_BE_ELECTED = 3
 
@@ -70,7 +68,7 @@ class Election:
             list[str]: A list of message contents in the correct order. Must be a string because the message is too
             long.
         """
-        with Path(f"{TEMPLATES_DIR}/election_announcement.txt", encoding="utf8").open() as file:
+        with Path(f"{TEMPLATES_DIR}/election_announcement.txt").open() as file:
             message = file.read()
 
             next_election = get_datetime(
@@ -98,7 +96,7 @@ class Election:
         Returns:
             str: Message to send.
         """
-        with Path(f"{TEMPLATES_DIR}/highcouncil_election.txt", encoding="utf8").open() as file:
+        with Path(f"{TEMPLATES_DIR}/highcouncil_election.txt").open() as file:
             message = file.read()
 
             return message.format(
@@ -116,7 +114,7 @@ class Election:
         Returns:
             str: Message to send.
         """
-        with Path(f"{TEMPLATES_DIR}/highcouncil_vote.txt", encoding="utf8").open() as file:
+        with Path(f"{TEMPLATES_DIR}/highcouncil_vote.txt").open() as file:
             message = file.read()
 
             return message.format(
@@ -131,7 +129,7 @@ class Election:
         Returns:
             str: Message to send
         """
-        with Path(f"{TEMPLATES_DIR}/vote_start.txt", encoding="utf8").open() as file:
+        with Path(f"{TEMPLATES_DIR}/vote_start.txt").open() as file:
             message = file.read()
             vote_end: datetime = get_datetime(settings.election_cycles[self.cycle].voting_end)
 
@@ -171,7 +169,10 @@ class Election:
     async def election_start_action(self) -> None:
         """Performs required actions for when the election cycle starts."""
         # create new channel for election
-        parent: nextcord.TextChannel = self.guild.get_channel(settings.channel_ids.voting)
+        parent = self.guild.get_channel(settings.channel_ids.voting)
+        if not isinstance(parent, nextcord.TextChannel):
+            dakiny_logger.error("Voting channel (id: %s) is not a text channel.", settings.channel_ids.voting)
+            return
         quarter_name: str = get_election_name()
 
         thread_msg: nextcord.Message = await parent.send(content=quarter_name)
@@ -192,9 +193,9 @@ class Election:
         # create new entry into database.
         with Session(engine) as session:
             new_cycle = ElectionInformation(
-                ElectionName=quarter_name,
+                name=quarter_name,
-                ThreadId=thread.id,
+                thread_id=thread.id,
-                CouncilNominationMsgId=nominate_msg.id,
+                council_nomination_msg_id=nominate_msg.id,
             )
             session.add(new_cycle)
             session.commit()
@@ -207,64 +208,98 @@ class Election:
         Removes write permission from the general clan member role.
         """
         message = self.build_vote_start()
-        election_thread: nextcord.Thread = self.guild.get_thread(self.get_election_info().ThreadId)
+
+        with Session(engine) as session:
+            stmt = select(ElectionInformation).where(col(ElectionInformation.name == get_election_name()))
+            if (election_info := session.scalar(stmt)) is None:
+                dakiny_logger.error("Unable to find election %s", get_election_name())
+                return
+            if (election_thread := self.guild.get_thread(election_info.thread_id)) is None:
+                dakiny_logger.error("Recorded thread id %s does not result in a thread.", election_info.thread_id)
+                return
+
             await election_thread.send(message)
+
             vote_msg = await election_thread.send(
                 build_votes_str(self.guild),
                 view=VoteView(self.guild),
             )
 
-        with Session(engine) as session:
+            election_info.council_vote_msg_id = vote_msg.id
-            stmt = select(ElectionInformation).where(ElectionInformation.ElectionName == get_election_name())
-            info = session.scalar(stmt)
-            info.CouncilVoteMsgId = vote_msg.id
             session.commit()
 
-            thread: nextcord.Thread = self.guild.get_thread(info.ThreadId)
+            if election_info.council_nomination_msg_id is None:
-            nominate_msg: nextcord.Message = await thread.fetch_message(info.CouncilNominationMsgId)
+                dakiny_logger.error("Election doesn't have a nomination message set.")
+                return
+
+            nominate_msg: nextcord.Message = await election_thread.fetch_message(
+                election_info.council_nomination_msg_id
+            )
             await nominate_msg.edit(view=None)
 
     async def voting_end_action(self) -> None:
         """Performs required actions for when voting time has ended."""
         message = self.build_highcouncil_election_announce()
-        council_elect_role: nextcord.Role = self.guild.get_role(settings.role_ids.clan_council_elect)
+        if (council_elect_role := self.guild.get_role(settings.role_ids.clan_council_elect)) is None:
-        election_thread: nextcord.Thread = self.guild.get_thread(self.get_election_info().ThreadId)
+            dakiny_logger.error("Clan council elect role not found (id: %s).", settings.role_ids.clan_council_elect)
-
+            return
-        nominate_msg: nextcord.Message = await election_thread.send(message, view=NominateHighCouncilView(self.guild))
 
         election_name = get_election_name()
 
         with Session(engine) as session:
+            stmt = select(ElectionInformation).where(col(ElectionInformation.name == election_name))
+            if (info := session.scalar(stmt)) is None:
+                dakiny_logger.error("Unable to find election %s", election_name)
+                return
+
+            if (election_thread := self.guild.get_thread(info.thread_id)) is None:
+                dakiny_logger.error("Recorded thread id %s does not result in a thread.", info.thread_id)
+                return
+
+            nominate_msg: nextcord.Message = await election_thread.send(
+                message, view=NominateHighCouncilView(self.guild)
+            )
+
             # remove vote button from vote message
-            stmt = select(ElectionInformation).where(ElectionInformation.ElectionName == election_name)
+            if not info.council_vote_msg_id:
-            info = session.scalar(stmt)
+                dakiny_logger.error("Council vote message id not set.")
-            vote_msg: nextcord.Message = await election_thread.fetch_message(info.CouncilVoteMsgId)
+                return
+            vote_msg: nextcord.Message = await election_thread.fetch_message(info.council_vote_msg_id)
             await vote_msg.edit(view=None)
 
             # add the highcouncil nomination message to election information
-            info.HighCouncilNominationMsgId = nominate_msg.id
+            info.high_council_nomination_msg_id = nominate_msg.id
             session.commit()
 
             # add the highcouncil role to everyone with 3 or more votes. set elected to true in database
             stmt = (
-                select(Votes.VotedForId)
+                select(col(Vote.voted_for_id))
-                .where(Votes.ElectionName == election_name)
+                .where(col(Vote.election_name == election_name))
-                .where(Votes.Type == VoteType.COUNCIL)
+                .where(col(Vote.type == VoteType.COUNCIL))
             )
             vote_count = collections.Counter(session.scalars(stmt))
 
             for user_id, votes in vote_count.items():
+                if user_id is None:
+                    # Abstained votes, ignore
+                    continue
                 if votes >= REQUIRED_VOTES_TO_BE_ELECTED:
                     stmt = (
-                        select(CouncilMembers)
+                        select(CouncilMember)
-                        .where(CouncilMembers.ElectionName == election_name)
+                        .where(col(CouncilMember.election_name == election_name))
-                        .where(CouncilMembers.MemberId == user_id)
+                        .where(col(CouncilMember.member_id == user_id))
                     )
-                    council_member = session.scalar(stmt)
+                    if (council_member := session.scalar(stmt)) is None:
-                    council_member.Elected = True
+                        dakiny_logger.error(
+                            "Council member (id: %s) for election %s not found.", user_id, election_name
+                        )
+                        return
+                    council_member.elected = True
                     session.commit()
 
-                    member: nextcord.Member = self.guild.get_member(user_id)
+                    if (member := self.guild.get_member(user_id)) is None:
+                        dakiny_logger.error("Member (id: %s) not found in guild.", user_id)
+                        return
 
                     await member.add_roles(
                         council_elect_role,
@@ -274,42 +309,58 @@ class Election:
     async def highcouncil_election_start_action(self) -> None:
         """Performs required actions for when the highcouncil is elected."""
         message = self.build_highcouncil_vote_start()
-        election_thread: nextcord.Thread = self.guild.get_thread(self.get_election_info().ThreadId)
+        with Session(engine) as session:
+            stmt = select(ElectionInformation).where(col(ElectionInformation.name == get_election_name()))
+            if (info := session.scalar(stmt)) is None:
+                dakiny_logger.error("Unknown election name %s.", get_election_name())
+                return
+
+            if (election_thread := self.guild.get_thread(info.thread_id)) is None:
+                dakiny_logger.error("Recorded thread id %s does not result in a thread.", info.thread_id)
+                return
+
             await election_thread.send(message)
             vote_msg = await election_thread.send(
                 build_highcouncil_votes_str(self.guild),
                 view=VoteHighcouncilView(self.guild),
             )
 
-        with Session(engine) as session:
+            info.high_council_vote_msg_id = vote_msg.id
-            stmt = select(ElectionInformation).where(ElectionInformation.ElectionName == get_election_name())
-            info = session.scalar(stmt)
-            info.HighCouncilVoteMsgId = vote_msg.id
             session.commit()
 
             # remove highcouncil nomination view
-            nomination_msg: nextcord.Message = await election_thread.fetch_message(info.HighCouncilNominationMsgId)
+            if info.high_council_nomination_msg_id is None:
+                dakiny_logger.error("Election doesn't have a nomination message set.")
+                return
+
+            nomination_msg: nextcord.Message = await election_thread.fetch_message(info.high_council_nomination_msg_id)
             await nomination_msg.edit(view=None)
 
     async def term_start_action(self) -> None:
         """Marks the election as over."""
         self.election_over = True
 
-        elect_role: nextcord.Role = self.guild.get_role(settings.role_ids.clan_council_elect)
+        if (elect_role := self.guild.get_role(settings.role_ids.clan_council_elect)) is None:
-        council_role: nextcord.Role = self.guild.get_role(settings.role_ids.clan_council)
+            dakiny_logger.error("Cannot find Elect Role.")
+            return
+        if (council_role := self.guild.get_role(settings.role_ids.clan_council)) is None:
+            dakiny_logger.error("Cannot find Council Role.")
+            return
 
         election_name = get_election_name()
 
         with Session(engine) as session:
             # role management
             stmt = (
-                select(CouncilMembers.MemberId)
+                select(col(CouncilMember.member_id))
-                .where(CouncilMembers.ElectionName == election_name)
+                .where(col(CouncilMember.election_name == election_name))
-                .where(CouncilMembers.Elected == true())
+                .where(col(CouncilMember.elected))
             )
 
             for member_id in session.scalars(stmt):
-                member = self.guild.get_member(member_id)
+                if (member := self.guild.get_member(member_id)) is None:
+                    dakiny_logger.warning("Cannot find member (id: %s) to manage roles.", member_id)
+                    continue
                 await member.remove_roles(
                     elect_role,
                     reason=f"{member.display_name} no longer needs the elect role.",
@@ -322,35 +373,36 @@ class Election:
             # announce high council members
             # ties are undefined behavior.
             stmt = (
-                select(Votes.VotedForId)
+                select(col(Vote.voted_for_id))
-                .where(Votes.ElectionName == election_name)
+                .where(col(Vote.election_name == election_name))
-                .where(Votes.Type == VoteType.DISCORDMANAGER)
+                .where(col(Vote.type == VoteType.DISCORDMANAGER))
             )
             dm_votes = collections.Counter(session.scalars(stmt))
             stmt = (
-                select(Votes.VotedForId)
+                select(col(Vote.voted_for_id))
-                .where(Votes.ElectionName == election_name)
+                .where(col(Vote.election_name == election_name))
-                .where(Votes.Type == VoteType.CLANLEADER)
+                .where(col(Vote.type == VoteType.CLANLEADER))
             )
             cl_votes = collections.Counter(session.scalars(stmt))
+            cl_votes.most_common()
 
             try:
-                dm_winner = max(dm_votes, key=dm_votes.get)
+                dm_winner = dm_votes.most_common(1).pop()[0]
                 stmt = (
-                    update(CouncilMembers)
+                    update(CouncilMember)
-                    .where(CouncilMembers.ElectionName == election_name)
+                    .where(col(CouncilMember.election_name == election_name))
-                    .where(CouncilMembers.MemberId == dm_winner)
+                    .where(col(CouncilMember.member_id == dm_winner))
                     .values(DiscordManager=True)
                 )
                 session.execute(stmt)
             except ValueError:
                 dm_winner = "No winner"
             try:
-                cl_winner = max(cl_votes, key=cl_votes.get)
+                cl_winner = cl_votes.most_common(1).pop()[0]
                 stmt = (
-                    update(CouncilMembers)
+                    update(CouncilMember)
-                    .where(CouncilMembers.ElectionName == election_name)
+                    .where(col(CouncilMember.election_name == election_name))
-                    .where(CouncilMembers.MemberId == cl_winner)
+                    .where(col(CouncilMember.member_id == cl_winner))
                     .values(ClanLeader=True)
                 )
                 session.execute(stmt)
@@ -359,14 +411,21 @@ class Election:
 
             session.commit()
 
-        election_thread: nextcord.Thread = self.guild.get_thread(self.get_election_info().ThreadId)
+        stmt = select(ElectionInformation).where(col(ElectionInformation.name == election_name))
+        if (info := session.scalar(stmt)) is None:
+            dakiny_logger.error("Could not find information for this election (%s).", election_name)
+            return
+
+        if (election_thread := self.guild.get_thread(info.thread_id)) is None:
+            dakiny_logger.error("Recorded thread id %s does not result in a thread.", info.thread_id)
+            return
+
+        if info.high_council_vote_msg_id is None:
+            dakiny_logger.error("No recorded High Council Voting Message.")
+            return
 
         # clear view from vote message
-        stmt = select(ElectionInformation.HighCouncilVoteMsgId).where(
+        vote_msg: nextcord.Message = await election_thread.fetch_message(info.high_council_vote_msg_id)
-            ElectionInformation.ElectionName == get_election_name()
-        )
-        vote_msg_id = session.scalar(stmt)
-        vote_msg: nextcord.Message = await election_thread.fetch_message(vote_msg_id)
         await vote_msg.edit(view=None)
 
         highcouncil_msg: str = f"Congratulations for our high council members!\nClan Leader: <@{cl_winner}>\n"
@@ -378,22 +437,23 @@ class Election:
         await election_thread.edit(archived=True, locked=True)
         await edit_clan_council_info(self.guild)
 
-    def get_election_info(self) -> ElectionInformation:
-        """
-        Gets information for current election.
-
-        Returns:
-            ElectionInformation: The information from the database about the current election.
-        """
-        with Session(engine) as session:
-            stmt = select(ElectionInformation).where(ElectionInformation.ElectionName == get_election_name())
-            return session.scalar(stmt)
-
     async def election_reminder(self) -> None:
         """Sends a message to general chat reminder clan about election."""
-        general: nextcord.TextChannel = self.guild.get_channel(settings.channel_ids.general)
+        general = self.guild.get_channel(settings.channel_ids.general)
+
+        if not isinstance(general, nextcord.TextChannel):
+            dakiny_logger.error(
+                "Recorded general channel (id: %s) is not a text channel.", settings.channel_ids.general
+            )
+            return
+
+        with Session(engine) as session:
+            stmt = select(ElectionInformation).where(col(ElectionInformation.name == get_election_name()))
+            if (info := session.scalar(stmt)) is None:
+                dakiny_logger.error("Could not find election %s.", get_election_name())
+                return
 
             message = f"Reminder to <@&{settings.role_ids.clan_member}>! Clan elections are happening! Head over "
-        message += f"to <#{self.get_election_info().ThreadId}> to cast your vote!"
+            message += f"to <#{info.thread_id}> to cast your vote!"
 
             await general.send(content=message)

dakiny_bot/classes/pyproject.toml

@@ -4,7 +4,3 @@ extend = "../../pyproject.toml"
 
 # Exclude a variety of commonly ignored directories.
 exclude = ["gitea_webhook.py"]
-
-[tool.ruff.lint]
-# ignore implicit-namespace-package in this directory.
-ignore = ["INP001"]