Add admin endpoints (#60)

* add endpoint & crud function for all officers

* comment out unused endpoints

* add website admin permission

* add endpoint for admin permission

Author: EarthenSky

src/main.py

@@ -3,6 +3,7 @@
 import auth.urls
 import database
 import officers.urls
+import permission.urls
 from fastapi import FastAPI
 
 import tests.urls
@@ -14,6 +15,8 @@
 
 app.include_router(auth.urls.router)
 app.include_router(officers.urls.router)
+app.include_router(permission.urls.router)
+
 app.include_router(tests.urls.router)
 
 @app.get("/")

src/officers/crud.py

@@ -31,6 +31,28 @@ async def most_recent_exec_term(db_session: database.DBSession, computing_id: st
     # TODO: can this be replaced with scalar to improve performance?
     return (await db_session.scalars(query)).first()
 
+# TODO: test this function
+async def current_officer_position(db_session: database.DBSession, computing_id: str) -> str | None:
+    """
+    Returns None if the user is not currently an officer
+    """
+    query = sqlalchemy.select(OfficerTerm)
+    query = query.where(OfficerTerm.computing_id == computing_id)
+    # TODO: assert this constraint at the SQL level, so that we don't even have to check it.
+    query = query.where(
+        # TODO: turn this query into a utility function, so it can be reused
+        OfficerTerm.is_filled_in
+        and (
+            # executives without a specified end_date are considered active
+            OfficerTerm.end_date is None
+            # check that today's timestamp is before (smaller than) the term's end date
+            or (datetime.today() <= OfficerTerm.end_date)
+        )
+    )
+    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]]:
     """
@@ -88,7 +110,7 @@ async def current_executive_team(db_session: database.DBSession, include_private
 
         officer_data[term.position] += [
             OfficerData(
-                is_current_officer = True,
+                is_active = True,
 
                 position = term.position,
                 start_date = term.start_date,
@@ -136,6 +158,59 @@ async def current_executive_team(db_session: database.DBSession, include_private
     return officer_data
 
 
+async def all_officer_terms(db_session: database.DBSession, include_private: bool) -> list[OfficerData]:
+    """
+    Orders officers recent first.
+
+    This could be a lot of data, so be careful.
+
+    TODO: optionally paginate data, so it's not so bad.
+    """
+    query = sqlalchemy.select(OfficerTerm)
+    query = query.order_by(OfficerTerm.start_date.desc())
+    officer_terms = (await db_session.scalars(query)).all()
+
+    officer_data_list = []
+    for term in officer_terms:
+        officer_info_query = sqlalchemy.select(OfficerInfo)
+        officer_info_query = officer_info_query.where(
+            OfficerInfo.computing_id == term.computing_id
+        )
+        officer_info = (await db_session.scalars(officer_info_query)).first()
+
+        officer_data_list += [
+            OfficerData(
+                is_active = (term.end_date is None) or (datetime.today() <= term.end_date),
+
+                position = term.position,
+                start_date = term.start_date,
+                end_date = term.end_date,
+
+                legal_name = officer_info.legal_name,
+                nickname = term.nickname,
+                discord_name = officer_info.discord_name,
+                discord_nickname = 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 = officer_info.phone_number,
+                    github_username = officer_info.github_username,
+                    google_drive_email = officer_info.google_drive_email,
+                ) if include_private else None,
+            )
+        ]
+
+    return officer_data_list
+
 # TODO: do we ever expect to need to remove officer info? Probably not? Just updating it.
 def update_officer_info(db_session: database.DBSession, officer_info_data: OfficerInfoData):
     """

src/officers/schemas.py

@@ -48,7 +48,7 @@ class OfficerPrivateData:
 
 @dataclass
 class OfficerData:
-    is_current_officer: bool
+    is_active: bool
 
     # an officer may have multiple positions, such as FroshWeekChair & DirectorOfEvents
     position: str
@@ -62,7 +62,6 @@ class OfficerData:
 
     favourite_course_0: str | None
     favourite_course_1: str | None
-
     favourite_language_0: str | None
     favourite_language_1: str | None
 
