Add CSV Export for PhD Applications in a Hiring Term (#576)

* Scaffolding for PhD hiring

* Add CSV export for PhD applications

backend/api/academics/hiring.py

@@ -176,7 +176,7 @@ def get_hiring_summary_csv(
     term_id: str,
     subject: User = Depends(registered_user),
     hiring_service: HiringService = Depends(),
-) -> Paginated[HiringAssignmentSummaryOverview]:
+) -> StreamingResponse:
     """
     Returns the state of hiring as a summary.
     """
@@ -202,7 +202,7 @@ def get_applicants_for_site_csv(
     course_site_id: int,
     subject: User = Depends(registered_user),
     hiring_service: HiringService = Depends(),
-) -> Paginated[HiringAssignmentSummaryOverview]:
+) -> StreamingResponse:
     """
     Returns the state of hiring as a summary.
     """
@@ -221,3 +221,56 @@ def get_applicants_for_site_csv(
     response.headers["Content-Disposition"] = "attachment; filename=export.csv"
     # Return the response
     return response
+
+
+@api.get("/summary/{term_id}/phd_applicants", tags=["Hiring"])
+def get_hiring_summary_csv(
+    term_id: str,
+    subject: User = Depends(registered_user),
+    hiring_service: HiringService = Depends(),
+) -> StreamingResponse:
+    """
+    Returns the state of hiring as a summary.
+    """
+    data = hiring_service.get_phd_applicants(subject, term_id)
+    # Create IO Stream
+    stream = io.StringIO()
+    # Create dictionary writer to convert objects to CSV rows
+    # Note: __dict__ converts the Pydantic model into a dictionary of key-value
+    # pairs, enabling access of the object's keys.
+    keys = [
+        "id",
+        "last_name",
+        "first_name",
+        "onyen",
+        "email",
+        "program_pursued",
+        "intro_video_url",
+        "student_preferences",
+        "instructor_preferences",
+    ]
+    wr = csv.DictWriter(stream, delimiter=",", fieldnames=keys)
+    wr.writeheader()
+    rows = []
+    for d in data:
+        rows.append(
+            {
+                "id": d.id,
+                "last_name": d.applicant.last_name,
+                "first_name": d.applicant.first_name,
+                "onyen": d.applicant.onyen,
+                "email": d.applicant.email,
+                "program_pursued": d.program_pursued,
+                "intro_video_url": d.intro_video_url,
+                "student_preferences": ", ".join(d.student_preferences),
+                "instructor_preferences": ", ".join(d.instructor_preferences),
+            }
+        )
+    wr.writerows(rows)
+    # Create HTTP response of type `text/csv`
+    response = StreamingResponse(iter([stream.getvalue()]), media_type="text/csv")
+    response.headers["Content-Disposition"] = (
+        f"attachment; filename=phd_applicants_{term_id}.csv"
+    )
+    # Return the response
+    return response

backend/models/academics/hiring/phd_application.py

@@ -0,0 +1,14 @@
+from pydantic import BaseModel
+from ...public_user import PublicUser
+from ..section import CatalogSectionIdentity
+
+
+class PhDApplicationReview(BaseModel):
+    id: int
+    applicant: PublicUser
+    applicant_name: str
+    advisor: str | None
+    program_pursued: str
+    intro_video_url: str
+    student_preferences: list[str]
+    instructor_preferences: list[str]

backend/services/academics/hiring.py

@@ -14,7 +14,7 @@
 from ...models.user import User
 from ...models.academics.section_member import RosterRole
 from ...entities import UserEntity
-from ...models.application import ApplicationUnderReview
+from ...models.application import ApplicationUnderReview, ApplicationOverview
 from ...entities.academics import SectionEntity, TermEntity
 from ...entities.office_hours import CourseSiteEntity
 from ...entities.academics.section_member_entity import SectionMemberEntity
@@ -35,6 +35,7 @@
     ApplicationReviewStatus,
     ApplicationReviewCsvRow,
 )
+from ...models.academics.hiring.phd_application import PhDApplicationReview
 from ...models.academics.hiring.hiring_assignment import *
 from ...models.academics.hiring.hiring_level import *
 
