add position to application pkey & implement crud

Author: EarthenSky

src/alembic/versions/243190df5588_create_election_tables.py

@@ -47,11 +47,11 @@ def upgrade() -> None:
         "election_nominee_application",
         sa.Column("computing_id", sa.String(length=32), nullable=False),
         sa.Column("nominee_election", sa.String(length=64), nullable=False),
+        sa.Column("position", sa.String(length=64), nullable=False),
         sa.Column("speech", sa.Text(), nullable=True),
-        sa.Column("position", sa.String(length=64), nullable=True),
         sa.ForeignKeyConstraint(["computing_id"], ["election_nominee.computing_id"]),
         sa.ForeignKeyConstraint(["nominee_election"], ["election.slug"]),
-        sa.PrimaryKeyConstraint("computing_id", "nominee_election")
+        sa.PrimaryKeyConstraint("computing_id", "nominee_election", "position")
     )
 
 

src/elections/crud.py

@@ -34,6 +34,7 @@ async def update_election(db_session: AsyncSession, new_election: Election) -> b
     You attempting to change the name or slug will fail. Instead, you must create a new election.
     """
     target_slug = new_election.slug
+    # TODO: does this check need to be performed?
     target_election = await get_election(db_session, target_slug)
 
     if target_election is None:
@@ -58,35 +59,55 @@ async def delete_election(db_session: AsyncSession, slug: str) -> None:
         .where(Election.slug == slug)
     )
 
+# TODO: switch to only using one of application or registration
 async def get_all_registrations(
     db_session: AsyncSession,
     computing_id: str,
     election_slug: str
 ) -> list[NomineeApplication] | None:
-    raise NotImplementedError("todo")
+    registrations = (await db_session.scalars(
+        sqlalchemy
+        .select(NomineeApplication)
+        .where(
+            NomineeApplication.computing_id == computing_id
+            and NomineeApplication.election_slug == election_slug
+        )
+    )).all()
+    return registrations
 
 async def add_registration(
     db_session: AsyncSession,
     initial_application: NomineeApplication
-) -> NomineeApplication:
-    raise NotImplementedError("todo")
+):
+    db_session.add(initial_application)
 
 async def update_registration(
     db_session: AsyncSession,
     initial_application: NomineeApplication
-) -> NomineeApplication:
-    raise NotImplementedError("todo")
+):
+    await db_session.execute(
+        sqlalchemy
+        .update(NomineeApplication)
+        .where(
+            NomineeApplication.computing_id == initial_application.computing_id
+            and NomineeApplication.nominee_election == initial_application.nominee_election
+            and NomineeApplication.position == initial_application.position
+        )
+        .values(initial_application.to_update_dict())
+    )
 
 async def delete_registration(
     db_session: AsyncSession,
     computing_id: str,
-    election_slug: str
+    election_slug: str,
+    position: str
 ):
     await db_session.execute(
         sqlalchemy
         .delete(NomineeApplication)
         .where(
             NomineeApplication.computing_id == computing_id
             and NomineeApplication.nominee_election == election_slug
+            and NomineeApplication.position == position
         )
     )

src/elections/tables.py

@@ -109,17 +109,29 @@ class NomineeApplication(Base):
     # TODO: add index for nominee_election?
     computing_id = Column(ForeignKey("election_nominee.computing_id"), primary_key=True)
     nominee_election = Column(ForeignKey("election.slug"), primary_key=True)
+    position = Column(String(64), primary_key=True)
+
     speech = Column(Text)
-    position = Column(String(64))
 
     __table_args__ = (
-        PrimaryKeyConstraint(computing_id, nominee_election),
+        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,
         }
+

src/elections/urls.py

@@ -257,7 +257,7 @@ async def delete_election(
     "/register/{election_name:str}",
     description="get your election registration(s)"
 )
-async def get_election_registration(
+async def get_election_registrations(
     request: Request,
     db_session: database.DBSession,
     election_name: str
@@ -289,19 +289,25 @@ async def get_election_registration(
 
 @router.post(
     "/register/{election_name:str}",
-    description="register for the election, but doesn't set any speeches or positions."
+    description="register for a specific position in this election, but doesn't set a speech"
 )
 async def register_in_election(
     request: Request,
     db_session: database.DBSession,
-    election_name: str
+    election_name: str,
+    position: str
 ):
     logged_in, _, computing_id = await is_logged_in(request, db_session)
     if not logged_in:
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
             detail="must be logged in to register in election"
         )
+    elif position not in OfficerPosition.position_list():
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"invalid position {position}"
+        )
 
     election_slug = _slugify(election_name)
     if await get_election(db_session, election_slug) is None:
@@ -321,20 +327,20 @@ async def register_in_election(
     await elections.crud.add_registration(db_session, NomineeApplication(
         computing_id=computing_id,
         nominee_election=election_slug,
-        speech=None,
-        position=None,
+        position=position,
+        speech=None
     ))
 
 @router.patch(
     "/register/{election_name:str}",
-    description="update your registration for an election"
+    description="update your speech for a specific position for an election"
 )
 async def update_registration(
     request: Request,
     db_session: database.DBSession,
     election_name: str,
-    speech: str | None,
     position: str,
+    speech: str | None,
 ):
     logged_in, _, computing_id = await is_logged_in(request, db_session)
     if not logged_in:
@@ -344,7 +350,7 @@ async def update_registration(
         )
     elif position not in OfficerPosition.position_list():
         raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
+            status_code=status.HTTP_400_BAD_REQUEST,
             detail=f"invalid position {position}"
         )
 
@@ -363,25 +369,31 @@ async def update_registration(
     await elections.crud.update_registration(db_session, NomineeApplication(
         computing_id=computing_id,
         nominee_election=election_slug,
-        speech=speech,
         position=position,
+        speech=speech
     ))
 
 @router.delete(
     "/register/{election_name:str}",
-    description="revoke your registration in the election"
+    description="revoke your registration for a specific position in this election"
 )
 async def delete_registration(
     request: Request,
     db_session: database.DBSession,
-    election_name: str
+    election_name: str,
+    position: str,
 ):
     logged_in, _, computing_id = await is_logged_in(request, db_session)
     if not logged_in:
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
             detail="must be logged in to delete election registeration"
         )
+    elif position not in OfficerPosition.position_list():
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"invalid position {position}"
+        )
 
     election_slug = _slugify(election_name)
     if await get_election(db_session, election_slug) is None:
@@ -395,4 +407,4 @@ async def delete_registration(
             detail="you are not yet registered in this election"
         )
 
-    await elections.crud.delete_registration(db_session, computing_id, election_slug)
+    await elections.crud.delete_registration(db_session, computing_id, election_slug, position)