Second Elections PR (#105)

* add urls.py functions for registration, add empty crud functions, update table names

* add election status & getting all elections

* add position to application pkey & implement crud

* add avaliable positions & make it configurable. clean code too

* registrations can only be made during the nomination period, and check if position is active

* add support for nominee info

* update names

* fix foreign key bug

* fix db null value issue when resetting test database

* fix sqlalchemy syntax error

* add test cases for election endpoints

* add mock test data to testDB

* cleanup endpoint logic

* upgrade alembic head

* fix ruff linting issues

* fix test file linting

* add additional test cases and cleanup endpoints

* update endpoint urls

* linting fixes

* code cleanup

* minor word corrections

* minor logic tweak

* deprecate update_registration endpoints

* cleanup linting errors

* minor tweaks

* add back patch registration endpoint and uodate logic

* update endpoint logic

---------

Co-authored-by: p-north <[email protected]>

Author: EarthenSky

src/alembic/versions/243190df5588_create_election_tables.py

@@ -28,34 +28,33 @@ def upgrade() -> None:
         sa.Column("datetime_start_nominations", sa.DateTime(), nullable=False),
         sa.Column("datetime_start_voting", sa.DateTime(), nullable=False),
         sa.Column("datetime_end_voting", sa.DateTime(), nullable=False),
+        sa.Column("available_positions", sa.Text(), nullable=False),
         sa.Column("survey_link", sa.String(length=300), nullable=True),
         sa.PrimaryKeyConstraint("slug")
     )
     op.create_table(
-        "election_nominee",
+        "election_nominee_info",
         sa.Column("computing_id", sa.String(length=32), nullable=False),
         sa.Column("full_name", sa.String(length=64), nullable=False),
-        sa.Column("facebook", sa.String(length=128), nullable=True),
+        sa.Column("linked_in", sa.String(length=128), nullable=True),
         sa.Column("instagram", sa.String(length=128), nullable=True),
         sa.Column("email", sa.String(length=64), nullable=True),
-        sa.Column("discord", sa.String(length=32), nullable=True),
-        sa.Column("discord_id", sa.String(length=32), nullable=True),
         sa.Column("discord_username", sa.String(length=32), nullable=True),
         sa.PrimaryKeyConstraint("computing_id")
     )
     op.create_table(
-        "nominee_application",
+        "election_nominee_application",
         sa.Column("computing_id", sa.String(length=32), nullable=False),
-        sa.Column("nominee_election", sa.String(length=32), nullable=False),
-        sa.Column("speech", sa.Text(), nullable=True),
+        sa.Column("nominee_election", sa.String(length=64), nullable=False),
         sa.Column("position", sa.String(length=64), nullable=False),
-        sa.ForeignKeyConstraint(["computing_id"], ["election_nominee.computing_id"]),
+        sa.Column("speech", sa.Text(), nullable=True),
+        sa.ForeignKeyConstraint(["computing_id"], ["election_nominee_info.computing_id"]),
         sa.ForeignKeyConstraint(["nominee_election"], ["election.slug"]),
-        sa.PrimaryKeyConstraint("computing_id", "nominee_election")
+        sa.PrimaryKeyConstraint("computing_id", "nominee_election", "position")
     )
 
 
 def downgrade() -> None:
-    op.drop_table("nominee_application")
-    op.drop_table("election_nominee")
+    op.drop_table("election_nominee_application")
+    op.drop_table("election_nominee_info")
     op.drop_table("election")

src/elections/crud.py

@@ -3,49 +3,146 @@
 import sqlalchemy
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from elections.tables import Election
+from elections.tables import Election, NomineeApplication, NomineeInfo
 
 _logger = logging.getLogger(__name__)
 
+async def get_all_elections(db_session: AsyncSession) -> list[Election] | None:
+    # TODO: can this return None?
+    election_list = (await db_session.scalars(
+        sqlalchemy
+        .select(Election)
+    )).all()
+    return election_list
+
 async def get_election(db_session: AsyncSession, election_slug: str) -> Election | None:
     return await db_session.scalar(
         sqlalchemy
         .select(Election)
         .where(Election.slug == election_slug)
     )
 
-async def create_election(db_session: AsyncSession, election: Election) -> None:
+async def create_election(db_session: AsyncSession, election: Election):
     """
     Creates a new election with given parameters.
     Does not validate if an election _already_ exists
     """
     db_session.add(election)
 
+async def update_election(db_session: AsyncSession, new_election: Election):
+    """
+    Attempting to change slug will fail. Instead, you must create a new election.
+    """
+    await db_session.execute(
+        sqlalchemy
+        .update(Election)
+        .where(Election.slug == new_election.slug)
+        .values(new_election.to_update_dict())
+    )
+
 async def delete_election(db_session: AsyncSession, slug: str) -> None:
     """
-    Deletes a given election by its slug.
-    Does not validate if an election exists
+    Deletes a given election by its slug. Does not validate if an election exists
     """
     await db_session.execute(
         sqlalchemy
         .delete(Election)
         .where(Election.slug == slug)
     )
 
