diff --git a/.gitignore b/.gitignore index e6c74964..c532a8dc 100644 --- a/.gitignore +++ b/.gitignore @@ -13,4 +13,5 @@ package-lock.json lms/public/frontend lms/www/lms.html lms/www/_lms.html -frappe-ui \ No newline at end of file +frappe-ui +frappe-semgrep-rules \ No newline at end of file diff --git a/frappe-semgrep-rules b/frappe-semgrep-rules deleted file mode 160000 index 239029b7..00000000 --- a/frappe-semgrep-rules +++ /dev/null @@ -1 +0,0 @@ -Subproject commit 239029b7ebaf2ec0e83222ca8bc668c8c76668f9 diff --git a/lms/command_palette.py b/lms/command_palette.py index 986e1780..01bb6bc5 100644 --- a/lms/command_palette.py +++ b/lms/command_palette.py @@ -16,7 +16,7 @@ def search_sqlite(query: str): return prepare_search_results(result) -def prepare_search_results(result): +def prepare_search_results(result: dict): groups = get_grouped_results(result) out = [] diff --git a/lms/hooks.py b/lms/hooks.py index 2a289ce1..71d47d8d 100644 --- a/lms/hooks.py +++ b/lms/hooks.py @@ -277,3 +277,4 @@ add_to_apps_screen = [ sqlite_search = ["lms.sqlite.LearningSearch"] auth_hooks = ["lms.auth.authenticate"] +require_type_annotated_api_methods = True diff --git a/lms/job/doctype/job_opportunity/job_opportunity.py b/lms/job/doctype/job_opportunity/job_opportunity.py index da9a357b..bc8d3e36 100644 --- a/lms/job/doctype/job_opportunity/job_opportunity.py +++ b/lms/job/doctype/job_opportunity/job_opportunity.py @@ -35,7 +35,7 @@ def update_job_openings(): @frappe.whitelist() -def report(job, reason): +def report(job: str, reason: str): system_managers = get_system_managers(only_name=True) user = frappe.db.get_value("User", frappe.session.user, "full_name") subject = _("User {0} has reported the job post {1}").format(user, job) diff --git a/lms/lms/api.py b/lms/lms/api.py index 931b23b7..7cee4f57 100644 --- a/lms/lms/api.py +++ b/lms/lms/api.py @@ -80,7 +80,7 @@ def get_translations(): @frappe.whitelist() -def validate_billing_access(billing_type, name): +def validate_billing_access(billing_type: str, name: str): doctype = "LMS Batch" if billing_type == "batch" else "LMS Course" access, message = verify_billing_access(doctype, name, billing_type) @@ -257,7 +257,7 @@ def get_branding(): @frappe.whitelist() -def get_unsplash_photos(keyword=None): +def get_unsplash_photos(keyword: str = None): from lms.unsplash import get_by_keyword, get_list if keyword: @@ -267,7 +267,7 @@ def get_unsplash_photos(keyword=None): @frappe.whitelist() -def get_evaluator_details(evaluator): +def get_evaluator_details(evaluator: str): frappe.only_for("Batch Evaluator") if not frappe.db.exists("Google Calendar", {"user": evaluator}): @@ -294,7 +294,7 @@ def get_evaluator_details(evaluator): @frappe.whitelist() -def get_certified_participants(filters=None, start=0, page_length=100): +def get_certified_participants(filters: dict = None, start: int = 0, page_length: int = 100): query = get_certification_query(filters) query = query.orderby("issue_date", order=frappe.qb.desc).offset(start).limit(page_length) participants = query.run(as_dict=True) @@ -306,7 +306,7 @@ def get_certified_participants(filters=None, start=0, page_length=100): return participants -def get_certified_participant_details(member): +def get_certified_participant_details(member: str): count = frappe.db.count("LMS Certificate", {"member": member}) details = frappe.db.get_value( "User", @@ -318,7 +318,7 @@ def get_certified_participant_details(member): return details -def get_certification_query(filters): +def get_certification_query(filters: dict = None): Certificate = frappe.qb.DocType("LMS Certificate") User = frappe.qb.DocType("User") @@ -348,7 +348,7 @@ def get_certification_query(filters): @frappe.whitelist() -def get_count_of_certified_members(filters=None): +def get_count_of_certified_members(filters: dict = None): query = get_certification_query(filters) result = query.run(as_dict=True) return len(result) or 0 @@ -424,7 +424,7 @@ def get_sidebar_settings(): @frappe.whitelist() -def update_sidebar_item(webpage, icon): +def update_sidebar_item(webpage: str, icon: str): frappe.only_for("Moderator") filters = { "web_page": webpage, @@ -443,7 +443,7 @@ def update_sidebar_item(webpage, icon): @frappe.whitelist() -def delete_sidebar_item(webpage): +def delete_sidebar_item(webpage: str): frappe.only_for("Moderator") return frappe.db.delete( "LMS Sidebar Item", @@ -457,7 +457,7 @@ def delete_sidebar_item(webpage): @frappe.whitelist() -def delete_lesson(lesson, chapter): +def delete_lesson(lesson: str, chapter: str): course = frappe.db.get_value("Course Chapter", chapter, "course") if not can_modify_course(course): frappe.throw(_("You do not have permission to delete this lesson."), frappe.PermissionError) @@ -477,7 +477,7 @@ def delete_lesson(lesson, chapter): @frappe.whitelist() -def update_lesson_index(lesson, sourceChapter, targetChapter, idx): +def update_lesson_index(lesson: str, sourceChapter: str, targetChapter: str, idx: int): course = frappe.db.get_value("Course Chapter", sourceChapter, "course") if not can_modify_course(course): frappe.throw(_("You do not have permission to modify this lesson."), frappe.PermissionError) @@ -488,7 +488,7 @@ def update_lesson_index(lesson, sourceChapter, targetChapter, idx): update_target_chapter(lesson, targetChapter, idx) -def update_source_chapter(lesson, chapter, idx, hasMoved=False): +def update_source_chapter(lesson: str, chapter: str, idx: int, hasMoved=False): lessons = frappe.get_all( "Lesson Reference", { @@ -507,7 +507,7 @@ def update_source_chapter(lesson, chapter, idx, hasMoved=False): update_index(lessons, chapter) -def update_target_chapter(lesson, chapter, idx): +def update_target_chapter(lesson: str, chapter: str, idx: int): lessons = frappe.get_all( "Lesson Reference", { @@ -531,7 +531,7 @@ def update_target_chapter(lesson, chapter, idx): update_index(lessons, chapter) -def update_index(lessons, chapter): +def update_index(lessons: list, chapter: str): for row in lessons: frappe.db.set_value( "Lesson Reference", {"lesson": row, "parent": chapter}, "idx", lessons.index(row) + 1 @@ -539,7 +539,7 @@ def update_index(lessons, chapter): @frappe.whitelist() -def update_chapter_index(chapter, course, idx): +def update_chapter_index(chapter: str, course: str, idx: int): """Update the index of a chapter within a course""" if not can_modify_course(course): @@ -562,7 +562,7 @@ def update_chapter_index(chapter, course, idx): @frappe.whitelist() -def get_members(start=0, search=""): +def get_members(start: int = 0, search: str = None): frappe.only_for(["Moderator"]) filters = {"enabled": 1, "name": ["not in", ["Administrator", "Guest"]]} or_filters = {} @@ -616,16 +616,16 @@ def check_app_permission(): @frappe.whitelist() def save_evaluation_details( - member, - course, - batch_name, - evaluator, - date, - start_time, - end_time, - status, - rating, - summary, + member: str, + course: str, + batch_name: str, + evaluator: str, + date: str, + start_time: str, + end_time: str, + status: str, + rating: float, + summary: str, ): """ Save evaluation details for a member against a course. @@ -662,14 +662,14 @@ def save_evaluation_details( @frappe.whitelist() def save_certificate_details( - member, - course, - batch_name, - evaluator, - issue_date, - expiry_date, - template, - published=True, + member: str, + course: str, + batch_name: str, + evaluator: str, + issue_date: str, + expiry_date: str, + template: str, + published: bool = True, ): """ Save certificate details for a member against a course. @@ -703,14 +703,14 @@ def save_certificate_details( @frappe.whitelist() -def delete_documents(doctype, documents): +def delete_documents(doctype: str, documents: list): frappe.only_for("Moderator") for doc in documents: frappe.delete_doc(doctype, doc) @frappe.whitelist() -def get_payment_gateway_details(payment_gateway): +def get_payment_gateway_details(payment_gateway: str): frappe.only_for("Moderator") gateway = frappe.get_doc("Payment Gateway", payment_gateway) @@ -741,7 +741,7 @@ def get_payment_gateway_details(payment_gateway): } -def get_transformed_fields(meta, data=None): +def get_transformed_fields(meta: list, data: dict = None): transformed_fields = [] for row in meta: if row.fieldtype not in ["Column Break", "Section Break"]: @@ -766,7 +766,7 @@ def get_transformed_fields(meta, data=None): @frappe.whitelist() -def get_new_gateway_fields(doctype): +def get_new_gateway_fields(doctype: str): frappe.only_for("Moderator") try: meta = frappe.get_meta(doctype).fields @@ -797,7 +797,7 @@ def update_course_statistics(): @frappe.whitelist() -def get_announcements(batch): +def get_announcements(batch: str): roles = frappe.get_roles() is_batch_student = frappe.db.exists( "LMS Batch Enrollment", {"batch": batch, "member": frappe.session.user} @@ -835,7 +835,7 @@ def get_announcements(batch): @frappe.whitelist() -def delete_course(course): +def delete_course(course: str): if not can_modify_course(course): frappe.throw(_("You do not have permission to delete this course."), frappe.PermissionError) @@ -872,7 +872,7 @@ def delete_course(course): @frappe.whitelist() -def delete_batch(batch): +def delete_batch(batch: str): if not can_modify_batch(batch): frappe.throw(_("You do not have permission to delete this batch."), frappe.PermissionError) @@ -885,7 +885,7 @@ def delete_batch(batch): frappe.db.delete("LMS Batch", batch) -def delete_batch_discussions(batch): +def delete_batch_discussions(batch: str): topics = frappe.get_all( "Discussion Topic", {"reference_doctype": "LMS Batch", "reference_docname": batch}, @@ -918,7 +918,7 @@ def give_discussions_permission(): @frappe.whitelist() -def upsert_chapter(title, course, is_scorm_package, scorm_package, name=None): +def upsert_chapter(title: str, course: str, is_scorm_package: bool, scorm_package: dict, name: str = None): if not can_modify_course(course): frappe.throw(_("You do not have permission to modify this chapter."), frappe.PermissionError) @@ -951,7 +951,7 @@ def upsert_chapter(title, course, is_scorm_package, scorm_package, name=None): return chapter -def extract_package(course, title, scorm_package): +def extract_package(course: str, title: str, scorm_package: dict): package = frappe.get_doc("File", scorm_package.name) zip_path = package.get_full_path() # check_for_malicious_code(zip_path) @@ -983,7 +983,7 @@ def check_for_malicious_code(zip_path): frappe.throw(_("Suspicious pattern found in {0}: {1}").format(file_name, pattern)) -def get_manifest_file(extract_path): +def get_manifest_file(extract_path: str): manifest_file = None for root, _dirs, files in os.walk(extract_path): for file in files: @@ -995,7 +995,7 @@ def get_manifest_file(extract_path): return manifest_file -def get_launch_file(extract_path): +def get_launch_file(extract_path: str): launch_file = None manifest_file = get_manifest_file(extract_path) @@ -1018,7 +1018,7 @@ def get_launch_file(extract_path): return launch_file -def add_lesson(title, chapter, course, idx): +def add_lesson(title: str, chapter: str, course: str, idx: int): lesson = frappe.new_doc("Course Lesson") lesson.update( { @@ -1043,7 +1043,7 @@ def add_lesson(title, chapter, course, idx): @frappe.whitelist() -def delete_chapter(chapter): +def delete_chapter(chapter: str): course = frappe.db.get_value("Course Chapter", chapter, "course") if not can_modify_course(course): frappe.throw(_("You do not have permission to delete this chapter."), frappe.PermissionError) @@ -1074,14 +1074,14 @@ def delete_chapter(chapter): i += 1 -def delete_scorm_package(scorm_package_path): +def delete_scorm_package(scorm_package_path: str): scorm_package_path = frappe.get_site_path("public", scorm_package_path[1:]) if os.path.exists(scorm_package_path): shutil.rmtree(scorm_package_path) @frappe.whitelist() -def mark_lesson_progress(course, chapter_number, lesson_number): +def mark_lesson_progress(course: str, chapter_number: int, lesson_number: int): chapter_name = frappe.get_value("Chapter Reference", {"parent": course, "idx": chapter_number}, "chapter") lesson_name = frappe.get_value( "Lesson Reference", {"parent": chapter_name, "idx": lesson_number}, "lesson" @@ -1090,7 +1090,7 @@ def mark_lesson_progress(course, chapter_number, lesson_number): @frappe.whitelist() -def get_heatmap_data(member, base_days=200): +def get_heatmap_data(member: str, base_days: int = 200): if not (has_course_instructor_role() or has_moderator_role() or has_evaluator_role()): frappe.throw(_("You do not have permission to access heatmap data."), frappe.PermissionError) @@ -1114,7 +1114,7 @@ def get_heatmap_data(member, base_days=200): } -def calculate_date_ranges(base_days): +def calculate_date_ranges(base_days: int): today = format_date(now(), "YYYY-MM-dd") day_today = get_datetime(today).strftime("%w") padding_end = 6 - cint(day_today) @@ -1128,11 +1128,11 @@ def calculate_date_ranges(base_days): return base_date, start_date, number_of_days, days -def initialize_date_count(days): +def initialize_date_count(days: list): return {format_date(day, "YYYY-MM-dd"): 0 for day in days} -def fetch_activity_data(member, start_date): +def fetch_activity_data(member: str, start_date: str): lesson_completions = frappe.get_all( "LMS Course Progress", fields=["creation"], @@ -1154,14 +1154,14 @@ def fetch_activity_data(member, start_date): return lesson_completions, quiz_submissions, assignment_submissions -def count_dates(data, date_count): +def count_dates(data: list, date_count: dict): for entry in data: date = format_date(entry.creation, "YYYY-MM-dd") if date in date_count: date_count[date] += 1 -def prepare_heatmap_data(start_date, number_of_days, date_count): +def prepare_heatmap_data(start_date: str, number_of_days: int, date_count: dict): days_of_week = ["Sun", "Mon", "Tue", "Wed", "Thu", "Fri", "Sat"] heatmap_data = {day: [] for day in days_of_week} week_count = -(number_of_days // -7) @@ -1198,13 +1198,13 @@ def prepare_heatmap_data(start_date, number_of_days, date_count): return formatted_heatmap_data, labels, total_activities, week_count -def get_week_difference(start_date, current_date): +def get_week_difference(start_date: str, current_date: str) -> int: diff_in_days = date_diff(current_date, start_date) return diff_in_days // 7 @frappe.whitelist() -def get_notifications(filters): +def get_notifications(filters: dict = None): filters = frappe._dict(filters or {}) filters.for_user = frappe.session.user notifications = frappe.get_all( @@ -1232,7 +1232,7 @@ def get_notifications(filters): return notifications -def update_user_details(notification): +def update_user_details(notification: dict) -> dict: if ( notification.document_details and len(notification.document_details.get("instructors", [])) @@ -1247,7 +1247,7 @@ def update_user_details(notification): return notification -def is_mention(notification): +def is_mention(notification: dict) -> bool: if notification.type == "Mention": return True if "mentioned you" in notification.subject.lower(): @@ -1255,7 +1255,7 @@ def is_mention(notification): return False -def update_document_details(notification): +def update_document_details(notification: dict) -> dict: if notification.document_type == "LMS Course": details = frappe.db.get_value( "LMS Course", notification.document_name, ["title", "video_link", "short_introduction"], as_dict=1 @@ -1304,7 +1304,7 @@ def get_lms_settings(): @frappe.whitelist() -def cancel_evaluation(evaluation): +def cancel_evaluation(evaluation: dict): evaluation = frappe._dict(evaluation) if evaluation.member != frappe.session.user: frappe.throw(_("You do not have permission to cancel this evaluation."), frappe.PermissionError) @@ -1336,7 +1336,7 @@ def cancel_evaluation(evaluation): @frappe.whitelist() -def get_certification_details(course): +def get_certification_details(course: str): membership = None filters = {"course": course, "member": frappe.session.user} @@ -1364,7 +1364,7 @@ def get_certification_details(course): @frappe.whitelist() -def save_role(user, role, value): +def save_role(user: str, role: str, value: int): frappe.only_for("Moderator") if cint(value): doc = frappe.get_doc( @@ -1384,7 +1384,7 @@ def save_role(user, role, value): @frappe.whitelist() -def add_an_evaluator(email): +def add_an_evaluator(email: str): frappe.only_for("Moderator") if not frappe.db.exists("User", email): user = frappe.new_doc("User") @@ -1406,7 +1406,7 @@ def add_an_evaluator(email): @frappe.whitelist() -def capture_user_persona(responses): +def capture_user_persona(responses: str): frappe.only_for("System Manager") data = frappe.parse_json(responses) data = json.dumps(data) @@ -1420,7 +1420,7 @@ def capture_user_persona(responses): @frappe.whitelist() -def get_meta_info(type, route): +def get_meta_info(type: str, route: str): if frappe.db.exists("Website Meta Tag", {"parent": f"{type}/{route}"}): meta_tags = frappe.get_all( "Website Meta Tag", @@ -1436,7 +1436,7 @@ def get_meta_info(type, route): @frappe.whitelist() -def update_meta_info(meta_type, route, meta_tags): +def update_meta_info(meta_type: str, route: str, meta_tags: list): frappe.only_for(["Course Creator", "Batch Evaluator", "Moderator"]) validate_meta_data_permissions(meta_type) validate_meta_tags(meta_tags) @@ -1473,12 +1473,12 @@ def update_meta_info(meta_type, route, meta_tags): create_meta_tag(tag_properties) -def validate_meta_tags(meta_tags): +def validate_meta_tags(meta_tags: list): if not isinstance(meta_tags, list): frappe.throw(_("Meta tags should be a list.")) -def create_meta(parent_name, tag_properties): +def create_meta(parent_name: str, tag_properties: dict): route_meta = frappe.new_doc("Website Route Meta") route_meta.update( { @@ -1489,13 +1489,13 @@ def create_meta(parent_name, tag_properties): route_meta.insert() -def create_meta_tag(tag_properties): +def create_meta_tag(tag_properties: dict): new_tag = frappe.new_doc("Website Meta Tag") new_tag.update(tag_properties) new_tag.insert() -def validate_meta_data_permissions(meta_type): +def validate_meta_data_permissions(meta_type: str): roles = frappe.get_roles() if meta_type == "courses": @@ -1508,14 +1508,14 @@ def validate_meta_data_permissions(meta_type): @frappe.whitelist() -def create_programming_exercise_submission(exercise, submission, code, test_cases): +def create_programming_exercise_submission(exercise: str, submission: str, code: str, test_cases: list): if submission == "new": return make_new_exercise_submission(exercise, code, test_cases) else: update_exercise_submission(submission, code, test_cases) -def make_new_exercise_submission(exercise, code, test_cases): +def make_new_exercise_submission(exercise: str, code: str, test_cases: list): submission = frappe.new_doc("LMS Programming Exercise Submission") submission.exercise = exercise submission.member = frappe.session.user @@ -1537,7 +1537,7 @@ def make_new_exercise_submission(exercise, code, test_cases): return submission.name -def update_exercise_submission(submission, code, test_cases): +def update_exercise_submission(submission: str, code: str, test_cases: list): member = frappe.db.get_value("LMS Programming Exercise Submission", submission, "member") if member != frappe.session.user: frappe.throw(_("You do not have permission to update this submission."), frappe.PermissionError) @@ -1547,7 +1547,7 @@ def update_exercise_submission(submission, code, test_cases): frappe.db.set_value("LMS Programming Exercise Submission", submission, {"status": status, "code": code}) -def get_exercise_status(test_cases): +def get_exercise_status(test_cases: list): if not test_cases: return "Failed" @@ -1557,7 +1557,7 @@ def get_exercise_status(test_cases): return "Failed" -def update_test_cases(test_cases, submission): +def update_test_cases(test_cases: list, submission: str): frappe.db.delete("LMS Test Case Submission", {"parent": submission}) for row in test_cases: test_case = frappe.new_doc("LMS Test Case Submission") @@ -1576,7 +1576,7 @@ def update_test_cases(test_cases, submission): @frappe.whitelist() -def track_video_watch_duration(lesson, videos): +def track_video_watch_duration(lesson: str, videos: list): """ Track the watch duration of videos in a lesson. """ @@ -1603,7 +1603,7 @@ def track_video_watch_duration(lesson, videos): track_new_watch_time(lesson, video) -def track_new_watch_time(lesson, video): +def track_new_watch_time(lesson: str, video: dict): doc = frappe.new_doc("LMS Video Watch Duration") doc.lesson = lesson doc.source = video.get("source") @@ -1613,7 +1613,7 @@ def track_new_watch_time(lesson, video): @frappe.whitelist() -def get_course_progress_distribution(course): +def get_course_progress_distribution(course: str): if not can_modify_course(course): frappe.throw( _("You do not have permission to access this course's progress data."), frappe.PermissionError @@ -1636,14 +1636,14 @@ def get_course_progress_distribution(course): } -def get_average_course_progress(progress_list): +def get_average_course_progress(progress_list: list): if not progress_list: return 0 average_progress = sum(progress_list) / len(progress_list) return flt(average_progress, frappe.get_system_settings("float_precision") or 3) -def get_progress_distribution(progressList): +def get_progress_distribution(progressList: list): distribution = [ { "name": "Just Started (0-30%)", @@ -1690,7 +1690,7 @@ def get_pwa_manifest(): @frappe.whitelist() -def get_profile_details(username): +def get_profile_details(username: str): details = frappe.db.get_value( "User", {"username": username}, @@ -1729,7 +1729,7 @@ def get_streak_info(): } -def fetch_activity_dates(user): +def fetch_activity_dates(user: str): doctypes = [ "LMS Course Progress", "LMS Quiz Submission", @@ -1744,7 +1744,7 @@ def fetch_activity_dates(user): return sorted({d.date() if hasattr(d, "date") else d for d in all_dates}) -def calculate_streaks(all_dates): +def calculate_streaks(all_dates: list): streak = 0 longest_streak = 0 prev_day = None @@ -1768,7 +1768,7 @@ def calculate_streaks(all_dates): return streak, longest_streak -def calculate_current_streak(all_dates, streak): +def calculate_current_streak(all_dates: list, streak: int): if not all_dates: return 0 @@ -2034,7 +2034,7 @@ def get_upcoming_batches(): @frappe.whitelist() -def delete_programming_exercise(exercise): +def delete_programming_exercise(exercise: str): frappe.only_for(["Moderator", "Course Creator"]) frappe.db.delete("LMS Programming Exercise Submission", {"exercise": exercise}) frappe.db.delete("LMS Programming Exercise", exercise) diff --git a/lms/lms/doctype/course_evaluator/course_evaluator.py b/lms/lms/doctype/course_evaluator/course_evaluator.py index e567dfca..fa127395 100644 --- a/lms/lms/doctype/course_evaluator/course_evaluator.py +++ b/lms/lms/doctype/course_evaluator/course_evaluator.py @@ -58,7 +58,7 @@ class CourseEvaluator(Document): @frappe.whitelist() -def get_schedule(course, batch=None): +def get_schedule(course: str, batch: str = None): evaluator = get_evaluator(course, batch) start_date = nowdate() end_date = get_schedule_range_end_date(start_date, batch) diff --git a/lms/lms/doctype/course_lesson/course_lesson.py b/lms/lms/doctype/course_lesson/course_lesson.py index 28105002..e891b9b0 100644 --- a/lms/lms/doctype/course_lesson/course_lesson.py +++ b/lms/lms/doctype/course_lesson/course_lesson.py @@ -46,7 +46,7 @@ class CourseLesson(Document): @frappe.whitelist() -def save_progress(lesson, course, scorm_details=None): +def save_progress(lesson: str, course: str, scorm_details: dict = None): """ Note: Pass the argument scorm_details as a dict if it is SCORM related save_progress """ diff --git a/lms/lms/doctype/lms_badge/lms_badge.py b/lms/lms/doctype/lms_badge/lms_badge.py index fe1a611b..4ca727f7 100644 --- a/lms/lms/doctype/lms_badge/lms_badge.py +++ b/lms/lms/doctype/lms_badge/lms_badge.py @@ -61,7 +61,7 @@ def eval_condition(doc, condition): @frappe.whitelist() -def assign_badge(badge): +def assign_badge(badge: str, user: str): badge = frappe._dict(json.loads(badge)) if not badge.event == "Auto Assign": return diff --git a/lms/lms/doctype/lms_batch/lms_batch.py b/lms/lms/doctype/lms_batch/lms_batch.py index b65fdf30..4da344da 100644 --- a/lms/lms/doctype/lms_batch/lms_batch.py +++ b/lms/lms/doctype/lms_batch/lms_batch.py @@ -203,15 +203,15 @@ def send_system_notification_for_published_batch(batch): @frappe.whitelist() def create_live_class( - batch_name, - zoom_account, - title, - duration, - date, - time, - timezone, - auto_recording, - description=None, + batch_name: str, + zoom_account: str, + title: str, + duration: int, + date: str, + time: str, + timezone: str, + auto_recording: str, + description: str = None, ): payload = { "topic": title, @@ -280,7 +280,7 @@ def authenticate(zoom_account): @frappe.whitelist() -def get_batch_timetable(batch): +def get_batch_timetable(batch: str): timetable = frappe.get_all( "LMS Batch Timetable", filters={"parent": batch}, diff --git a/lms/lms/doctype/lms_batch_enrollment/lms_batch_enrollment.py b/lms/lms/doctype/lms_batch_enrollment/lms_batch_enrollment.py index c1b4a75f..6b78b9c2 100644 --- a/lms/lms/doctype/lms_batch_enrollment/lms_batch_enrollment.py +++ b/lms/lms/doctype/lms_batch_enrollment/lms_batch_enrollment.py @@ -106,7 +106,7 @@ class LMSBatchEnrollment(Document): @frappe.whitelist() -def send_confirmation_email(doc): +def send_confirmation_email(doc: str): if isinstance(doc, str): doc = frappe._dict(json.loads(doc)) diff --git a/lms/lms/doctype/lms_certificate/lms_certificate.py b/lms/lms/doctype/lms_certificate/lms_certificate.py index ec9f5231..ca35d11a 100644 --- a/lms/lms/doctype/lms_certificate/lms_certificate.py +++ b/lms/lms/doctype/lms_certificate/lms_certificate.py @@ -123,7 +123,7 @@ def is_certified(course): @frappe.whitelist() -def create_certificate(course): +def create_certificate(course: str): if is_certified(course): return frappe.db.get_value( "LMS Certificate", certificate, ["name", "course", "template"], as_dict=True diff --git a/lms/lms/doctype/lms_certificate_evaluation/lms_certificate_evaluation.py b/lms/lms/doctype/lms_certificate_evaluation/lms_certificate_evaluation.py index 6e4c8cfa..d8589108 100644 --- a/lms/lms/doctype/lms_certificate_evaluation/lms_certificate_evaluation.py +++ b/lms/lms/doctype/lms_certificate_evaluation/lms_certificate_evaluation.py @@ -25,7 +25,7 @@ def has_website_permission(doc, ptype, user, verbose=False): @frappe.whitelist() -def create_lms_certificate(source_name, target_doc=None): +def create_lms_certificate(source_name: str, target_doc: dict = None): doc = get_mapped_doc( "LMS Certificate Evaluation", source_name, diff --git a/lms/lms/doctype/lms_certificate_request/lms_certificate_request.py b/lms/lms/doctype/lms_certificate_request/lms_certificate_request.py index 33a781ce..b9fc97fb 100644 --- a/lms/lms/doctype/lms_certificate_request/lms_certificate_request.py +++ b/lms/lms/doctype/lms_certificate_request/lms_certificate_request.py @@ -174,7 +174,7 @@ def schedule_evals(): @frappe.whitelist() -def setup_calendar_event(eval): +def setup_calendar_event(eval: str): if isinstance(eval, str): eval = frappe._dict(json.loads(eval)) @@ -186,7 +186,7 @@ def setup_calendar_event(eval): update_meeting_details(eval, event, calendar) -def create_event(eval): +def create_event(eval: dict): event = frappe.get_doc( { "doctype": "Event", @@ -199,7 +199,7 @@ def create_event(eval): return event -def add_participants(eval, event): +def add_participants(eval: dict, event: frappe._dict): participants = [eval.member, eval.evaluator] for participant in participants: contact_name = frappe.db.get_value("Contact", {"email_id": participant}, "name") @@ -216,7 +216,7 @@ def add_participants(eval, event): ).save() -def update_meeting_details(eval, event, calendar): +def update_meeting_details(eval: dict, event: frappe._dict, calendar: str): event.reload() event.update( { @@ -232,7 +232,7 @@ def update_meeting_details(eval, event, calendar): @frappe.whitelist() -def create_lms_certificate_evaluation(source_name, target_doc=None): +def create_lms_certificate_evaluation(source_name: str, target_doc: dict = None): frappe.only_for(["Moderator", "Batch Evaluator", "System Manager"]) doc = get_mapped_doc( "LMS Certificate Request", diff --git a/lms/lms/doctype/lms_quiz/lms_quiz.py b/lms/lms/doctype/lms_quiz/lms_quiz.py index fc1ca4b0..9837e125 100644 --- a/lms/lms/doctype/lms_quiz/lms_quiz.py +++ b/lms/lms/doctype/lms_quiz/lms_quiz.py @@ -93,7 +93,7 @@ class LMSQuiz(Document): return result[0] -def set_total_marks(questions): +def set_total_marks(questions: list) -> int: marks = 0 for question in questions: marks += question.get("marks") @@ -101,7 +101,7 @@ def set_total_marks(questions): @frappe.whitelist() -def quiz_summary(quiz, results): +def quiz_summary(quiz: str, results: str): results = results and json.loads(results) percentage = 0 @@ -141,7 +141,7 @@ def quiz_summary(quiz, results): } -def process_results(results, quiz_details): +def process_results(results: list, quiz_details: dict): score = 0 is_open_ended = False @@ -188,7 +188,7 @@ def process_results(results, quiz_details): } -def _save_file(match): +def _save_file(match: re.Match) -> str: data = match.group(1).split("data:")[1] headers, content = data.split(",") mtype = headers.split(";", 1)[0] @@ -231,7 +231,7 @@ def get_corrupted_image_msg(): return _("Image: Corrupted Data Stream") -def create_submission(quiz, results, score_out_of, passing_percentage): +def create_submission(quiz: str, results: list, score_out_of: int, passing_percentage: float): submission = frappe.new_doc("LMS Quiz Submission") # Score and percentage are calculated by the controller function submission.update( @@ -250,7 +250,7 @@ def create_submission(quiz, results, score_out_of, passing_percentage): return submission -def save_progress_after_quiz(quiz_details, percentage): +def save_progress_after_quiz(quiz_details: dict, percentage: float): if percentage >= quiz_details.passing_percentage and quiz_details.lesson and quiz_details.course: save_progress(quiz_details.lesson, quiz_details.course) elif not quiz_details.passing_percentage: @@ -258,7 +258,7 @@ def save_progress_after_quiz(quiz_details, percentage): @frappe.whitelist() -def check_answer(question, type, answers): +def check_answer(question: str, type: str, answers: str): answers = json.loads(answers) if type == "Choices": return check_choice_answers(question, answers) @@ -266,7 +266,7 @@ def check_answer(question, type, answers): return check_input_answers(question, answers[0]) -def check_choice_answers(question, answers): +def check_choice_answers(question: str, answers: list): fields = ["multiple"] is_correct = [] for num in range(1, 5): @@ -286,7 +286,7 @@ def check_choice_answers(question, answers): return is_correct -def check_input_answers(question, answer): +def check_input_answers(question: str, answer: str): fields = [] for num in range(1, 5): fields.append(f"possibility_{cstr(num)}") diff --git a/lms/lms/payments.py b/lms/lms/payments.py index 8c2e1834..1cc3c78d 100644 --- a/lms/lms/payments.py +++ b/lms/lms/payments.py @@ -19,18 +19,18 @@ def validate_currency(payment_gateway, currency): @frappe.whitelist() def get_payment_link( - doctype, - docname, - title, - amount, - discount_amount, - gst_amount, - currency, - address, - redirect_to, - payment_for_certificate, - coupon_code=None, - coupon=None, + doctype: str, + docname: str, + title: str, + amount: float, + discount_amount: float, + gst_amount: float, + currency: str, + address: dict, + redirect_to: str, + payment_for_certificate: int, + coupon_code: str = None, + coupon: str = None, ): payment_gateway = get_payment_gateway() address = frappe._dict(address) @@ -73,7 +73,7 @@ def get_payment_link( return url -def create_order(payment_gateway, payment_details, controller): +def create_order(payment_gateway: str, payment_details: dict, controller): if payment_gateway != "Razorpay": return @@ -81,7 +81,7 @@ def create_order(payment_gateway, payment_details, controller): payment_details.update({"order_id": order.get("id")}) -def get_amount_with_gst(amount, gst_amount): +def get_amount_with_gst(amount: float, gst_amount: float) -> float: amount_with_gst = 0 if gst_amount: amount_with_gst = amount + gst_amount @@ -90,17 +90,17 @@ def get_amount_with_gst(amount, gst_amount): def record_payment( - address, - doctype, - docname, - amount, - original_amount, - currency, - amount_with_gst=0, - discount_amount=0, - payment_for_certificate=0, - coupon_code=None, - coupon=None, + address: dict, + doctype: str, + docname: str, + amount: float, + original_amount: float, + currency: str, + amount_with_gst: float = 0, + discount_amount: float = 0, + payment_for_certificate: int = 0, + coupon_code: str = None, + coupon: str = None, ): address = frappe._dict(address) address_name = save_address(address) @@ -138,7 +138,7 @@ def record_payment( return payment_doc -def save_address(address): +def save_address(address: dict) -> str: filters = {"email_id": frappe.session.user} exists = frappe.db.exists("Address", filters) if exists: diff --git a/lms/lms/telemetry.py b/lms/lms/telemetry.py deleted file mode 100644 index 51dcd03b..00000000 --- a/lms/lms/telemetry.py +++ /dev/null @@ -1,12 +0,0 @@ -import frappe -from frappe.utils.telemetry import POSTHOG_HOST_FIELD, POSTHOG_PROJECT_FIELD - - -@frappe.whitelist() -def get_posthog_settings(): - return { - "posthog_project_id": frappe.conf.get(POSTHOG_PROJECT_FIELD), - "posthog_host": frappe.conf.get(POSTHOG_HOST_FIELD), - "enable_telemetry": frappe.get_system_settings("enable_telemetry"), - "telemetry_site_age": frappe.utils.telemetry.site_age(), - } diff --git a/lms/lms/user.py b/lms/lms/user.py index d6535977..be33b110 100644 --- a/lms/lms/user.py +++ b/lms/lms/user.py @@ -24,7 +24,7 @@ def after_insert(doc, method): @frappe.whitelist(allow_guest=True) -def sign_up(email, full_name, verify_terms, user_category): +def sign_up(email: str, full_name: str, verify_terms: bool, user_category: str): if is_signup_disabled(): frappe.throw(_("Sign Up is disabled"), _("Not Allowed")) @@ -75,7 +75,7 @@ def sign_up(email, full_name, verify_terms, user_category): return 2, _("Please ask your administrator to verify your sign-up") -def set_country_from_ip(login_manager=None, user=None): +def set_country_from_ip(login_manager: object = None, user: str = None): if not user and login_manager: user = login_manager.user user_country = frappe.db.get_value("User", user, "country") @@ -85,7 +85,7 @@ def set_country_from_ip(login_manager=None, user=None): return -def on_login(login_manager): +def on_login(): default_app = frappe.db.get_single_value("System Settings", "default_app") if default_app == "lms": frappe.local.response["home_page"] = get_lms_route() diff --git a/lms/lms/utils.py b/lms/lms/utils.py index f9c127c1..a08a830c 100644 --- a/lms/lms/utils.py +++ b/lms/lms/utils.py @@ -44,11 +44,11 @@ def get_lms_route(path=""): return f"{base}/{path.lstrip('/')}" -def extend_bootinfo(bootinfo): +def extend_bootinfo(bootinfo: dict): bootinfo["lms_path"] = get_lms_path() -def slugify(title, used_slugs=None): +def slugify(title: str, used_slugs: list = None): """Converts title to a slug. If a list of used slugs is specified, it will make sure the generated slug @@ -78,13 +78,13 @@ def slugify(title, used_slugs=None): count = count + 1 -def generate_slug(title, doctype): +def generate_slug(title: str, doctype: str): result = frappe.get_all(doctype, fields=["name"]) slugs = {row["name"] for row in result} return slugify(title, used_slugs=slugs) -def get_membership(course, member=None): +def get_membership(course: str, member: str = None): if not member: member = frappe.session.user @@ -110,7 +110,7 @@ def get_membership(course, member=None): return False -def get_chapters(course): +def get_chapters(course: str): """Returns all chapters of this course.""" if not course: return [] @@ -126,7 +126,7 @@ def get_chapters(course): return chapters -def get_lessons(course, chapter=None, get_details=True, progress=False): +def get_lessons(course: str, chapter: str = None, get_details: bool = True, progress: bool = False): """If chapter is passed, returns lessons of only that chapter. Else returns lessons of all chapters of the course""" lessons = [] @@ -146,7 +146,7 @@ def get_lessons(course, chapter=None, get_details=True, progress=False): return lessons if get_details else lesson_count -def get_lesson_details(chapter, progress=False): +def get_lesson_details(chapter: dict, progress: bool = False): lessons = [] lesson_list = frappe.get_all( "Lesson Reference", {"parent": chapter.name}, ["lesson", "idx"], order_by="idx" @@ -182,7 +182,7 @@ def get_lesson_details(chapter, progress=False): return lessons -def get_lesson_icon(body, content): +def get_lesson_icon(body: str, content: str): if content: content = json.loads(content) @@ -222,7 +222,7 @@ def get_lesson_icon(body, content): return "icon-list" -def get_instructors(doctype, docname): +def get_instructors(doctype: str, docname: str): instructor_details = [] instructors = frappe.get_all( "Course Instructor", @@ -243,7 +243,7 @@ def get_instructors(doctype, docname): return instructor_details -def get_average_rating(course): +def get_average_rating(course: str): ratings = [review.rating for review in get_reviews(course)] if not len(ratings): return None @@ -252,7 +252,7 @@ def get_average_rating(course): @frappe.whitelist(allow_guest=True) @rate_limit(limit=500, seconds=60 * 60) -def get_reviews(course): +def get_reviews(course: str): reviews = frappe.get_all( "LMS Course Review", {"course": course}, @@ -274,7 +274,7 @@ def get_reviews(course): return reviews -def get_lesson_index(lesson_name): +def get_lesson_index(lesson_name: str) -> str: """Returns the {chapter_index}.{lesson_index} for the lesson.""" lesson = frappe.db.get_value("Lesson Reference", {"lesson": lesson_name}, ["idx", "parent"], as_dict=True) if not lesson: @@ -287,13 +287,13 @@ def get_lesson_index(lesson_name): return f"{chapter.idx}-{lesson.idx}" -def get_lesson_url(course, lesson_number): +def get_lesson_url(course: str, lesson_number: str): if not lesson_number: return return get_lms_route(f"courses/{course}/learn/{lesson_number}") -def get_progress(course, lesson, member=None): +def get_progress(course: str, lesson: str, member: str = None): if not member: member = frappe.session.user @@ -304,7 +304,7 @@ def get_progress(course, lesson, member=None): ) -def get_course_progress(course, member=None): +def get_course_progress(course: str, member: str = None): """Returns the course progress of the session user""" lesson_count = get_lessons(course, get_details=False) if not lesson_count: @@ -317,7 +317,7 @@ def get_course_progress(course, member=None): return flt(((completed_lessons / lesson_count) * 100), precision) -def is_instructor(course): +def is_instructor(course: str) -> bool: instructors = get_instructors("LMS Course", course) for instructor in instructors: if instructor.name == frappe.session.user: @@ -325,7 +325,7 @@ def is_instructor(course): return False -def has_course_instructor_role(member=None): +def has_course_instructor_role(member: str = None): return frappe.db.get_value( "Has Role", {"parent": member or frappe.session.user, "role": "Course Creator"}, @@ -333,7 +333,7 @@ def has_course_instructor_role(member=None): ) -def has_moderator_role(member=None): +def has_moderator_role(member: str = None): return frappe.db.get_value( "Has Role", {"parent": member or frappe.session.user, "role": "Moderator"}, @@ -341,7 +341,7 @@ def has_moderator_role(member=None): ) -def has_evaluator_role(member=None): +def has_evaluator_role(member: str = None): return frappe.db.get_value( "Has Role", {"parent": member or frappe.session.user, "role": "Batch Evaluator"}, @@ -349,7 +349,7 @@ def has_evaluator_role(member=None): ) -def has_student_role(member=None): +def has_student_role(member: str = None): return frappe.db.get_value( "Has Role", {"parent": member or frappe.session.user, "role": "LMS Student"}, @@ -376,7 +376,7 @@ def get_courses_under_review(): ) -def validate_image(path): +def validate_image(path: str) -> str: if path and "/private" in path: frappe.db.set_value( "File", @@ -388,7 +388,7 @@ def validate_image(path): return path -def handle_notifications(doc, method): +def handle_notifications(doc: object, method: str): topic = frappe.db.get_value( "Discussion Topic", doc.topic, @@ -402,7 +402,7 @@ def handle_notifications(doc, method): notify_mentions_via_email(doc, topic) -def get_course_details_for_notification(topic): +def get_course_details_for_notification(topic: dict): users = [] course = frappe.db.get_value("Course Lesson", topic.reference_docname, "course") course_title = frappe.db.get_value("LMS Course", course, "title") @@ -419,7 +419,7 @@ def get_course_details_for_notification(topic): return subject, link, users -def get_batch_details_for_notification(topic): +def get_batch_details_for_notification(topic: dict): users = [] batch_title = frappe.db.get_value("LMS Batch", topic.reference_docname, "title") subject = _("New comment in batch {0}").format(batch_title) @@ -435,7 +435,7 @@ def get_batch_details_for_notification(topic): return subject, link, users -def create_notification_log(doc, topic): +def create_notification_log(doc: object, topic: dict): if topic.reference_doctype == "Course Lesson": subject, link, users = get_course_details_for_notification(topic) else: @@ -459,7 +459,7 @@ def create_notification_log(doc, topic): make_notification_logs(notification, users) -def notify_mentions_on_portal(doc, topic): +def notify_mentions_on_portal(doc: object, topic: dict): mentions = extract_mentions(doc.reply) if not mentions: return @@ -495,7 +495,7 @@ def notify_mentions_on_portal(doc, topic): make_notification_logs(notification, user) -def notify_mentions_via_email(doc, topic): +def notify_mentions_via_email(doc: object, topic: dict): outgoing_email_account = frappe.get_cached_value( "Email Account", {"default_outgoing": 1, "enable_outgoing": 1}, "name" ) @@ -542,7 +542,7 @@ def notify_mentions_via_email(doc, topic): ) -def get_lesson_count(course): +def get_lesson_count(course: str) -> int: lesson_count = 0 chapters = frappe.get_all("Chapter Reference", {"parent": course}, ["chapter"]) for chapter in chapters: @@ -554,10 +554,10 @@ def get_lesson_count(course): @frappe.whitelist(allow_guest=True) @rate_limit(limit=500, seconds=60 * 60) def get_chart_data( - chart_name, - timegrain="Daily", - from_date=None, - to_date=None, + chart_name: str, + timegrain: str = "Daily", + from_date: str = None, + to_date: str = None, ): from_date, to_date = get_chart_date_range(from_date, to_date) chart = frappe.get_doc("Dashboard Chart", chart_name) @@ -578,7 +578,7 @@ def get_chart_data( return data -def get_chart_date_range(from_date, to_date): +def get_chart_date_range(from_date: str, to_date: str): if not from_date: from_date = add_months(getdate(), -1) if not to_date: @@ -590,7 +590,7 @@ def get_chart_date_range(from_date, to_date): return from_date, to_date -def get_chart_filters(doctype, chart, datefield, from_date, to_date): +def get_chart_filters(doctype: str, chart: object, datefield: str, from_date: str, to_date: str): version = get_frappe_version() if version.startswith("15.") or version.startswith("14."): filters = [([chart.document_type, "docstatus", "<", 2, False])] @@ -606,7 +606,9 @@ def get_chart_filters(doctype, chart, datefield, from_date, to_date): return filters -def get_chart_details(doctype, datefield, value_field, chart, from_date, to_date): +def get_chart_details( + doctype: str, datefield: str, value_field: str, chart: object, from_date: str, to_date: str +): filters = get_chart_filters(doctype, chart, datefield, from_date, to_date) version = get_frappe_version() if version.startswith("15.") or version.startswith("14."): @@ -641,7 +643,7 @@ def get_course_completion_data(): ] -def get_evaluator(course, batch=None): +def get_evaluator(course: str, batch: str = None): evaluator = None if batch: evaluator = frappe.db.get_value( @@ -654,7 +656,7 @@ def get_evaluator(course, batch=None): return evaluator -def check_multicurrency(amount, currency, country=None, amount_usd=None): +def check_multicurrency(amount: float, currency: str, country: str = None, amount_usd: float = None): settings = frappe.get_single("LMS Settings") show_usd_equivalent = settings.show_usd_equivalent @@ -697,7 +699,7 @@ def check_multicurrency(amount, currency, country=None, amount_usd=None): return rounded(amount), currency -def apply_gst(amount, country=None): +def apply_gst(amount: float, country: str = None) -> tuple: gst_applied = 0 apply_gst = frappe.db.get_single_value("LMS Settings", "apply_gst") @@ -711,7 +713,7 @@ def apply_gst(amount, country=None): return amount, gst_applied -def get_current_exchange_rate(source, target="USD"): +def get_current_exchange_rate(source: str, target: str = "USD") -> float: url = f"https://api.frankfurter.app/latest?from={source}&to={target}" response = requests.request("GET", url) @@ -728,7 +730,7 @@ def guest_access_allowed(): @frappe.whitelist(allow_guest=True) @rate_limit(limit=500, seconds=60 * 60) -def get_courses(filters=None, start=0): +def get_courses(filters: dict = None, start: int = 0) -> list: """Returns the list of courses.""" if not guest_access_allowed(): @@ -758,7 +760,7 @@ def get_courses(filters=None, start=0): return courses -def get_course_card_details(courses): +def get_course_card_details(courses: list) -> list: for course in courses: course.instructors = get_instructors("LMS Course", course.name) @@ -771,7 +773,7 @@ def get_course_card_details(courses): return courses -def get_course_or_filters(filters): +def get_course_or_filters(filters: dict) -> dict: or_filters = {} or_filters.update({"title": filters.get("title")}) or_filters.update({"short_introduction": filters.get("title")}) @@ -780,7 +782,7 @@ def get_course_or_filters(filters): return or_filters -def update_course_filters(filters): +def update_course_filters(filters: dict) -> tuple: or_filters = {} show_featured = False @@ -813,7 +815,7 @@ def update_course_filters(filters): return filters, or_filters, show_featured -def get_enrollment_details(courses): +def get_enrollment_details(courses: list) -> list: for course in courses: filters = { "course": course.name, @@ -831,7 +833,7 @@ def get_enrollment_details(courses): return courses -def get_featured_courses(filters, or_filters, fields): +def get_featured_courses(filters: dict, or_filters: dict, fields: list) -> list: filters.update({"featured": 1}) featured_courses = frappe.get_all( "LMS Course", @@ -874,7 +876,7 @@ def get_course_fields(): @frappe.whitelist(allow_guest=True) @rate_limit(limit=500, seconds=60 * 60) -def get_course_details(course): +def get_course_details(course: str): if not guest_access_allowed(): return {} @@ -915,7 +917,7 @@ def get_course_details(course): return course_details -def get_categorized_courses(courses): +def get_categorized_courses(courses: list) -> dict: live, upcoming, new, enrolled, created, under_review = [], [], [], [], [], [] for course in courses: @@ -951,7 +953,7 @@ def get_categorized_courses(courses): @frappe.whitelist(allow_guest=True) -def get_course_outline(course, progress=False): +def get_course_outline(course: str, progress: bool = False) -> list: """Returns the course outline.""" if not guest_access_allowed(): @@ -983,7 +985,7 @@ def get_course_outline(course, progress=False): @frappe.whitelist(allow_guest=True) @rate_limit(limit=500, seconds=60 * 60) -def get_lesson(course, chapter, lesson): +def get_lesson(course: str, chapter: int, lesson: int) -> dict: if not guest_access_allowed(): return {} @@ -1067,7 +1069,7 @@ def get_lesson(course, chapter, lesson): return lesson_details -def get_video_details(lesson_name): +def get_video_details(lesson_name: str) -> list: return frappe.get_all( "LMS Video Watch Duration", {"lesson": lesson_name, "member": frappe.session.user}, @@ -1075,7 +1077,7 @@ def get_video_details(lesson_name): ) -def get_neighbour_lesson(course, chapter, lesson): +def get_neighbour_lesson(course: str, chapter: int, lesson: int) -> dict: numbers = [] current = f"{chapter}.{lesson}" chapters = frappe.get_all("Chapter Reference", {"parent": course}, ["idx", "chapter"]) @@ -1097,7 +1099,7 @@ def get_neighbour_lesson(course, chapter, lesson): @frappe.whitelist(allow_guest=True) @rate_limit(limit=500, seconds=60 * 60) -def get_batch_details(batch): +def get_batch_details(batch: str): if not guest_access_allowed(): return {} @@ -1168,7 +1170,7 @@ def get_batch_details(batch): return batch_details -def categorize_batches(batches): +def categorize_batches(batches: list) -> dict: upcoming, archived, private, enrolled = [], [], [], [] for batch in batches: @@ -1212,7 +1214,7 @@ def get_country_code(): @frappe.whitelist() -def get_question_details(question): +def get_question_details(question: str) -> dict: fields = ["question", "type", "multiple"] for i in range(1, 5): fields.append(f"option_{i}") @@ -1224,7 +1226,7 @@ def get_question_details(question): @frappe.whitelist(allow_guest=True) @rate_limit(limit=500, seconds=60 * 60) -def get_batch_courses(batch): +def get_batch_courses(batch: str) -> list: if not guest_access_allowed(): return [] @@ -1240,7 +1242,7 @@ def get_batch_courses(batch): @frappe.whitelist() -def get_assessments(batch): +def get_assessments(batch: str) -> list: member = frappe.session.user assessments = frappe.get_all( "LMS Assessment", @@ -1262,7 +1264,7 @@ def get_assessments(batch): return assessments -def get_assignment_details(assessment, member): +def get_assignment_details(assessment: dict, member: str) -> dict: assessment.title = frappe.db.get_value("LMS Assignment", assessment.assessment_name, "title") existing_submission = frappe.db.exists( @@ -1293,7 +1295,7 @@ def get_assignment_details(assessment, member): return assessment -def get_quiz_details(assessment, member): +def get_quiz_details(assessment: dict, member: str) -> dict: assessment_details = frappe.db.get_value( "LMS Quiz", assessment.assessment_name, ["title", "passing_percentage"], as_dict=1 ) @@ -1325,7 +1327,7 @@ def get_quiz_details(assessment, member): return assessment -def get_exercise_details(assessment, member): +def get_exercise_details(assessment: dict, member: str) -> dict: assessment.title = frappe.db.get_value("LMS Programming Exercise", assessment.assessment_name, "title") filters = {"member": member, "exercise": assessment.assessment_name} @@ -1349,7 +1351,7 @@ def get_exercise_details(assessment, member): @frappe.whitelist() -def get_batch_assessment_count(batch): +def get_batch_assessment_count(batch: str) -> int: frappe.only_for(["Moderator", "Batch Evaluator"]) if not frappe.db.exists("LMS Batch", batch): frappe.throw(_("The specified batch does not exist.")) @@ -1357,7 +1359,9 @@ def get_batch_assessment_count(batch): @frappe.whitelist() -def get_batch_students(filters, offset=0, limit_start=0, limit_page_length=None, limit=None): +def get_batch_students( + filters: dict, offset: int = 0, limit_start: int = 0, limit_page_length: int = None, limit: int = None +): # limit_start and limit_page_length are used for backward compatibility start = limit_start or offset page_length = limit_page_length or limit @@ -1386,7 +1390,7 @@ def get_batch_students(filters, offset=0, limit_start=0, limit_page_length=None, return students -def get_course_completion_stats(batch): +def get_course_completion_stats(batch: str) -> list: """Get completion counts per course in batch""" BatchCourse = frappe.qb.DocType("Batch Course") BatchEnrollment = frappe.qb.DocType("LMS Batch Enrollment") @@ -1409,7 +1413,7 @@ def get_course_completion_stats(batch): return [{"task": row.title, "value": row.completed or 0} for row in rows] -def get_assignment_pass_stats(batch): +def get_assignment_pass_stats(batch: str) -> list: """Get pass counts per assignment in batch""" Assessment = frappe.qb.DocType("LMS Assessment") Assignment = frappe.qb.DocType("LMS Assignment") @@ -1438,7 +1442,7 @@ def get_assignment_pass_stats(batch): return [{"task": row.title, "value": row.passed or 0} for row in rows] -def get_quiz_pass_stats(batch): +def get_quiz_pass_stats(batch: str) -> list: """Get pass counts per quiz in batch""" Assessment = frappe.qb.DocType("LMS Assessment") Quiz = frappe.qb.DocType("LMS Quiz") @@ -1467,7 +1471,7 @@ def get_quiz_pass_stats(batch): @frappe.whitelist() -def get_batch_chart_data(batch): +def get_batch_chart_data(batch: str) -> list: """Get completion counts per course and assessment""" if not can_modify_batch(batch): frappe.throw(_("You are not authorized to view the chart data of this batch.")) @@ -1477,7 +1481,7 @@ def get_batch_chart_data(batch): return get_course_completion_stats(batch) + get_assignment_pass_stats(batch) + get_quiz_pass_stats(batch) -def get_batch_student_details(student): +def get_batch_student_details(student: dict) -> dict: details = frappe.db.get_value( "User", student.member, @@ -1490,7 +1494,7 @@ def get_batch_student_details(student): return details -def calculate_student_progress(batch, details): +def calculate_student_progress(batch: str, details: dict): batch_courses = frappe.get_all("Batch Course", {"parent": batch}, ["course", "title"]) assessments = frappe.get_all( "LMS Assessment", @@ -1514,7 +1518,7 @@ def calculate_student_progress(batch, details): details.progress = 0 -def calculate_course_progress(batch_courses, details): +def calculate_course_progress(batch_courses: list, details: dict): course_progress = [] details.courses = frappe._dict() @@ -1533,7 +1537,7 @@ def calculate_course_progress(batch_courses, details): ) -def calculate_assessment_progress(assessments, details): +def calculate_assessment_progress(assessments: list, details: dict): assessments_completed = 0 details.assessments = frappe._dict() @@ -1552,7 +1556,7 @@ def calculate_assessment_progress(assessments, details): ) -def has_submitted_assessment(assessment, assessment_type, member=None): +def has_submitted_assessment(assessment: str, assessment_type: str, member: str = None): if not member: member = frappe.session.user @@ -1607,7 +1611,7 @@ def has_submitted_assessment(assessment, assessment_type, member=None): ) -def can_access_topic(doctype, docname): +def can_access_topic(doctype: str, docname: str) -> bool: is_student = False if doctype == "Course Lesson": course = frappe.db.get_value("Course Lesson", docname, "course") @@ -1624,7 +1628,7 @@ def can_access_topic(doctype, docname): @frappe.whitelist() -def get_discussion_topics(doctype, docname, single_thread): +def get_discussion_topics(doctype: str, docname: str, single_thread: bool = False): if not can_access_topic(doctype, docname): frappe.throw(_("You are not authorized to view the discussion topics for this item.")) @@ -1654,7 +1658,7 @@ def get_discussion_topics(doctype, docname, single_thread): return topics -def create_discussion_topic(doctype, docname): +def create_discussion_topic(doctype: str, docname: str) -> str: doc = frappe.new_doc("Discussion Topic") doc.update( { @@ -1668,7 +1672,7 @@ def create_discussion_topic(doctype, docname): @frappe.whitelist() -def get_discussion_replies(topic): +def get_discussion_replies(topic: str): doctype = frappe.db.get_value("Discussion Topic", topic, "reference_doctype") if not can_access_topic(doctype, topic): frappe.throw(_("You are not authorized to view the discussion replies for this topic.")) @@ -1689,7 +1693,7 @@ def get_discussion_replies(topic): @frappe.whitelist() -def get_order_summary(doctype, docname, coupon=None, country=None): +def get_order_summary(doctype: str, docname: str, coupon: str = None, country: str = None): details = get_paid_course_details(docname) if doctype == "LMS Course" else get_paid_batch_details(docname) details.amount, details.currency = check_multicurrency( @@ -1708,7 +1712,7 @@ def get_order_summary(doctype, docname, coupon=None, country=None): return details -def get_paid_course_details(docname): +def get_paid_course_details(docname: str) -> dict: details = frappe.db.get_value( "LMS Course", docname, @@ -1730,7 +1734,7 @@ def get_paid_course_details(docname): return details -def get_paid_batch_details(docname): +def get_paid_batch_details(docname: str) -> dict: details = frappe.db.get_value( "LMS Batch", docname, @@ -1744,7 +1748,7 @@ def get_paid_batch_details(docname): return details -def adjust_amount_for_coupon(details, coupon, doctype, docname): +def adjust_amount_for_coupon(details: dict, coupon: str, doctype: str, docname: str): if not coupon: return discount_amount, subtotal, coupon_name = apply_coupon(doctype, docname, coupon, details.amount) @@ -1754,7 +1758,7 @@ def adjust_amount_for_coupon(details, coupon, doctype, docname): details.coupon = coupon_name -def get_gst_details(details, country): +def get_gst_details(details: dict, country: str): if details.currency != "INR": return @@ -1762,7 +1766,7 @@ def get_gst_details(details, country): details.gst_amount_formatted = fmt_money(details.gst_applied, 0, details.currency) -def apply_coupon(doctype, docname, code, base_amount): +def apply_coupon(doctype: str, docname: str, code: str, base_amount: float): coupon_name = frappe.db.exists("LMS Coupon", {"code": code, "enabled": 1}) if not coupon_name: frappe.throw(_("The coupon code '{0}' is invalid.").format(code)) @@ -1792,7 +1796,7 @@ def apply_coupon(doctype, docname, code, base_amount): return discount_amount, subtotal, coupon_name -def validate_coupon(code, coupon): +def validate_coupon(code: str, coupon: dict): if coupon.expires_on and getdate(coupon.expires_on) < getdate(): frappe.throw(_("This coupon has expired.")) @@ -1800,7 +1804,7 @@ def validate_coupon(code, coupon): frappe.throw(_("This coupon has reached its maximum usage limit.")) -def validate_coupon_applicability(doctype, docname, coupon_name): +def validate_coupon_applicability(doctype: str, docname: str, coupon_name: str): applicable_item = frappe.db.exists( "LMS Coupon Item", {"parent": coupon_name, "reference_doctype": doctype, "reference_name": docname} ) @@ -1812,7 +1816,7 @@ def validate_coupon_applicability(doctype, docname, coupon_name): ) -def calculate_discount_amount(base_amount, coupon): +def calculate_discount_amount(base_amount: float, coupon: dict) -> float: discount_amount = 0 if coupon.discount_type == "Percentage": @@ -1824,7 +1828,7 @@ def calculate_discount_amount(base_amount, coupon): @frappe.whitelist() -def get_lesson_creation_details(course, chapter, lesson): +def get_lesson_creation_details(course: str, chapter: int, lesson: int) -> dict: frappe.only_for(["Moderator", "Course Creator"]) chapter_name = frappe.db.get_value("Chapter Reference", {"parent": course, "idx": chapter}, "chapter") lesson_name = frappe.db.get_value("Lesson Reference", {"parent": chapter_name, "idx": lesson}, "lesson") @@ -1855,7 +1859,7 @@ def get_lesson_creation_details(course, chapter, lesson): @frappe.whitelist() -def get_roles(name): +def get_roles(name: str) -> dict: frappe.only_for(["Moderator", "Batch Evaluator"]) return { "moderator": has_moderator_role(name), @@ -1865,11 +1869,11 @@ def get_roles(name): } -def publish_notifications(doc, method): +def publish_notifications(doc: dict, method: str): frappe.publish_realtime("publish_lms_notifications", user=doc.for_user, after_commit=True) -def update_payment_record(doctype, docname): +def update_payment_record(doctype: str, docname: str): request = get_integration_requests(doctype, docname) if len(request): @@ -1888,7 +1892,7 @@ def update_payment_record(doctype, docname): enroll_in_batch(docname, data.payment) -def get_integration_requests(doctype, docname): +def get_integration_requests(doctype: str, docname: str): return frappe.get_all( "Integration Request", { @@ -1902,13 +1906,13 @@ def get_integration_requests(doctype, docname): ) -def get_payment_doc(payment_name): +def get_payment_doc(payment_name: str) -> dict: return frappe.db.get_value( "LMS Payment", payment_name, ["name", "coupon", "payment_for_certificate"], as_dict=True ) -def update_payment_details(data): +def update_payment_details(data: dict): payment_id = get_payment_id(data) frappe.db.set_value( @@ -1922,7 +1926,7 @@ def update_payment_details(data): ) -def get_payment_id(data): +def get_payment_id(data: dict) -> str: payment_gateway = data.get("payment_gateway") if payment_gateway == "Razorpay": payment_id = "razorpay_payment_id" @@ -1933,7 +1937,7 @@ def get_payment_id(data): return payment_id -def update_coupon_redemption(payment_doc): +def update_coupon_redemption(payment_doc: dict): if payment_doc.coupon: redemption_count = frappe.db.get_value("LMS Coupon", payment_doc.coupon, "redemption_count") or 0 @@ -1945,7 +1949,7 @@ def update_coupon_redemption(payment_doc): ) -def enroll_in_course(course, payment_name): +def enroll_in_course(course: str, payment_name: str): if not frappe.db.exists("LMS Enrollment", {"member": frappe.session.user, "course": course}): enrollment = frappe.new_doc("LMS Enrollment") payment = frappe.db.get_value("LMS Payment", payment_name, ["name", "source"], as_dict=True) @@ -1961,7 +1965,7 @@ def enroll_in_course(course, payment_name): @frappe.whitelist() -def enroll_in_batch(batch, payment_name=None): +def enroll_in_batch(batch: str, payment_name: str = None): if not frappe.db.exists("LMS Batch", batch): frappe.throw(_("The specified batch does not exist.")) @@ -1969,7 +1973,7 @@ def enroll_in_batch(batch, payment_name=None): create_enrollment(batch, payment_doc) -def get_payment_details(payment_name): +def get_payment_details(payment_name: str) -> dict: payment_doc = None if payment_name: payment_doc = frappe.db.get_value( @@ -1978,7 +1982,7 @@ def get_payment_details(payment_name): return payment_doc -def create_enrollment(batch, payment_doc=None): +def create_enrollment(batch: str, payment_doc: dict = None): new_student = frappe.new_doc("LMS Batch Enrollment") new_student.update( { @@ -1997,7 +2001,7 @@ def create_enrollment(batch, payment_doc=None): new_student.save() -def update_certificate_purchase(course, payment_name): +def update_certificate_purchase(course: str, payment_name: str): frappe.db.set_value( "LMS Enrollment", {"member": frappe.session.user, "course": course}, @@ -2044,7 +2048,7 @@ def get_programs(): @frappe.whitelist() -def get_program_details(program_name): +def get_program_details(program_name: str) -> dict: if not guest_access_allowed(): frappe.throw(_("Please login to view program details.")) @@ -2088,7 +2092,7 @@ def get_program_details(program_name): @frappe.whitelist() -def enroll_in_program(program): +def enroll_in_program(program: str): validate_program_enrollment(program) if not frappe.db.exists("LMS Program Member", {"parent": program, "member": frappe.session.user}): @@ -2104,7 +2108,7 @@ def enroll_in_program(program): program_member.save(ignore_permissions=True) -def validate_program_enrollment(program): +def validate_program_enrollment(program: str): published = frappe.db.get_value("LMS Program", program, "published") if not published: frappe.throw(_("You cannot enroll in an unpublished program.")) @@ -2112,7 +2116,7 @@ def validate_program_enrollment(program): @frappe.whitelist(allow_guest=True) @rate_limit(limit=500, seconds=60 * 60) -def get_batches(filters=None, start=0, order_by="start_date"): +def get_batches(filters: dict = None, start: int = 0, order_by: str = "start_date"): if not guest_access_allowed(): return [] @@ -2156,7 +2160,7 @@ def get_batches(filters=None, start=0, order_by="start_date"): return batches -def filter_batches_based_on_start_time(batches, filters): +def filter_batches_based_on_start_time(batches: list, filters: dict) -> list: batchType = get_batch_type(filters) if batchType == "upcoming": batches_to_remove = [ @@ -2175,7 +2179,7 @@ def filter_batches_based_on_start_time(batches, filters): return batches -def get_batch_type(filters): +def get_batch_type(filters: dict) -> str: start_date_filter = filters.get("start_date") batchType = None if start_date_filter: @@ -2188,7 +2192,7 @@ def get_batch_type(filters): return batchType -def get_batch_card_details(batches): +def get_batch_card_details(batches: list) -> list: for batch in batches: batch.instructors = get_instructors("LMS Batch", batch.name) students_count = frappe.db.count("LMS Batch Enrollment", {"batch": batch.name}) @@ -2205,7 +2209,7 @@ def get_batch_card_details(batches): return batches -def get_palette(full_name): +def get_palette(full_name: str) -> list: """ Returns a color unique to each member for Avatar""" @@ -2229,7 +2233,7 @@ def get_palette(full_name): @frappe.whitelist(allow_guest=True) @rate_limit(limit=500, seconds=60 * 60) -def get_related_courses(course): +def get_related_courses(course: str) -> list: if not guest_access_allowed(): return [] @@ -2245,7 +2249,7 @@ def persona_captured(): frappe.db.set_single_value("LMS Settings", "persona_captured", 1) -def validate_discussion_reply(doc, method): +def validate_discussion_reply(doc: dict, method: str): topic = frappe.db.get_value( "Discussion Topic", doc.topic, ["reference_doctype", "reference_docname"], as_dict=True ) @@ -2257,7 +2261,7 @@ def validate_discussion_reply(doc, method): validate_batch_access(topic.reference_docname) -def validate_course_access(lesson): +def validate_course_access(lesson: str): if not frappe.db.exists("Course Lesson", lesson): frappe.throw(_("The lesson does not exist.")) @@ -2273,7 +2277,7 @@ def validate_course_access(lesson): frappe.throw(_("You do not have access to this course.")) -def validate_batch_access(batch): +def validate_batch_access(batch: str): if not frappe.db.exists("LMS Batch", batch): frappe.throw(_("The batch does not exist.")) @@ -2290,7 +2294,7 @@ def validate_batch_access(batch): frappe.throw(_("You do not have access to this batch.")) -def can_modify_course(course): +def can_modify_course(course: str) -> bool: is_instructor = frappe.db.exists( "Course Instructor", {"instructor": frappe.session.user, "parent": course, "parenttype": "LMS Course"}, @@ -2300,7 +2304,7 @@ def can_modify_course(course): return True -def can_modify_batch(batch): +def can_modify_batch(batch: str) -> bool: is_instructor = frappe.db.exists( "Course Instructor", {