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

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,107 @@ 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:
+            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,
                 )
 
-            identity_in_orgreg = matching_ids[0]
+    @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,
+            )
 
-            # Check if the id exists
-            identify_in_db = (
-                self.processed[ou.ou_id]
-                .identifiers.filter(
-                    source=extra_id["source"],
-                    name=extra_id["type"],
-                )
-                .first()
+        return matching_ids[0]
+
+    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]]

greg/tests/management/test_import_from_orgreg.py

@@ -34,6 +34,7 @@ def orgreg_response():
                 valid_from=datetime.date(year=2020, month=2, day=4),
                 external_keys=[],
                 name={"nob": "foo"},
+                acronym={"nob": "foo_acronym"},
             ),
             OrgUnit(
                 ou_id=3,
@@ -42,6 +43,7 @@ def orgreg_response():
                 parent=2,
                 name={"nob": "bar"},
                 external_keys=[],
+                acronym={"nob": "foo_acronym2"},
             ),
             OrgUnit(
                 ou_id=2,
@@ -49,6 +51,7 @@ def orgreg_response():
                 parent=1,
                 name={"eng": "baz"},
                 external_keys=[],
+                acronym={"nob": "foo_acronym3"},
             ),
         ]
     )
@@ -92,3 +95,26 @@ def test_run_twice(requests_mock, orgreg_response):
 
     call_command("import_from_orgreg")
     assert OrganizationalUnit.objects.all().count() == 3
+
+
+@pytest.mark.django_db
+def test_import_acronym(requests_mock, orgreg_response):
+    requests_mock.get("https://example.com/fake/ou/", text=orgreg_response.json())
+
+    settings.ORGREG_ACRONYMS.append("nob")
+
+    call_command("import_from_orgreg")
+    assert OrganizationalUnit.objects.all().count() == 3
+
+    assert (
+        OuIdentifier.objects.get(orgunit__id=1, name="acronym_nob").value
+        == "foo_acronym"
+    )
+    assert (
+        OuIdentifier.objects.get(orgunit__id=2, name="acronym_nob").value
+        == "foo_acronym2"
+    )
+    assert (
+        OuIdentifier.objects.get(orgunit__id=3, name="acronym_nob").value
+        == "foo_acronym3"
+    )

gregsite/settings/base.py

@@ -301,3 +301,7 @@ ORGREG_NAME = "orgreg_id"
 # Extra ids to be imported from orgreg, list of dict with source/type.
 # [{"source": "sapuio", "type": "legacy_stedkode"}]
 ORGREG_EXTRA_IDS = []
+
+# Acronyms to be imported as identifiers from orgreg. List of acronym values, supported ones are: nob, nno, eng
+# Example list: ["nob"]
+ORGREG_ACRONYMS = []