wip: add endpoint information for officer terms and officer info POST

Author: jbriones1

src/nominees/urls.py

@@ -8,7 +8,7 @@
     NomineeInfoUpdateParams,
 )
 from nominees.tables import NomineeInfo
-from utils.urls import admin_or_raise
+from utils.urls import AdminTypeEnum, admin_or_raise
 
 router = APIRouter(
     prefix="/nominee",
@@ -30,7 +30,7 @@ async def get_nominee_info(
     computing_id: str
 ):
     # Putting this one behind the admin wall since it has contact information
-    await admin_or_raise(request, db_session)
+    await admin_or_raise(request, db_session, AdminTypeEnum.Election)
     nominee_info = await nominees.crud.get_nominee_info(db_session, computing_id)
     if nominee_info is None:
         raise HTTPException(
@@ -56,7 +56,7 @@ async def provide_nominee_info(
     computing_id: str
 ):
     # TODO: There needs to be a lot more validation here.
-    await admin_or_raise(request, db_session)
+    await admin_or_raise(request, db_session, AdminTypeEnum.Election)
 
     updated_data = {}
     # Only update fields that were provided

src/officers/crud.py

@@ -98,6 +98,19 @@ async def get_officer_info_or_raise(db_session: database.DBSession, computing_id
         raise HTTPException(status_code=404, detail=f"officer_info for computing_id={computing_id} does not exist yet")
     return officer_term
 
+async def get_new_officer_info_or_raise(db_session: database.DBSession, computing_id: str) -> OfficerInfo:
+    """
+    This check is for after a create/update
+    """
+    officer_term = await db_session.scalar(
+        sqlalchemy
+        .select(OfficerInfo)
+        .where(OfficerInfo.computing_id == computing_id)
+    )
+    if officer_term is None:
+        raise HTTPException(status_code=500, detail=f"failed to fetch {computing_id} after update")
+    return officer_term
+
 async def get_officer_terms(
     db_session: database.DBSession,
     computing_id: str,

src/officers/models.py

@@ -5,14 +5,14 @@
 from officers.constants import OFFICER_LEGAL_NAME_MAX, OfficerPositionEnum
 
 
-class OfficerBaseModel(BaseModel):
+class OfficerInfoBaseModel(BaseModel):
     # TODO (#71): compute this using SFU's API & remove from being uploaded
     legal_name: str = Field(..., max_length=OFFICER_LEGAL_NAME_MAX)
     position: OfficerPositionEnum
     start_date: date
     end_date: date | None = None
 
-class PublicOfficerResponse(OfficerBaseModel):
+class PublicOfficerInfoResponse(OfficerInfoBaseModel):
     """
     Response when fetching public officer data
     """
@@ -24,7 +24,7 @@ class PublicOfficerResponse(OfficerBaseModel):
     biography: str | None = None
     csss_email: str
 
-class PrivateOfficerResponse(PublicOfficerResponse):
+class PrivateOfficerInfoResponse(PublicOfficerInfoResponse):
     """
     Response when fetching private officer data
     """
@@ -33,11 +33,40 @@ class PrivateOfficerResponse(PublicOfficerResponse):
     github_username: str | None = None
     google_drive_email: str | None = None
 
+class OfficerSelfUpdate(BaseModel):
+    """
+    Used when an Officer is updating their own information
+    """
+    nickname: str | None = None
+    discord_id: str | None = None
+    discord_name: str | None = None
+    discord_nickname: str | None = None
+    biography: str | None = None
+    phone_number: str | None = None
+    github_username: str | None = None
+    google_drive_email: str | None = None
+
+class OfficerUpdate(OfficerSelfUpdate):
+    """
+    Used when an admin is updating an Officer's info
+    """
+    legal_name: str | None = Field(None, max_length=OFFICER_LEGAL_NAME_MAX)
+    position: OfficerPositionEnum | None = None
+    start_date: date | None = None
+    end_date: date | None = None
+
 class OfficerTermBaseModel(BaseModel):
     computing_id: str
     position: OfficerPositionEnum
     start_date: date
 
+class OfficerTermCreate(OfficerTermBaseModel):
+    """
+    Params to create a new Officer Term
+    """
+    legal_name: str
+
+
 class OfficerTermResponse(OfficerTermBaseModel):
     id: int
     end_date: date | None = None

src/officers/tables.py

@@ -21,6 +21,7 @@
 )
 from database import Base
 from officers.constants import OFFICER_LEGAL_NAME_MAX, OFFICER_POSITION_MAX, OfficerPositionEnum
+from officers.models import OfficerSelfUpdate, OfficerUpdate
 
 
 # A row represents an assignment of a person to a position.
@@ -149,6 +150,11 @@ def serializable_dict(self) -> dict:
             "google_drive_email": self.google_drive_email,
         }
 
+    def update_from_params(self, params: OfficerUpdate | OfficerSelfUpdate):
+        update_data = params.model_dump(exclude_unset=True)
+        for k, v in update_data.items():
+            setattr(update_data, k, v)
+
     def is_filled_in(self):
         return (
             self.computing_id is not None

src/officers/urls.py

@@ -5,12 +5,19 @@
 import database
 import officers.crud
 import utils
-from officers.models import OfficerTermResponse, PrivateOfficerResponse, PublicOfficerResponse
+from officers.models import (
+    OfficerSelfUpdate,
+    OfficerTermCreate,
+    OfficerTermResponse,
+    OfficerUpdate,
+    PrivateOfficerInfoResponse,
+    PublicOfficerInfoResponse,
+)
 from officers.tables import OfficerInfo, OfficerTerm
 from officers.types import InitialOfficerInfo, OfficerInfoUpload, OfficerTermUpload
 from permission.types import OfficerPrivateInfo, WebsiteAdmin
-from utils.shared_models import DetailModel
-from utils.urls import logged_in_or_raise
+from utils.shared_models import DetailModel, SuccessResponse
+from utils.urls import admin_or_raise, is_website_admin, logged_in_or_raise
 
 router = APIRouter(
     prefix="/officers",
@@ -42,7 +49,7 @@ async def _has_officer_private_info_access(
 @router.get(
     "/current",
     description="Get information about the current officers. With no authorization, only get basic info.",
-    response_model=list[PrivateOfficerResponse] | list[PublicOfficerResponse],
+    response_model=list[PrivateOfficerInfoResponse] | list[PublicOfficerInfoResponse],
     operation_id="get_current_officers"
 )
 async def current_officers(
@@ -62,9 +69,9 @@ async def current_officers(
 @router.get(
     "/all",
     description="Information for all execs from all exec terms",
-    response_model=list[PrivateOfficerResponse] | list[PublicOfficerResponse],
+    response_model=list[PrivateOfficerInfoResponse] | list[PublicOfficerInfoResponse],
     responses={
-        401: { "description": "not authorized to view private info", "model": DetailModel }
+        403: { "description": "not authorized to view private info", "model": DetailModel }
     },
     operation_id="get_all_officers"
 )
@@ -97,7 +104,7 @@ async def all_officers(
     responses={
         401: { "description": "not authorized to view private info", "model": DetailModel }
     },
-    operation_id="get_all_officers"
+    operation_id="get_officer_terms_by_id"
 )
 async def get_officer_terms(
     request: Request,
@@ -121,8 +128,13 @@ async def get_officer_terms(
     ])
 
 @router.get(
-    "/info/{computing_id}",
+    "/info/{computing_id:str}",
     description="Get officer info for the current user, if they've ever been an exec. Only admins can get info about another user.",
+    response_model=PrivateOfficerInfoResponse,
+    responses={
+        403: { "description": "not authorized to view author user info", "model": DetailModel }
+    },
+    operation_id="get_officer_info_by_id"
 )
 async def get_officer_info(
     request: Request,
@@ -145,28 +157,27 @@ async def get_officer_info(
         Only the sysadmin, president, or DoA can submit this request. It will usually be the DoA.
         Updates the system with a new officer, and enables the user to login to the system to input their information.
     """,
+    response_model=SuccessResponse,
+    responses={
+        403: { "description": "must be a website admin", "model": DetailModel },
+        500: { "model": DetailModel },
+    },
+    operation_id="create_officer_term"
 )
 async def new_officer_term(
     request: Request,
     db_session: database.DBSession,
-    officer_info_list: list[InitialOfficerInfo] = Body(), # noqa: B008
+    officer_info_list: list[OfficerTermCreate],
 ):
-    """
-    If the current computing_id is not already an officer, officer_info will be created for them.
-    """
-    for officer_info in officer_info_list:
-        officer_info.valid_or_raise()
-
-    _, session_computing_id = await logged_in_or_raise(request, db_session)
-    await WebsiteAdmin.has_permission_or_raise(db_session, session_computing_id)
+    await admin_or_raise(request, db_session)
 
     for officer_info in officer_info_list:
         # if user with officer_info.computing_id has never logged into the website before,
         # a site_user tuple will be created for them.
         await officers.crud.create_new_officer_info(db_session, OfficerInfo(
             computing_id = officer_info.computing_id,
             # TODO (#71): use sfu api to get legal name from officer_info.computing_id
-            legal_name = "default name",
+            legal_name = officer_info.legal_name,
             phone_number = None,
 
             discord_id = None,
@@ -183,47 +194,43 @@ async def new_officer_term(
         ))
 
     await db_session.commit()
-    return PlainTextResponse("ok")
+    return JSONResponse({ "success": True })
 
 @router.patch(
-    "/info/{computing_id}",
+    "/info/{computing_id:str}",
     description="""
         After election, officer computing ids are input into our system.
         If you have been elected as a new officer, you may authenticate with SFU CAS,
         then input your information & the valid token for us. Admins may update this info.
-    """
+    """,
+    response_model=OfficerPrivateInfo,
+    responses={
+        403: { "description": "must be a website admin", "model": DetailModel },
+        500: { "description": "failed to fetch after update", "model": DetailModel },
+    },
+    operation_id="create_officer_term"
 )
 async def update_info(
     request: Request,
     db_session: database.DBSession,
     computing_id: str,
-    officer_info_upload: OfficerInfoUpload = Body() # noqa: B008
+    officer_info_upload: OfficerUpdate | OfficerSelfUpdate
 ):
-    officer_info_upload.valid_or_raise()
-    _, session_computing_id = await logged_in_or_raise(request, db_session)
+    is_site_admin, _, session_computing_id = await is_website_admin(request, db_session)
 
-    if computing_id != session_computing_id:
-        await WebsiteAdmin.has_permission_or_raise(
-            db_session, session_computing_id,
-            errmsg="must have website admin permissions to update another user"
-        )
+    if computing_id != session_computing_id and not is_site_admin:
+        raise HTTPException(status_code=403, detail="you may not update other officers")
 
     old_officer_info = await officers.crud.get_officer_info_or_raise(db_session, computing_id)
-    validation_failures, corrected_officer_info = await officer_info_upload.validate(computing_id, old_officer_info)
+    old_officer_info.update_from_params(officer_info_upload)
+    await officers.crud.update_officer_info(db_session, old_officer_info)
 
     # TODO (#27): log all important changes just to a .log file & persist them for a few years
 
-    success = await officers.crud.update_officer_info(db_session, corrected_officer_info)
-    if not success:
-        raise HTTPException(status_code=400, detail="officer_info does not exist yet, please create the officer info entry first")
-
     await db_session.commit()
 
-    updated_officer_info = await officers.crud.get_officer_info_or_raise(db_session, computing_id)
-    return JSONResponse({
-        "officer_info": updated_officer_info.serializable_dict(),
-        "validation_failures": validation_failures,
-    })
+    updated_officer_info = await officers.crud.get_new_officer_info_or_raise(db_session, computing_id)
+    return JSONResponse(updated_officer_info)
 
 @router.patch(
     "/term/{term_id}",

src/registrations/urls.py

@@ -18,7 +18,7 @@
 )
 from registrations.tables import NomineeApplication
 from utils.shared_models import DetailModel, SuccessResponse
-from utils.urls import admin_or_raise, logged_in_or_raise, slugify
+from utils.urls import AdminTypeEnum, admin_or_raise, logged_in_or_raise, slugify
 
 router = APIRouter(
     prefix="/registration",
@@ -74,7 +74,7 @@ async def register_in_election(
     body: NomineeApplicationParams,
     election_name: str
 ):
-    await admin_or_raise(request, db_session)
+    await admin_or_raise(request, db_session, AdminTypeEnum.Election)
 
     if body.position not in OfficerPositionEnum:
         raise HTTPException(
@@ -157,7 +157,7 @@ async def update_registration(
     computing_id: str,
     position: OfficerPositionEnum
 ):
-    await admin_or_raise(request, db_session)
+    await admin_or_raise(request, db_session, AdminTypeEnum.Election)
 
     if body.position not in OfficerPositionEnum:
         raise HTTPException(
@@ -221,7 +221,7 @@ async def delete_registration(
     position: OfficerPositionEnum,
     computing_id: str
 ):
-    await admin_or_raise(request, db_session)
+    await admin_or_raise(request, db_session, AdminTypeEnum.Election)
 
     if position not in OfficerPositionEnum:
         raise HTTPException(

src/utils/urls.py

@@ -1,4 +1,5 @@
 import re
+from enum import Enum
 
 from fastapi import HTTPException, Request, status
 
@@ -8,6 +9,11 @@
 from permission.types import ElectionOfficer, WebsiteAdmin
 
 
+class AdminTypeEnum(Enum):
+    Full = 1
+    Election = 2
+
+
 # TODO: move other utils into this module
 def slugify(text: str) -> str:
     """Creates a unique slug based on text passed in. Assumes non-unicode text."""
@@ -49,7 +55,8 @@ async def get_current_user(request: Request, db_session: database.DBSession) ->
 
     return session_id, session_computing_id
 
-async def admin_or_raise(request: Request, db_session: database.DBSession) -> tuple[str, str]:
+# TODO: Add an election admin version that checks the election attempting to be modified as well
+async def admin_or_raise(request: Request, db_session: database.DBSession, admintype: AdminTypeEnum = AdminTypeEnum.Full) -> tuple[str, str]:
     session_id, computing_id = await get_current_user(request, db_session)
     if not session_id or not computing_id:
         raise HTTPException(
@@ -58,11 +65,20 @@ async def admin_or_raise(request: Request, db_session: database.DBSession) -> tu
         )
 
     # where valid means election officer or website admin
-    if (await ElectionOfficer.has_permission(db_session, computing_id)) or (await WebsiteAdmin.has_permission(db_session, computing_id)):
+    if (await WebsiteAdmin.has_permission(db_session, computing_id)) or (admintype is AdminTypeEnum.Election and await ElectionOfficer.has_permission(db_session, computing_id)):
         return session_id, computing_id
-    else:
-        raise HTTPException(
-            status_code=status.HTTP_403_FORBIDDEN,
-            detail="must be an admin"
-        )
 
+    raise HTTPException(
+        status_code=status.HTTP_403_FORBIDDEN,
+        detail="must be an admin"
+    )
+
+async def is_website_admin(request: Request, db_session: database.DBSession) -> tuple[bool, str | None, str | None]:
+    session_id, computing_id = await get_current_user(request, db_session)
+    if session_id is None or computing_id is None:
+        return False, session_id, computing_id
+
+    if (await WebsiteAdmin.has_permission(db_session, computing_id)):
+        return True, session_id, computing_id
+
+    return False, session_id, computing_id