Feature/mx 1677 orcid extractor

CHANGELOG.md

@@ -9,6 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+  - Connector class for retrieving ORCID data by ID or name
+  - methods for extracting data from orcid
+  - methods to transform from OcidPerson to mex person
+  - model class for orcid data
+  - unit tests for orcid connector
+
 ### Changes
 
 ### Deprecated

assets/raw-data/primary-sources/primary-sources.json

@@ -34,5 +34,14 @@
                 "value": "Wikidata APIs"
             }
         ]
+    },
+    {
+        "identifier": "orcid",
+        "title": [
+            {
+                "language": "en",
+                "value": "Open Researcher Contributor Identification Initiative"
+            }
+        ]
     }
 ]

mex/common/orcid/connector.py

@@ -0,0 +1,89 @@
+from typing import Any
+
+from mex.common.connector.http import HTTPConnector
+from mex.common.exceptions import EmptySearchResultError, FoundMoreThanOneError
+from mex.common.settings import BaseSettings
+
+
+class OrcidConnector(HTTPConnector):
+    """Connector class for querying Orcid records."""
+
+    def _set_url(self) -> None:
+        """Set url of the host."""
+        settings = BaseSettings.get()
+        self.url = str(settings.orcid_api_url)
+
+    def _check_availability(self) -> None:
+        """Send a GET request to verify the host is available."""
+        url = f"{self.url.rstrip('/')}/search"
+        response = self._send_request("HEAD", url=url, params={})
+        response.raise_for_status()
+
+    def check_orcid_id_exists(self, orcid_id: str) -> bool:
+        """Search for an ORCID person by ORCID ID."""
+        query_dict = {"orcid": orcid_id}
+        response = self.fetch(query_dict)
+        return bool(response.get("num-found", 0))
+
+    @staticmethod
+    def build_query(filters: dict[str, Any]) -> str:
+        """Construct the ORCID API query string."""
+        return " AND ".join([f"{key}:{value}" for key, value in filters.items()])
+
+    def fetch(self, filters: dict[str, Any]) -> dict[str, Any]:
+        """Perform a search query against the ORCID API."""
+        query = OrcidConnector.build_query(filters)
+        return self.request(method="GET", endpoint="search", params={"q": query})
+
+    @staticmethod
+    def get_data_by_id(orcid_id: str) -> dict[str, Any]:
+        """Retrieve data by UNIQUE ORCID ID.
+
+        Args:
+            orcid_id: Unique identifier in ORCID system.
+
+        Returns:
+            Personal data of the single matching id.
+        """
+        orcidapi = OrcidConnector.get()
+        # or endpoint = f"{orcid_id}/person"
+        endpoint = f"{orcid_id}/record"
+        return dict(orcidapi.request(method="GET", endpoint=endpoint))
+
+    @staticmethod
+    def get_data_by_name(
+        given_names: str = "*",
+        family_name: str = "*",
+        **filters: str,
+    ) -> dict[str, Any]:
+        """Get ORCID record of a single person for the given filters.
+
+        Args:
+            self: Connector.
+            given_names: Given name of a person, defaults to non-null
+            family_name: Surname of a person, defaults to non-null
+            **filters: Key-value pairs representing ORCID search filters.
+
+        Raises:
+            EmptySearchResultError
+            FoundMoreThanOneError
+
+        Returns:
+            Orcid data of the single matching person by name.
+        """
+        if given_names:
+            filters["given-names"] = given_names
+        if family_name:
+            filters["family-name"] = family_name
+        orcidapi = OrcidConnector.get()
+        search_response = orcidapi.fetch(filters=filters)
+        num_found = search_response.get("num-found", 0)
+        if num_found == 0:
+            msg = f"Cannot find orcid person for filters {filters}'"
+            raise EmptySearchResultError(msg)
+        if num_found > 1:
+            msg = f"Found multiple orcid persons for filters {filters}'"
+            raise FoundMoreThanOneError(msg)
+
+        orcid_id = search_response["result"][0]["orcid-identifier"]["path"]
+        return OrcidConnector.get_data_by_id(orcid_id)

mex/common/orcid/extract.py

