Feat 144 grouped election endpoints

src/elections/crud.py

@@ -1,9 +1,13 @@
 from collections.abc import Sequence
+from datetime import datetime
 
 import sqlalchemy
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from candidates.tables import CandidateDB
+from elections.models import ElectionNomineeSummary, ElectionResponse
 from elections.tables import ElectionDB
+from nominees.tables import NomineeInfoDB
 
 
 async def get_all_elections(db_session: AsyncSession) -> Sequence[ElectionDB]:
@@ -37,3 +41,121 @@ 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_elections_with_nominees(
+    db_session: AsyncSession,
+    at_time: datetime,
+    has_permission: bool,
+) -> list[ElectionResponse]:
+    """
+    Fetches all elections and their nominees in a single query.
+    Returns a list of ElectionResponse with candidates embedded.
+    """
+    query = (
+        sqlalchemy.select(ElectionDB, CandidateDB, NomineeInfoDB)
+        .outerjoin(CandidateDB, ElectionDB.slug == CandidateDB.nominee_election)
+        .outerjoin(NomineeInfoDB, CandidateDB.computing_id == NomineeInfoDB.computing_id)
+    )
+
+    rows = (await db_session.execute(query)).all()
+
+    elections_by_slug = {}
+    for election, candidate, nominee_info in rows:
+        if election.slug not in elections_by_slug:
+            elections_by_slug[election.slug] = ElectionResponse(
+                slug=election.slug,
+                name=election.name,
+                type=election.type,
+                datetime_start_nominations=election.datetime_start_nominations,
+                datetime_start_voting=election.datetime_start_voting,
+                datetime_end_voting=election.datetime_end_voting,
+                available_positions=election.available_positions,
+                status=election.status(at_time),
+                survey_link=election.survey_link if has_permission else None,
+                candidates=[],
+            )
+
+        if (candidate is None) or (nominee_info is None) or (nominee_info.full_name is None):
+            continue
+
+        if has_permission:
+            nominee = ElectionNomineeSummary(
+                full_name=nominee_info.full_name,
+                position=candidate.position,
+                speech=candidate.speech or "No speech provided by this candidate",
+                computing_id=nominee_info.computing_id,
+                linked_in=nominee_info.linked_in,
+                instagram=nominee_info.instagram,
+                email=nominee_info.email,
+                discord_username=nominee_info.discord_username,
+            )
+        else:
+            nominee = ElectionNomineeSummary(
+                full_name=nominee_info.full_name,
+                position=candidate.position,
+                speech=candidate.speech or "No speech provided by this candidate",
+            )
+        elections_by_slug[election.slug].candidates.append(nominee)
+
+    return list(elections_by_slug.values())
+
+
+async def get_election_nominees(
+    db_session: AsyncSession,
+    election_slug: str,
+    has_permission: bool,
+) -> list[ElectionNomineeSummary]:
+    """
+    Fetches all nominees for a single election in one JOIN query.
+    Only fetches contact fields when has_permission is True.
+    """
+    if has_permission:
+        query = (
+            sqlalchemy.select(
+                CandidateDB.position,
+                CandidateDB.speech,
+                NomineeInfoDB.full_name,
+                NomineeInfoDB.computing_id,
+                NomineeInfoDB.linked_in,
+                NomineeInfoDB.instagram,
+                NomineeInfoDB.email,
+                NomineeInfoDB.discord_username,
+            )
+            .outerjoin(NomineeInfoDB, CandidateDB.computing_id == NomineeInfoDB.computing_id)
+            .where(CandidateDB.nominee_election == election_slug)
+        )
+    else:
+        query = (
+            sqlalchemy.select(
+                CandidateDB.position,
+                CandidateDB.speech,
+                NomineeInfoDB.full_name,
+            )
+            .outerjoin(NomineeInfoDB, CandidateDB.computing_id == NomineeInfoDB.computing_id)
+            .where(CandidateDB.nominee_election == election_slug)
+        )
+
+    rows = (await db_session.execute(query)).all()
+
+    candidates_list = []
+    for row in rows:
+        if has_permission:
+            candidate_entry = 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:
+            candidate_entry = ElectionNomineeSummary(
+                full_name=row.full_name,
+                position=row.position,
+                speech=row.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,38 @@ 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]
+    has_permission = await is_user_election_admin(computing_id, db_session)
+
+    if with_nominees:
+        election_responses = await elections.crud.get_all_elections_with_nominees(
+            db_session, current_time, has_permission
+        )
+        if not election_responses:
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="no election found")
+        election_metadata_list = [
+            election.model_dump(mode="json", exclude_none=True) for election in election_responses
+        ]
     else:
-        election_metadata_list = [election.public_details(current_time) for election in election_list]
+        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")
+        election_metadata_list = []
+        for election in election_list:
+            if has_permission:
+                election_metadata_list.append(election.private_details(current_time))
+            else:
+                election_metadata_list.append(election.public_details(current_time))
 
     return JSONResponse(election_metadata_list)
 
@@ -100,14 +114,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 +135,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.slug, 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/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())