Finalize Officer Terms (#86)

* add support for having multiple exec terms at once

* officers that have not started yet are not considered active

* add support for measuring exec term length

* fix bugs

* clean naming

Author: EarthenSky

README.md

@@ -25,3 +25,7 @@ If you're planning to read through the source code, please check out this projec
     - `misc/` for anything that can't be easily categorized or is very small
     - `officers/` for officer contact information + photos
 - `test/` for html pages which interact with the backend's local api
+
+## Linter
+
+We use `ruff 0.6.9` as our linter, which you can run with `ruff check --fix`. If you use a different version, it may be inconsistent with our CI checks.

src/auth/crud.py

@@ -7,52 +7,49 @@
 from auth.tables import SiteUser, UserSession
 from auth.types import SiteUserData
 
+_logger = logging.getLogger(__name__)
 
 async def create_user_session(db_session: AsyncSession, session_id: str, computing_id: str):
     """
     Updates the past user session if one exists, so no duplicate sessions can ever occur.
 
     Also, adds the new user to the SiteUser table if it's their first time logging in.
     """
-    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 = await db_session.scalar(
+        sqlalchemy
+        .select(UserSession)
+        .where(UserSession.computing_id == computing_id)
+    )
+    existing_user = await db_session.scalar(
+        sqlalchemy
+        .select(SiteUser)
+        .where(SiteUser.computing_id == computing_id)
+    )
+
+    if existing_user is None:
+        if existing_user_session is not None:
+            # log this strange case that shouldn't be possible
+            _logger.warning(f"User session {session_id} exists for non-existent user {computing_id} ... !")
+
+        # add new user to User table if it's their first time logging in
+        db_session.add(SiteUser(
+            computing_id=computing_id,
+            first_logged_in=datetime.now(),
+            last_logged_in=datetime.now()
+        ))
+
+    if existing_user_session is not None:
         existing_user_session.issue_time = datetime.now()
         existing_user_session.session_id = session_id
-        query = sqlalchemy.select(SiteUser).where(SiteUser.computing_id == computing_id)
-        existing_user = (await db_session.scalars(query)).first()
-        if existing_user is None:
-            # log this strange case
-            _logger = logging.getLogger(__name__)
-            _logger.warning(f"User session {session_id} exists for non-existent user {computing_id}!")
-            # create a user for this session
-            new_user = SiteUser(
-                computing_id=computing_id,
-                first_logged_in=datetime.now(),
-                last_logged_in=datetime.now()
-            )
-            db_session.add(new_user)
-        else:
+        if existing_user is not None:
             # update the last time the user logged in to now
-            existing_user.last_logged_in=datetime.now()
+            existing_user.last_logged_in = datetime.now()
     else:
-        # add new user to User table if it's their first time logging in
-        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 = SiteUser(
-                computing_id=computing_id,
-                first_logged_in=datetime.now(),
-                last_logged_in=datetime.now()
-            )
-            db_session.add(new_user)
-
-        new_user_session = UserSession(
-            issue_time=datetime.now(),
+        db_session.add(UserSession(
             session_id=session_id,
             computing_id=computing_id,
-        )
-        db_session.add(new_user_session)
+            issue_time=datetime.now(),
+        ))
 
 
 async def remove_user_session(db_session: AsyncSession, session_id: str) -> dict:

src/blog/crud.py

@@ -1,15 +1,12 @@
-import dataclasses
 import logging
 from datetime import date, datetime
-from typing import Optional
 
 import sqlalchemy
 from sqlalchemy import func
 
 import database
 from blog.models import BlogPosts
 
-_logger = logging.getLogger(__name__)
 
 async def create_new_entry(
     db_session: database.DBSession,

src/constants.py

@@ -2,7 +2,7 @@
 
 # TODO(future): replace new.sfucsss.org with sfucsss.org during migration
 # TODO(far-future): branch-specific root IP addresses (e.g., devbranch.sfucsss.org)
-FRONTEND_ROOT_URL = "http://127.0.0.1:8000" if os.environ.get("LOCAL") == "true" else "https://new.sfucsss.org"
+FRONTEND_ROOT_URL = "http://localhost:8080" if os.environ.get("LOCAL") == "true" else "https://new.sfucsss.org"
 GITHUB_ORG_NAME = "CSSS-Test-Organization" if os.environ.get("LOCAL") == "true" else "CSSS"
 
 W3_GUILD_ID = "1260652618875797504"

src/cron/daily.py

@@ -7,7 +7,7 @@
 import google_api
 import utils
 from database import _db_session
-from officers.crud import all_officer_terms, get_user_by_username, officer_terms
+from officers.crud import all_officer_data, get_user_by_username
 
 _logger = logging.getLogger(__name__)
 
@@ -18,7 +18,7 @@ async def update_google_permissions(db_session):
 
     # TODO: for performance, only include officers with recent end-date (1 yr)
     # but measure performance first
-    for term in await all_officer_terms(db_session):
+    for term in await all_officer_data(db_session):
         if utils.is_active(term):
             # TODO: if google drive permission is not active, update them
             pass
@@ -31,7 +31,7 @@ async def update_google_permissions(db_session):
 async def update_github_permissions(db_session):
     github_permissions, team_id_map = github.all_permissions()
 
-    for term in await all_officer_terms(db_session):
+    for term in await all_officer_data(db_session):
         new_teams = (
             # move all active officers to their respective teams
             github.officer_teams(term.position)

src/load_test_db.py

@@ -3,7 +3,7 @@
 # python load_test_db.py
 
 import asyncio
-from datetime import date, datetime, timedelta
+from datetime import date, timedelta
 
 import sqlalchemy
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -60,6 +60,7 @@ async def reset_db(engine):
 async def load_test_auth_data(db_session: AsyncSession):
     await create_user_session(db_session, "temp_id_314", "abc314")
     await update_site_user(db_session, "temp_id_314", "www.my_profile_picture_url.ca/test")
+    await db_session.commit()
 
 async def load_test_officers_data(db_session: AsyncSession):
     print("login the 3 users, putting them in the site users table")
@@ -109,7 +110,6 @@ async def load_test_officers_data(db_session: AsyncSession):
     ))
     await db_session.commit()
 
-    # TODO: will id autoincrement?
     await create_new_officer_term(db_session, OfficerTerm(
         computing_id="abc11",
 
@@ -213,11 +213,11 @@ async def load_test_officers_data(db_session: AsyncSession):
     await db_session.commit()
 
 async def load_sysadmin(db_session: AsyncSession):
-    print("loading new sysadmin")
     # put your computing id here for testing purposes
     SYSADMIN_COMPUTING_ID = "gsa92"
 
-    await create_user_session(db_session, "temp_id_4", SYSADMIN_COMPUTING_ID)
+    print(f"loading new sysadmin '{SYSADMIN_COMPUTING_ID}'")
+    await create_user_session(db_session, f"temp_id_{SYSADMIN_COMPUTING_ID}", SYSADMIN_COMPUTING_ID)
     await create_new_officer_info(db_session, OfficerInfo(
         legal_name="Gabe Schulz",
         discord_id=None,
@@ -246,6 +246,40 @@ async def load_sysadmin(db_session: AsyncSession):
         biography="The systems are good o7",
         photo_url=None,
     ))
+    await create_new_officer_term(db_session, OfficerTerm(
+        computing_id=SYSADMIN_COMPUTING_ID,
+
+        position=OfficerPosition.FIRST_YEAR_REPRESENTATIVE,
+        start_date=date.today() - timedelta(days=(365*3)),
+        end_date=date.today() - timedelta(days=(365*2)),
+
+        nickname="Gabe",
+        favourite_course_0="MACM 101",
+        favourite_course_1="CMPT 125",
+
+        favourite_pl_0="C#",
+        favourite_pl_1="C++",
+
+        biography="o hey fellow kids \n\n\n I can newline",
+        photo_url=None,
+    ))
+    await create_new_officer_term(db_session, OfficerTerm(
+        computing_id=SYSADMIN_COMPUTING_ID,
+
+        position=OfficerPosition.SYSTEM_ADMINISTRATOR,
+        start_date=date.today() - timedelta(days=365),
+        end_date=None,
+
+        nickname="Gabe",
+        favourite_course_0="CMPT 379",
+        favourite_course_1="CMPT 295",
+
+        favourite_pl_0="Rust",
+        favourite_pl_1="C",
+
+        biography="The systems are good o7",
+        photo_url=None,
+    ))
     await db_session.commit()
 
 async def async_main(sessionmanager):

src/officers/constants.py

@@ -1,9 +1,3 @@
-import logging
-from enum import Enum
-from typing import Self
-
-_logger = logging.getLogger(__name__)
-
 class OfficerPosition:
     PRESIDENT = "president"
     VICE_PRESIDENT = "vice-president"
@@ -32,6 +26,13 @@ class OfficerPosition:
     def position_list() -> list[str]:
         return _OFFICER_POSITION_LIST
 
+    @staticmethod
+    def length_in_semesters(position: str) -> int | None:
+        # TODO: ask the committee to maintain a json file with all the important details from the constitution
+        # (I can create the version version of the file)
+        """How many semester position is active for, according to the CSSS Constitution"""
+        return _LENGTH_MAP[position]
+
     @staticmethod
     def to_email(position: str) -> str | None:
         return _EMAIL_MAP.get(position, None)
@@ -43,13 +44,13 @@ def num_active(position: str) -> int | None:
         """
         # None means there can be any number active
         if (
-            position == OfficerPosition.ExecutiveAtLarge
-            or position == OfficerPosition.FirstYearRepresentative
+            position == OfficerPosition.EXECUTIVE_AT_LARGE
+            or position == OfficerPosition.FIRST_YEAR_REPRESENTATIVE
         ):
             return 2
         elif (
-            position == OfficerPosition.FroshWeekChair
-            or position == OfficerPosition.SocialMediaManager
+            position == OfficerPosition.FROSH_WEEK_CHAIR
+            or position == OfficerPosition.SOCIAL_MEDIA_MANAGER
         ):
             return None
         else:
@@ -61,38 +62,38 @@ def is_signer(position: str) -> bool:
         If the officer is a signing authority of the CSSS
         """
         return (
-            position == OfficerPosition.President
-            or position == OfficerPosition.VicePresident
-            or position == OfficerPosition.Treasurer
-            or position == OfficerPosition.DirectorOfResources
-            or position == OfficerPosition.DirectorOfEvents
+            position == OfficerPosition.PRESIDENT
+            or position == OfficerPosition.VICE_PRESIDENT
+            or position == OfficerPosition.TREASURER
+            or position == OfficerPosition.DIRECTOR_OF_RESOURCES
+            or position == OfficerPosition.DIRECTOR_OF_EVENTS
         )
 
     @staticmethod
     def expected_positions() -> list[str]:
         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,
+            OfficerPosition.PRESIDENT,
+            OfficerPosition.VICE_PRESIDENT,
+            OfficerPosition.TREASURER,
+
+            OfficerPosition.DIRECTOR_OF_RESOURCES,
+            OfficerPosition.DIRECTOR_OF_EVENTS,
+            OfficerPosition.DIRECTOR_OF_EDUCATIONAL_EVENTS,
+            OfficerPosition.ASSISTANT_DIRECTOR_OF_EVENTS,
+            OfficerPosition.DIRECTOR_OF_COMMUNICATIONS,
+            #OfficerPosition.DIRECTOR_OF_OUTREACH, # TODO: when https://github.com/CSSS/documents/pull/9/files merged
+            OfficerPosition.DIRECTOR_OF_MULTIMEDIA,
+            OfficerPosition.DIRECTOR_OF_ARCHIVES,
+            OfficerPosition.EXECUTIVE_AT_LARGE,
             # TODO: expect these only during fall & spring semesters. Also, TODO: this todo is correct...
-            #FirstYearRepresentative,
+            #OfficerPosition.FIRST_YEAR_REPRESENTATIVE,
 
             #ElectionsOfficer,
-            OfficerPosition.SFSSCouncilRepresentative,
-            OfficerPosition.FroshWeekChair,
+            OfficerPosition.SFSS_COUNCIL_REPRESENTATIVE,
+            OfficerPosition.FROSH_WEEK_CHAIR,
 
-            OfficerPosition.SystemAdministrator,
-            OfficerPosition.Webmaster,
+            OfficerPosition.SYSTEM_ADMINISTRATOR,
+            OfficerPosition.WEBMASTER,
         ]
 
 _EMAIL_MAP = {
@@ -120,6 +121,34 @@ def expected_positions() -> list[str]:
     OfficerPosition.SOCIAL_MEDIA_MANAGER: "N/A",
 }
 
+# TODO: when an officer's start date is modified, update the end date as well if it's defined in this list
+# a number of semesters (a semester begins on the 1st of each four month period, starting january)
+# None, means that the length of the position does not have a set length in semesters
+_LENGTH_MAP = {
+    OfficerPosition.PRESIDENT: 3,
+    OfficerPosition.VICE_PRESIDENT: 3,
+    OfficerPosition.TREASURER: 3,
+
+    OfficerPosition.DIRECTOR_OF_RESOURCES: 3,
+    OfficerPosition.DIRECTOR_OF_EVENTS: 3,
+    OfficerPosition.DIRECTOR_OF_EDUCATIONAL_EVENTS: 3,
+    OfficerPosition.ASSISTANT_DIRECTOR_OF_EVENTS: 3,
+    OfficerPosition.DIRECTOR_OF_COMMUNICATIONS: 3,
+    #OfficerPosition.DIRECTOR_OF_OUTREACH: 3,
+    OfficerPosition.DIRECTOR_OF_MULTIMEDIA: 3,
+    OfficerPosition.DIRECTOR_OF_ARCHIVES: 3,
+    OfficerPosition.EXECUTIVE_AT_LARGE: 1,
+    OfficerPosition.FIRST_YEAR_REPRESENTATIVE: 2,
+
+    OfficerPosition.ELECTIONS_OFFICER: None,
+    OfficerPosition.SFSS_COUNCIL_REPRESENTATIVE: 3,
+    OfficerPosition.FROSH_WEEK_CHAIR: None,
+
+    OfficerPosition.SYSTEM_ADMINISTRATOR: None,
+    OfficerPosition.WEBMASTER: None,
+    OfficerPosition.SOCIAL_MEDIA_MANAGER: None,
+}
+
 _OFFICER_POSITION_LIST = [
     OfficerPosition.PRESIDENT,
     OfficerPosition.VICE_PRESIDENT,