test: 💍 added try except to make the code more robust

.gitignore

@@ -1,3 +1,4 @@
 .env
 __pycache__/
-/node_modules
+/node_modules
+*.csv

Inquirer.py

@@ -1,6 +1,10 @@
+import os
+from dotenv import load_dotenv
 import inquirer
 from MyDashboard import MyDashboard
+load_dotenv()
 
+script_env = os.environ['SCRIPT_ENV']
 database_id = "eeee053ef4ec44e6a666382bd54a1b39"
 
 
@@ -9,21 +13,46 @@ class Inquirer:
     def __init__(self, dashboard: MyDashboard):
         self.dashboard = dashboard
 
-        self.commands_to_execute = {
-            'I want to update my students\'s progress on notion':
-                [{"fn": self.dashboard.set_page_id_to_students, "args": [database_id]},
-                 {"fn": self.dashboard.update_students_progress, "args": []}]}
         self.questions = [
             inquirer.List('command',
                           message="What do you want to do?",
                           choices=[
-                              'I want to update my students\'s progress on notion'],
+                              'I want to update my students\'s progress on notion',
+                              'I want to set attendance for my students on udacity\'s dashboard'],
                           ),
         ]
+        self.answers = {"command": "", "session_idx": "",
+                        "file_path": "", "state": ""}
 
-        #self.answers = inquirer.prompt(self.questions)
+        self.commands_to_execute = {
+            'I want to update my students\'s progress on notion':
+                [{"fn": self.dashboard.set_page_id_to_students, "args": [database_id]},
+                 {"fn": self.dashboard.update_students_progress, "args": []},
+                 ],
+            'I want to set attendance for my students on udacity\'s dashboard':
+                [{"fn": self.get_file_path_from_user, "args": []}, ],
+        }
+
+        if script_env and script_env == 'dev':
+            self.answers.update(inquirer.prompt(self.questions))
+            self.execute_command(self.answers['command'])
+
+    def get_file_path_from_user(self):
+        self.question = [
+            inquirer.Text(
+                'session_idx', message='Please enter the index of the sessions',),
+
+            inquirer.List('state',
+                          message="What state do you want to set for these students?",
+                          choices=['present', 'absent', 'excused'],
+                          ),
+        ]
 