@@ -0,0 +1,39 @@
+from mex.common.orcid.connector import OrcidConnector
+from mex.common.orcid.models.person import OrcidRecord
+from mex.common.orcid.transform import map_orcid_data_to_orcid_record
+
+
+def get_orcid_record_by_name(
+    given_names: str = "*", family_name: str = "*"
+) -> OrcidRecord:
+    """Returns Orcidrecord of a single person for the given filters.
+
+    Args:
+        given_names: Given name of a person, defaults to non-null
+        family_name: Surname of a person, defaults to non-null
+        **filters: Key-value pairs representing ORCID search filters.
+
+    Raises:
+        EmptySearchResultError
+        FoundMoreThanOneError
+
+    Returns:
+        Orcidrecord of the matching person by name.
+    """
+    orcid_data = OrcidConnector.get_data_by_name(
+        given_names=given_names, family_name=family_name
+    )
+    return map_orcid_data_to_orcid_record(orcid_data)
+
+
+def get_orcid_record_by_id(orcid_id: str) -> OrcidRecord:
+    """Returns Orcidrecord by UNIQUE ORCID ID.
+
+    Args:
+        orcid_id: Unique identifier in ORCID system.
+
+    Returns:
+        Orcidrecord of the matching id.
+    """
+    orcid_data = OrcidConnector.get_data_by_id(orcid_id=orcid_id)
+    return map_orcid_data_to_orcid_record(orcid_data)

mex/common/orcid/models/person.py

@@ -0,0 +1,56 @@
+from pydantic import Field
+
+from mex.common.models import BaseModel
+
+
+class OrcidIdentifier(BaseModel):
+    """Model class for OrcidID."""
+
+    path: str
+    uri: str
+
+
+class OrcidEmail(BaseModel):
+    """Model class for Orcid email."""
+
+    email: list[str]
+
+
+class OrcidEmails(BaseModel):
+    """Model class for Orcid emails."""
+
+    email: list[OrcidEmail]
+
+
+class OrcidFamilyName(BaseModel):
+    """Model class for orcid family names."""
+
+    value: str
+
+
+class OrcidGivenNames(BaseModel):
+    """Model class for Orcid given names."""
+
+    value: str
+
+
+class OrcidName(BaseModel):
+    """Model class for Orcid name."""
+
+    family_name: OrcidFamilyName = Field(alias="family-name")
+    given_names: OrcidGivenNames = Field(alias="given-names")
+    visibility: str
+
+
+class OrcidPerson(BaseModel):
+    """Model class for Orcid person."""
+
+    emails: OrcidEmails
+    name: OrcidName
+
+
+class OrcidRecord(BaseModel):
+    """Model class for Orcid record."""
+
+    orcid_identifier: OrcidIdentifier = Field(alias="orcid-identifier")
+    person: OrcidPerson

mex/common/orcid/transform.py

@@ -0,0 +1,46 @@
+from typing import Any
+
+from mex.common.models import (
+    ExtractedPerson,
+)
+from mex.common.orcid.models.person import OrcidRecord
+from mex.common.primary_source.helpers import get_all_extracted_primary_sources
+
+
+def map_orcid_data_to_orcid_record(orcid_data: dict[str, Any]) -> OrcidRecord:
+    """Wraps orcid data into an OrcidRecord."""
+    return OrcidRecord.model_validate(orcid_data)
+
+
+def transform_orcid_person_to_mex_person(
+    orcid_record: OrcidRecord,
+) -> ExtractedPerson:
+    """Transforms a single ORCID person to an ExtractedPerson.
+
+    Args:
+        orcid_record: OrcidRecord object of a person.
+
+    Returns:
+        ExtractedPerson.
+    """
+    primary_source = get_all_extracted_primary_sources()["orcid"]
+    had_primary_source = primary_source.stableTargetId
+
+    id_in_primary_source = orcid_record.orcid_identifier.path
+    email = orcid_record.person.emails.email[0].email
+    if orcid_record.person.name.visibility == "public":
+        given_names = orcid_record.person.name.given_names.value
+        family_name = orcid_record.person.name.family_name.value
+    else:
+        given_names = None
+        family_name = None
+    orcid_id = orcid_record.orcid_identifier.uri
+
+    return ExtractedPerson(
+        identifierInPrimarySource=id_in_primary_source,
+        hadPrimarySource=had_primary_source,
+        givenName=given_names,
+        familyName=family_name,
+        orcidId=orcid_id,
+        email=email,
+    )

