clean & fix POST /term

Author: EarthenSky

src/officers/crud.py

@@ -14,6 +14,8 @@
 
 _logger = logging.getLogger(__name__)
 
+# NOTE: this module should do any data validation; that should be done in the urls.py or higher layer
+
 async def most_recent_officer_term(db_session: database.DBSession, computing_id: str) -> OfficerTerm | None:
     """
     Returns the most recent OfficerTerm an exec has held
@@ -82,7 +84,7 @@ async def current_officers(
 async def all_officers(
     db_session: database.DBSession,
     include_private_data: bool,
-    include_future_terms: bool,
+    include_future_terms: bool
 ) -> list[OfficerData]:
     """
     This could be a lot of data, so be careful
@@ -138,19 +140,17 @@ async def get_officer_terms(
     )
     if not include_future_terms:
         query = utils.has_started_term(query)
-
     if max_terms is not None:
         query.limit(max_terms)
 
     return (await db_session.scalars(query)).all()
 
 async def get_officer_term_by_id(db_session: database.DBSession, term_id: int) -> OfficerTerm:
-    query = (
+    officer_term = await db_session.scalar(
         sqlalchemy
         .select(OfficerTerm)
         .where(OfficerTerm.id == term_id)
     )
-    officer_term = await db_session.scalar(query)
     if officer_term is None:
         raise HTTPException(status_code=400, detail=f"Could not find officer_term with id={term_id}")
     return officer_term
@@ -159,16 +159,13 @@ async def create_new_officer_info(
     db_session: database.DBSession,
     new_officer_info: OfficerInfo
 ) -> bool:
-    """
-    Return False if the officer already exists
-    """
-    query = (
+    """Return False if the officer already exists & don't do anything."""
+    existing_officer_info = await db_session.scalar(
         sqlalchemy
         .select(OfficerInfo)
         .where(OfficerInfo.computing_id == new_officer_info.computing_id)
     )
-    stored_officer_info = await db_session.scalar(query)
-    if stored_officer_info is not None:
+    if existing_officer_info is not None:
         return False
 
     db_session.add(new_officer_info)
@@ -178,13 +175,9 @@ async def create_new_officer_term(
     db_session: database.DBSession,
     new_officer_term: OfficerTerm
 ):
-    # TODO: does this check need to be here?
-    # if new_officer_term.position not in OfficerPosition.position_list():
-    #     raise HTTPException(status_code=500)
-
-    # when creating a new position, assign a default end date if one exists
     position_length = OfficerPosition.length_in_semesters(new_officer_term.position)
     if position_length is not None:
+        # when creating a new position, assign a default end date if one exists
         new_officer_term.end_date = semesters.step_semesters(
             semesters.current_semester_start(new_officer_term.start_date),
             position_length,

src/officers/tables.py

@@ -27,7 +27,9 @@
 class OfficerTerm(Base):
     __tablename__ = "officer_term"
 
+    # TODO: change primary key to computing_id, position, start_date?
     id = Column(Integer, primary_key=True, autoincrement=True)
+
     computing_id = Column(
         String(COMPUTING_ID_LEN),
         ForeignKey("site_user.computing_id"),

src/officers/types.py

@@ -18,12 +18,10 @@ class OfficerInfoUpload:
     github_username: None | str = None
     google_drive_email: None | str = None
 
-    def validate(self) -> None | HTTPException:
-        if self.legal_name is not None and self.legal_name == "":
-            return HTTPException(status_code=400, detail="legal name must not be empty")
+    def valid_or_raise(self):
         # TODO: more checks
-        else:
-            return None
+        if self.legal_name is not None and self.legal_name == "":
+            raise HTTPException(status_code=400, detail="legal name must not be empty")
 
     def to_officer_info(self, computing_id: str, discord_id: str | None, discord_nickname: str | None) -> OfficerInfo:
         return OfficerInfo(
@@ -58,8 +56,7 @@ class OfficerTermUpload:
     # NOTE: changing the name of this variable without changing all instances is breaking
     photo_url: None | str = None
 
-    def validate(self):
-        """input validation"""
+    def valid_or_raise(self):
         # NOTE: An officer can change their own data for terms that are ongoing.
         if self.position not in OfficerPosition.position_list():
             raise HTTPException(status_code=400, detail=f"invalid new position={self.position}")

src/officers/urls.py

@@ -167,51 +167,66 @@ async def get_officer_info(
 
     return JSONResponse(officer_info.serializable_dict())
 
+# TODO: move this to types?
 @dataclass
 class InitialOfficerInfo:
     computing_id: str
     position: str
     start_date: date
 
+    def valid_or_raise(self):
+        if len(self.computing_id) > COMPUTING_ID_MAX:
+            raise HTTPException(status_code=400, detail=f"computing_id={self.computing_id} is too large")
+        elif self.computing_id == "":
+            raise HTTPException(status_code=400, detail="computing_id cannot be empty")
+        elif self.position not in OfficerPosition.position_list():
+            raise HTTPException(status_code=400, detail=f"invalid position={self.position}")
+
 @router.post(
     "/term",
-    description="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.",
+    description="""
+        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.
+    """,
 )
 async def new_officer_term(
     request: Request,
     db_session: database.DBSession,
-    officer_info_list: list[InitialOfficerInfo] = Body(),  # noqa: B008
+    officer_info_list: list[InitialOfficerInfo] = Body(), # noqa: B008
 ):
     """
     If the current computing_id is not already an officer, officer_info will be created for them.
     """
     for officer_info in officer_info_list:
-        if len(officer_info.computing_id) > COMPUTING_ID_MAX:
-            raise HTTPException(status_code=400, detail=f"computing_id={officer_info.computing_id} is too large")
-        elif officer_info.position not in OfficerPosition.position_list():
-            raise HTTPException(status_code=400, detail=f"invalid position={officer_info.position}")
+        officer_info.valid_or_raise()
 
-    WebsiteAdmin.validate_request(db_session, request)
+    _, session_computing_id = logged_in_or_raise(request, db_session)
+    WebsiteAdmin.has_permission_or_raise(db_session, session_computing_id)
 
     for officer_info in officer_info_list:
-        await officers.crud.create_new_officer_info(
-            db_session,
-            # TODO: do I need this object atm?
-            OfficerInfoUpload(
-                # TODO: use sfu api to get legal name
-                legal_name = "default name",
-            ).to_officer_info(officer_info.computing_id, None, None),
-        )
-        # TODO: update create_new_officer_term to be the same as create_new_officer_info
+        await officers.crud.create_new_officer_info(db_session, OfficerInfo(
+            computing_id = officer_info.computing_id,
+            # TODO: use sfu api to get legal name from officer_info.computing_id
+            legal_name = "default name",
+            phone_number = None,
+
+            discord_id = None,
+            discord_name = None,
+            discord_nickname = None,
+
+            google_drive_email = None,
+            github_username = None,
+        ))
         await officers.crud.create_new_officer_term(db_session, OfficerTerm(
             computing_id = officer_info.computing_id,
             position = officer_info.position,
             # TODO: remove the hours & seconds (etc.) from start_date
+            # TODO: start_date should be a Date, not a Datetime
             start_date = officer_info.start_date,
         ))
 
     await db_session.commit()
-    return PlainTextResponse("ok")
+    return PlainTextResponse("success")
 
 @router.patch(
     "/info/{computing_id}",
@@ -227,9 +242,7 @@ async def update_info(
     computing_id: str,
     officer_info_upload: OfficerInfoUpload = Body(), # noqa: B008
 ):
-    http_exception = officer_info_upload.validate()
-    if http_exception is not None:
-        raise http_exception
+    officer_info_upload.valid_or_raise()
 
     session_id = request.cookies.get("session_id", None)
     if session_id is None:
@@ -320,7 +333,7 @@ async def update_term(
     term_id: int,
     officer_term_upload: OfficerTermUpload = Body(), # noqa: B008
 ):
-    officer_term_upload.validate()
+    officer_term_upload.valid_or_raise()
 
     # Refactor all of these gets & raises into small functions
     session_id = request.cookies.get("session_id", None)
@@ -370,9 +383,9 @@ async def update_term(
         "validation_failures": [], # none for now, but may be important later
     })
 
-@router.post(
-    "/remove/{term_id}",
-    description="Remove the specified officer term. Only website admins can run this endpoint. BE CAREFUL WITH THIS!"
+@router.delete(
+    "/term/{term_id}",
+    description="Remove the specified officer term. Only website admins can run this endpoint. BE CAREFUL WITH THIS!",
 )
 async def remove_officer():
     # TODO: this

src/permission/types.py

@@ -52,15 +52,6 @@ async def has_permission(db_session: database.DBSession, computing_id: str) -> b
         return False
 
     @staticmethod
-    async def validate_request(db_session: database.DBSession, request: Request) -> bool:
-        """
-        Checks if the provided request satisfies these permissions, and raises the neccessary
-        exceptions if not
-        """
-        session_id = request.cookies.get("session_id", None)
-        if session_id is None:
-            raise HTTPException(status_code=401, detail="must be logged in")
-        else:
-            computing_id = await auth.crud.get_computing_id(db_session, session_id)
-            if not await WebsiteAdmin.has_permission(db_session, computing_id):
-                raise HTTPException(status_code=401, detail="must have website admin permissions")
+    async def has_permission_or_raise(db_session: database.DBSession, computing_id: str) -> bool:
+        if not await WebsiteAdmin.has_permission(db_session, computing_id):
+            raise HTTPException(status_code=401, detail="must have website admin permissions")