@@ -149,13 +150,14 @@ def create_missing_course_sites_for_term(self, subject: User, term_id: str) -> b
         Creates missing course sites for a given term.
         """
         self._permission.enforce(
-            subject, "hiring.create_missing_course_sites_for_term", f"course_sites/*"
+            subject,
+            "hiring.create_missing_course_sites_for_term",
+            f"course_sites/term:{term_id}",
         )
 
         # Get a list of all sections that are not associated with course sites
         section_query = select(SectionEntity).where(
-            SectionEntity.term_id == term_id,
-            SectionEntity.course_site_id.is_(None)
+            SectionEntity.term_id == term_id, SectionEntity.course_site_id.is_(None)
         )
         joint: dict[tuple[str, str], list[SectionEntity]] = {}
         for section in self._session.scalars(section_query).all():
@@ -182,6 +184,73 @@ def create_missing_course_sites_for_term(self, subject: User, term_id: str) -> b
         self._session.commit()
         return True
 
+    def get_phd_applicants(
+        self, subject: User, term_id: str
+    ) -> list[PhDApplicationReview]:
+        self._permission.enforce(
+            subject, "hiring.get_phd_applicants", f"course_sites/term:{term_id}"
+        )
+
+        query = select(ApplicationEntity).where(
+            ApplicationEntity.term_id == term_id,
+            ApplicationEntity.type == "gta",
+            ApplicationEntity.program_pursued.in_({"PhD", "PhD (ABD)"}),
+        )
+        all = self._session.scalars(query).all()
+
+        # Create the models
+        phd_applications = {}
+        for application in all:
+            phd_application = PhDApplicationReview(
+                id=application.id,
+                applicant=application.user.to_public_model(),
+                applicant_name=application.user.full_name(),
+                advisor=application.advisor,
+                program_pursued=application.program_pursued,
+                intro_video_url=application.intro_video_url,
+                student_preferences=[],
+                instructor_preferences=[],
+            )
+            phd_applications[application.id] = phd_application
+
+        sections_query = select(SectionEntity).where(SectionEntity.term_id == term_id)
+        sections = {
+            section.id: section
+            for section in self._session.scalars(sections_query).all()
+        }
+
+        # Grab student preferences of sections
+        application_ids = list(phd_applications.keys())
+        section_application_query = (
+            select(section_application_table)
+            .where(section_application_table.c.application_id.in_(application_ids))
+            .order_by(section_application_table.c.preference)
+        )
+        for section_application in self._session.execute(section_application_query):
+            preference, section_id, application_id = section_application
+            phd_applications[application_id].student_preferences.append(
+                f"{preference}. {sections[section_id].course_id}.{sections[section_id].number}"
+            )
+
+        # Grab instructor preferences of applications
+        instructor_review_query = (
+            select(ApplicationReviewEntity)
+            .where(ApplicationReviewEntity.application_id.in_(application_ids))
+            .where(ApplicationReviewEntity.status == ApplicationReviewStatus.PREFERRED)
+            .options(
+                joinedload(ApplicationReviewEntity.course_site).joinedload(
+                    CourseSiteEntity.sections
+                )
+            )
+        )
+        instructor_preferences = self._session.scalars(instructor_review_query).all()
+        for review in instructor_preferences:
+            phd_applications[review.application_id].instructor_preferences.append(
+                f"{review.preference}. {review.course_site.sections[0].course_id}.{review.course_site.sections[0].number}"
+            )
+
+        return list(phd_applications.values())
+
     def _load_course_site(self, course_site_id: int) -> CourseSiteEntity:
         """
         Loads a course site given a subject and course site ID.
@@ -744,7 +813,7 @@ def get_hiring_summary_overview(
         )
 
         # 3. Fetch data and build summary model
-        length = self._session.scalar(count_query)
+        length = self._session.scalar(count_query) or 0
         assignment_entities = self._session.scalars(assignment_query).unique().all()
 
         return Paginated(

backend/test/services/academics/hiring/hiring_data.py

@@ -111,8 +111,28 @@
     term_id=term_data.current_term.id,
 )
 
-new_application = Application(
+application_five = Application(
     id=5,
+    type="gta",
+    user_id=user_data.root.id,
+    academic_hours=12,
+    extracurriculars="Many extracurriculars",
+    expected_graduation="Soon",
+    program_pursued="PhD",
+    other_programs="None",
+    gpa=3.8,
+    comp_gpa=4.0,
+    comp_227=Comp227.EITHER,
+    intro_video_url="https://www.youtube.com/watch?v=d6O6kyqjcYo",
+    prior_experience="None",
+    service_experience="None",
+    additional_experience="None",
+    preferred_sections=[],
+    term_id=term_data.current_term.id,
+)
+
+new_application = Application(
+    id=6,
     type="new_uta",
     user_id=user_data.ambassador.id,
     academic_hours=12,
@@ -146,7 +166,13 @@
     term_id=term_data.current_term.id,
 )
 
-applications = [application_one, application_two, application_three, application_four]
+applications = [
+    application_one,
+    application_two,
+    application_three,
+    application_four,
+    application_five,
+]
 
 application_associations = [
     (application_one, section_data.comp_301_001_current_term, 0),
@@ -155,6 +181,7 @@
     (application_two, section_data.comp_110_001_current_term, 0),
     (application_three, section_data.comp_110_001_current_term, 0),
     (application_four, section_data.comp_110_001_current_term, 0),
+    (application_five, section_data.comp_301_001_current_term, 0),
 ]
 
 review_one = ApplicationReview(

backend/test/services/academics/hiring/hiring_test.py

@@ -294,3 +294,12 @@ def test_create_missing_course_sites_for_term(
     hiring_svc.create_missing_course_sites_for_term(user, term.id)
     overview_post = hiring_svc.get_hiring_admin_overview(user, term.id)
     assert len(overview_post.sites) > len(overview_pre.sites)
+
+
+def test_get_phd_applicants(hiring_svc: HiringService):
+    user = user_data.root
+    term = term_data.current_term
+    applicants = hiring_svc.get_phd_applicants(user, term.id)
+    assert len(applicants) > 0
+    for applicant in applicants:
+        assert applicant.program_pursued in {"PhD", "PhD (ABD)"}