implement endpoint for getting officer info

src/alembic/env.py

@@ -1,7 +1,6 @@
 import asyncio
 from logging.config import fileConfig
 
-# ruff: noqa: F401
 import auth.models
 import database
 import officers.models

src/auth/crud.py

@@ -2,47 +2,47 @@
 from typing import Optional
 
 import sqlalchemy
-from auth import models
+from auth.models import SiteUser, UserSession
 from sqlalchemy.ext.asyncio import AsyncSession
 
 
 async def create_user_session(db_session: AsyncSession, session_id: str, computing_id: str) -> None:
     """
     Updates the past user session if one exists, so no duplicate sessions can ever occur.
 
-    Also, adds the new user to the User table if it's their first time logging in.
+    Also, adds the new user to the SiteUser table if it's their first time logging in.
     """
-    query = sqlalchemy.select(models.UserSession).where(models.UserSession.computing_id == computing_id)
+    query = sqlalchemy.select(UserSession).where(UserSession.computing_id == computing_id)
     existing_user_session = (await db_session.scalars(query)).first()
     if existing_user_session:
         existing_user_session.issue_time = datetime.now()
         existing_user_session.session_id = session_id
     else:
-        new_user_session = models.UserSession(
+        new_user_session = UserSession(
             issue_time=datetime.now(),
             session_id=session_id,
             computing_id=computing_id,
         )
         db_session.add(new_user_session)
 
         # add new user to User table if it's their first time logging in
-        query = sqlalchemy.select(models.User).where(models.User.computing_id == computing_id)
+        query = sqlalchemy.select(SiteUser).where(SiteUser.computing_id == computing_id)
         existing_user = (await db_session.scalars(query)).first()
         if existing_user is None:
-            new_user = models.User(
+            new_user = SiteUser(
                 computing_id=computing_id,
             )
             db_session.add(new_user)
 
 
 async def remove_user_session(db_session: AsyncSession, session_id: str) -> dict:
-    query = sqlalchemy.select(models.UserSession).where(models.UserSession.session_id == session_id)
+    query = sqlalchemy.select(UserSession).where(UserSession.session_id == session_id)
     user_session = await db_session.scalars(query)
-    await db_session.delete(user_session.first())  # TODO: what to do with this result?
+    await db_session.delete(user_session.first())  # TODO: what to do with this result that we're awaiting?
 
 
 async def check_session_validity(db_session: AsyncSession, session_id: str) -> dict:
-    query = sqlalchemy.select(models.UserSession).where(models.UserSession.session_id == session_id)
+    query = sqlalchemy.select(UserSession).where(UserSession.session_id == session_id)
     existing_user_session = (await db_session.scalars(query)).first()
 
     if existing_user_session:
@@ -52,19 +52,15 @@ async def check_session_validity(db_session: AsyncSession, session_id: str) -> d
 
 
 async def get_computing_id(db_session: AsyncSession, session_id: str) -> str | None:
-    query = sqlalchemy.select(models.UserSession).where(models.UserSession.session_id == session_id)
+    query = sqlalchemy.select(UserSession).where(UserSession.session_id == session_id)
     existing_user_session = (await db_session.scalars(query)).first()
-
-    if existing_user_session:
-        return existing_user_session.computing_id
-    else:
-        return None
+    return existing_user_session.computing_id if existing_user_session else None
 
 
 # remove all out of date user sessions
 async def task_clean_expired_user_sessions(db_session: AsyncSession) -> None:
     one_day_ago = datetime.now() - timedelta(days=0.5)
 
-    query = sqlalchemy.delete(models.UserSession).where(models.UserSession.issue_time < one_day_ago)
+    query = sqlalchemy.delete(UserSession).where(UserSession.issue_time < one_day_ago)
     await db_session.execute(query)
     await db_session.commit()

src/auth/models.py

@@ -17,10 +17,10 @@ class UserSession(Base):
         String(SESSION_ID_LEN), nullable=False, unique=True
     )  # the space needed to store 256 bytes in base64
 
-    site_user = relationship("User")
+    site_user = relationship("SiteUser")
 
 
-class User(Base):
+class SiteUser(Base):
     # user is a reserved word in postgres
     # see: https://stackoverflow.com/questions/22256124/cannot-create-a-database-table-named-user-in-postgresql
     __tablename__ = "site_user"
@@ -34,9 +34,9 @@ class User(Base):
         unique=True,
     )  # technically a max of 8 digits https://www.sfu.ca/computing/about/support/tips/sfu-userid.html
 
-    officer_term = relationship("OfficerTerm")
+    officer_term = relationship("OfficerTerm", back_populates="site_user")
     officer_info = relationship("OfficerInfo")
 
     # TODO: (#13) add two new columns for storing the initial date & last date logged in.
     # When running the migration, you'll want to decide on some random date to be the default for users who've logged
-    # before but haven't
+    # before but haven't ...

src/main.py

