Outline scraper function prototypes

I'm basically following Single Responsibility Principle as I'm writing this new SIS scraper.

The course_search() method, previously get_courses_in_subject(), now gives the caller more control over how results should be sorted.

There are a lot of other miscellaneous changes included with this commit, but they're all basically for the sake of SRP.

Author: ramonechen

app/scrapers/sis_scraper.py

@@ -1,16 +1,26 @@
 import asyncio
 import json
 import time
+from enum import Enum
 
 import aiohttp
 import bs4
 
 
+class CourseColumn(str, Enum):
+    COURSE_TITLE = "courseTitle"
+    SUBJECT_DESCRIPTION = "subjectDescription"
+    COURSE_NUMBER = "courseNumber"
+    SECTION = "sequenceNumber"
+    CRN = "courseReferenceNumber"
+    TERM = "term"
+
+
 async def get_subjects(
     session: aiohttp.ClientSession, term: str
 ) -> list[dict[str, str]]:
     """
-    Fetches the list of subjects for a given term from the SIS API.
+    Fetches the list of subjects for a given term from SIS.
 
     Returned data format is as follows:
     [
@@ -22,56 +32,98 @@ async def get_subjects(
     ]
     """
     url = "https://sis9.rpi.edu/StudentRegistrationSsb/ssb/classSearch/get_subject"
-    async with session.get(
-        url, params={"term": term, "offset": 1, "max": 100}
-    ) as response:
+    params = {"term": term, "offset": 1, "max": 100}
+    async with session.get(url, params=params) as response:
         response.raise_for_status()
         data = await response.json()
         return data
 
 
-async def reset_subject_search(session: aiohttp.ClientSession, term: str) -> None:
+async def reset_course_search(session: aiohttp.ClientSession, term: str) -> None:
     """
-    Resets the subject search state on the SIS server.
+    Resets the term and subject search state on the SIS server.
 
-    Must be re-called before each new attempt to fetch courses from a different subject.
-    Otherwise, the server will continue returning results from the last subject accessed.
+    Must be called before each attempt to fetch courses from a subject in the given term.
+    Otherwise, the server will continue returning the same results from the last subject
+    accessed, or no data if attempting to access data from a different term.
     """
     url = "https://sis9.rpi.edu/StudentRegistrationSsb/ssb/term/search?mode=search"
-    async with session.get(url, params={"term": term}) as response:
+    params = {"term": term}
+    async with session.get(url, params=params) as response:
         response.raise_for_status()
 
 
-async def get_courses_in_subject(
-    session: aiohttp.ClientSession, term: str, subject: str
+async def course_search(
+    session: aiohttp.ClientSession,
+    term: str,
+    subject: str,
+    max_size: int = 1000,
+    sort_column: CourseColumn = CourseColumn.SUBJECT_DESCRIPTION,
+    sort_asc: bool = True,
 ) -> list[dict[str, str]]:
     """
-    Fetches the list of courses for a given subject and term from the SIS API.
+    Fetches the list of courses for a given subject and term from SIS.
 
-    reset_subject_search() must be called once before each call to this function.
+    The term and subject search state on the SIS server must be reset before each call
+    to this function.
     """
-    url = "https://sis9.rpi.edu/StudentRegistrationSsb/ssb/searchResults/searchResults"
+    url = "https://sis9.rpi.edu/StudentRegistrationSsb/ssb/searchResults/searchResults?pageOffset=0"
     params = {
         "txt_subject": subject,
         "txt_term": term,
-        "pageOffset": 0,
-        "pageMaxSize": 1000,
-        "sortColumn": "subjectDescription",
-        "sortDirection": "asc",
+        "pageMaxSize": max_size,
+        "sortColumn": sort_column,
+        "sortDirection": "asc" if sort_asc else "desc",
     }
     async with session.get(url, params=params) as response:
         response.raise_for_status()
         data = await response.json()
         return data
 
 
+async def get_course_details(session: aiohttp.ClientSession, term: str, crn: str):
+    url = (
+        "https://sis9.rpi.edu/StudentRegistrationSsb/ssb/searchResults/getClassDetails"
+    )
+
+
+async def get_course_description(session: aiohttp.ClientSession, term: str, crn: str):
+    url = "https://sis9.rpi.edu/StudentRegistrationSsb/ssb/searchResults/getCourseDescription"
+
+
+async def get_course_attributes(session: aiohttp.ClientSession, term: str, crn: str):
+    url = "https://sis9.rpi.edu/StudentRegistrationSsb/ssb/searchResults/getSectionAttributes"
+
+
+async def get_course_restrictions(session: aiohttp.ClientSession, term: str, crn: str):
+    url = (
+        "https://sis9.rpi.edu/StudentRegistrationSsb/ssb/searchResults/getRestrictions"
+    )
+
+
+async def get_course_corequisites(session: aiohttp.ClientSession, term: str, crn: str):
+    url = (
+        "https://sis9.rpi.edu/StudentRegistrationSsb/ssb/searchResults/getCorequisites"
+    )
+
+
+async def get_course_prerequisites(session: aiohttp.ClientSession, term: str, crn: str):
+    url = "https://sis9.rpi.edu/StudentRegistrationSsb/ssb/searchResults/getSectionPrerequisites"
+
+
+async def get_course_crosslists(session: aiohttp.ClientSession, term: str, crn: str):
+    url = (
+        "https://sis9.rpi.edu/StudentRegistrationSsb/ssb/searchResults/getXlstSections"
+    )
+
+
 async def main():
     """
     A JSESSIONID cookie is required before accessing any course data, which can be
     obtained on the first request to any SIS page. The cookie should automatically be
     included in subsequent requests made with the same aiohttp session.
 
-    reset_subject_search() must be called before each new attempt to fetch courses from
+    reset_course_search() must be called before each new attempt to fetch courses from
     a different subject.
     """
     async with aiohttp.ClientSession(
@@ -93,9 +145,10 @@ async def main():
     ) as session:
         term = "202409"
         subjects = await get_subjects(session, term)
-        await reset_subject_search(session, term)
-        data = await get_courses_in_subject(session, term, subjects[0]["code"])
-        print(json.dumps(data, indent=4))
+        await reset_course_search(session, term)
+        data = await course_search(session, term, subjects[0]["code"])
+        # print(json.dumps(data, indent=4))
+    return True
 
 
 if __name__ == "__main__":