-async def update_election(db_session: AsyncSession, new_election: Election) -> bool:
-    """
-    You attempting to change the name or slug will fail. Instead, you must create a new election.
-    """
-    target_slug = new_election.slug
-    target_election = await get_election(db_session, target_slug)
-
-    if target_election is None:
-        return False
-    else:
-        await db_session.execute(
-            sqlalchemy
-            .update(Election)
-            .where(Election.slug == target_slug)
-            .values(new_election.to_update_dict())
+# ------------------------------------------------------- #
+
+# TODO: switch to only using one of application or registration
+async def get_all_registrations(
+    db_session: AsyncSession,
+    computing_id: str,
+    election_slug: str
+) -> list[NomineeApplication] | None:
+    registrations = (await db_session.scalars(
+        sqlalchemy
+        .select(NomineeApplication)
+        .where(
+            (NomineeApplication.computing_id == computing_id)
+            & (NomineeApplication.nominee_election == election_slug)
         )
-        return True
+    )).all()
+    return registrations
+
+async def get_all_registrations_in_election(
+    db_session: AsyncSession,
+    election_slug: str,
+) -> list[NomineeApplication] | None:
+    registrations = (await db_session.scalars(
+        sqlalchemy
+        .select(NomineeApplication)
+        .where(
+            NomineeApplication.nominee_election == election_slug
+        )
+    )).all()
+    return registrations
+
+async def add_registration(
+    db_session: AsyncSession,
+    initial_application: NomineeApplication
+):
+    db_session.add(initial_application)
+
+async def update_registration(
+    db_session: AsyncSession,
+    initial_application: NomineeApplication
+):
+    await db_session.execute(
+        sqlalchemy
+        .update(NomineeApplication)
+        .where(
+            (NomineeApplication.computing_id == initial_application.computing_id)
+            & (NomineeApplication.nominee_election == initial_application.nominee_election)
+            & (NomineeApplication.position == initial_application.position)
+        )
+        .values(initial_application.to_update_dict())
+    )
+
+async def delete_registration(
+    db_session: AsyncSession,
+    computing_id: str,
+    election_slug: str,
+    position: str
+):
+    await db_session.execute(
+        sqlalchemy
+        .delete(NomineeApplication)
+        .where(
+            (NomineeApplication.computing_id == computing_id)
+            & (NomineeApplication.nominee_election == election_slug)
+            & (NomineeApplication.position == position)
+        )
+    )
+
+# ------------------------------------------------------- #
+
+async def get_nominee_info(
+    db_session: AsyncSession,
+    computing_id: str,
+) -> NomineeInfo | None:
+    return await db_session.scalar(
+        sqlalchemy
+        .select(NomineeInfo)
+        .where(NomineeInfo.computing_id == computing_id)
+    )
+
+async def create_nominee_info(
+    db_session: AsyncSession,
+    info: NomineeInfo,
+):
+    db_session.add(info)
+
+async def update_nominee_info(
+    db_session: AsyncSession,
+    info: NomineeInfo,
+):
+    await db_session.execute(
+        sqlalchemy
+        .update(NomineeInfo)
+        .where(NomineeInfo.computing_id == info.computing_id)
+        .values(info.to_update_dict())
+    )

src/elections/tables.py

@@ -1,3 +1,5 @@
+from datetime import datetime
+
 from sqlalchemy import (
     Column,
     DateTime,
@@ -14,9 +16,20 @@
     DISCORD_NICKNAME_LEN,
 )
 from database import Base
+from officers.constants import COUNCIL_REP_ELECTION_POSITIONS, GENERAL_ELECTION_POSITIONS
 
+# If you wish to add more elections & defaults, please see `create_election`
 election_types = ["general_election", "by_election", "council_rep_election"]
 
+DEFAULT_POSITIONS_GENERAL_ELECTION = ",".join(GENERAL_ELECTION_POSITIONS)
+DEFAULT_POSITIONS_BY_ELECTION = ",".join(GENERAL_ELECTION_POSITIONS)
+DEFAULT_POSITIONS_COUNCIL_REP_ELECTION = ",".join(COUNCIL_REP_ELECTION_POSITIONS)
+
+STATUS_BEFORE_NOMINATIONS = "before_nominations"
+STATUS_NOMINATIONS = "nominations"
+STATUS_VOTING = "voting"
+STATUS_AFTER_VOTING = "after_voting"
+
 MAX_ELECTION_NAME = 64
 MAX_ELECTION_SLUG = 64
 
@@ -30,9 +43,13 @@ class Election(Base):
     datetime_start_nominations = Column(DateTime, nullable=False)
     datetime_start_voting = Column(DateTime, nullable=False)
     datetime_end_voting = Column(DateTime, nullable=False)
+
+    # a csv list of positions which must be elements of OfficerPosition
+    available_positions = Column(Text, nullable=False)
     survey_link = Column(String(300))
 
-    def serializable_dict(self) -> dict:
+    def private_details(self, at_time: datetime) -> dict:
+        # is serializable
         return {
             "slug": self.slug,
             "name": self.name,
@@ -42,10 +59,28 @@ def serializable_dict(self) -> dict:
             "datetime_start_voting": self.datetime_start_voting.isoformat(),
             "datetime_end_voting": self.datetime_end_voting.isoformat(),
 
+            "status": self.status(at_time),
+            "available_positions": self.available_positions,
             "survey_link": self.survey_link,
         }
 
-    def public_details(self) -> dict:
+    def public_details(self, at_time: datetime) -> dict:
+        # is serializable
+        return {
+            "slug": self.slug,
+            "name": self.name,
+            "type": self.type,
+
+            "datetime_start_nominations": self.datetime_start_nominations.isoformat(),
+            "datetime_start_voting": self.datetime_start_voting.isoformat(),
+            "datetime_end_voting": self.datetime_end_voting.isoformat(),
+
+            "status": self.status(at_time),
+            "available_positions": self.available_positions,
+        }
+
+    def public_metadata(self, at_time: datetime) -> dict:
+        # is serializable
         return {
             "slug": self.slug,
             "name": self.name,
@@ -54,6 +89,8 @@ def public_details(self) -> dict:
             "datetime_start_nominations": self.datetime_start_nominations.isoformat(),
             "datetime_start_voting": self.datetime_start_voting.isoformat(),
             "datetime_end_voting": self.datetime_end_voting.isoformat(),
+
+            "status": self.status(at_time),
         }
 
     def to_update_dict(self) -> dict:
@@ -66,31 +103,85 @@ def to_update_dict(self) -> dict:
             "datetime_start_voting": self.datetime_start_voting,
             "datetime_end_voting": self.datetime_end_voting,
 
+            "available_positions": self.available_positions,
             "survey_link": self.survey_link,
         }
 