mex/common/settings.py

@@ -186,6 +186,11 @@ def get(cls) -> Self:
         "services ",
         validation_alias="MEX_WEB_USER_AGENT",
     )
+    orcid_api_url: AnyUrl = Field(
+        Url("https://orcid"),
+        description="URL of orcid api.",
+        validation_alias="MEX_ORCID_API_URL",
+    )
 
     def text(self) -> str:
         """Dump the current settings into a readable table."""

mex/common/testing/plugin.py

@@ -12,13 +12,15 @@
 from typing import Any, cast
 from unittest.mock import MagicMock, Mock
 
+import pytest
 import requests
 from langdetect import DetectorFactory
 from pydantic import AnyUrl
-from requests import Response
+from requests import HTTPError, Response
 
 from mex.common.connector import CONNECTOR_STORE
 from mex.common.models import ExtractedPrimarySource
+from mex.common.orcid.connector import OrcidConnector
 from mex.common.primary_source.helpers import get_all_extracted_primary_sources
 from mex.common.settings import SETTINGS_STORE, BaseSettings
 from mex.common.wikidata.connector import (
@@ -191,3 +193,67 @@ def get_wikidata_item_details_by_id(
         "get_wikidata_item_details_by_id",
         get_wikidata_item_details_by_id,
     )
+
+
+@pytest.fixture
+def orcid_person_raw() -> dict[str, Any]:
+    """Return a raw orcid person."""
+    with open(Path(__file__).parent / "test_data" / "orcid_person_raw.json") as fh:
+        return cast(dict[str, Any], json.load(fh))
+
+
+@pytest.fixture
+def orcid_multiple_matches() -> dict[str, Any]:
+    """Return a raw orcid person."""
+    with open(
+        Path(__file__).parent / "test_data" / "orcid_multiple_matches.json"
+    ) as fh:
+        return cast(dict[str, Any], json.load(fh))
+
+
+@pytest.fixture
+def mocked_orcid(
+    monkeypatch: pytest.MonkeyPatch,
+    orcid_person_raw: dict[str, Any],
+    orcid_multiple_matches: dict[str, Any],
+) -> None:
+    """Mock orcid connector."""
+    response_query = Mock(spec=Response, status_code=200)
+
+    session = MagicMock(spec=requests.Session)
+    session.get = MagicMock(side_effect=[response_query])
+
+    def mocked_init(self: OrcidConnector) -> None:
+        self.session = session
+
+    monkeypatch.setattr(OrcidConnector, "__init__", mocked_init)
+
+    def check_orcid_id_exists(_self: OrcidConnector, _orcid_id: str) -> bool:
+        return _orcid_id == "0009-0004-3041-5706"
+
+    monkeypatch.setattr(OrcidConnector, "check_orcid_id_exists", check_orcid_id_exists)
+
+    def fetch(_self: OrcidConnector, filters: dict[str, Any]) -> dict[str, Any]:
+        if filters.get("given-names") == "John":
+            return {"num-found": 1, "result": [orcid_person_raw]}
+        if filters.get("given-names") == "Multiple":
+            return orcid_multiple_matches
+        return {"result": None, "num-found": 0}
+
+    monkeypatch.setattr(OrcidConnector, "fetch", fetch)
+
+    def get_data_by_id(orcid_id: str) -> dict[str, Any]:
+        if orcid_id == "0009-0004-3041-5706":
+            return orcid_person_raw
+        msg = "404 Not Found"
+        raise HTTPError(msg)
+
+    monkeypatch.setattr(OrcidConnector, "get_data_by_id", staticmethod(get_data_by_id))
+
+    def build_query(filters: dict[str, Any]) -> str:
+        """Construct the ORCID API query string."""
+        if "givennames" in filters:
+            return "givennames:Josiah AND familyname:Carberry"
+        return "given-names:Josiah AND family-name:Carberry"
+
+    monkeypatch.setattr(OrcidConnector, "build_query", staticmethod(build_query))