Skip to content
Snippets Groups Projects

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

Merged Tore.Brede requested to merge GREG-123_acronym_nob_import_option into master
3 files
+ 122
52
Compare changes
  • Side-by-side
  • Inline
Files
3
@@ -12,12 +12,13 @@ orgreg/v3/ as the url argument (note the trailing slash).
@@ -12,12 +12,13 @@ orgreg/v3/ as the url argument (note the trailing slash).
"""
"""
import datetime
import datetime
import logging
import logging
from typing import Union, Mapping, Dict
from typing import Union, Mapping, Dict, Optional
import orgreg_client
import orgreg_client
from django.conf import settings
from django.conf import settings
from django.core.management.base import BaseCommand
from django.core.management.base import BaseCommand
from orgreg_client import OrgUnit
from orgreg_client import OrgUnit
 
from orgreg_client.models import ExternalKey
from greg.models import OrganizationalUnit, OuIdentifier
from greg.models import OrganizationalUnit, OuIdentifier
@@ -34,68 +35,107 @@ class Command(BaseCommand):
@@ -34,68 +35,107 @@ class Command(BaseCommand):
):
):
"""Upsert any configured extra IDs from orgreg."""
"""Upsert any configured extra IDs from orgreg."""
for extra_id in settings.ORGREG_EXTRA_IDS:
for extra_id in settings.ORGREG_EXTRA_IDS:
identity_in_orgreg = self._get_external_key_from_ou(extra_id, ou)
matching_ids = [
if identity_in_orgreg is not None:
x
self._upsert_identifier(
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"],
extra_id["source"],
 
extra_id["type"],
 
identity_in_orgreg.value,
ou.ou_id,
ou.ou_id,
)
)
continue
if len(matching_ids) > 1:
# Acronyms can also be used as identifiers
# External_ids
for acronym in settings.ORGREG_ACRONYMS:
logger.warning(
if acronym == "nob" and ou.acronym.nob is not None:
"Found multiple ids matching type: %s source: %s in OrgReg. Using the first one: %s",
self._upsert_identifier(
extra_id["type"],
settings.ORGREG_SOURCE,
extra_id["source"],
"acronym_" + acronym,
matching_ids[0].value,
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
return matching_ids[0]
identify_in_db = (
self.processed[ou.ou_id]
def _upsert_identifier(
.identifiers.filter(
self, source: str, identity_type: str, identity_in_orgreg_value: str, ou_id: int
source=extra_id["source"],
):
name=extra_id["type"],
# Check if the id exists
)
identify_in_db = (
.first()
self.processed[ou_id]
 
.identifiers.filter(
 
source=source,
 
name=identity_type,
)
)
 
.first()
 
)
if identify_in_db:
if identify_in_db:
if identify_in_db.value != identity_in_orgreg.value:
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],
)
logger.info(
logger.info(
"Added new id to ou: %s, type: %s, source: %s value: %s",
"Updating id: source: %s, type: %s, old_id: %s, new_id %s",
ou.ou_id,
source,
extra_id["type"],
identity_type,
extra_id["source"],
identify_in_db.value,
identity_in_orgreg["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(
def _get_or_create_and_set_values(
self, ou: OrgUnit, values: Mapping[str, Union[str, int, bool]]
self, ou: OrgUnit, values: Mapping[str, Union[str, int, bool]]
Loading