Support Pensieve

Author: qria

main.py

@@ -2,10 +2,12 @@
 
 import src.handle_email_queue as handle_email
 import src.handle_flush_gradescope as handle_flush
+import src.handle_flush_pensieve as handle_flush_pensieve_module
 import src.handle_form_submit as handle_form
+from src.environment import Environment
 from src.errors import KnownError
 from src.slack import SlackManager
-from src.utils import Environment, truncate
+from src.utils import truncate
 
 
 def handle_email_queue(request):
@@ -44,6 +46,24 @@ def handle_flush_gradescope(request):
         Environment.clear()
 
 
+def handle_flush_pensieve(request):
+    request_json = request.get_json()
+    print("handle_flush_pensieve called on payload: " + json.dumps(request_json))
+    try:
+        handle_flush_pensieve_module.handle_flush_pensieve(request_json=request_json)
+        return {"success": True}
+    except KnownError as e:
+        print("Known Error Occurred: " + str(e) + f" (Request: {request_json})")
+        SlackManager().send_error(str(e) + f" (Request: {truncate(request_json)})")
+        return {"success": False, "error": str(e)}
+    except Exception as e:
+        print("Internal Error Occurred: " + str(e) + f" (Request: {request_json})")
+        SlackManager().send_error("Internal error: " + str(e) + f" (Request: {truncate(request_json)})")
+        raise
+    finally:
+        Environment.clear()
+
+
 def handle_form_submit(request):
     request_json = request.get_json()
     print("handle_form_submit called on payload: " + json.dumps(request_json))

src/assignments.py

@@ -24,13 +24,15 @@ def __init__(
         hard_due_date: Optional[datetime],
         partner: bool,
         gradescope: List[str],
+        pensieve: List[str],
     ) -> None:
         self.name = name
         self.id = id
         self.due_date = due_date
         self.hard_due_date = hard_due_date
         self.partner = partner
         self.gradescope = gradescope
+        self.pensieve = pensieve
 
     def is_past_due(self, request_time: str):
         """
@@ -67,6 +69,9 @@ def is_partner_assignment(self):
     def get_gradescope_assignment_urls(self) -> List[str]:
         return self.gradescope
 
+    def get_pensieve_assignment_urls(self) -> List[str]:
+        return self.pensieve
+
 
 class AssignmentList:
     """
@@ -86,6 +91,7 @@ def __init__(self, sheet: Sheet) -> None:
                         hard_due_date=cast_date(row.get("hard_due_date", ""), optional=True),
                         partner=cast_bool(row["partner"]),
                         gradescope=cast_list_str(row.get("gradescope", "")),
+                        pensieve=cast_list_str(row.get("pensieve", "")),
                     )
                 )
 

src/environment.py

@@ -2,6 +2,7 @@
 from typing import Optional
 
 from dotenv import dotenv_values
+
 from src.sheets import Sheet
 
 PREFIX = "flextensions_"
@@ -19,6 +20,7 @@
 DEFAULT_EMAIL_SIGNATURE = "{} Staff".format(DEFAULT_COURSE_NAME)
 
 DEFAULT_EXTEND_GRADESCOPE_ASSIGNMENTS = "No"
+DEFAULT_EXTEND_PENSIEVE_ASSIGNMENTS = "No"
 
 class Environment:
     @staticmethod
@@ -110,6 +112,18 @@ def get_gradescope_email() -> Optional[str]:
     def get_gradescope_password() -> Optional[str]:
         return Environment._safe_get("GRADESCOPE_PASSWORD")
 
+    @staticmethod
+    def get_extend_pensieve_assignments() -> Optional[str]:
+        return Environment._safe_get("EXTEND_PENSIEVE_ASSIGNMENTS", DEFAULT_EXTEND_PENSIEVE_ASSIGNMENTS)
+
+    @staticmethod
+    def get_pensieve_email() -> Optional[str]:
+        return Environment._safe_get("PENSIEVE_EMAIL")
+
+    @staticmethod
+    def get_pensieve_api_token() -> Optional[str]:
+        return Environment._safe_get("PENSIEVE_API_TOKEN")
+
     @staticmethod
     def get_spreadsheet_url() -> Optional[str]:
         return Environment._safe_get("SPREADSHEET_URL")

src/handle_email_queue.py

@@ -5,10 +5,13 @@
 from src.environment import Environment
 from src.errors import ConfigurationError
 from src.gradescope import Gradescope
+from src.pensieve import Pensieve
 from src.record import EMAIL_STATUS_IN_QUEUE, StudentRecord
