Update: Various Elections API changes, Part 2

src/elections/crud.py

@@ -1,11 +1,13 @@
+from collections.abc import Sequence
+
 import sqlalchemy
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from elections.tables import Election, NomineeApplication, NomineeInfo
+from officers.types import OfficerPositionEnum
 
 
-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)
@@ -54,7 +56,7 @@ async def get_all_registrations_of_user(
     db_session: AsyncSession,
     computing_id: str,
     election_slug: str
-) -> list[NomineeApplication] | None:
+) -> Sequence[NomineeApplication] | None:
     registrations = (await db_session.scalars(
         sqlalchemy
         .select(NomineeApplication)
@@ -65,10 +67,27 @@ async def get_all_registrations_of_user(
     )).all()
     return registrations
 
+async def get_one_registration_in_election(
+    db_session: AsyncSession,
+    computing_id: str,
+    election_slug: str,
+    position: OfficerPositionEnum,
+) -> NomineeApplication | None:
+    registration = (await db_session.scalar(
+        sqlalchemy
+        .select(NomineeApplication)
+        .where(
+            NomineeApplication.computing_id == computing_id,
+            NomineeApplication.nominee_election == election_slug,
+            NomineeApplication.position == position
+        )
+    ))
+    return registration
+
 async def get_all_registrations_in_election(
     db_session: AsyncSession,
     election_slug: str,
-) -> list[NomineeApplication] | None:
+) -> Sequence[NomineeApplication] | None:
     registrations = (await db_session.scalars(
         sqlalchemy
         .select(NomineeApplication)
@@ -103,7 +122,7 @@ async def delete_registration(
     db_session: AsyncSession,
     computing_id: str,
     election_slug: str,
-    position: str
+    position: OfficerPositionEnum
 ):
     await db_session.execute(
         sqlalchemy

src/elections/models.py

@@ -1,23 +1,74 @@
 from enum import StrEnum
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
+
+from officers.types import OfficerPositionEnum
 
 
 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 CandidateModel(BaseModel):
+    position: str
+    full_name: str
+    linked_in: str
+    instagram: str
+    email: str
+    discord_username: str
+    speech: str
+
+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[str]
+    status: ElectionStatusEnum
+
+    survey_link: str | None = Field(None, description="Only available to admins")
+    candidates: list[CandidateModel] | 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[str] | 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[str] | None = None
+    survey_link: str | None = None
+
+class NomineeApplicationParams(BaseModel):
+    computing_id: str
+    position: OfficerPositionEnum
+
+class NomineeApplicationUpdateParams(BaseModel):
+    position: OfficerPositionEnum | None = None
+    speech: str | None = None
+
+class NomineeApplicationModel(BaseModel):
+    computing_id: str
+    nominee_election: str
+    position: str
+    speech: str | None = None
+
 class NomineeInfoModel(BaseModel):
     computing_id: str
     full_name: str
@@ -26,8 +77,10 @@ class NomineeInfoModel(BaseModel):
     email: str
     discord_username: str
 
-class NomineeApplicationModel(BaseModel):
-    computing_id: str
-    nominee_election: str
-    position: str
-    speech: str
+class NomineeInfoUpdateParams(BaseModel):
+    full_name: str | None = None
+    linked_in: str | None = None
+    instagram: str | None = None
+    email: str | None = None
+    discord_username: str | None = None
+

src/elections/tables.py

@@ -1,34 +1,26 @@
 from datetime import datetime
 
 from sqlalchemy import (
-    Column,
     DateTime,
     ForeignKey,
     PrimaryKeyConstraint,
     String,
     Text,
 )
+from sqlalchemy.ext.hybrid import hybrid_property
+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,
+    ElectionUpdateParams,
+    NomineeApplicationUpdateParams,
+)
+from officers.types import OfficerPositionEnum
 
 MAX_ELECTION_NAME = 64
 MAX_ELECTION_SLUG = 64
@@ -37,16 +29,26 @@ 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)
+    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[str] = mapped_column(String(64), default="general_election")
+    datetime_start_nominations: Mapped[datetime] = mapped_column(DateTime, nullable=False)
+    datetime_start_voting: Mapped[datetime] = mapped_column(DateTime, nullable=False)
+    datetime_end_voting: Mapped[datetime] = mapped_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))
+    _available_positions: Mapped[str] = mapped_column("available_positions", Text, nullable=False)
+    survey_link: Mapped[str | None] = mapped_column(String(300))
+
+    @hybrid_property
+    def available_positions(self) -> str: # pyright: ignore
+        return self._available_positions
+
+    @available_positions.setter
+    def available_positions(self, value: str | list[str]) -> None:
+        if isinstance(value, list):
+            value = ",".join(value)
+        self._available_positions = value
 
     def private_details(self, at_time: datetime) -> dict:
         # is serializable
