Update: Various Elections API changes, Part 2

src/elections/crud.py

@@ -1,11 +1,12 @@
+from collections.abc import Sequence
+
 import sqlalchemy
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from elections.tables import Election, NomineeApplication, NomineeInfo
+from elections.tables import Election
 
 
-async def get_all_elections(db_session: AsyncSession) -> list[Election]:
-    # TODO: can this return None?
+async def get_all_elections(db_session: AsyncSession) -> Sequence[Election]:
     election_list = (await db_session.scalars(
         sqlalchemy
         .select(Election)
@@ -46,100 +47,3 @@ async def delete_election(db_session: AsyncSession, slug: str) -> None:
         .delete(Election)
         .where(Election.slug == slug)
     )
-
-# ------------------------------------------------------- #
-
-# TODO: switch to only using one of application or registration
-async def get_all_registrations_of_user(
-    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)
-        )
-    )).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/models.py

@@ -1,33 +1,49 @@
 from enum import StrEnum
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
+
+from officers.constants import OfficerPositionEnum
+from registrations.models import RegistrationModel
 
 
 class ElectionTypeEnum(StrEnum):
     GENERAL = "general_election"
     BY_ELECTION = "by_election"
     COUNCIL_REP = "council_rep_election"
 
-class ElectionModel(BaseModel):
+class ElectionStatusEnum(StrEnum):
+    BEFORE_NOMINATIONS = "before_nominations"
+    NOMINATIONS = "nominations"
+    VOTING = "voting"
+    AFTER_VOTING = "after_voting"
+
+class ElectionResponse(BaseModel):
     slug: str
     name: str
     type: ElectionTypeEnum
     datetime_start_nominations: str
     datetime_start_voting: str
     datetime_end_voting: str
-    available_positions: str
+    available_positions: list[OfficerPositionEnum]
+    status: ElectionStatusEnum
+
+    survey_link: str | None = Field(None, description="Only available to admins")
+    candidates: list[RegistrationModel] | None = Field(None, description="Only available to admins")
+
+class ElectionParams(BaseModel):
+    name: str
+    type: ElectionTypeEnum
+    datetime_start_nominations: str
+    datetime_start_voting: str
+    datetime_end_voting: str
+    available_positions: list[OfficerPositionEnum] | None = None
+    survey_link: str | None = None
+
+class ElectionUpdateParams(BaseModel):
+    type: ElectionTypeEnum | None = None
+    datetime_start_nominations: str | None = None
+    datetime_start_voting: str | None = None
+    datetime_end_voting: str | None = None
+    available_positions: list[OfficerPositionEnum] | None = None
     survey_link: str | None = None
 
-class NomineeInfoModel(BaseModel):
-    computing_id: str
-    full_name: str
-    linked_in: str
-    instagram: str
-    email: str
-    discord_username: str
-
-class NomineeApplicationModel(BaseModel):
-    computing_id: str
-    nominee_election: str
-    position: str
-    speech: str

src/elections/tables.py

@@ -1,34 +1,19 @@
 from datetime import datetime
 
 from sqlalchemy import (
-    Column,
     DateTime,
-    ForeignKey,
-    PrimaryKeyConstraint,
     String,
-    Text,
 )
+from sqlalchemy.orm import Mapped, mapped_column
 
-from constants import (
-    COMPUTING_ID_LEN,
-    DISCORD_ID_LEN,
-    DISCORD_NAME_LEN,
-    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"
+from elections.models import (
+    ElectionStatusEnum,
+    ElectionTypeEnum,
+    ElectionUpdateParams,
+)
+from officers.constants import OfficerPositionEnum
+from utils.types import StringList
 
 MAX_ELECTION_NAME = 64
 MAX_ELECTION_SLUG = 64
@@ -37,16 +22,19 @@ class Election(Base):
     __tablename__ = "election"
 
     # Slugs are unique identifiers
-    slug = Column(String(MAX_ELECTION_SLUG), primary_key=True)
-    name = Column(String(MAX_ELECTION_NAME), nullable=False)
-    type = Column(String(64), default="general_election")
-    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))
+    slug: Mapped[str] = mapped_column(String(MAX_ELECTION_SLUG), primary_key=True)
+    name: Mapped[str] = mapped_column(String(MAX_ELECTION_NAME), nullable=False)
+    type: Mapped[ElectionTypeEnum] = mapped_column(String(64), default=ElectionTypeEnum.GENERAL)
+    datetime_start_nominations: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+    datetime_start_voting: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+    datetime_end_voting: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+
+    # a comma-separated string of positions which must be elements of OfficerPosition
+    # By giving it the type `StringList`, the database entry will automatically be marshalled to the correct form
+    # DB -> Python: str -> list[str]
+    # Python -> DB: list[str] -> str
+    available_positions: Mapped[list[OfficerPositionEnum]] = mapped_column(StringList(), nullable=False,)
+    survey_link: Mapped[str | None] = mapped_column(String(300))
 
     def private_details(self, at_time: datetime) -> dict:
         # is serializable
@@ -107,81 +95,27 @@ def to_update_dict(self) -> dict:
             "survey_link": self.survey_link,
         }
 
+    def update_from_params(self, params: ElectionUpdateParams):
+        update_data = params.model_dump(
+            exclude_unset=True,
+            exclude={"datetime_start_nominations", "datetime_start_voting", "datetime_end_voting"}
+        )
+        for k, v in update_data.items():
+            setattr(self, k, v)
+        if params.datetime_start_nominations:
+            self.datetime_start_nominations = datetime.fromisoformat(params.datetime_start_nominations)
+        if params.datetime_start_voting:
+            self.datetime_start_voting = datetime.fromisoformat(params.datetime_start_voting)
+        if params.datetime_end_voting:
+            self.datetime_end_voting = datetime.fromisoformat(params.datetime_end_voting)
+
     def status(self, at_time: datetime) -> str:
         if at_time <= self.datetime_start_nominations:
-            return STATUS_BEFORE_NOMINATIONS
+            return ElectionStatusEnum.BEFORE_NOMINATIONS
         elif at_time <= self.datetime_start_voting:
-            return STATUS_NOMINATIONS
+            return ElectionStatusEnum.NOMINATIONS
         elif at_time <= self.datetime_end_voting:
-            return STATUS_VOTING
+            return ElectionStatusEnum.VOTING
         else:
-            return STATUS_AFTER_VOTING
-
-class NomineeInfo(Base):
-    __tablename__ = "election_nominee_info"
-
-    computing_id = Column(String(COMPUTING_ID_LEN), primary_key=True)
-    full_name = Column(String(64), nullable=False)
-    linked_in = Column(String(128))
-    instagram = Column(String(128))
-    email = Column(String(64))
-    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__ = "election_nominee_application"
-
-    # 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)
-
-    __table_args__ = (
-        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,
-        }
+            return ElectionStatusEnum.AFTER_VOTING