@@ -2,12 +2,12 @@
 from auth import auth
 from fastapi import FastAPI
 
-# from officers import officers
+from officers import officers
 from tests import tests
 
 app = FastAPI(lifespan=database.lifespan, title="CSSS Site Backend")
 app.include_router(auth.router)
-# app.include_router(officers.router)
+app.include_router(officers.router)
 app.include_router(tests.router)
 
 

src/officers/constants.py

@@ -0,0 +1,142 @@
+import logging
+from enum import Enum
+from typing import Self
+
+_logger = logging.getLogger(__name__)
+
+
+class OfficerPosition(Enum):
+    President = "president"
+    VicePresident = "vice-president"
+    Treasurer = "treasurer"
+
+    DirectorOfResources = "director of resources"
+    DirectorOfEvents = "director of events"
+    DirectorOfEducationalEvents = "director of educational events"
+    AssistantDirectorOfEvents = "assistant director of events"
+    DirectorOfCommunications = "director of communications"
+    #DirectorOfOutreach = "director of outreach"
+    DirectorOfMultimedia = "director of multimedia"
+    DirectorOfArchives = "director of archives"
+    ExecutiveAtLarge = "executive at large"
+    FirstYearRepresentative = "first year representative"
+
+    ElectionsOfficer = "elections officer"
+    SFSSCouncilRepresentative = "sfss council representative"
+    FroshWeekChair = "frosh week chair"
+
+    SystemAdministrator = "system administrator"
+    Webmaster = "webmaster"
+    SocialMediaManager = "social media manager"
+
+    @staticmethod
+    def from_string(position: str) -> Self | None:
+        for item in OfficerPosition:
+            if position == item.value:
+                return item
+
+        _logger.warning(f"Unknown OfficerPosition position = {position}. reporting N/A.")
+        return None
+
+    def to_string(self) -> str:
+        return self.value
+
+    def to_email(self) -> str:
+        match self:
+            case OfficerPosition.President:
+                return "csss-president-current@sfu.ca"
+            case OfficerPosition.VicePresident:
+                return "csss-vp-current@sfu.ca"
+            case OfficerPosition.Treasurer:
+                return "csss-treasurer-current@sfu.ca"
+
+            case OfficerPosition.DirectorOfResources:
+                return "csss-dor-current@sfu.ca"
+            case OfficerPosition.DirectorOfEvents:
+                return "csss-doe-current@sfu.ca"
+            case OfficerPosition.DirectorOfEducationalEvents:
+                return "csss-doee-current@sfu.ca"
+            case OfficerPosition.AssistantDirectorOfEvents:
+                return "csss-adoe-current@sfu.ca"
+            case OfficerPosition.DirectorOfCommunications:
+                return "csss-doc-current@sfu.ca"
+            case OfficerPosition.DirectorOfMultimedia:
+                return "csss-domm-current@sfu.ca"
+            case OfficerPosition.DirectorOfArchives:
+                return "csss-doa-current@sfu.ca"
+            case OfficerPosition.ExecutiveAtLarge:
+                return "csss-eal-current@sfu.ca"
+            case OfficerPosition.FirstYearRepresentative:
+                return "csss-fyr-current@sfu.ca"
+
+            case OfficerPosition.ElectionsOfficer:
+                return "csss-elections@sfu.ca"
+            case OfficerPosition.SFSSCouncilRepresentative:
+                return "csss-councilrep@sfu.ca"
+            case OfficerPosition.FroshWeekChair:
+                return "csss-froshchair@sfu.ca"
+
+            case OfficerPosition.SystemAdministrator:
+                return "csss-sysadmin@sfu.ca"
+            case OfficerPosition.Webmaster:
+                return "csss-webmaster@sfu.ca"
+            case OfficerPosition.SocialMediaManager:
+                return "N/A"
+
+    def num_active(self) -> int | None:
+        """
+        The number of executive positions active at a given time
+        """
+        # None means there can be any number active
+        if (
+            self == OfficerPosition.ExecutiveAtLarge 
+            or self == OfficerPosition.FirstYearRepresentative
+        ):
+            return 2
+        elif (
+            self == OfficerPosition.FroshWeekChair 
+            or OfficerPosition.SocialMediaManager
+        ):
+            # TODO: configure this value in a database table somewhere?
+            return None
+        else: 
+            return 1
+
+    def is_signer(self) -> bool:
+        """
+        If the officer is a signing authority of the CSSS
+        """
+        return (
+            self == OfficerPosition.President
+            or self == OfficerPosition.VicePresident
+            or self == OfficerPosition.Treasurer
+            or self == OfficerPosition.DirectorOfResources
+            or self == OfficerPosition.DirectorOfEvents
+        )
+
+    @staticmethod
+    def expected_positions() -> list[Self]:
+        return [
+            OfficerPosition.President,
+            OfficerPosition.VicePresident,
+            OfficerPosition.Treasurer,
+
+            OfficerPosition.DirectorOfResources,
+            OfficerPosition.DirectorOfEvents,
+            OfficerPosition.DirectorOfEducationalEvents,
+            OfficerPosition.AssistantDirectorOfEvents,
+            OfficerPosition.DirectorOfCommunications,
+            #DirectorOfOutreach, # TODO: when https://github.com/CSSS/documents/pull/9/files merged
+            OfficerPosition.DirectorOfMultimedia,
+            OfficerPosition.DirectorOfArchives,
+            OfficerPosition.ExecutiveAtLarge,
+            # TODO: expect these only during fall & spring semesters. Also, TODO: this todo is correct...
+            #FirstYearRepresentative, 
+
+            #ElectionsOfficer,
+            OfficerPosition.SFSSCouncilRepresentative,
+            OfficerPosition.FroshWeekChair,
+
+            OfficerPosition.SystemAdministrator,
+            OfficerPosition.Webmaster,
+        ]

