GREG-123: Adding option for acronym identifier import from orgreg

UiB want to make use of acronym from orgreg to identify the units. Adding option for importing these as ouidentifiers.

greg/management/commands/import_from_orgreg.py

@@ -12,12 +12,13 @@ orgreg/v3/ as the url argument (note the trailing slash).
 """
 import datetime
 import logging
-from typing import Union, Mapping, Dict
+from typing import Union, Mapping, Dict, Optional
 
 import orgreg_client
 from django.conf import settings
 from django.core.management.base import BaseCommand
 from orgreg_client import OrgUnit
+from orgreg_client.models import ExternalKey
 
 from greg.models import OrganizationalUnit, OuIdentifier
 
@@ -34,68 +35,110 @@ class Command(BaseCommand):
     ):
         """Upsert any configured extra IDs from orgreg."""
         for extra_id in settings.ORGREG_EXTRA_IDS:
-
-            matching_ids = [
-                x
-                for x in ou.external_keys
-                if x.source_system == extra_id["source"] and x.type == extra_id["type"]
-            ]
-
-            if not matching_ids:
-                logger.warning(
-                    "No %s id from %s found in OrgReg for ou %s",
-                    extra_id["type"],
+            identity_in_orgreg = self._get_external_key_from_ou(extra_id, ou)
+            if identity_in_orgreg is not None:
+                self._upsert_identifier(
                     extra_id["source"],
+                    extra_id["type"],
+                    identity_in_orgreg.value,
                     ou.ou_id,
                 )
-                continue
 
-            if len(matching_ids) > 1:
-                # External_ids
-                logger.warning(
-                    "Found multiple ids matching type: %s source: %s in OrgReg. Using the first one: %s",
-                    extra_id["type"],
-                    extra_id["source"],
-                    matching_ids[0].value,
-                )
+        # Acronyms can also be used as identifiers
+        for acronym in settings.ORGREG_ACRONYMS:
+            self.handle_acronym_identifier(acronym, ou)
+
+    def handle_acronym_identifier(self, acronym, ou):
+        if acronym == "nob" and ou.acronym.nob is not None:
+            self._upsert_identifier(
+                settings.ORGREG_SOURCE,
+                "acronym_" + acronym,
+                ou.acronym.nob,
+                ou.ou_id,
+            )
+        if acronym == "eng" and ou.acronym.eng is not None:
+            self._upsert_identifier(
+                settings.ORGREG_SOURCE,
+                "acronym_" + acronym,
+                ou.acronym.eng,
+                ou.ou_id,
+            )
+        if acronym == "nno" and ou.acronym.nno is not None:
+            self._upsert_identifier(
+                settings.ORGREG_SOURCE,
+                "acronym_" + acronym,
+                ou.acronym.nno,
+                ou.ou_id,
+            )
+
+    @staticmethod
+    def _get_external_key_from_ou(
+        extra_id: Dict[str, str], ou: OrgUnit
+    ) -> Optional[ExternalKey]:
+        matching_ids = [
+            x
+            for x in ou.external_keys
+            if x.source_system == extra_id["source"] and x.type == extra_id["type"]
+        ]
+
+        if not matching_ids:
+            logger.warning(
+                "No %s id from %s found in OrgReg for ou %s",
+                extra_id["type"],
+                extra_id["source"],
+                ou.ou_id,
+            )
+            return None
+
+        if len(matching_ids) > 1:
+            # External_ids
+            logger.warning(
+                "Found multiple ids matching type: %s source: %s in OrgReg. Using the first one: %s",
+                extra_id["type"],
+                extra_id["source"],
+                matching_ids[0].value,
+            )
 
-            identity_in_orgreg = matching_ids[0]
+        return matching_ids[0]
 
-            # Check if the id exists
-            identify_in_db = (
-                self.processed[ou.ou_id]
-                .identifiers.filter(
-                    source=extra_id["source"],
-                    name=extra_id["type"],
-                )
-                .first()
+    def _upsert_identifier(
+        self, source: str, identity_type: str, identity_in_orgreg_value: str, ou_id: int
+    ):
+        # Check if the id exists
+        identify_in_db = (
+            self.processed[ou_id]
+            .identifiers.filter(
+                source=source,
+                name=identity_type,
             )
+            .first()
+        )
 
-            if identify_in_db:
-                if identify_in_db.value != identity_in_orgreg.value:
-                    logger.info(
-                        "Updating id: source: %s, type: %s, old_id: %s, new_id %s",
-                        extra_id["source"],
-                        extra_id["type"],
-                        identify_in_db.value,
-                        identity_in_orgreg.value,
-                    )
-                    identify_in_db.value = identity_in_orgreg.value
-                    identify_in_db.save()
-            else:
-                OuIdentifier.objects.create(
-                    name=extra_id["type"],
-                    source=extra_id["source"],
-                    value=identity_in_orgreg["value"],
-                    orgunit=self.processed[ou.ou_id],
-                )
+        if identify_in_db:
+            if identify_in_db.value != identity_in_orgreg_value:
                 logger.info(
-                    "Added new id to ou: %s, type: %s, source: %s value: %s",
-                    ou.ou_id,
-                    extra_id["type"],
-                    extra_id["source"],
-                    identity_in_orgreg["value"],
+                    "Updating id: source: %s, type: %s, old_id: %s, new_id %s",
+                    source,
+                    identity_type,
+                    identify_in_db.value,
+                    identity_in_orgreg_value,
                 )
+                identify_in_db.value = identity_in_orgreg_value
+                identify_in_db.save()
+        else:
+            OuIdentifier.objects.create(
+                name=identity_type,
+                source=source,
+                value=identity_in_orgreg_value,
+                orgunit=self.processed[ou_id],
+            )
+            logger.info(
+                "Added new id to ou: %s, type: %s, source: %s value: %s",
+                ou_id,
+                identity_type,
+                source,
+                identity_in_orgreg_value,
+            )
 
     def _get_or_create_and_set_values(
         self, ou: OrgUnit, values: Mapping[str, Union[str, int, bool]]