chore: added type hints to all whitelisted functions

This commit is contained in:
Jannat Patel
2026-02-06 16:01:49 +05:30
parent 1bbdff9aaf
commit cd0d4c413d
19 changed files with 266 additions and 273 deletions

View File

@@ -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 = []

View File

@@ -277,3 +277,4 @@ add_to_apps_screen = [
sqlite_search = ["lms.sqlite.LearningSearch"]
auth_hooks = ["lms.auth.authenticate"]
require_type_annotated_api_methods = True

View File

@@ -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)

View File

@@ -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)

View File

@@ -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)

View File

@@ -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
"""

View File

@@ -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

View File

@@ -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},

View File

@@ -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))

View File

@@ -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

View File

@@ -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,

View File

@@ -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",

View File

@@ -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)}")

View File

@@ -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:

View File

@@ -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(),
}

View File

@@ -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()

View File

@@ -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",
{