Second Elections PR

src/alembic/versions/243190df5588_create_election_tables.py

@@ -28,34 +28,35 @@ def upgrade() -> None:
         sa.Column("datetime_start_nominations", sa.DateTime(), nullable=False),
         sa.Column("datetime_start_voting", sa.DateTime(), nullable=False),
         sa.Column("datetime_end_voting", sa.DateTime(), nullable=False),
+        sa.Column("avaliable_positions", sa.Text(), nullable=False),
         sa.Column("survey_link", sa.String(length=300), nullable=True),
         sa.PrimaryKeyConstraint("slug")
     )
     op.create_table(
-        "election_nominee",
+        "election_nominee_info",
         sa.Column("computing_id", sa.String(length=32), nullable=False),
         sa.Column("full_name", sa.String(length=64), nullable=False),
-        sa.Column("facebook", sa.String(length=128), nullable=True),
+        sa.Column("linked_in", sa.String(length=128), nullable=True),
         sa.Column("instagram", sa.String(length=128), nullable=True),
         sa.Column("email", sa.String(length=64), nullable=True),
-        sa.Column("discord", sa.String(length=32), nullable=True),
-        sa.Column("discord_id", sa.String(length=32), nullable=True),
+        #sa.Column("discord", sa.String(length=32), nullable=True),
+        #sa.Column("discord_id", sa.String(length=32), nullable=True),
         sa.Column("discord_username", sa.String(length=32), nullable=True),
         sa.PrimaryKeyConstraint("computing_id")
     )
     op.create_table(
-        "nominee_application",
+        "election_nominee_application",
         sa.Column("computing_id", sa.String(length=32), nullable=False),
-        sa.Column("nominee_election", sa.String(length=32), nullable=False),
-        sa.Column("speech", sa.Text(), nullable=True),
+        sa.Column("nominee_election", sa.String(length=64), nullable=False),
         sa.Column("position", sa.String(length=64), nullable=False),
-        sa.ForeignKeyConstraint(["computing_id"], ["election_nominee.computing_id"]),
+        sa.Column("speech", sa.Text(), nullable=True),
+        sa.ForeignKeyConstraint(["computing_id"], ["election_nominee_info.computing_id"]),
         sa.ForeignKeyConstraint(["nominee_election"], ["election.slug"]),
-        sa.PrimaryKeyConstraint("computing_id", "nominee_election")
+        sa.PrimaryKeyConstraint("computing_id", "nominee_election", "position")
     )
 
 
 def downgrade() -> None:
-    op.drop_table("nominee_application")
-    op.drop_table("election_nominee")
+    op.drop_table("election_nominee_application")
+    op.drop_table("election_nominee_info")
     op.drop_table("election")

src/alembic/versions/46d14891e1a9_elections.py

@@ -0,0 +1,27 @@
+"""elections
+
+Revision ID: 46d14891e1a9
+Revises: 243190df5588
+Create Date: 2025-08-19 21:58:08.035067
+
+"""
+from collections.abc import Sequence
+from typing import Union
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "46d14891e1a9"
+down_revision: str | None = "243190df5588"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    pass

src/elections/crud.py

@@ -3,49 +3,146 @@
 import sqlalchemy
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from elections.tables import Election
+from elections.tables import Election, NomineeApplication, NomineeInfo
 
 _logger = logging.getLogger(__name__)
 
+async def get_all_elections(db_session: AsyncSession) -> list[Election] | None:
+    # TODO: can this return None?
+    election_list = (await db_session.scalars(
+        sqlalchemy
+        .select(Election)
+    )).all()
+    return election_list
+
 async def get_election(db_session: AsyncSession, election_slug: str) -> Election | None:
     return await db_session.scalar(
         sqlalchemy
         .select(Election)
         .where(Election.slug == election_slug)
     )
 
-async def create_election(db_session: AsyncSession, election: Election) -> None:
+async def create_election(db_session: AsyncSession, election: Election):
     """
     Creates a new election with given parameters.
     Does not validate if an election _already_ exists
     """
     db_session.add(election)
 
+async def update_election(db_session: AsyncSession, new_election: Election):
+    """
+    Attempting to change slug will fail. Instead, you must create a new election.
+    """
+    await db_session.execute(
+        sqlalchemy
+        .update(Election)
+        .where(Election.slug == new_election.slug)
+        .values(new_election.to_update_dict())
+    )
+
 async def delete_election(db_session: AsyncSession, slug: str) -> None:
     """
-    Deletes a given election by its slug.
-    Does not validate if an election exists
+    Deletes a given election by its slug. Does not validate if an election exists
     """
     await db_session.execute(
         sqlalchemy
         .delete(Election)
         .where(Election.slug == slug)
     )
 
-async def update_election(db_session: AsyncSession, new_election: Election) -> bool:
-    """
-    You attempting to change the name or slug will fail. Instead, you must create a new election.
-    """
-    target_slug = new_election.slug
-    target_election = await get_election(db_session, target_slug)
-
-    if target_election is None:
-        return False
-    else:
-        await db_session.execute(
-            sqlalchemy
-            .update(Election)
-            .where(Election.slug == target_slug)
-            .values(new_election.to_update_dict())
+# ------------------------------------------------------- #
+
+# 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:
+    registrations = (await db_session.scalars(
+        sqlalchemy
+        .select(NomineeApplication)
+        .where(
+            (NomineeApplication.computing_id == computing_id)
+            & (NomineeApplication.nominee_election == election_slug)
         )
-        return True
+    )).all()
+    return registrations
+
+async def get_all_registrations_in_election(
+    db_session: AsyncSession,
+    election_slug: str,
+) -> list[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: str
+):
+    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,
+) -> NomineeInfo | None:
+    return await db_session.scalar(
+        sqlalchemy
+        .select(NomineeInfo)
+        .where(NomineeInfo.computing_id == computing_id)
+    )
+
+async def create_nominee_info(
+    db_session: AsyncSession,
+    info: NomineeInfo,
+):
+    db_session.add(info)
+
+async def update_nominee_info(
+    db_session: AsyncSession,
+    info: NomineeInfo,
+):
+    await db_session.execute(
+        sqlalchemy
+        .update(NomineeInfo)
+        .where(NomineeInfo.computing_id == info.computing_id)
+        .values(info.to_update_dict())
+    )