wip: move Nominee Application stuff to its own directory

Author: jbriones1

src/elections/crud.py

@@ -3,8 +3,7 @@
 import sqlalchemy
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from elections.tables import Election, NomineeApplication, NomineeInfo
-from officers.constants import OfficerPositionEnum
+from elections.tables import Election, NomineeInfo
 
 
 async def get_all_elections(db_session: AsyncSession) -> Sequence[Election]:
@@ -51,91 +50,6 @@ async def delete_election(db_session: AsyncSession, slug: str) -> None:
 
 # ------------------------------------------------------- #
 
-# 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
-) -> Sequence[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_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,
-) -> Sequence[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: OfficerPositionEnum
-):
-    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,

src/elections/models.py

@@ -3,6 +3,7 @@
 from pydantic import BaseModel, Field
 
 from officers.constants import OfficerPositionEnum
+from registrations.models import RegistrationModel
 
 
 class ElectionTypeEnum(StrEnum):
@@ -16,15 +17,6 @@ class ElectionStatusEnum(StrEnum):
     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
@@ -36,7 +28,7 @@ class ElectionResponse(BaseModel):
     status: ElectionStatusEnum
 
     survey_link: str | None = Field(None, description="Only available to admins")
-    candidates: list[CandidateModel] | None = Field(None, description="Only available to admins")
+    candidates: list[RegistrationModel] | None = Field(None, description="Only available to admins")
 
 class ElectionParams(BaseModel):
     name: str
@@ -55,20 +47,6 @@ class ElectionUpdateParams(BaseModel):
     available_positions: list[OfficerPositionEnum] | 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: OfficerPositionEnum
-    speech: str | None = None
-
 class NomineeInfoModel(BaseModel):
     computing_id: str
     full_name: str

src/elections/tables.py

@@ -2,10 +2,7 @@
 
 from sqlalchemy import (
     DateTime,
-    ForeignKey,
-    PrimaryKeyConstraint,
     String,
-    Text,
 )
 from sqlalchemy.orm import Mapped, mapped_column
 
@@ -17,7 +14,6 @@
 from elections.models import (
     ElectionStatusEnum,
     ElectionUpdateParams,
-    NomineeApplicationUpdateParams,
 )
 from officers.constants import OfficerPositionEnum
 from utils.types import StringList
@@ -162,40 +158,3 @@ def serialize(self) -> dict:
             "discord_username": self.discord_username,
         }
 
-class NomineeApplication(Base):
-    __tablename__ = "election_nominee_application"
-
-    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: Mapped[str | None] = mapped_column(Text)
-
-    __table_args__ = (
-        PrimaryKeyConstraint(computing_id, nominee_election, position),
-    )
-
-    def serialize(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,
-        }
-
-    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)
-
-

src/elections/urls.py

@@ -4,26 +4,24 @@
 from fastapi.responses import JSONResponse
 
 import database
-import elections
 import elections.crud
 import elections.tables
+import registrations.crud
 from elections.models import (
     ElectionParams,
     ElectionResponse,
     ElectionTypeEnum,
     ElectionUpdateParams,
-    NomineeInfoModel,
-    NomineeInfoUpdateParams,
 )
-from elections.tables import Election, NomineeInfo
+from elections.tables import Election
 from officers.constants import COUNCIL_REP_ELECTION_POSITIONS, GENERAL_ELECTION_POSITIONS, OfficerPositionEnum
 from permission.types import ElectionOfficer, WebsiteAdmin
 from utils.shared_models import DetailModel, SuccessResponse
