Feat 144 grouped election endpoints

src/elections/crud.py

@@ -3,7 +3,12 @@
 import sqlalchemy
 from sqlalchemy.ext.asyncio import AsyncSession
 
+import candidates.crud
+import nominees.crud
+from candidates.tables import CandidateDB
+from elections.models import ElectionNomineeSummary
 from elections.tables import ElectionDB
+from nominees.tables import NomineeInfoDB
 
 
 async def get_all_elections(db_session: AsyncSession) -> Sequence[ElectionDB]:
@@ -37,3 +42,96 @@ async def delete_election(db_session: AsyncSession, slug: str) -> None:
     Deletes a given election by its slug. Does not validate if an election exists
     """
     await db_session.execute(sqlalchemy.delete(ElectionDB).where(ElectionDB.slug == slug))
+
+
+async def get_all_nominees_by_election(
+    db_session: AsyncSession,
+    has_permission: bool,
+) -> dict[str, list[ElectionNomineeSummary]]:
+    """
+    Fetches all nominees across all elections in a JOIN query.
+    Returns a dict mapping election slug -> list of ElectionNomineeSummary.
+    Only fetches contact fields (computing_id, linked_in, etc.) when has_permission is True.
+    """
+    if has_permission:
+        query = sqlalchemy.select(
+            CandidateDB.nominee_election,
+            CandidateDB.position,
+            CandidateDB.speech,
+            NomineeInfoDB.full_name,
+            NomineeInfoDB.computing_id,
+            NomineeInfoDB.linked_in,
+            NomineeInfoDB.instagram,
+            NomineeInfoDB.email,
+            NomineeInfoDB.discord_username,
+        ).join(NomineeInfoDB, CandidateDB.computing_id == NomineeInfoDB.computing_id)
+    else:
+        query = sqlalchemy.select(
+            CandidateDB.nominee_election,
+            CandidateDB.position,
+            CandidateDB.speech,
+            NomineeInfoDB.full_name,
+        ).join(NomineeInfoDB, CandidateDB.computing_id == NomineeInfoDB.computing_id)
+
+    rows = (await db_session.execute(query)).all()
+
+    nominees_by_election: dict[str, list[ElectionNomineeSummary]] = {}
+    for row in rows:
+        if has_permission:
+            nominee = ElectionNomineeSummary(
+                full_name=row.full_name,
+                position=row.position,
+                speech=row.speech or "No speech provided by this candidate",
+                computing_id=row.computing_id,
+                linked_in=row.linked_in,
+                instagram=row.instagram,
+                email=row.email,
+                discord_username=row.discord_username,
+            )
+        else:
+            nominee = ElectionNomineeSummary(
+                full_name=row.full_name,
+                position=row.position,
+                speech=row.speech or "No speech provided by this candidate",
+            )
+        if row.nominee_election not in nominees_by_election:
+            nominees_by_election[row.nominee_election] = []
+        nominees_by_election[row.nominee_election].append(nominee)
+
+    return nominees_by_election
+
+
+async def _get_election_nominees(
+    db_session: AsyncSession,
+    election_row: ElectionDB,
+    has_permission: bool,
+) -> list[ElectionNomineeSummary]:
+    candidates_list = []
+    all_nominations = await candidates.crud.get_all_candidates_in_election(db_session, election_row.slug)
+    if not all_nominations:
+        return []
+    for nomination in all_nominations:
+        # NOTE: if a nominee does not input their legal name, they are not considered a nominee
+        nominee_info = await nominees.crud.get_nominee_info(db_session, nomination.computing_id)
+        if nominee_info is None:
+            continue
+
+        if has_permission:
+            candidate_entry = ElectionNomineeSummary(
+                full_name=nominee_info.full_name,
+                position=nomination.position,
+                speech=nomination.speech or "No speech provided by this candidate",
+                computing_id=nomination.computing_id,
+                linked_in=nominee_info.linked_in,
+                instagram=nominee_info.instagram,
+                email=nominee_info.email,
+                discord_username=nominee_info.discord_username,
+            )
+        else:
+            candidate_entry = ElectionNomineeSummary(
+                full_name=nominee_info.full_name,
+                position=nomination.position,
+                speech=nomination.speech or "No speech provided by this candidate",
+            )
+        candidates_list.append(candidate_entry)
+    return candidates_list

src/elections/models.py

@@ -3,7 +3,6 @@
 
 from pydantic import BaseModel, Field
 
-from candidates.models import Candidate
 from officers.constants import OfficerPositionEnum
 
 
@@ -20,6 +19,17 @@ class ElectionStatusEnum(StrEnum):
     AFTER_VOTING = "after_voting"
 
 
+class ElectionNomineeSummary(BaseModel):
+    full_name: str
+    position: OfficerPositionEnum
+    speech: str
+    computing_id: str | None = None
+    linked_in: str | None = None
+    instagram: str | None = None
+    email: str | None = None
+    discord_username: str | None = None
+
+
 class ElectionResponse(BaseModel):
     slug: str
     name: str
@@ -32,7 +42,10 @@ class ElectionResponse(BaseModel):
 
     # Private fields
     survey_link: str | None = Field(None, description="Only available to admins")
-    candidates: list[Candidate] | None = Field(None, description="Only available to admins")
+    candidates: list[ElectionNomineeSummary] | None = Field(
+        None,
+        description="Included when with_nominees is true, contact fields only for election admins",
+    )
 
 
 class ElectionParams(BaseModel):

src/elections/urls.py

@@ -1,6 +1,6 @@
 import datetime
 
-from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi import APIRouter, Depends, HTTPException, Query, status
 from fastapi.responses import JSONResponse
 from sqlalchemy.exc import IntegrityError
 
@@ -74,24 +74,40 @@ def _raise_if_bad_election_data(
 
 @router.get(
     "",
-    description="Returns a list of all election & their status",
+    description="Return a list of all elections, their statuses and nominees (if requested)",
     response_model=list[ElectionResponse],
     responses={status.HTTP_404_NOT_FOUND: {"description": "No election found", "model": DetailModel}},
     operation_id="get_all_elections",
 )
 async def list_elections(
     computing_id: SessionUser,
     db_session: database.DBSession,
+    with_nominees: bool = Query(False),
 ):
     election_list = await elections.crud.get_all_elections(db_session)
     if election_list is None or len(election_list) == 0:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="no election found")
 
     current_time = datetime.datetime.now(datetime.UTC)
-    if await is_user_election_admin(computing_id, db_session):
-        election_metadata_list = [election.private_details(current_time) for election in election_list]
-    else:
-        election_metadata_list = [election.public_details(current_time) for election in election_list]
+    election_metadata_list = []
+    has_permission = await is_user_election_admin(computing_id, db_session)
+
+    nominees_by_election = {}
+    if with_nominees:
+        nominees_by_election = await elections.crud.get_all_nominees_by_election(db_session, has_permission)
+
+    for election in election_list:
+        if has_permission:
+            election_data = election.private_details(current_time)
+        else:
+            election_data = election.public_details(current_time)
+        if with_nominees:
+            election_nominees = nominees_by_election.get(election.slug, [])
+            candidates_list = []
+            for nominee in election_nominees:
+                candidates_list.append(nominee.model_dump(exclude_none=True))
+            election_data["candidates"] = candidates_list
+        election_metadata_list.append(election_data)
 
     return JSONResponse(election_metadata_list)
 
@@ -100,14 +116,18 @@ async def list_elections(
     "/{election_name}",
     description="""
     Retrieves the election data for an election by name.
