Improve JSON importer code

evap/staff/importers/json.py

@@ -1,3 +1,4 @@
+import json
 import logging
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta
@@ -128,7 +129,7 @@ def get_log(self) -> str:
 class JSONImporter:
     DATETIME_FORMAT = "%d.%m.%Y %H:%M"
 
-    def __init__(self, semester: Semester):
+    def __init__(self, semester: Semester) -> None:
         self.semester = semester
         self.user_profile_map: dict[str, UserProfile] = {}
         self.course_type_cache: dict[str, CourseType] = {}
@@ -155,7 +156,7 @@ def _get_degree(self, name: str) -> Degree:
     def _get_user_profiles(self, data: list[ImportRelated]) -> list[UserProfile]:
         return [self.user_profile_map[related["gguid"]] for related in data]
 
-    def _import_students(self, data: list[ImportStudent]):
+    def _import_students(self, data: list[ImportStudent]) -> None:
         for entry in data:
             email = clean_email(entry["email"])
             user_profile, __, changes = update_or_create_with_changes(
@@ -183,7 +184,7 @@ def _import_students(self, data: list[ImportStudent]):
 
             self.user_profile_map[entry["gguid"]] = user_profile
 
-    def _import_lecturers(self, data: list[ImportLecturer]):
+    def _import_lecturers(self, data: list[ImportLecturer]) -> None:
         for entry in data:
             email = clean_email(entry["email"])
             user_profile, __ = UserProfile.objects.update_or_create(
@@ -281,7 +282,9 @@ def _import_evaluation(self, course: Course, data: ImportEvent) -> Evaluation:
 
         return evaluation
 
-    def _import_contribution(self, evaluation: Evaluation, data: ImportRelated):
+    def _import_contribution(
+        self, evaluation: Evaluation, data: ImportRelated
+    ) -> tuple[Contribution, bool, dict[str, tuple[any, any]]]:
         user_profile = self.user_profile_map[data["gguid"]]
 
         contribution, created, changes = update_or_create_with_changes(
@@ -291,32 +294,32 @@ def _import_contribution(self, evaluation: Evaluation, data: ImportRelated):
         )
         return contribution, created, changes
 
-    def _import_events(self, data: list[ImportEvent]):
+    def _import_events(self, data: list[ImportEvent]) -> None:
         # Divide in two lists so corresponding courses are imported before their exams
-        normal_events = filter(lambda e: not e["isexam"], data)
-        exam_events = filter(lambda e: e["isexam"], data)
+        normal_events = (event for event in data if not event["isexam"])
+        exam_events = (event for event in data if event["isexam"])
 
         for event in normal_events:
-            event: ImportEvent
-
             course = self._import_course(event)
 
             self._import_evaluation(course, event)
 
         for event in exam_events:
-            event: ImportEvent
-
             course = self.course_map[event["relatedevents"]["gguid"]]
 
             self._import_evaluation(course, event)
 
-    def _process_log(self):
+    def _process_log(self) -> None:
         log = self.statistics.get_log()
         logger.info(log)
 
     @transaction.atomic
-    def import_json(self, data: ImportDict):
+    def import_dict(self, data: ImportDict) -> None:
         self._import_students(data["students"])
         self._import_lecturers(data["lecturers"])
         self._import_events(data["events"])
         self._process_log()
+
+    def import_json(self, data: str) -> None:
+        data = json.loads(data)
+        self.import_dict(data)

evap/staff/log_handler.py

@@ -5,6 +5,7 @@
 
 def mail_managers(subject, message, fail_silently=False, connection=None, html_message=None):
     """Send a message to the managers, as defined by the MANAGERS setting."""
+    # pylint: disable=import-outside-toplevel
     from evap.evaluation.models import UserProfile
 
     managers = UserProfile.objects.filter(groups__name="Manager", email__isnull=False)

evap/staff/tests/test_json_importer.py

@@ -85,7 +85,7 @@ def test_import_existing_students(self):
         importer = JSONImporter(self.semester)
         importer._import_students(self.students)
 
-        assert UserProfile.objects.all().count() == 2
+        self.assertEqual(UserProfile.objects.all().count(), 2)
 
         user_profile.refresh_from_db()
 
@@ -244,7 +244,7 @@ def test_import_courses_evaluation_approved(self):
         self.assertEqual(len(importer.statistics.attempted_changes), 1)
 
     def test_import_courses_update(self):
-        importer = self._import()
+        self._import()
 
         self.assertEqual(Course.objects.all().count(), 1)
         course = Course.objects.all()[0]

evap/staff/tools.py

@@ -2,6 +2,7 @@
 from collections.abc import Iterable
 from datetime import date, datetime, timedelta
 from enum import Enum
+from typing import TypeVar
 
 from django.conf import settings
 from django.contrib import messages
@@ -383,11 +384,14 @@ def user_edit_link(user_id):
     )
 
 
+T = TypeVar("T")
+
+
 def update_or_create_with_changes(
-    model: type[Model],
+    model: type[T],
     defaults=None,
     **kwargs,
-) -> tuple[Model, bool, dict[str, tuple[any, any]]]:
+) -> tuple[T, bool, dict[str, tuple[any, any]]]:
     """Do update_or_create and track changed values."""
 
     if not defaults:

