diff --git a/greg/management/commands/import_from_orgreg.py b/greg/management/commands/import_from_orgreg.py
index 2411ed04921d25484d88af2c451043c80b262504..cc5d35a598404480fa7cee92bedde4158dbb3686 100644
--- a/greg/management/commands/import_from_orgreg.py
+++ b/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]]
diff --git a/greg/tests/management/test_import_from_orgreg.py b/greg/tests/management/test_import_from_orgreg.py
index d7a39917ce00965d33fabc83ba004dd0664a49a3..1c132b4391934ae8cf1d07cce16c0ce9610e603f 100644
--- a/greg/tests/management/test_import_from_orgreg.py
+++ b/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_acronym3"},
),
OrgUnit(
ou_id=2,
@@ -49,6 +51,7 @@ def orgreg_response():
parent=1,
name={"eng": "baz"},
external_keys=[],
+ acronym={"nob": "foo_acronym2"},
),
]
)
@@ -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"
+ )
diff --git a/gregsite/settings/base.py b/gregsite/settings/base.py
index 2d9f564218e9de15077db1aa6556a9587d399705..ec1ffa9990cfa36a4a4f76a34e699d7217abc0d8 100644
--- a/gregsite/settings/base.py
+++ b/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 = []