From 9e921e0744551801d74656d25ef5ee8feccd857c Mon Sep 17 00:00:00 2001 From: Jan Gaura Date: Tue, 16 Apr 2024 09:59:16 +0200 Subject: [PATCH] Removed possible infinite loop in INBUS integration Various typing enhancements in INBUS integration, code reformat, and import change. --- api/views.py | 5 ++-- common/inbus/inbus.py | 70 +++++++++++++++++++++++++++---------------- common/inbus/utils.py | 2 +- 3 files changed, 48 insertions(+), 29 deletions(-) diff --git a/api/views.py b/api/views.py index 8a26ee8e..b4af15dd 100644 --- a/api/views.py +++ b/api/views.py @@ -22,7 +22,7 @@ import common.bulk_import from common.bulk_import import ImportException -from common import inbus +from common.inbus import inbus from pathlib import Path from shutil import copytree, ignore_patterns from django.utils.dateparse import parse_datetime @@ -608,7 +608,8 @@ def import_activities(request): subject = post['subject'] activities_id = post['activities'] - activities = [ inbus.inbus.concrete_activity(activity_id) for activity_id in activities_id ] + activities = [inbus.concrete_activity(activity_id) for activity_id in activities_id] + activities = [concrete_activity for concrete_activity in activities if concrete_activity is not None] semester = Semester.objects.get(pk=semester_id) try: diff --git a/common/inbus/inbus.py b/common/inbus/inbus.py index 02cd8fc9..e3851441 100644 --- a/common/inbus/inbus.py +++ b/common/inbus/inbus.py @@ -21,42 +21,50 @@ def person_by_login(login: str) -> dto.PersonSimple | None: if 'login' not in person_json: return None - person_simple = dto.PersonSimple(login=person_json["login"].upper(), first_name=person_json.get('firstName', ''), second_name=person_json.get('secondName', ''), + person_simple = dto.PersonSimple(login=person_json["login"].upper(), first_name=person_json.get('firstName', ''), + second_name=person_json.get('secondName', ''), full_name=person_json.get('fullName', ''), email=person_json.get('email', '')) return person_simple -def subject_versions(department_id: dto.DepartmentId = 386) -> List[dto.SubjectVersion]: +def subject_versions(department_id: dto.DepartmentId = dto.DepartmentId(386)) -> List[dto.SubjectVersion]: """ Get list of all subjects and their versions by department. Here `386` is Department of Computer Science. """ - results_per_page = 20 + results_per_page: int = 20 subject_versions = [] - offset = 0 + offset: int = 0 while True: url = urllib.parse.urljoin(config.INBUS_SERVICE_EDISON_URL, 'edu/subjectVersions') subject_versions_resp = utils.inbus_request(url, {'departmentId': department_id, 'offset': offset, 'limit': results_per_page}) - subject_versions_json = subject_versions_resp.json() + if subject_versions_resp: + subject_versions_json = subject_versions_resp.json() - for subject_version_json in subject_versions_json: - subject_json = subject_version_json['subject'] - subject_guarantee = serde.from_dict(dto.Person, subject_json['guarantee']) - subject_version_guarantee = serde.from_dict(dto.Person, subject_version_json['guarantee']) - subject = dto.Subject(subjectId=subject_json['subjectId'], code=subject_json['code'], abbrev=subject_json['abbrev'], - title=subject_json['title'], guarantee=subject_guarantee) - subject_version = dto.SubjectVersion(subjectVersionId=subject_version_json['subjectVersionId'], subject=subject, - subjectVersionCompleteCode=subject_version_json['subjectVersionCompleteCode'], guarantee=subject_version_guarantee) + for subject_version_json in subject_versions_json: + subject_json = subject_version_json['subject'] + subject_guarantee = serde.from_dict(dto.Person, subject_json['guarantee']) + subject_version_guarantee = serde.from_dict(dto.Person, subject_version_json['guarantee']) + subject = dto.Subject(subjectId=subject_json['subjectId'], code=subject_json['code'], + abbrev=subject_json['abbrev'], title=subject_json['title'], + guarantee=subject_guarantee) + subject_version = dto.SubjectVersion(subjectVersionId=subject_version_json['subjectVersionId'], + subject=subject, + subjectVersionCompleteCode=subject_version_json['subjectVersionCompleteCode'], + guarantee=subject_version_guarantee) - subject_versions.append(subject_version) + subject_versions.append(subject_version) - results = len(subject_versions_json) + results: int = len(subject_versions_json) - offset += results + offset += results - if results == 0: + if results == 0: + break + + else: break return subject_versions @@ -69,29 +77,39 @@ def schedule_subject_by_version_id(subject_version_id: dto.SubjectVersionId) -> url = urllib.parse.urljoin(config.INBUS_SERVICE_EDISON_URL, 'schedule') concrete_activities_resp = utils.inbus_request(url, {'subjectVersionId': subject_version_id, 'semesterId': 129}) - concrete_activities_json = concrete_activities_resp.json() + concrete_activities = [] + + if concrete_activities_resp: + concrete_activities_json = concrete_activities_resp.json() - concrete_activities = [ serde.from_dict(dto.ConcreteActivity, concrete_activity_dict) for concrete_activity_dict in concrete_activities_json ] + concrete_activities = [ serde.from_dict(dto.ConcreteActivity, concrete_activity_dict) for concrete_activity_dict in concrete_activities_json ] - return concrete_activities + return dto.SubjectVersionSchedule(concrete_activities) -def concrete_activity(concrete_activity_id: dto.ConcreteActivityId) -> dto.ConcreteActivity: +def concrete_activity(concrete_activity_id: dto.ConcreteActivityId) -> dto.ConcreteActivity | None: url = urllib.parse.urljoin(config.INBUS_SERVICE_EDISON_URL, f'schedule/{concrete_activity_id}') concrete_activity_resp = utils.inbus_request(url, {}) - concrete_activity_json = concrete_activity_resp.json() - concrete_activity = serde.from_dict(dto.ConcreteActivity, concrete_activity_json) + if concrete_activity_resp: + concrete_activity_json = concrete_activity_resp.json() - return concrete_activity + concrete_activity: dto.ConcreteActivity = serde.from_dict(dto.ConcreteActivity, concrete_activity_json) + return concrete_activity + else: + return None def students_in_concrete_activity(concrete_activity_id: dto.ConcreteActivityId) -> List[dto.StudyRelation]: url = urllib.parse.urljoin(config.INBUS_SERVICE_EDISON_URL, f'schedule/{concrete_activity_id}/studyRelations') study_relation_resp = utils.inbus_request(url, {}) - study_relation_json = study_relation_resp.json() - study_relations = [ serde.from_dict(dto.StudyRelation, study_relation) for study_relation in study_relation_json ] + study_relations: List[dto.StudyRelation] = [] + + if study_relation_resp: + study_relation_json = study_relation_resp.json() + + study_relations = [serde.from_dict(dto.StudyRelation, study_relation) for study_relation in study_relation_json] return study_relations diff --git a/common/inbus/utils.py b/common/inbus/utils.py index 97fdd10e..e0402aad 100644 --- a/common/inbus/utils.py +++ b/common/inbus/utils.py @@ -47,7 +47,7 @@ def is_response_ok_or_new_token_(response: requests.Response) -> bool: return False -def inbus_request(url, params: Dict = None) -> requests.Response | None: +def inbus_request(url, params: Dict | None = None) -> requests.Response | None: if params is None: params = {} token = inbus_token()