@@ -60,7 +62,7 @@ def private_details(self, at_time: datetime) -> dict:
             "datetime_end_voting": self.datetime_end_voting.isoformat(),
 
             "status": self.status(at_time),
-            "available_positions": self.available_positions,
+            "available_positions": self._available_positions,
             "survey_link": self.survey_link,
         }
 
@@ -76,7 +78,7 @@ def public_details(self, at_time: datetime) -> dict:
             "datetime_end_voting": self.datetime_end_voting.isoformat(),
 
             "status": self.status(at_time),
-            "available_positions": self.available_positions,
+            "available_positions": self._available_positions,
         }
 
     def public_metadata(self, at_time: datetime) -> dict:
@@ -99,33 +101,47 @@ def to_update_dict(self) -> dict:
             "name": self.name,
             "type": self.type,
 
-            "datetime_start_nominations": self.datetime_start_nominations,
-            "datetime_start_voting": self.datetime_start_voting,
-            "datetime_end_voting": self.datetime_end_voting,
+            "datetime_start_nominations": self.datetime_start_nominations.date(),
+            "datetime_start_voting": self.datetime_start_voting.date(),
+            "datetime_end_voting": self.datetime_end_voting.date(),
 
-            "available_positions": self.available_positions,
+            "available_positions": self._available_positions,
             "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
+            return ElectionStatusEnum.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))
+    computing_id: Mapped[str] = mapped_column(String(COMPUTING_ID_LEN), primary_key=True)
+    full_name: Mapped[str] = mapped_column(String(64), nullable=False)
+    linked_in: Mapped[str] = mapped_column(String(128))
+    instagram: Mapped[str] = mapped_column(String(128))
+    email: Mapped[str] = mapped_column(String(64))
+    discord_username: Mapped[str] = mapped_column(String(DISCORD_NICKNAME_LEN))
 
     def to_update_dict(self) -> dict:
         return {
@@ -138,7 +154,7 @@ def to_update_dict(self) -> dict:
             "discord_username": self.discord_username,
         }
 
-    def as_serializable(self) -> dict:
+    def serialize(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
@@ -156,18 +172,17 @@ def as_serializable(self) -> dict:
 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)
+    computing_id: Mapped[str] = mapped_column(ForeignKey("election_nominee_info.computing_id"), primary_key=True)
+    nominee_election: Mapped[str] = mapped_column(ForeignKey("election.slug"), primary_key=True)
+    position: Mapped[OfficerPositionEnum] = mapped_column(String(64), primary_key=True)
 
-    speech = Column(Text)
+    speech: Mapped[str | None] = mapped_column(Text)
 
     __table_args__ = (
         PrimaryKeyConstraint(computing_id, nominee_election, position),
     )
 
-    def serializable_dict(self) -> dict:
+    def serialize(self) -> dict:
         return {
             "computing_id": self.computing_id,
             "nominee_election": self.nominee_election,
@@ -185,3 +200,9 @@ def to_update_dict(self) -> dict:
             "speech": self.speech,
         }
 
+    def update_from_params(self, params: NomineeApplicationUpdateParams):
+        update_data = params.model_dump(exclude_unset=True)
+        for k, v in update_data.items():
+            setattr(self, k, v)
+
+