optimize various bits around scan profiles

octopoes/octopoes/core/service.py

@@ -264,21 +264,19 @@ def recalculate_scan_profiles(self, valid_time: datetime) -> None:
         # fetch all scan profiles
         all_scan_profiles = self.scan_profile_repository.list_scan_profiles(None, valid_time=valid_time)
 
-        # cache all declared
-        all_declared_scan_profiles = {
-            scan_profile for scan_profile in all_scan_profiles if isinstance(scan_profile, DeclaredScanProfile)
-        }
-        # cache all inherited
-        inherited_scan_profiles = {
-            scan_profile.reference: scan_profile
-            for scan_profile in all_scan_profiles
-            if isinstance(scan_profile, InheritedScanProfile)
-        }
-
-        # track all scan level assignments
-        assigned_scan_levels: dict[Reference, ScanLevel] = {
-            scan_profile.reference: scan_profile.level for scan_profile in all_declared_scan_profiles
-        }
+        all_declared_scan_profiles: set[DeclaredScanProfile] = set()
+        inherited_scan_profiles: dict[Reference, InheritedScanProfile] = {}
+        assigned_scan_levels: dict[Reference, ScanLevel] = {}
+        source_scan_profile_references: set[Reference] = set()
+
+        # fill profile caches
+        for scan_profile in all_scan_profiles:
+            if isinstance(scan_profile, DeclaredScanProfile):
+                all_declared_scan_profiles.add(scan_profile)
+                assigned_scan_levels[scan_profile.reference] = scan_profile.level
+                source_scan_profile_references.add(scan_profile.reference)
+            elif isinstance(scan_profile, InheritedScanProfile):
+                inherited_scan_profiles[scan_profile.reference] = scan_profile
 
         for current_level in range(4, 0, -1):
             # start point: all scan profiles with current level + all higher scan levels
@@ -331,7 +329,6 @@ def recalculate_scan_profiles(self, valid_time: datetime) -> None:
 
         # Save all assigned scan levels
         update_count = 0
-        source_scan_profile_references = {sp.reference for sp in all_declared_scan_profiles}
         for reference, scan_level in assigned_scan_levels.items():
             # Skip source scan profiles
             if reference in source_scan_profile_references:

octopoes/octopoes/repositories/scan_profile_repository.py

@@ -15,6 +15,8 @@
 from octopoes.xtdb.client import XTDBSession
 from octopoes.xtdb.query_builder import generate_pull_query
 
+scan_profile_adapter = TypeAdapter(ScanProfile)
+
 
 class ScanProfileRepository(Repository):
     def __init__(self, event_manager: EventManager):
@@ -62,7 +64,7 @@ def serialize(cls, scan_profile: ScanProfile) -> dict[str, Any]:
 
     @classmethod
     def deserialize(cls, data: dict[str, Any]) -> ScanProfileBase:
-        return TypeAdapter(ScanProfile).validate_python(data)
+        return scan_profile_adapter.validate_python(data)
 
     def list_scan_profiles(self, scan_profile_type: str | None, valid_time: datetime) -> list[ScanProfileBase]:
         where = {"type": self.object_type}