remove overcomplication, fix semantics, more DRY. For /all & /current

Author: EarthenSky

src/officers/crud.py

@@ -96,46 +96,52 @@ async def current_executive_team(db_session: database.DBSession, include_private
 
     Returns a mapping between officer position and officer terms
     """
-
-    query = sqlalchemy.select(OfficerTerm)
+    query = (
+        sqlalchemy
+        .select(OfficerTerm)
+        .order_by(OfficerTerm.start_date.desc())
+    )
     query = utils.is_active_officer(query)
-    query = query.order_by(OfficerTerm.start_date.desc())
 
     officer_terms = (await db_session.scalars(query)).all()
-    num_officers = {}
+    ##num_officers = {}
     officer_data = {}
-
     for term in officer_terms:
+        """
         if term.position not in OfficerPosition.position_list():
             _logger.warning(
                 f"Unknown OfficerTerm.position={term.position} in database. Ignoring in request."
             )
             continue
+        """
 
-        officer_info_query = sqlalchemy.select(OfficerInfo)
-        officer_info_query = officer_info_query.where(
-            OfficerInfo.computing_id == term.computing_id
+        officer_info_query = (
+            sqlalchemy
+            .select(OfficerInfo)
+            .where(OfficerInfo.computing_id == term.computing_id)
         )
         officer_info = (await db_session.scalars(officer_info_query)).first()
         if officer_info is None:
             # TODO: make sure there are daily checks that this data actually exists
             continue
 
         if term.position not in officer_data:
-            num_officers[term.position] = 0
+            ##num_officers[term.position] = 0
             officer_data[term.position] = []
 
+        """
         num_officers[term.position] += 1
-        # TODO: move this to a ~~daily cronjob~~ SQL model checking
+        # TODO: move this to a daily cronjob
         if num_officers[term.position] > OfficerPosition.num_active(term.position):
             # If there are more active positions than expected, log it to a file
             _logger.warning(
                 f"There are more active {term.position} positions in the OfficerTerm than expected "
                 f"({num_officers[term.position]} > {OfficerPosition.num_active(term.position)})"
             )
-
+        """
         officer_data[term.position] += [OfficerData.from_data(term, officer_info, include_private, is_active=True)]
 
+    """
     # validate & warn if there are any data issues
     # TODO: decide whether we should enforce empty instances or force the frontend to deal with it
     for position in OfficerPosition.expected_positions():
@@ -151,39 +157,39 @@ async def current_executive_team(db_session: database.DBSession, include_private
                 f"Unexpected number of {position} entries "
                 f"({len(officer_data[position])} entries) in current_executive_team response."
             )
-
+    """
     return officer_data
 
 async def all_officer_data(
     db_session: database.DBSession,
-    include_private: bool,
-    view_only_filled_in: bool,
+    include_private_data: bool,
+    view_not_started_officer_terms: bool,
 ) -> list[OfficerData]:
     """
-    This could be a lot of data, so be careful.
-
-    TODO: optionally paginate data, so it's not so bad.
+    This could be a lot of data, so be careful
     """
-    query = sqlalchemy.select(OfficerTerm)
-    if view_only_filled_in:
-        query = OfficerTerm.sql_is_filled_in(query)
-    # Ordered recent first
-    query = query.order_by(OfficerTerm.start_date.desc())
-    officer_terms = (await db_session.scalars(query)).all()
+    # NOTE: paginate data if needed
+    query = (
+        sqlalchemy
+        .select(OfficerTerm)
+        # Ordered recent first
+        .order_by(OfficerTerm.start_date.desc())
+    )
+    if not view_not_started_officer_terms:
+        query = utils.has_term_started(query)
 
     officer_data_list = []
+    officer_terms = (await db_session.scalars(query)).all()
     for term in officer_terms:
-        officer_info_query = (
+        officer_info = await db_session.scalar(
             sqlalchemy
             .select(OfficerInfo)
             .where(OfficerInfo.computing_id == term.computing_id)
         )
-        officer_info = await db_session.scalar(officer_info_query)
-
         officer_data_list += [OfficerData.from_data(
             term,
             officer_info,
-            include_private,
+            include_private_data,
             utils.is_active_term(term)
         )]
 

src/officers/tables.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 from sqlalchemy import (
-    # Boolean,
     Column,
     DateTime,
     ForeignKey,

src/officers/types.py

@@ -5,7 +5,6 @@
 
 from fastapi import HTTPException
 
-from constants import COMPUTING_ID_MAX
 from officers.constants import OfficerPosition
 from officers.tables import OfficerInfo, OfficerTerm
 
@@ -117,7 +116,7 @@ class OfficerData:
 
     csss_email: str | None
     biography: str | None
-    photo_url: str | None  # some urls get big...
+    photo_url: str | None
 
     private_data: OfficerPrivateData | None
 
@@ -133,7 +132,7 @@ def serializable_dict(self):
     def from_data(
         term: OfficerTerm,
         officer_info: OfficerInfo,
-        include_private: bool,
+        include_private_data: bool,
         is_active: bool,
     ) -> OfficerData:
         return OfficerData(
@@ -162,5 +161,5 @@ def from_data(
                 phone_number = officer_info.phone_number,
                 github_username = officer_info.github_username,
                 google_drive_email = officer_info.google_drive_email,
-            ) if include_private else None,
+            ) if include_private_data else None,
         )

src/officers/urls.py

@@ -1,6 +1,6 @@
 import logging
 from dataclasses import dataclass
-from datetime import date, datetime
+from datetime import date
 
 import sqlalchemy
 from fastapi import APIRouter, Body, HTTPException, Request
@@ -25,7 +25,22 @@
     tags=["officers"],
 )
 
-# TODO: combine the following two endpoints
+async def has_officer_private_info_access(
+    request: Request,
+    db_session: database.DBSession,
+) -> tuple[None | str, None | str, bool]:
+    # determine if the user has access to this private data
+    session_id = request.cookies.get("session_id", None)
+    if session_id is None:
+        return None, None, False
+
+    computing_id = await auth.crud.get_computing_id(db_session, session_id)
+    if computing_id is None:
+        return session_id, None, False
+
+    has_private_access = await OfficerPrivateInfo.has_permission(db_session, computing_id)
+    return session_id, computing_id, has_private_access
+
 @router.get(
     "/current",
     description="Get information about all the officers. More information is given if you're authenticated & have access to private executive data.",
@@ -35,13 +50,7 @@ async def current_officers(
     request: Request,
     db_session: database.DBSession,
 ):
-    # determine if the user has access to this private data
-    session_id = request.cookies.get("session_id", None)
-    if session_id is None:
-        has_private_access = False
-    else:
-        computing_id = await auth.crud.get_computing_id(db_session, session_id)
-        has_private_access = await OfficerPrivateInfo.has_permission(db_session, computing_id)
+    _, _, has_private_access = await has_officer_private_info_access(request, db_session)
 
     current_executives = await officers.crud.current_executive_team(db_session, has_private_access)
     json_current_executives = {
@@ -54,36 +63,26 @@ async def current_officers(
 
 @router.get(
     "/all",
-    description="Information from all exec terms. If year is not included, all years will be returned. If semester is not included, all semesters that year will be returned. If semester is given, but year is not, return all years and all semesters.",
+    description="Information for all execs from all exec terms",
 )
 async def all_officers(
     request: Request,
     db_session: database.DBSession,
-    view_only_filled_in: bool = True,
+    # Officer terms for officers which have not yet started their term yet are considered private,
+    # and may only be accessed by that officer and executives.
+    view_not_started_officer_terms: bool = False,
 ):
-    async def has_access(db_session: database.DBSession, request: Request) -> bool:
-        # determine if user has access to this private data
-        session_id = request.cookies.get("session_id", None)
-        if session_id is None:
-            return False
-
-        computing_id = await auth.crud.get_computing_id(db_session, session_id)
-        if computing_id is None:
-            return False
-        else:
-            has_private_access = await OfficerPrivateInfo.has_permission(db_session, computing_id)
-            is_website_admin = await WebsiteAdmin.has_permission(db_session, computing_id)
-
-            if not view_only_filled_in and (session_id is None or not is_website_admin):
-                raise HTTPException(status_code=401, detail="must have private access to view not filled in terms")
-
-            return has_private_access
-
-    has_private_access = await has_access(db_session, request)
-
-    all_officer_data = await officers.crud.all_officer_data(db_session, has_private_access, view_only_filled_in)
-    all_officer_data = [officer_data.serializable_dict() for officer_data in all_officer_data]
-    return JSONResponse(all_officer_data)
+    _, computing_id, has_private_access = await has_officer_private_info_access(request, db_session)
+    if view_not_started_officer_terms:
+        is_website_admin = (computing_id is not None) and (await WebsiteAdmin.has_permission(db_session, computing_id))
+        if not is_website_admin:
+            raise HTTPException(status_code=401, detail="only website admins can view all executive terms that have not started yet")
+
+    all_officer_data = await officers.crud.all_officer_data(db_session, has_private_access, not view_not_started_officer_terms)
+    return JSONResponse([
+        officer_data.serializable_dict()
+        for officer_data in all_officer_data
+    ])
 
 @router.get(
     "/terms/{computing_id}",

src/utils.py

@@ -6,7 +6,7 @@
 # we can't use and/or in sql expressions, so we must use these functions
 from sqlalchemy.sql.expression import and_, or_
 
-from officers.tables import OfficerInfo, OfficerTerm
+from officers.tables import OfficerTerm
 
 
 def is_iso_format(date_str: str) -> bool:
@@ -17,6 +17,10 @@ def is_iso_format(date_str: str) -> bool:
         return False
 
 def is_active_officer(query: Select) -> Select:
+    """
+    An active officer is one who is currently part of the CSSS officer team.
+    That is, they are not tentative, or past.
+    """
     query = query.where(
         and_(
             # cannot be an officer who has not started yet
@@ -43,6 +47,9 @@ def is_active_term(term: OfficerTerm) -> bool:
         )
     )
 
+def has_term_started(term: OfficerTerm) -> bool:
+    return term.start_date <= datetime.today()
+
 def is_past_term(term: OfficerTerm) -> bool:
     """Any term which has concluded"""
     return (