-        # self.execute_command(self.answers['command'])
+        answers = inquirer.prompt(self.question)
+        self.answers.update(answers)
+        self.dashboard.set_attendance_for_session(
+            self.answers['session_idx'],  self.answers['state']),
+        print(self.answers)
 
     def add_question(self, question):
         '''

MyDashboard.py

@@ -1,14 +1,21 @@
 import json
-from concurrent.futures import ThreadPoolExecutor
+import os
+
+import requests
+from dotenv import load_dotenv
+from joblib import Parallel, delayed
 
 from crawlers.UdacityCrawler import UdacityCrawler
 from NotionClient import NotionClient
-from joblib import Parallel, delayed
+from utils.read_data_from_csv import read_emails_from_csv
+
+load_dotenv()
 
 
 class MyDashboard:
     students = {}
     sessions = []
+    headers = {"Authorization": "Bearer " + os.environ["JWT_TOKEN"]}
 
     def __init__(self, sessions: list[str]):
         self.notion_client = NotionClient()
@@ -17,6 +24,8 @@ def __init__(self, sessions: list[str]):
         print('--- Started loading students ---')
         self.students = self.udacity_client.get_students_for_my_sessions(
             sessions)
+
+        print(len(self.students))
         self.set_students_progress()
         print('--- Finished loading students ---')
 
@@ -38,13 +47,17 @@ def set_students_progress(self):
             # self.students_progress.update(single_res)
 
     def get_student_progress(self, student_key: str):
+        if(student_key not in self.students):
+            return None
         return self.students[student_key]['completion_rate']
 
     def add_property_to_student(self, student_key: str, property_name: str, property_value: str):
+        if(student_key not in self.students):
+            return
         self.students[student_key][property_name] = property_value
 
     def set_page_id_to_students(self, database_id: str):
-        print('Started setting page id to students', database_id)
+        print('Started setting page id to students')
         all_pages_in_db = self.notion_client.get_pages_per_database(
             database_id, {})
         for page in all_pages_in_db['results']:
@@ -57,16 +70,18 @@ def set_page_id_to_students(self, database_id: str):
         print('Finished setting page id to students')
 
     def update_student_progress(self, student: dict):
-        page_id = student['page_id']
-        completion_rate = student['completion_rate']
-        completion_rate_payload = {"properties": {"Completion Rate":
-                                                  {
-                                                      "type": "number",
-                                                      "number": completion_rate
-                                                  }
-                                                  }}
-        self.notion_client.update_property(
-            page_id, completion_rate_payload)
+        try:
+            page_id = student['page_id']
+            completion_rate = student['completion_rate']
+            completion_rate_payload = {"properties":
+                                       {"Completion Rate":
+                                        {"type": "number",
+                                         "number": completion_rate}}}
+            self.notion_client.update_property(
+                page_id, completion_rate_payload)
+        except Exception as e:
+            print('Error updating completion rate', e)
+            return
 
     def update_students_progress(self):
         print('Started updating students progress')
@@ -76,23 +91,51 @@ def update_students_progress(self):
 
         print('Finished updating students progress')
 
-    def get_student_key_with_email(self, student_email: str):
+    def get_student_with_email(self, student_email: str):
         students = list(self.students.values())
-        print(students[0])
         student = filter(
             lambda student: student['student']['email'] == student_email, students)
 
-        return list(student)[0]['student']['key'] if student else None
-
-    def get_student_session_id(self, student_email: str):
-        pass
+        return list(student)[0] if student else None
 
-    def mark_student_as_excused(self, student_key: str):
-        pass
-
-    def store_attendance_for_student(self, student_key: str):
-        url = "https://www.notion.so/attendance-sheet-for-udacity-students-b9f9f8f8f9f94f9c9f9f9f9f9f9f9f9f"
-        pass
+    def get_student_key_with_email(self, student_email: str):
+        student = self.get_student_with_email(student_email)
+
+        return student['student']['key'] if student else None
+
+    def mark_student_as_excused(self, student_email: str):
+        student = self.get_student_with_email(student_email)
+        student_key = student['student']['key']
+        student_session_id = student['enrollment']['session_id']
+        print(student_session_id)
+        url = f'https://connect-dashboard.udacity.com/api/uhome/v1/sessions/{student_session_id}/instances/42376/excuse_attendance'
+        payload = {student_key: student_key}
+        request = requests.post(url,
+                                json=payload, headers=self.headers)
+
+        return request.text
+
+    def set_attendance_for_student(self, student_email: str, session_index: int, state: str):
+        student = self.get_student_with_email(student_email)
+        self.print_json(student)
+        student_session_id = student['enrollment']['session_id']
+        attendance_instance = student['attendances'][session_index]
+        session_insance_id = attendance_instance['instance_id']
+        attendance_id = attendance_instance['id']
+
+        payload = {"state": state}
+        url = f'https://connect-dashboard.udacity.com/api/uhome/v1/sessions/{student_session_id}/instances/{session_insance_id}/attendances/{attendance_id}'
+        print(url)
+        res = requests.patch(url, json=payload, headers=self.headers)
+
+        return res.json()
+
+    def set_attendance_for_session(self, session_idx, state: str):
+        int_session_idx = int(session_idx)
+        emails = read_emails_from_csv()
+        responses = [self.set_attendance_for_student(
+            email, int_session_idx, state) for email in emails]
+        return responses
 
     def print_json(self, json_object: json):
         print(json.dumps(json_object, indent=4, ))

main.py

@@ -1,31 +1,31 @@
 
-import json
 import time
 
 from dotenv import load_dotenv
-from Inquirer import Inquirer
-
+import inquirer
 
+from Inquirer import Inquirer
 from MyDashboard import MyDashboard
-from NotionClient import NotionClient
-from joblib import Parallel, delayed
+from utils.read_data_from_csv import read_emails_from_csv
 
 load_dotenv()
 
 
 if __name__ == '__main__':
 
-    # https://www.notion.so/9462e15053ff4bbc871ad7ab91d9f145?v=632479bcc69b41ce8ff355d5c72dcdfb
-    # https://www.notion.so/812372d76eb4492f81089f79f3cddefa?v=aed9d7af707b4527a37fda7295ec7a34
-    # https://www.notion.so/eeee053ef4ec44e6a666382bd54a1b39?v=2ad8662f2b2a465186d8854289950e9b
-    # https://www.notion.so/812372d76eb4492f81089f79f3cddefa?v=aed9d7af707b4527a37fda7295ec7a34
-    # https://www.notion.so/eeee053ef4ec44e6a666382bd54a1b39?v=2ad8662f2b2a465186d8854289950e9b
-
     print('--- Started the script ---')
+
+    start = time.process_time()
+    '''  '''
     my_dashboard = MyDashboard(['4725', '4731', '4732'])
-    # print(my_dashboard.print_json(my_dashboard.get_students()))
-    student_key = my_dashboard.get_student_key_with_email(
-        'abdelaty.magdi@gmail.com')
-    print(student_key)
-    #inq = Inquirer(my_dashboard)
+
+    ''' inquirer.Path('file_path', message='Please type the csv file path',
+                  path_type=inquirer.Path.DIRECTORY)
+    answers = inquirer.prompt([inquirer.Path('file_path', message='Please type the csv file path',
+                                             path_type=inquirer.Path.DIRECTORY)]) '''
+
+    # todo: add a method to create an attendance record if it doesn't exist
+    inq = Inquirer(my_dashboard)
     #inq.execute_command('I want to update my students\'s progress on notion')
+
+    print('Finished in: ', time.process_time() - start)

utils/read_data_from_csv.py

@@ -0,0 +1,14 @@
+
+import csv
+
+
+def read_emails_from_csv(file_path: str = 'attendance.csv'):
+    with open(file_path, 'r') as file:
+        csvreader = csv.reader(file)
+        emails = []
+        for row in csvreader:
+            # print(row)
+            if row[2] == 'Email':
+                continue
+            emails.append(row[2])
+    return emails