First Elections PR

src/alembic/env.py

@@ -3,6 +3,7 @@
 
 import auth.tables
 import database
+import elections.tables
 import officers.tables
 from alembic import context
 from sqlalchemy import pool

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 = "166f3772fce7"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    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),
+    sa.Column("date", sa.DateTime(), nullable=False),
+    sa.Column("end_date", sa.DateTime(), nullable=True),
+    sa.Column("websurvey", 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")
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table("nominee_application")
+    op.drop_table("election_nominee")
+    op.drop_table("election")
+    # ### end Alembic commands ###

src/elections/crud.py

@@ -0,0 +1,68 @@
+import dataclasses
+import logging
+from datetime import datetime
+
+import database
+import sqlalchemy
+from elections.tables import Election
+from officers.constants import OfficerPosition
+from officers.tables import OfficerInfo, OfficerTerm
+
+_logger = logging.getLogger(__name__)
+
+async def get_election(db_session: database.DBSession, election_slug: str) -> Election | None:
+    query = sqlalchemy.select(Election)
+    query = query.where(Election.slug == election_slug)
+    result = (await db_session.execute(query)).scalar()
+    db_session.commit()
+    return result
+
+async def create_election(params: dict[str, datetime], db_session: database.DBSession) -> 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"],
+               websurvey=params["websurvey"])
+    db_session.add(election)
+    await db_session.commit()
+
+async def delete_election(slug: str, db_session: database.DBSession) -> 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)
+    await db_session.commit()
+
+async def update_election(params: dict[str, datetime], db_session: database.DBSession) -> 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["websurvey"] is not None:
+        election.websurvey = params["websurvey"]
+
+    await db_session.commit()
+
+
+    # query = sqlalchemy.update(Election).where(Election.slug == params["slug"]).values(election)
+    # await db_session.execute(query)
+

src/elections/tables.py

@@ -0,0 +1,58 @@
+from datetime import datetime
+
+from constants import (
+    COMPUTING_ID_LEN,
+    DISCORD_ID_LEN,
+    DISCORD_NAME_LEN,
+    DISCORD_NICKNAME_LEN,
+)
+from database import Base
+from sqlalchemy import (
+    Column,
+    DateTime,
+    ForeignKey,
+    PrimaryKeyConstraint,
+    String,
+    Text,
+)
+
+
+# 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)
+    # Can be one of (general_election: General Election, by_election: By-Election, council_rep_election: Council Rep Election)
+    type = Column(String(64), default="general_election")
+    date = Column(DateTime, nullable=False)
+    end_date = Column(DateTime)
+    websurvey = 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
+from enum import Enum
+
+import auth
+import auth.crud
+import database
+import elections
+import requests  # TODO: make this async
+import xmltodict
+from constants import root_ip_address
+from fastapi import APIRouter, BackgroundTasks, FastAPI, HTTPException, Request, status
+from fastapi.exceptions import RequestValidationError
+from permission import types
+
+_logger = logging.getLogger(__name__)
+
+router = APIRouter(
+    prefix="/elections",
+    tags=["elections"],
+)
+
+class ElectionType(Enum):
+    GENERAL_ELECTION = "general_election"
+    BY_ELECTION = "by_election"
+    COUNCIL_REP_ELECTION = "council_rep_election"
+
+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,
+    websurvey: 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 [e.value for e in ElectionType]:
+        raise RequestValidationError()
+
+    params = {
+        "slug" : _slugify(name),
+        "name": name,
+        "officer_id" : await auth.crud.get_computing_id(db_session, session_id),
+        "type": election_type,
+        "date": date,
+        "end_date": end_date,
+        "websurvey": websurvey
+    }
+
+    await elections.crud.create_election(params, db_session)
+
+    # 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)
+
+@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,
+    websurvey: 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 = {
+            "slug" : slug,
+            "name" : name,
+            "officer_id" : await auth.crud.get_computing_id(db_session, session_id),
+            "type": election_type,
+            "date": date,
+            "end_date": end_date,
+            "websurvey": websurvey
+        }
+        await elections.crud.update_election(params, db_session)
+
+
+@router.get(
+    "/test"
+)
+async def test():
+    return {"error": "lol"}

src/main.py

@@ -2,9 +2,13 @@
 
 import auth.urls
 import database
+import elections.urls
 import officers.urls
 import permission.urls
-from fastapi import FastAPI
+from fastapi import FastAPI, Request, status
+from fastapi.encoders import jsonable_encoder
+from fastapi.exceptions import RequestValidationError
+from fastapi.responses import JSONResponse
 
 import tests.urls
 
@@ -18,7 +22,21 @@
 app.include_router(permission.urls.router)
 
 app.include_router(tests.urls.router)
+app.include_router(elections.urls.router)
 
 @app.get("/")
 async def read_root():
     return {"message": "Hello! You might be lost, this is actually the sfucsss.org's backend api."}
+
+@app.exception_handler(RequestValidationError)
+async def validation_exception_handler(
+    request: Request,
+    exception: RequestValidationError,
+):
+    return JSONResponse(
+        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+        content=jsonable_encoder({
+            "detail": exception.errors(),
+            "body": exception.body,
+        })
+    )