test_data.json

@@ -0,0 +1,117 @@
+{
+  "students": [
+    {
+      "gguid": "0x1",
+      "email": "[email protected]",
+      "name": "1",
+      "christianname": "1"
+    },
+    {
+      "gguid": "0x2",
+      "email": "[email protected]",
+      "name": "2",
+      "christianname": "2"
+    }
+  ],
+  "lecturers": [
+    {
+      "gguid": "0x3",
+      "email": "[email protected]",
+      "name": "3",
+      "christianname": "3",
+      "titlefront": "Prof. Dr."
+    },
+    {
+      "gguid": "0x4",
+      "email": "[email protected]",
+      "name": "4",
+      "christianname": "4",
+      "titlefront": "Dr."
+    }
+  ],
+  "events": [
+    {
+      "gguid": "0x5",
+      "lvnr": 1,
+      "title": "Prozessorientierte Informationssysteme",
+      "title_en": "Process-oriented information systems",
+      "type": "Vorlesung",
+      "isexam": false,
+      "courses": [
+        {
+          "cprid": "BA-Inf",
+          "scale": "GRADE_PARTICIPATION"
+        },
+        {
+          "cprid": "MA-Inf",
+          "scale": "GRADE_PARTICIPATION"
+        }
+      ],
+      "relatedevents": {
+        "gguid": "0x6"
+      },
+      "appointments": [
+        {
+          "begin": "15.04.2024 10:15",
+          "end": "15.07.2024 11:45"
+        }
+      ],
+      "lecturers": [
+        {
+          "gguid": "0x3"
+        }
+      ],
+      "students": [
+        {
+          "gguid": "0x1"
+        },
+        {
+          "gguid": "0x2"
+        }
+      ]
+    },
+    {
+      "gguid": "0x6",
+      "lvnr": 2,
+      "title": "Prozessorientierte Informationssysteme",
+      "title_en": "Process-oriented information systems",
+      "type": "Klausur",
+      "isexam": true,
+      "courses": [
+        {
+          "cprid": "BA-Inf",
+          "scale": ""
+        },
+        {
+          "cprid": "MA-Inf",
+          "scale": ""
+        }
+      ],
+      "relatedevents": {
+        "gguid": "0x5"
+      },
+      "appointments": [
+        {
+          "begin": "29.07.2024 10:15",
+          "end": "29.07.2024 11:45"
+        }
+      ],
+      "lecturers": [
+        {
+          "gguid": "0x3"
+        },
+        {
+          "gguid": "0x4"
+        }
+      ],
+      "students": [
+        {
+          "gguid": "0x1"
+        },
+        {
+          "gguid": "0x2"
+        }
+      ]
+    }
+  ]
+}