@@ -128,7 +127,7 @@ class OfficerPositionData_Upload(BaseModel):
 """
 # TODO: this is for another api call, where the doa should be able to change any aspect of the person's configuration, just in case
 class OfficerDataUpdate_Upload(BaseModel):
-    is_current_officer: bool
+    is_active: bool
 
     # an officer may have multiple positions, such as FroshWeekChair & DirectorOfEvents
     position: str

src/officers/urls.py

@@ -43,19 +43,37 @@ async def current_officers(
     return JSONResponse(json_current_executives)
 
 
-# TODO: test this error afterwards
-@router.get("/please_error", description="Raises an error & should send an email to the sysadmin")
-async def raise_error():
-    raise ValueError("This is an error, you're welcome")
-
-
 @router.get(
-    "/past",
-    description="Information from past 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.",
+    "/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.",
 )
-async def past_officers():
-    return {"officers": "none"}
+async def all_officers(
+    request: Request,
+    db_session: database.DBSession,
+):
+    # determine if 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)
 
+    all_officers = await officers.crud.all_officers(db_session, include_private=has_private_access)
+    """
+    json_current_executives = {
+        position: [
+            officer_data.serializable_dict() for officer_data in officer_data_list
+        ] for position, officer_data_list in all_officers.items()
+    }
+    """
+    return JSONResponse(all_officers)
+
+"""
+# TODO: test this error later
+@router.get("/please_error", description="Raises an error & should send an email to the sysadmin")
+async def raise_error():
+    raise ValueError("This is an error, you're welcome")
 
 @router.post(
     "/enter_info",
@@ -68,15 +86,12 @@ async def enter_info():
 
     return {}
 
-
-"""
 @router.get(
     "/my_info",
     description="Get info about whether you are still an executive or not / what your position is.",
 )
 async def my_info():
     return {}
-"""
 
 
 @router.post(
@@ -101,3 +116,4 @@ async def remove_officer():
 )
 async def update_officer():
     return {}
+"""

src/permission/permission.py

@@ -1,8 +1,10 @@
 from datetime import UTC, datetime, timezone
+from typing import ClassVar
 
 import database
 import officers.crud
 from data.semesters import current_semester_start, step_semesters
+from officers.constants import OfficerPosition
 
 
 class OfficerPrivateInfo:
@@ -23,3 +25,23 @@ async def has_permission(db_session: database.DBSession, computing_id: str) -> b
         cutoff_date = step_semesters(semester_start, -NUM_SEMESTERS)
 
         return most_recent_exec_term > cutoff_date
+
+class WebsiteAdmin:
+    WEBSITE_ADMIN_POSITIONS: ClassVar[list[OfficerPosition]] = [
+        OfficerPosition.President,
+        OfficerPosition.VicePresident,
+        OfficerPosition.DirectorOfArchives,
+        OfficerPosition.SystemAdministrator,
+        OfficerPosition.Webmaster,
+    ]
+
+    @staticmethod
+    async def has_permission(db_session: database.DBSession, computing_id: str) -> bool:
+        """
+        A website admin has to be one of the following positions, and
+        """
+        position = await officers.crud.current_officer_position(db_session, computing_id)
+        if position is None:
+            return False
+
+        return position in WebsiteAdmin.WEBSITE_ADMIN_POSITIONS

src/permission/types.py

@@ -1,8 +1,29 @@
-from fastapi import APIRouter
+import auth.crud
+import database
+from fastapi import APIRouter, Request
+from fastapi.responses import JSONResponse
+
+from permission.types import WebsiteAdmin
 
 router = APIRouter(
     prefix="/permission",
     tags=["permission"],
 )
 
 # TODO: add an endpoint for viewing permissions that exist & what levels they can have & what levels each person has
+@router.get(
+    "/is_admin",
+    description="checks if the current user has the admin permission"
+)
+async def is_admin(
+    request: Request,
+    db_session: database.DBSession,
+):
+    session_id = request.cookies.get("session_id", None)
+    if session_id is None:
+        return JSONResponse({"is_admin": False})
+
+    # what if user doesn't have a computing id?
+    computing_id = await auth.crud.get_computing_id(db_session, session_id)
+    is_admin_permission = await WebsiteAdmin.has_permission(db_session, computing_id)
+    return JSONResponse({"is_admin": is_admin_permission})

src/permission/urls.py

@@ -4,7 +4,7 @@
 import pytest
 from database import SQLALCHEMY_TEST_DATABASE_URL, DatabaseSessionManager
 from officers.constants import OfficerPosition
-from officers.crud import current_executive_team, most_recent_exec_term
+from officers.crud import all_officer_terms, current_executive_team, most_recent_exec_term
 
 # TODO: setup a database on the CI machine & run this as a unit test then (since
 # this isn't really an integration test)
@@ -55,6 +55,10 @@ async def test__read_execs(database_setup):
         assert next(iter(current_exec_team.values()))[0].private_data is not None
         assert next(iter(current_exec_team.values()))[0].private_data.computing_id == "abc33"
 
+        all_terms = await all_officer_terms(db_session, include_private=True)
+        assert len(all_terms) == 3
+
+
 #async def test__update_execs(database_setup):
 #    # TODO: the second time an update_officer_info call occurs, the user should be updated with info
 #    pass