src/officers/crud.py

@@ -1,22 +1,110 @@
-from typing import Optional
+import logging
+
+import sqlalchemy
 
 import database
-from models import OfficerTerm
 
-# rom . import schemas
+from officers.models import OfficerTerm
+from officers.constants import OfficerPosition
+from officers.schemas import OfficerData, OfficerPrivateData
+
+_logger = logging.getLogger(__name__)
 
 
-def most_recent_exec_term(db_session: database.DBSession, computing_id: str) -> OfficerTerm | None:
+async def most_recent_exec_term(db_session: database.DBSession, computing_id: str) -> OfficerTerm | None:
     """
     Returns the most recent OfficerTerm an exec has had
     """
 
-    query = db_session.query(OfficerTerm)
-    query = query.filter(OfficerTerm.computing_id == computing_id)
+    query = sqlalchemy.select(OfficerTerm)
+    query = query.where(OfficerTerm.computing_id == computing_id)
+    query = query.order_by(OfficerTerm.start_date.desc())
+    query = query.limit(1)
+
+    # TODO: can this be replaced with scalar to improve performance?
+    return (await db_session.scalars(query)).first()
+
+async def current_executive_team(db_session: database.DBSession, include_private: bool) -> dict[str, list[OfficerData]]:
+    """
+    Get info about officers that satisfy is_active. Go through all active & complete officer terms.
+    """
+
+    query = sqlalchemy.select(OfficerTerm)
+    query = query.filter(OfficerTerm.is_active and OfficerTerm.is_complete)
     query = query.order_by(OfficerTerm.start_date.desc())
 
-    # TODO: confirm that the result is an instance of OfficerTerm (or None)
-    return query.first()
+    officer_terms = (await db_session.scalars(query)).all()
+    num_officers = {}
+    officer_data = {}
+
+    for term in officer_terms:
+        if term.position not in [officer.value for officer in OfficerPosition]:
+            _logger.warning(
+                f"Unknown OfficerTerm.position={term.position} in database. Ignoring in request"
+            )
+            continue
+
+        if term.position not in officer_data:
+            num_officers[term.position] = 1
+            officer_data[term.position] = []
+        else:
+            num_officers[term.position] += 1
+            if num_officers[term.position] > OfficerPrivateData.from_string(term.position).num_active():
+                # 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]} > {OfficerPrivateData.from_string(term.position).num_active()})"
+                )
+
+        officer_data[term.position] += [
+            OfficerData(
+                is_current_officer = True,
+
+                position = term.position,
+                start_date = term.start_date,
+                end_date = term.end_date,
+
+                legal_name = term.site_user.officer_info.legal_name,
+                nickname = term.nickname,
+                discord_name = term.site_user.officer_info.discord_name,
+                discord_nickname = term.site_user.officer_info.discord_nickname,
+
+                favourite_course_0 = term.favourite_course_0,
+                favourite_course_1 = term.favourite_course_1,
+
+                favourite_language_0 = term.favourite_pl_0,
+                favourite_language_1 = term.favourite_pl_1,
+
+                csss_email = OfficerPosition.from_string(term.position).to_email(),
+                biography = term.biography,
+                photo_url = term.photo_url,
+
+                private_data = OfficerPrivateData(
+                    computing_id = term.computing_id,
+                    phone_number = term.site_user.officer_info.phone_number,
+                    github_username = term.site_user.officer_info.github_username,
+                    google_drive_email = term.site_user.officer_info.google_drive_email,
+                ) if include_private else None,
+            )
+        ]
+
+    # 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():
+        if position.to_string() not in officer_data:
+            _logger.warning(
+                f"Expected position={position.to_string()} in response current_executive_team."
+            )
+        elif (
+            position.num_active is not None 
+            and len(officer_data[position.to_string()]) != position.num_active
+        ):
+            _logger.warning(
+                f"Unexpected number of {position.to_string()} entries "
+                f"({len(officer_data[position.to_string()])} entries) in current_executive_team response."
+            )
+
+    return officer_data
 
 
 """