update formatting & fix some small access bugs

Author: EarthenSky

src/alembic/versions/243190df5588_create_election_tables.py

@@ -20,7 +20,8 @@
 
 
 def upgrade() -> None:
-    op.create_table("election",
+    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),
@@ -30,7 +31,8 @@ def upgrade() -> None:
         sa.Column("survey_link", sa.String(length=300), nullable=True),
         sa.PrimaryKeyConstraint("slug")
     )
-    op.create_table("election_nominee",
+    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),
@@ -41,13 +43,14 @@ def upgrade() -> None:
         sa.Column("discord_username", sa.String(length=32), nullable=True),
         sa.PrimaryKeyConstraint("computing_id")
     )
-    op.create_table("nominee_application",
+    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.ForeignKeyConstraint(["computing_id"], ["election_nominee.computing_id"]),
+        sa.ForeignKeyConstraint(["nominee_election"], ["election.slug"]),
         sa.PrimaryKeyConstraint("computing_id", "nominee_election")
     )
 

src/elections/crud.py

@@ -7,9 +7,8 @@
 
 import database
 from elections.tables import Election
-from officers.constants import OfficerPosition
-from officers.tables import OfficerInfo, OfficerTerm
 
+_logger = logging.getLogger(__name__)
 
 @dataclass
 class ElectionParameters:
@@ -24,39 +23,38 @@ class ElectionParameters:
     end_datetime: datetime
     survey_link: str
 
-
-_logger = logging.getLogger(__name__)
-
 async def get_election(db_session: AsyncSession, election_slug: str) -> Election | None:
-    query = (
+    return await db_session.scalar(
         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,
-               start_datetime=params.start_datetime,
-               end_datetime=params.end_datetime,
-               survey_link=params.survey_link)
-    db_session.add(election)
+    db_session.add(Election(
+        slug=params.slug,
+        name=params.name,
+        officer_id=params.officer_id,
+        type=params.type,
+        start_datetime=params.start_datetime,
+        end_datetime=params.end_datetime,
+        survey_link=params.survey_link
+    ))
 
 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)
+    await db_session.execute(
+        sqlalchemy
+        .delete(Election)
+        .where(Election.slug == slug)
+    )
 
 async def update_election(db_session: AsyncSession, params: ElectionParameters) -> None:
     """
@@ -66,7 +64,12 @@ async def update_election(db_session: AsyncSession, params: ElectionParameters)
     Does not validate if an election _already_ exists
     """
 
-    election = (await db_session.execute(sqlalchemy.select(Election).filter_by(slug=params.slug))).scalar_one()
+    election = (await db_session.execute(
+        sqlalchemy
+        .select(Election)
+        # TODO: what is filter_by?
+        .filter_by(slug=params.slug)
+    )).scalar_one()
 
     if params.start_datetime is not None:
         election.start_datetime = params.start_datetime
@@ -77,6 +80,9 @@ async def update_election(db_session: AsyncSession, params: ElectionParameters)
     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)
-
+    await db_session.execute(
+        sqlalchemy
+        .update(Election)
+        .where(Election.slug == params.slug)
+        .values(election)
+    )

src/elections/tables.py

@@ -1,5 +1,3 @@
-from datetime import datetime
-
 from sqlalchemy import (
     Column,
     DateTime,

src/elections/urls.py

@@ -1,14 +1,9 @@
-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 import APIRouter, HTTPException, Request, status
 from fastapi.exceptions import RequestValidationError
 from tables import election_types
 
@@ -17,7 +12,7 @@
 import database
 import elections
 from constants import root_ip_address
-from permission import types
+from permission.types import ElectionOfficer
 
 _logger = logging.getLogger(__name__)
 
@@ -27,21 +22,27 @@
 )
 
 def _slugify(
-        text: str
+    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:
+    request: Request,
+    db_session: database.DBSession,
+) -> tuple[bool, str, str]:
+    session_id = request.cookies.get("session_id", None)
+    if session_id is None:
+        return False, None, None
+
     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
+    if computing_id is None:
+        return False, None, None
+
+    has_permission = await ElectionOfficer.has_permission(db_session, computing_id)
+    return has_permission, session_id, computing_id
 
 @router.get(
     "/create_election",
@@ -59,9 +60,11 @@ async def create_election(
     """
     aaa
     """
-    session_id = request.cookies.get("session_id", None)
-    user_auth = await _validate_user(db_session, session_id)
-    if user_auth is False:
+    if election_type not in election_types:
+        raise RequestValidationError()
+
+    is_valid_user, session_id, _ = await _validate_user(request, db_session)
+    if not is_valid_user:
         # let's workshop how we actually wanna handle this
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
@@ -73,9 +76,6 @@ async def create_election(
     if start_datetime is None:
         start_datetime = datetime.now()
 
-    if election_type not in election_types:
-        raise RequestValidationError()
-
     params = ElectionParameters(
         _slugify(name),
         name,
@@ -93,17 +93,16 @@ async def create_election(
     return {}
 
 @router.get(
-        "/delete_election",
-        description="Deletes an election from the database"
+    "/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:
+    is_valid_user, _, _ = await _validate_user(request, db_session)
+    if not is_valid_user:
         # let's workshop how we actually wanna handle this
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
@@ -130,16 +129,15 @@ async def update_election(
     end_datetime: 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:
+    is_valid_user, session_id, _ = await _validate_user(request, db_session)
+    if not is_valid_user:
         # 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:
+    elif slug is not None:
         params = ElectionParameters(
             _slugify(name),
             name,
@@ -152,7 +150,6 @@ async def update_election(
         await elections.crud.update_election(params, db_session)
         await db_session.commit()
 
-
 @router.get(
     "/test"
 )