-from src.sheets import SHEET_ASSIGNMENTS, SHEET_ENVIRONMENT_VARIABLES, SHEET_STUDENT_RECORDS, BaseSpreadsheet
+from src.sheets import (SHEET_ASSIGNMENTS, SHEET_ENVIRONMENT_VARIABLES,
+                        SHEET_STUDENT_RECORDS, BaseSpreadsheet)
 from src.slack import SlackManager
 
+
 def handle_email_queue(request_json):
     if "spreadsheet_url" not in request_json:
         raise ConfigurationError("handle_email_queue expects spreadsheet_url parameter")
@@ -55,6 +58,12 @@ def handle_email_queue(request_json):
                 for warning in warnings:
                     slack.add_warning(warning)
 
+            if Pensieve.is_enabled():
+                pensieve_client = Pensieve()
+                warnings = student.apply_extensions_pensieve(assignments=assignments, pensieve=pensieve_client)
+                for warning in warnings:
+                    slack.add_warning(warning)
+
     if len(emails) == 0:
         slack.send_message("Sent zero emails from the queue...was it empty?")
     else:

src/handle_flush_pensieve.py

@@ -0,0 +1,64 @@
+from src.assignments import AssignmentList
+from src.environment import Environment
+from src.errors import ConfigurationError
+from src.pensieve import Pensieve
+from src.record import StudentRecord
+from src.sheets import (SHEET_ASSIGNMENTS, SHEET_ENVIRONMENT_VARIABLES,
+                        SHEET_STUDENT_RECORDS, BaseSpreadsheet)
+from src.slack import SlackManager
+
+
+def handle_flush_pensieve(request_json):
+    if "spreadsheet_url" not in request_json:
+        raise ConfigurationError("handle_flush_pensieve expects spreadsheet_url parameter")
+
+    # Get pointers to sheets.
+    base = BaseSpreadsheet(spreadsheet_url=request_json["spreadsheet_url"])
+    sheet_assignments = base.get_sheet(SHEET_ASSIGNMENTS)
+    sheet_records = base.get_sheet(SHEET_STUDENT_RECORDS)
+    sheet_env_vars = base.get_sheet(SHEET_ENVIRONMENT_VARIABLES)
+
+    # Set up environment variables.
+    Environment.configure_env_vars(sheet_env_vars)
+
+    # Fetch assignments.
+    assignments = AssignmentList(sheet=sheet_assignments)
+
+    # Fetch records.
+    records = sheet_records.get_all_records()
+
+    slack = SlackManager()
+
+    pensieve = Pensieve()
+
+    all_warnings = []
+    successes = []
+    failures = []
+    for i, table_record in enumerate(records):
+        student = StudentRecord(table_index=i, table_record=table_record, sheet=sheet_records)
+        if student.should_flush_pensieve():
+            w = student.apply_extensions_pensieve(assignments=assignments, pensieve=pensieve)
+            if len(w) > 0:
+                failures.append(student.get_email())
+                all_warnings.extend(w)
+            else:
+                successes.append(student.get_email())
+                student.set_flush_pensieve_status_success()
+
+        student.flush()
+
+    for warning in all_warnings:
+        slack.add_warning(warning)
+
+    summary = "Flush Pensieve Summary:" + "\n"
+    if len(successes) > 0:
+        summary += "\n" + "*Successes:* " + ", ".join(successes)
+    if len(failures) > 0:
+        summary += "\n" + "*Failures:* " + ", ".join(failures)
+    if len(successes) + len(failures) == 0:
+        summary += (
+            "\n"
+            + "No student records processed. To process a student record, create a `flush_pensieve` column on the Roster sheet, and set the value to TRUE for each record you would like to flush to Pensieve."
+        )
+
+    slack.send_message(summary)

src/pensieve.py

@@ -0,0 +1,60 @@
+from typing import List
+
+import requests
+
+from src.environment import Environment
+from src.errors import KnownError
+from src.utils import truncate
+
+
+class Pensieve:
+    """
+    Pensieve client to apply extensions for a student on an assignment.
+    """
+
+    def __init__(self) -> None:
+        email = Environment.get_pensieve_email()
+        token = Environment.get_pensieve_api_token()
+
+        if not email:
+            raise KnownError("PENSIEVE_EMAIL must be set to use Pensieve.")
+        if not token:
+            raise KnownError("PENSIEVE_API_TOKEN must be set to use Pensieve.")
+
+        self.base_url = "https://api.pensieve.co"
+        self.email = email
+        self.token = token
+
+    @staticmethod
+    def is_enabled():
+        return Environment.get_extend_pensieve_assignments() in ["Yes", "TRUE"]
+
+    def apply_extension(self, assignment_name: str, assignment_urls: List[str], email: str, num_days: int) -> List[str]:
+        warnings = []
+        course_name = Environment.get_course_name()
+
+        for assignment_url in assignment_urls:
+            prefix = '[{}] [{}{}] [{}] [{}] '.format(
+                email, course_name + ' ', assignment_name, assignment_url, num_days)
+            try:
+                resp = requests.post(
+                    f"{self.base_url}/api/b2s/v1/external-client/grant-extension",
+                    headers={"Authorization": f"Bearer {self.token}", "Content-Type": "application/json"},
+                    json={
+                        "assignment_url": assignment_url,
+                        "student_email": email,
+                        "num_days": num_days,
+                    },
+                    timeout=30,
+                )
+                if resp.status_code != 200:
+                    raise Exception(f"HTTP {resp.status_code}: {truncate(resp.text)}")
+                data = resp.json() if resp.headers.get("content-type", "").startswith("application/json") else {}
+                if not data or not data.get("success", False):
+                    raise Exception(truncate(data))
+            except Exception as err:
+                warnings.append(
+                    prefix
+                    + f"failed to extend assignment in Pensieve: internal Pensieve error occurred ({truncate(err)})"
+                )
+        return warnings