-from utils.urls import admin_or_raise, get_current_user, slugify
+from utils.urls import get_current_user, slugify
 
 router = APIRouter(
-    prefix="/elections",
-    tags=["elections"],
+    prefix="/election",
+    tags=["election"],
 )
 
 async def get_user_permissions(
@@ -34,7 +32,7 @@ async def get_user_permissions(
     if not session_id or not computing_id:
         return False, None, None
 
-    # where valid means elections officer or website admin
+    # where valid means election officer or website admin
     has_permission = await ElectionOfficer.has_permission(db_session, computing_id)
     if not has_permission:
         has_permission = await WebsiteAdmin.has_permission(db_session, computing_id)
@@ -84,14 +82,14 @@ def _raise_if_bad_election_data(
             detail=f"election slug '{slug}' is too long",
         )
 
-# elections ------------------------------------------------------------- #
+# election ------------------------------------------------------------- #
 
 @router.get(
     "",
-    description="Returns a list of all elections & their status",
+    description="Returns a list of all election & their status",
     response_model=list[ElectionResponse],
     responses={
-        404: { "description": "No elections found" }
+        404: { "description": "No election found" }
     },
     operation_id="get_all_elections"
 )
@@ -104,7 +102,7 @@ async def list_elections(
     if election_list is None or len(election_list) == 0:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
-            detail="no elections found"
+            detail="no election found"
         )
 
     current_time = datetime.now()
@@ -126,7 +124,7 @@ async def list_elections(
     description="""
     Retrieves the election data for an election by name.
     Returns private details when the time is allowed.
-    If user is an admin or elections officer, returns computing ids for each candidate as well.
+    If user is an admin or election officer, returns computing ids for each candidate as well.
     """,
     response_model=ElectionResponse,
     responses={
@@ -152,7 +150,7 @@ async def get_election(
     if current_time >= election.datetime_start_voting or is_valid_user:
 
         election_json = election.private_details(current_time)
-        all_nominations = await elections.crud.get_all_registrations_in_election(db_session, slugified_name)
+        all_nominations = await registrations.crud.get_all_registrations_in_election(db_session, slugified_name)
         if not all_nominations:
             raise HTTPException(
                 status_code=status.HTTP_404_NOT_FOUND,
@@ -364,93 +362,3 @@ async def delete_election(
 
     old_election = await elections.crud.get_election(db_session, slugified_name)
     return JSONResponse({"success": old_election is None})
-
-# nominee info ------------------------------------------------------------- #
-
-@router.get(
-    "/nominee/{computing_id:str}",
-    description="Nominee info is always publically tied to elections, so be careful!",
-    response_model=NomineeInfoModel,
-    responses={
-        404: { "description": "nominee doesn't exist" }
-    },
-    operation_id="get_nominee"
-)
-async def get_nominee_info(
-    request: Request,
-    db_session: database.DBSession,
-    computing_id: str
-):
-    # Putting this one behind the admin wall since it has contact information
-    await admin_or_raise(request, db_session)
-    nominee_info = await elections.crud.get_nominee_info(db_session, computing_id)
-    if nominee_info is None:
-        raise HTTPException(
-            status_code=status.HTTP_404_NOT_FOUND,
-            detail="nominee doesn't exist"
-        )
-
-    return JSONResponse(nominee_info.serialize())
-
-@router.patch(
-    "/nominee/{computing_id:str}",
-    description="Will create or update nominee info. Returns an updated copy of their nominee info.",
-    response_model=NomineeInfoModel,
-    responses={
-        500: { "description": "Failed to retrieve updated nominee." }
-    },
-    operation_id="update_nominee"
-)
-async def provide_nominee_info(
-    request: Request,
-    db_session: database.DBSession,
-    body: NomineeInfoUpdateParams,
-    computing_id: str
-):
-    # TODO: There needs to be a lot more validation here.
-    await admin_or_raise(request, db_session)
-
-    updated_data = {}
-    # Only update fields that were provided
-    if body.full_name is not None:
-        updated_data["full_name"] = body.full_name
-    if body.linked_in is not None:
-       updated_data["linked_in"] = body.linked_in
-    if body.instagram is not None:
-        updated_data["instagram"] = body.instagram
-    if body.email is not None:
-        updated_data["email"] = body.email
-    if body.discord_username is not None:
-        updated_data["discord_username"] = body.discord_username
-
-    existing_info = await elections.crud.get_nominee_info(db_session, computing_id)
-    # if not already existing, create it
-    if not existing_info:
-        # unpack dictionary and expand into NomineeInfo class
-        new_nominee_info = NomineeInfo(computing_id=computing_id, **updated_data)
-        # create a new nominee
-        await elections.crud.create_nominee_info(db_session, new_nominee_info)
-    # else just update the partial data
-    else:
-        merged_data = {
-            "computing_id": computing_id,
-            "full_name": existing_info.full_name,
-            "linked_in": existing_info.linked_in,
-            "instagram": existing_info.instagram,
-            "email": existing_info.email,
-            "discord_username": existing_info.discord_username,
-        }
-        #  update the dictionary with new data
-        merged_data.update(updated_data)
-        updated_nominee_info = NomineeInfo(**merged_data)
-        await elections.crud.update_nominee_info(db_session, updated_nominee_info)
-
-    await db_session.commit()
-
-    nominee_info = await elections.crud.get_nominee_info(db_session, computing_id)
-    if not nominee_info:
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail="failed to get updated nominee"
-        )
-    return JSONResponse(nominee_info.serialize())