First Elections PR (#63)

* Added election tables revision to alembic

* Initial elections model implementation

* Added ElectionOfficer class, created has_permission() to check whether a user is a current elections officer
* Working commit containing fns to create, delete, and update elections

* Changed parameter orders to be consistent with other crud functions in the project. Changed database.DBSession -> AsyncSession to match other crud functions.

---------

Co-authored-by: Sean Chan <[email protected]>

Author: EarthenSky

src/alembic/env.py

@@ -8,6 +8,7 @@
 import auth.tables
 import blog.tables
 import database
+import elections.tables
 import officers.tables
 from alembic import context
 

src/alembic/versions/243190df5588_create_election_tables.py

@@ -0,0 +1,61 @@
+"""create election tables
+
+Revision ID: 243190df5588
+Revises: 43f71e4bd6fc
+Create Date: 2024-08-10 08:32:54.037614
+
+"""
+from collections.abc import Sequence
+from typing import Union
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "243190df5588"
+down_revision: str | None = "2a6ea95342dc"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "election",
+        sa.Column("slug", sa.String(length=64), nullable=False),
+        sa.Column("name", sa.String(length=64), nullable=False),
+        sa.Column("type", sa.String(length=64), default="general_election"),
+        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("survey_link", sa.String(length=300), nullable=True),
+        sa.PrimaryKeyConstraint("slug")
+    )
+    op.create_table(
+        "election_nominee",
+        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("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_username", sa.String(length=32), nullable=True),
+        sa.PrimaryKeyConstraint("computing_id")
+    )
+    op.create_table(
+        "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("position", sa.String(length=64), nullable=False),
+        sa.ForeignKeyConstraint(["computing_id"], ["election_nominee.computing_id"]),
+        sa.ForeignKeyConstraint(["nominee_election"], ["election.slug"]),
+        sa.PrimaryKeyConstraint("computing_id", "nominee_election")
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("nominee_application")
+    op.drop_table("election_nominee")
+    op.drop_table("election")

src/elections/crud.py

@@ -0,0 +1,51 @@
+import logging
+
+import sqlalchemy
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from elections.tables import Election
+
+_logger = logging.getLogger(__name__)
+
+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:
+    """
+    Creates a new election with given parameters.
+    Does not validate if an election _already_ exists
+    """
+    db_session.add(election)
+
+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(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())
+        )
+        return True

src/elections/tables.py

@@ -0,0 +1,96 @@
+from sqlalchemy import (
+    Column,
+    DateTime,
+    ForeignKey,
+    PrimaryKeyConstraint,
+    String,
+    Text,
+)
+
+from constants import (
+    COMPUTING_ID_LEN,
+    DISCORD_ID_LEN,
+    DISCORD_NAME_LEN,
+    DISCORD_NICKNAME_LEN,
+)
+from database import Base
+
+election_types = ["general_election", "by_election", "council_rep_election"]
+
+MAX_ELECTION_NAME = 64
+MAX_ELECTION_SLUG = 64
+
+class Election(Base):
+    __tablename__ = "election"
+
+    # Slugs are unique identifiers
+    slug = Column(String(MAX_ELECTION_SLUG), primary_key=True)
+    name = Column(String(MAX_ELECTION_NAME), nullable=False)
+    type = Column(String(64), default="general_election")
+    datetime_start_nominations = Column(DateTime, nullable=False)
+    datetime_start_voting = Column(DateTime, nullable=False)
+    datetime_end_voting = Column(DateTime, nullable=False)
+    survey_link = Column(String(300))
+
+    def serializable_dict(self) -> dict:
+        return {
+            "slug": self.slug,
+            "name": self.name,
+            "type": self.type,
+
+            "datetime_start_nominations": self.datetime_start_nominations.isoformat(),
+            "datetime_start_voting": self.datetime_start_voting.isoformat(),
+            "datetime_end_voting": self.datetime_end_voting.isoformat(),
+
+            "survey_link": self.survey_link,
+        }
+
+    def public_details(self) -> dict:
+        return {
+            "slug": self.slug,
+            "name": self.name,
+            "type": self.type,
+
+            "datetime_start_nominations": self.datetime_start_nominations.isoformat(),
+            "datetime_start_voting": self.datetime_start_voting.isoformat(),
+            "datetime_end_voting": self.datetime_end_voting.isoformat(),
+        }
+
+    def to_update_dict(self) -> dict:
+        return {
+            "slug": self.slug,
+            "name": self.name,
+            "type": self.type,
+
+            "datetime_start_nominations": self.datetime_start_nominations,
+            "datetime_start_voting": self.datetime_start_voting,
+            "datetime_end_voting": self.datetime_end_voting,
+
+            "survey_link": self.survey_link,
+        }
+
+# Each row represents a nominee of a given election
+class Nominee(Base):
+    __tablename__ = "election_nominee"
+
+    # Previously named sfuid
+    computing_id = Column(String(COMPUTING_ID_LEN), primary_key=True)
+    full_name = Column(String(64), nullable=False)
+    facebook = Column(String(128))
+    instagram = Column(String(128))
+    email = Column(String(64))
+    discord = Column(String(DISCORD_NAME_LEN))
+    discord_id = Column(String(DISCORD_ID_LEN))
+    discord_username = Column(String(DISCORD_NICKNAME_LEN))
+
+class NomineeApplication(Base):
+    __tablename__ = "nominee_application"
+
+    computing_id = Column(ForeignKey("election_nominee.computing_id"), primary_key=True)
+    nominee_election = Column(ForeignKey("election.slug"), primary_key=True)
+    speech = Column(Text)
+    position = Column(String(64), nullable=False)
+
+    __table_args__ = (
+        PrimaryKeyConstraint(computing_id, nominee_election),
+    )