-    Returns private details when the time is allowed.
-    If user is an admin or election officer, returns computing ids for each candidate as well.
+    If user is an admin or election officer, returns complete nominee information (if requested).
     """,
     response_model=ElectionResponse,
     responses={404: {"description": "Election of that name doesn't exist", "model": DetailModel}},
     operation_id="get_election_by_name",
 )
-async def get_election(db_session: database.DBSession, computing_id: SessionUser, election_name: str):
+async def get_election(
+    db_session: database.DBSession,
+    computing_id: SessionUser,
+    election_name: str,
+    with_nominees: bool = Query(False),
+):
     current_time = datetime.datetime.now(datetime.UTC)
     slugified_name = slugify(election_name)
     election = await elections.crud.get_election(db_session, slugified_name)
@@ -117,41 +137,16 @@ async def get_election(db_session: database.DBSession, computing_id: SessionUser
         )
 
     has_permission = await is_user_election_admin(computing_id, db_session)
-    if current_time >= election.datetime_start_voting or has_permission:
+    if has_permission:
         election_json = election.private_details(current_time)
-        all_nominations = await candidates.crud.get_all_candidates_in_election(db_session, slugified_name)
-        if not all_nominations:
-            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="no candidates found")
-        election_json["candidates"] = []
-
-        available_positions_list = election.available_positions
-        for nomination in all_nominations:
-            if nomination.position not in available_positions_list:
-                # ignore any positions that are **no longer** active
-                continue
-
-            # NOTE: if a nominee does not input their legal name, they are not considered a nominee
-            nominee_info = await nominees.crud.get_nominee_info(db_session, nomination.computing_id)
-            if nominee_info is None:
-                continue
-
-            candidate_entry = {
-                "position": nomination.position,
-                "full_name": nominee_info.full_name,
-                "linked_in": nominee_info.linked_in,
-                "instagram": nominee_info.instagram,
-                "email": nominee_info.email,
-                "discord_username": nominee_info.discord_username,
-                "speech": ("No speech provided by this candidate" if nomination.speech is None else nomination.speech),
-            }
-            if has_permission:
-                candidate_entry["computing_id"] = nomination.computing_id
-            election_json["candidates"].append(candidate_entry)
-
-        # after the voting period starts, all election data becomes public
-        return JSONResponse(election_json)
     else:
         election_json = election.public_details(current_time)
+    if with_nominees:
+        nominees = await elections.crud._get_election_nominees(db_session, election, has_permission)
+        candidates_list = []
+        for nominee in nominees:
+            candidates_list.append(nominee.model_dump(exclude_none=True))
+        election_json["candidates"] = candidates_list
 
     return JSONResponse(election_json)
 

src/nominees/urls.py

@@ -84,7 +84,7 @@ async def get_nominee_info(db_session: database.DBSession, computing_id: str):
     operation_id="delete_nominee",
     dependencies=[Depends(perm_election)],
 )
-async def delete_nominee_info(db_session: database.tDBSession, computing_id: str):
+async def delete_nominee_info(db_session: database.DBSession, computing_id: str):
     try:
         await nominees.crud.delete_nominee_info(db_session, computing_id)
         await db_session.commit()

src/officers/crud.py

@@ -177,7 +177,7 @@ async def get_active_officer_terms(
     )
     query = utils.is_active_officer(query)
     if positions:
-        query.where(OfficerTermDB.position.in_(positions))
+        query = query.where(OfficerTermDB.position.in_(positions))
 
     return list((await db_session.scalars(query)).all())