feature/mx-1664-use-wikidata-helper

CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 
 ### Changes
+- extractors now use wikidata helper function
 
 ### Deprecated
 

mex/extractors/biospecimen/extract.py

@@ -9,10 +9,12 @@
 from mex.common.ldap.connector import LDAPConnector
 from mex.common.ldap.models.person import LDAPPerson
 from mex.common.logging import watch
-from mex.common.wikidata.extract import search_organization_by_label
-from mex.common.wikidata.models.organization import WikidataOrganization
+from mex.common.types import MergedOrganizationIdentifier
 from mex.extractors.biospecimen.models.source import BiospecimenResource
 from mex.extractors.settings import Settings
+from mex.extractors.wikidata.helpers import (
+    get_wikidata_extracted_organization_id_by_name,
+)
 
 
 @watch
@@ -42,20 +44,24 @@ def extract_biospecimen_contacts_by_email(
 
 def extract_biospecimen_organizations(
     biospecimen_resources: list[BiospecimenResource],
-) -> dict[str, WikidataOrganization]:
+) -> dict[str, MergedOrganizationIdentifier]:
     """Search and extract organization from wikidata.
 
     Args:
         biospecimen_resources: Iterable of biospecimen resources
 
     Returns:
-        dict with WikidataOrganization by externe partner
+        dict with WikidataOrganization ID by externe partner
     """
     return {
-        resource.externe_partner: org
+        resource.externe_partner: org_id
         for resource in biospecimen_resources
         if resource.externe_partner
-        and (org := search_organization_by_label(resource.externe_partner))
+        and (
+            org_id := get_wikidata_extracted_organization_id_by_name(
+                resource.externe_partner
+            )
+        )
     }
 
 

mex/extractors/biospecimen/main.py

@@ -24,9 +24,6 @@
 from mex.extractors.pipeline import asset, run_job_in_process
 from mex.extractors.settings import Settings
 from mex.extractors.sinks import load
-from mex.extractors.wikidata.extract import (
-    get_merged_organization_id_by_query_with_transform_and_load,
-)
 
 
 @asset(group_name="biospecimen", deps=["extracted_primary_source_mex"])
@@ -72,20 +69,15 @@ def extracted_biospecimen_resources(
     unit_stable_target_ids_by_synonym: dict[str, MergedOrganizationalUnitIdentifier],
     extracted_organization_rki: ExtractedOrganization,
     extracted_synopse_activities: list[ExtractedActivity],
-    extracted_primary_source_wikidata: ExtractedPrimarySource,
 ) -> list[ExtractedResource]:
     """Transform biospecimen resources to extracted resources and load them to the sinks."""  # noqa: E501
     settings = Settings.get()
     resource_mapping = transform_mapping_data_to_model(
         extract_mapping_data(settings.biospecimen.mapping_path / "resource.yaml"),
         ExtractedResource,
     )