src/record.py

@@ -10,10 +10,9 @@
 from src.environment import Environment
 from src.errors import StudentRecordError
 from src.gradescope import Gradescope
+from src.pensieve import Pensieve
 from src.sheets import Sheet
-from src.utils import cast_bool 
-
-import json
+from src.utils import cast_bool
 
 APPROVAL_STATUS_REQUESTED_MEETING = "Requested Meeting"
 APPROVAL_STATUS_PENDING = "Pending"
@@ -101,7 +100,16 @@ def set_flush_gradescope_status_success(self):
         if "flush_gradescope" in self.sheet.get_headers():
             self.queue_write_back(col_key="flush_gradescope", col_value=False)
 
-    def count_requests(self, assignments=AssignmentList):
+    def should_flush_pensieve(self):
+        if "flush_pensieve" in self.sheet.get_headers():
+            return cast_bool(self.table_record["flush_pensieve"])
+        return False
+
+    def set_flush_pensieve_status_success(self):
+        if "flush_pensieve" in self.sheet.get_headers():
+            self.queue_write_back(col_key="flush_pensieve", col_value=False)
+
+    def count_requests(self, assignments: AssignmentList) -> int:
         count = 0
         for assignment_id in assignments.get_all_ids():
             if self.get_request(assignment_id=assignment_id) is not None:
@@ -141,7 +149,7 @@ def flush(self):
         if self.table_index == -1:
             values = [self.write_queue.get(header) for header in headers]
             # minus 1 to account for header row
-            self.table_index = self.sheet.num_entries - 1 
+            self.table_index = self.sheet.num_entries - 1
             self.sheet.append_row(values=values, value_input_option="USER_ENTERED")
 
             # Update local table_record object for email.
@@ -164,7 +172,7 @@ def apply_extensions(self, assignments: AssignmentList, gradescope: Gradescope)
         warnings = []
         for assignment in assignments:
             num_days = self.get_request(assignment_id=assignment.get_id())
-            course_name = Environment.safe_get("COURSE_NAME", "")
+            course_name = Environment.get_course_name()
 
             if num_days:
 
@@ -193,6 +201,40 @@ def apply_extensions(self, assignments: AssignmentList, gradescope: Gradescope)
 
         return warnings
 
+    def apply_extensions_pensieve(self, assignments: AssignmentList, pensieve: Pensieve) -> List[str]:
+
+        warnings = []
+        for assignment in assignments:
+            num_days = self.get_request(assignment_id=assignment.get_id())
+            course_name = Environment.get_course_name()
+
+            if num_days:
+
+                if len(assignment.get_pensieve_assignment_urls()) == 0:
+                    print(
+                        "[{}{}] could not extend assignment deadline for {} (assignment URL's not set).".format(
+                            course_name + " ", assignment.get_name(), self.get_email()))
+                    continue
+
+                elif not assignment.get_due_date():
+                    warnings.append(
+                        "[{} {}] could not extend assignment deadline for {} (deadline not set).".format(
+                            course_name + " ", assignment.get_name(), self.get_email()))
+                    continue
+
+                else:
+                    print("Extending assignments (Pensieve): [{}{}] {}".format(
+                        course_name + " ", assignment.get_name(), str(assignment.get_pensieve_assignment_urls())))
+                    pensieve_warnings = pensieve.apply_extension(
+                        assignment_name=assignment.get_name(),
+                        assignment_urls=assignment.get_pensieve_assignment_urls(),
+                        email=self.get_email(),
+                        num_days=num_days,
+                    )
+                    warnings.extend(pensieve_warnings)
+
+        return warnings
+
     @staticmethod
     def from_email(email: str, sheet_records: Sheet) -> StudentRecord:
         email = email.lower()