-# Each row represents a nominee of a given election
-class Nominee(Base):
-    __tablename__ = "election_nominee"
+    def status(self, at_time: datetime) -> str:
+        if at_time <= self.datetime_start_nominations:
+            return STATUS_BEFORE_NOMINATIONS
+        elif at_time <= self.datetime_start_voting:
+            return STATUS_NOMINATIONS
+        elif at_time <= self.datetime_end_voting:
+            return STATUS_VOTING
+        else:
+            return STATUS_AFTER_VOTING
+
+class NomineeInfo(Base):
+    __tablename__ = "election_nominee_info"
 
-    # Previously named sfuid
     computing_id = Column(String(COMPUTING_ID_LEN), primary_key=True)
     full_name = Column(String(64), nullable=False)
-    facebook = Column(String(128))
+    linked_in = Column(String(128))
     instagram = Column(String(128))
     email = Column(String(64))
-    discord = Column(String(DISCORD_NAME_LEN))
-    discord_id = Column(String(DISCORD_ID_LEN))
     discord_username = Column(String(DISCORD_NICKNAME_LEN))
 
+    def to_update_dict(self) -> dict:
+        return {
+            "computing_id": self.computing_id,
+            "full_name": self.full_name,
+
+            "linked_in": self.linked_in,
+            "instagram": self.instagram,
+            "email": self.email,
+            "discord_username": self.discord_username,
+        }
+
+    def as_serializable(self) -> dict:
+        # NOTE: this function is currently the same as to_update_dict since the contents
+        # have a different invariant they're upholding, which may cause them to change if a
+        # new property is introduced. For example, dates must be converted into strings
+        # to be serialized, but must not for update dictionaries.
+        return {
+            "computing_id": self.computing_id,
+            "full_name": self.full_name,
+
+            "linked_in": self.linked_in,
+            "instagram": self.instagram,
+            "email": self.email,
+            "discord_username": self.discord_username,
+        }
+
 class NomineeApplication(Base):
-    __tablename__ = "nominee_application"
+    __tablename__ = "election_nominee_application"
 
-    computing_id = Column(ForeignKey("election_nominee.computing_id"), primary_key=True)
+    # TODO: add index for nominee_election?
+    computing_id = Column(ForeignKey("election_nominee_info.computing_id"), primary_key=True)
     nominee_election = Column(ForeignKey("election.slug"), primary_key=True)
+    position = Column(String(64), primary_key=True)
+
     speech = Column(Text)
-    position = Column(String(64), nullable=False)
 
     __table_args__ = (
-        PrimaryKeyConstraint(computing_id, nominee_election),
+        PrimaryKeyConstraint(computing_id, nominee_election, position),
     )
+
+    def serializable_dict(self) -> dict:
+        return {
+            "computing_id": self.computing_id,
+            "nominee_election": self.nominee_election,
+            "position": self.position,
+
+            "speech": self.speech,
+        }
+
+    def to_update_dict(self) -> dict:
+        return {
+            "computing_id": self.computing_id,
+            "nominee_election": self.nominee_election,
+            "position": self.position,
+
+            "speech": self.speech,
+        }
+