-    biospecimen_organizations = extract_biospecimen_organizations(biospecimen_resources)
-    extracted_organizations = (
-        get_merged_organization_id_by_query_with_transform_and_load(
-            biospecimen_organizations, extracted_primary_source_wikidata
-        )
-    )
+    extracted_organizations = extract_biospecimen_organizations(biospecimen_resources)
+
     mex_sources = list(
         transform_biospecimen_resource_to_mex_resource(
             biospecimen_resources,

mex/extractors/biospecimen/transform.py

@@ -49,7 +49,7 @@ def transform_biospecimen_resource_to_mex_resource(
     person_stable_target_id_by_email = {
         str(p.email[0]): Identifier(p.stableTargetId) for p in mex_persons
     }
-    sysnopse_stable_target_id_by_studien_id = {
+    synopse_stable_target_id_by_studien_id = {
         activity.identifierInPrimarySource: activity.stableTargetId
         for activity in extracted_synopse_activities
     }
@@ -96,7 +96,7 @@ def transform_biospecimen_resource_to_mex_resource(
                 contact.append(k)
             elif k := unit_stable_target_ids_by_synonym.get(kontakt):
                 contact.append(k)
-        was_generated_by = sysnopse_stable_target_id_by_studien_id.get(
+        was_generated_by = synopse_stable_target_id_by_studien_id.get(
             resource.studienbezug[0], None
         )
         if resource.weiterfuehrende_dokumentation_url_oder_dateipfad:

mex/extractors/blueant/extract.py

@@ -4,11 +4,13 @@
 from mex.common.ldap.connector import LDAPConnector
 from mex.common.ldap.models.person import LDAPPerson
 from mex.common.logging import watch
-from mex.common.wikidata.extract import search_organization_by_label
-from mex.common.wikidata.models.organization import WikidataOrganization
+from mex.common.types import MergedOrganizationIdentifier
 from mex.extractors.blueant.connector import BlueAntConnector
 from mex.extractors.blueant.models.source import BlueAntSource
 from mex.extractors.settings import Settings
+from mex.extractors.wikidata.helpers import (
+    get_wikidata_extracted_organization_id_by_name,
+)
 
 
 @watch
@@ -96,19 +98,19 @@ def remove_prefixes_from_name(name: str) -> str:
 
 def extract_blueant_organizations(
     blueant_sources: list[BlueAntSource],
-) -> dict[str, WikidataOrganization]:
+) -> dict[str, MergedOrganizationIdentifier]:
     """Search and extract organization from wikidata.
 
     Args:
         blueant_sources: Iterable of blueant sources
 
     Returns:
-        Dict with organization label and WikidataOrganization
+        Dict with organization label and WikidataOrganization ID
     """
     return {
-        name: org
+        name: org_id
         for source in blueant_sources
         for name in source.client_names
         if name not in ["Robert Koch-Institut", "RKI"]
-        and (org := search_organization_by_label(name))
+        and (org_id := get_wikidata_extracted_organization_id_by_name(name))
     }

mex/extractors/blueant/main.py

@@ -28,9 +28,6 @@
 from mex.extractors.pipeline import asset, run_job_in_process
 from mex.extractors.settings import Settings
 from mex.extractors.sinks import load
-from mex.extractors.wikidata.extract import (
-    get_merged_organization_id_by_query_with_transform_and_load,
-)
 
 
 @asset(group_name="blueant", deps=["extracted_primary_source_mex"])
@@ -81,15 +78,10 @@ def blueant_project_leaders_by_employee_id(
 
 @asset(group_name="blueant")
 def blueant_organization_ids_by_query_string(
-    extracted_primary_source_wikidata: ExtractedPrimarySource,
     blueant_sources: list[BlueAntSource],
 ) -> dict[str, MergedOrganizationIdentifier]:
     """Extract organizations for blueant from wikidata and group them by query."""
-    wikidata_organizations_by_query = extract_blueant_organizations(blueant_sources)
-
-    return get_merged_organization_id_by_query_with_transform_and_load(
-        wikidata_organizations_by_query, extracted_primary_source_wikidata
-    )
+    return extract_blueant_organizations(blueant_sources)
 
 
 @asset(group_name="blueant")

mex/extractors/datscha_web/extract.py

@@ -4,10 +4,12 @@
 from mex.common.ldap.models.person import LDAPPersonWithQuery
 from mex.common.ldap.transform import analyse_person_string
 from mex.common.logging import watch
-from mex.common.wikidata.extract import search_organization_by_label
-from mex.common.wikidata.models.organization import WikidataOrganization
+from mex.common.types import MergedOrganizationIdentifier
 from mex.extractors.datscha_web.connector import DatschaWebConnector
 from mex.extractors.datscha_web.models.item import DatschaWebItem
+from mex.extractors.wikidata.helpers import (
+    get_wikidata_extracted_organization_id_by_name,
+)
 
 
 @watch
@@ -52,7 +54,7 @@ def extract_datscha_web_source_contacts(
 
 def extract_datscha_web_organizations(
     datscha_web_items: Iterable[DatschaWebItem],
-) -> dict[str, WikidataOrganization]:
+) -> dict[str, MergedOrganizationIdentifier]:
     """Search and extract organization from wikidata.
 
     Args:
@@ -62,12 +64,14 @@ def extract_datscha_web_organizations(
         Dict with keys DatschaWebItem.Auftragsverarbeiter,
             DatschaWebItem.Empfaenger_der_Daten_im_Drittstaat, and
             DatschaWebItem.Empfaenger_der_verarbeiteten_uebermittelten_oder_offengelegten_Daten,
-            and values: WikidataOrganization
+            and values: MergedOrganizationIdentifier
     """
     partner_to_org_map = {}
     for item in datscha_web_items:
         for partner in item.get_partners():
             if partner and partner != "None":
-                if organization := search_organization_by_label(partner):
+                if organization := get_wikidata_extracted_organization_id_by_name(
+                    partner
+                ):
                     partner_to_org_map[partner] = organization
     return partner_to_org_map

mex/extractors/datscha_web/main.py

@@ -27,9 +27,6 @@
 from mex.extractors.pipeline import asset, run_job_in_process
 from mex.extractors.settings import Settings
 from mex.extractors.sinks import load
-from mex.extractors.wikidata.extract import (
-    get_merged_organization_id_by_query_with_transform_and_load,
-)
 
 
 @asset(group_name="datscha_web", deps=["extracted_primary_source_mex"])
@@ -82,16 +79,9 @@ def datscha_web_person_ids_by_query_string(
 @asset(group_name="datscha_web")
 def datscha_web_organization_ids_by_query_string(
     extracted_datscha_web_items: list[DatschaWebItem],
-    extracted_primary_source_wikidata: ExtractedPrimarySource,
 ) -> dict[str, MergedOrganizationIdentifier]:
     """Extract organizations for Datscha Web from wikidata and group them by query."""
-    wikidata_organizations_by_query = extract_datscha_web_organizations(
-        extracted_datscha_web_items
-    )
-
-    return get_merged_organization_id_by_query_with_transform_and_load(
-        wikidata_organizations_by_query, extracted_primary_source_wikidata
-    )
+    return extract_datscha_web_organizations(extracted_datscha_web_items)
 
 
 @asset(group_name="datscha_web")

mex/extractors/ff_projects/extract.py

@@ -11,13 +11,15 @@
 from mex.common.ldap.transform import analyse_person_string
 from mex.common.logging import watch
 from mex.common.types import (
+    MergedOrganizationIdentifier,
     TemporalEntity,
     TemporalEntityPrecision,
 )
-from mex.common.wikidata.extract import search_organization_by_label
-from mex.common.wikidata.models.organization import WikidataOrganization
 from mex.extractors.ff_projects.models.source import FFProjectsSource
 from mex.extractors.settings import Settings
+from mex.extractors.wikidata.helpers import (
+    get_wikidata_extracted_organization_id_by_name,
+)
 
 
 @watch
@@ -198,22 +200,26 @@ def extract_ff_project_authors(
 
 def extract_ff_projects_organizations(
     ff_projects_sources: Iterable[FFProjectsSource],
-) -> dict[str, WikidataOrganization]:
+) -> dict[str, MergedOrganizationIdentifier]:
     """Search and extract organization from wikidata.
 
     Args:
         ff_projects_sources: Iterable of ff-project sources
 
     Returns:
-        Dict with organization label and WikidataOrganization
+        Dict with organization label and WikidataOrganization ID
     """
     return {
-        zuwendungs_oder_auftraggeber: org
+        zuwendungs_oder_auftraggeber: org_id
         for source in ff_projects_sources
         if source.zuwendungs_oder_auftraggeber
         and source.zuwendungs_oder_auftraggeber != "Sonderforschung"
         for zuwendungs_oder_auftraggeber in source.zuwendungs_oder_auftraggeber.split(
             "/"
         )
-        if (org := search_organization_by_label(zuwendungs_oder_auftraggeber))
+        if (
+            org_id := get_wikidata_extracted_organization_id_by_name(
+                zuwendungs_oder_auftraggeber
+            )
+        )
     }

mex/extractors/ff_projects/main.py

@@ -28,9 +28,6 @@
 from mex.extractors.pipeline import asset, run_job_in_process
 from mex.extractors.settings import Settings
 from mex.extractors.sinks import load
-from mex.extractors.wikidata.extract import (
-    get_merged_organization_id_by_query_with_transform_and_load,
-)
 
 
 @asset(group_name="ff_projects", deps=["extracted_primary_source_mex"])
@@ -86,17 +83,10 @@ def ff_projects_person_ids_by_query_string(
 
 @asset(group_name="ff_projects")
 def ff_projects_organization_ids_by_query_string(
-    extracted_primary_source_wikidata: ExtractedPrimarySource,
     ff_projects_sources: list[FFProjectsSource],
 ) -> dict[str, MergedOrganizationIdentifier]:
     """Extract organizations for FF Projects from wikidata and group them by query."""
-    wikidata_organizations_by_query = extract_ff_projects_organizations(
-        ff_projects_sources
-    )
-
-    return get_merged_organization_id_by_query_with_transform_and_load(
-        wikidata_organizations_by_query, extracted_primary_source_wikidata
-    )
+    return extract_ff_projects_organizations(ff_projects_sources)
 
 
 @asset(group_name="ff_projects")

mex/extractors/grippeweb/extract.py

@@ -3,10 +3,12 @@
 from mex.common.ldap.connector import LDAPConnector
 from mex.common.ldap.models.actor import LDAPActor
 from mex.common.ldap.models.person import LDAPPerson
-from mex.common.wikidata.extract import search_organization_by_label
-from mex.common.wikidata.models.organization import WikidataOrganization
+from mex.common.types import MergedOrganizationIdentifier
 from mex.extractors.grippeweb.connector import QUERY_BY_TABLE_NAME, GrippewebConnector
 from mex.extractors.mapping.types import AnyMappingModel
+from mex.extractors.wikidata.helpers import (
+    get_wikidata_extracted_organization_id_by_name,
+)
 
 
 def extract_columns_by_table_and_column_name() -> dict[str, dict[str, list[Any]]]:
@@ -71,23 +73,23 @@ def extract_ldap_persons(
 
 def extract_grippeweb_organizations(
     grippeweb_resource_mappings: list[AnyMappingModel],
-) -> dict[str, WikidataOrganization]:
+) -> dict[str, MergedOrganizationIdentifier]:
     """Search and extract grippeweb organization from wikidata.
 
     Args:
         grippeweb_resource_mappings: grippeweb resource mapping models
 
     Returns:
         Dict with keys: mapping default values
-            and values: WikidataOrganization
+            and values: MergedOrganizationIdentifier
     """
     organization_by_name = {}
     for resource in grippeweb_resource_mappings:
         if external_partner_dict := resource.externalPartner:
             external_partner = external_partner_dict[0].mappingRules[0].forValues[0]
-            if org := search_organization_by_label(external_partner):
+            if org := get_wikidata_extracted_organization_id_by_name(external_partner):
                 organization_by_name[external_partner] = org
         publisher_name = resource.publisher[0].mappingRules[0].forValues[0]
-        if publisher := search_organization_by_label(publisher_name):
+        if publisher := get_wikidata_extracted_organization_id_by_name(publisher_name):
             organization_by_name[publisher_name] = publisher
     return organization_by_name

mex/extractors/grippeweb/main.py

@@ -45,9 +45,6 @@
 from mex.extractors.settings import Settings
 from mex.extractors.sinks import load
 from mex.extractors.sumo.transform import get_contact_merged_ids_by_emails
-from mex.extractors.wikidata.extract import (
-    get_merged_organization_id_by_query_with_transform_and_load,
-)
 
 
 @asset(group_name="grippeweb", deps=["extracted_primary_source_mex"])
@@ -148,17 +145,12 @@ def extracted_mex_persons_grippeweb(
 @asset(group_name="grippeweb")
 def grippeweb_organization_ids_by_query_string(
     grippeweb_resource_mappings: list[dict[str, Any]],
-    extracted_primary_source_wikidata: ExtractedPrimarySource,
 ) -> dict[str, MergedOrganizationIdentifier]:
     """Extract organizations for grippeweb from wikidata and group them by query."""
-    wikidata_organizations_by_query = extract_grippeweb_organizations(
+    return extract_grippeweb_organizations(
         transform_mapping_data_to_models(grippeweb_resource_mappings, ExtractedResource)
     )
 
-    return get_merged_organization_id_by_query_with_transform_and_load(
-        wikidata_organizations_by_query, extracted_primary_source_wikidata
-    )
-
 
 @asset(group_name="grippeweb")
 def extracted_access_platform_grippeweb(