First Elections PR

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,58 @@
+"""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=32), nullable=False),
+        sa.Column("name", sa.String(length=32), nullable=False),
+        sa.Column("officer_id", sa.String(length=32), nullable=False),
+        sa.Column("type", sa.String(length=64), nullable=True, default="general_election"),
+        sa.Column("date", sa.DateTime(), nullable=False),
+        sa.Column("end_date", sa.DateTime(), nullable=True),
+        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=64), nullable=True),
+        sa.Column("instagram", sa.String(length=64), 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=18), 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,82 @@
+import logging
+from dataclasses import dataclass
+from datetime import datetime
+
+import sqlalchemy
+from sqlalchemy.ext.asyncio import AsyncSession
+
+import database
+from elections.tables import Election
+from officers.constants import OfficerPosition
+from officers.tables import OfficerInfo, OfficerTerm
+
+
+@dataclass
+class ElectionParameters:
+    """
+    Dataclass encompassing the data that can go into an Election.
+    """
+    slug: str
+    name: str
+    officer_id: str
+    type: str
+    date: datetime
+    end_date: datetime
+    survey_link: str
+
+
+_logger = logging.getLogger(__name__)
+
+async def get_election(db_session: AsyncSession, election_slug: str) -> Election | None:
+    query = (
+        sqlalchemy
+        .select(Election)
+        .where(Election.slug == election_slug)
+    )
+    result = await db_session.scalar(query)
+    return result
+
+async def create_election(db_session: AsyncSession, params: ElectionParameters) -> None:
+    """
+    Creates a new election with given parameters.
+    Does not validate if an election _already_ exists
+    """
+    election = Election(slug=params.slug,
+               name=params.name,
+               officer_id=params.officer_id,
+               type=params.type,
+               date=params.date,
+               end_date=params.end_date,
+               survey_link=params.survey_link)
+    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
+    """
+    query = sqlalchemy.delete(Election).where(Election.slug == slug)
+    await db_session.execute(query)
+
+async def update_election(db_session: AsyncSession, params: ElectionParameters) -> None:
+    """
+    Updates an election with the provided parameters.
+    Take care as this will replace values with None if not populated.
+    You _cannot_ change the name or slug, you should instead delete and create a new election.
+    Does not validate if an election _already_ exists
+    """
+
+    election = (await db_session.execute(sqlalchemy.select(Election).filter_by(slug=params.slug))).scalar_one()
+
+    if params.date is not None:
+        election.date = params.date
+    if params.type is not None:
+        election.type = params.type
+    if params.end_date is not None:
+        election.end_date = params.end_date
+    if params.survey_link is not None:
+        election.survey_link = params.survey_link
+
+    query = sqlalchemy.update(Election).where(Election.slug == params.slug).values(election)
+    await db_session.execute(query)
+

src/elections/tables.py

@@ -0,0 +1,59 @@
+from datetime import datetime
+
+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"]
+
+# Each row represents an instance of an
+class Election(Base):
+    __tablename__ = "election"
+
+    # Slugs are unique identifiers
+    slug = Column(String(32), primary_key=True)
+    name = Column(String(32), nullable=False)
+    officer_id = Column(String(COMPUTING_ID_LEN), nullable=False)
+    type = Column(String(64), default="general_election")
+    date = Column(DateTime, nullable=False)
+    end_date = Column(DateTime)
+    survey_link = Column(String(300))
+
+# 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(64))
+    instagram = Column(String(64))
+    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),
+    )

src/elections/urls.py

@@ -0,0 +1,160 @@
+import base64
+import logging
+import os
+import re
+import urllib.parse
+from datetime import datetime
+
+import requests  # TODO: make this async
+import xmltodict
+from crud import ElectionParameters
+from fastapi import APIRouter, BackgroundTasks, FastAPI, HTTPException, Request, status
+from fastapi.exceptions import RequestValidationError
+from tables import election_types
+
+import auth
+import auth.crud
+import database
+import elections
+from constants import root_ip_address
+from permission import types
+
+_logger = logging.getLogger(__name__)
+
+router = APIRouter(
+    prefix="/elections",
+    tags=["elections"],
+)
+
+def _slugify(
+        text: str
+) -> str:
+    """
+    Creates a unique slug based on text passed in. Assumes non-unicode text.
+    """
+    return re.sub(r"[\W_]+", "-", text)
+
+async def _validate_user(
+        db_session: database.DBSession,
+        session_id: str
+) -> dict:
+    computing_id = await auth.crud.get_computing_id(db_session, session_id)
+    # Assuming now user is validated
+    result = await types.ElectionOfficer.has_permission(db_session, computing_id)
+    return result
+
+@router.get(
+    "/create_election",
+    description="Creates an election and places it in the database",
+)
+async def create_election(
+    request: Request,
+    db_session: database.DBSession,
+    name: str,
+    election_type: str,
+    date: datetime | None = None,
+    end_date: datetime | None = None,
+    survey_link: str | None = None
+):
+    """
+    aaa
+    """
+    session_id = request.cookies.get("session_id", None)
+    user_auth = await _validate_user(db_session, session_id)
+    if user_auth is False:
+        # let's workshop how we actually wanna handle this
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="You do not have permission to access this resource",
+            headers={"WWW-Authenticate": "Basic"},
+        )
+
+    # Default start time should be now unless specified otherwise
+    if date is None:
+        date = datetime.now()
+
+    if election_type not in election_types:
+        raise RequestValidationError()
+
+    params = ElectionParameters(
+        _slugify(name),
+        name,
+        await auth.crud.get_computing_id(db_session, session_id),
+        election_type,
+        date,
+        end_date,
+        survey_link
+    )
+
+    await elections.crud.create_election(params, db_session)
+    await db_session.commit()
+
+    # TODO: create a suitable json response
+    return {}
+
+@router.get(
+        "/delete_election",
+        description="Deletes an election from the database"
+)
+async def delete_election(
+    request: Request,
+    db_session: database.DBSession,
+    slug: str
+):
+    session_id = request.cookies.get("session_id", None)
+    user_auth = await _validate_user(db_session, session_id)
+    if user_auth is False:
+        # let's workshop how we actually wanna handle this
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="You do not have permission to access this resource",
+            headers={"WWW-Authenticate": "Basic"},
+        )
+
+    if slug is not None:
+        await elections.crud.delete_election(slug, db_session)
+        await db_session.commit()
+
+@router.get(
+    "/update_election",
+    description="""Updates an election in the database.
+                   Note that this does not allow you to change the _name_ of an election as this would generate a new slug."""
+)
+async def update_election(
+    request: Request,
+    db_session: database.DBSession,
+    slug: str,
+    name: str,
+    election_type: str,
+    date: datetime | None = None,
+    end_date: datetime | None = None,
+    survey_link: str | None = None
+):
+    session_id = request.cookies.get("session_id", None)
+    user_auth = await _validate_user(db_session, session_id)
+    if user_auth is False:
+        # let's workshop how we actually wanna handle this
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="You do not have permission to access this resource",
+            headers={"WWW-Authenticate": "Basic"},
+        )
+    if slug is not None:
+        params = ElectionParameters(
+            _slugify(name),
+            name,
+            await auth.crud.get_computing_id(db_session, session_id),
+            election_type,
+            date,
+            end_date,
+            survey_link
+        )
+        await elections.crud.update_election(params, db_session)
+        await db_session.commit()
+
+
+@router.get(
+    "/test"
+)
+async def test():
+    return {"error": "lol"}