Upstream ROR implementation (#471)

Author: cthoyt

pyproject.toml

@@ -86,6 +86,7 @@ dependencies = [
     "umls_downloader>=0.1.3",
     "clinicaltrials_downloader>=0.0.2",
     "nih-reporter-downloader>=0.0.1",
+    "ror-downloader",
     "typing_extensions",
     "rdflib",
     "obographs>=0.0.8",

src/pyobo/sources/ror.py

@@ -2,18 +2,14 @@
 
 from __future__ import annotations
 
-import datetime
-import json
 import logging
-import zipfile
 from collections.abc import Iterable
-from functools import lru_cache
-from pathlib import Path
-from typing import Any, Literal, NamedTuple, TypeAlias
+from typing import Any
 
 import bioregistry
-import zenodo_client
-from pydantic import BaseModel, ValidationError
+import ror_downloader
+from pydantic import ValidationError
+from ror_downloader import OrganizationType
 from tqdm.auto import tqdm
 
 from pyobo.struct import Obo, Reference, Term
@@ -29,16 +25,12 @@
 )
 
 __all__ = [
-    "OrganizationType",
-    "RORStatus",
-    "get_ror_records",
-    "get_ror_status",
+    "RORGetter",
     "get_ror_to_country_geonames",
 ]
 
 logger = logging.getLogger(__name__)
 PREFIX = "ror"
-ROR_ZENODO_RECORD_ID = "17953395"
 
 # Constants
 ORG_CLASS = Reference(prefix="OBI", identifier="0000245", name="organization")
@@ -70,7 +62,7 @@ class RORGetter(Obo):
     root_terms = [CITY_CLASS, ORG_CLASS]
 
     def __post_init__(self):
-        self.data_version, _url, _path = get_ror_status()
+        self.data_version, _url, _path = ror_downloader.get_version_info()
         super().__post_init__()
 
     def iter_terms(self, force: bool = False) -> Iterable[Term]:
@@ -83,18 +75,6 @@ def iter_terms(self, force: bool = False) -> Iterable[Term]:
         yield from iterate_ror_terms(force=force)
 
 
-OrganizationType: TypeAlias = Literal[
-    "education",
-    "facility",
-    "funder",
-    "company",
-    "government",
-    "healthcare",
-    "archive",
-    "nonprofit",
-    "other",
-]
-
 ROR_ORGANIZATION_TYPE_TO_OBI: dict[OrganizationType, Term] = {
     "education": Term.default(PREFIX, "education", "educational organization"),
     "facility": Term.default(PREFIX, "facility", "facility"),
@@ -115,127 +95,9 @@ def iter_terms(self, force: bool = False) -> Iterable[Term]:
 _MISSED_ORG_TYPES: set[str] = set()
 
 
-class LocationDetails(BaseModel):
-    """The location details slot in the ROR schema."""
-
-    continent_code: str
-    continent_name: str
-    country_code: str
-    country_name: str
-    country_subdivision_code: str | None = None
-    country_subdivision_name: str | None = None
-    lat: float
-    lng: float
-    name: str
-
-
-class Location(BaseModel):
-    """The lcoation slot in the ROR schema."""
-
-    geonames_id: int
-    geonames_details: LocationDetails
-
-
-class ExternalID(BaseModel):
-    """The external ID slot in the ROR schema."""
-
-    type: str
-    all: list[str]
-    preferred: str | None = None
-
-
-class Link(BaseModel):
-    """The link slot in the ROR schema."""
-
-    type: str
-    value: str
-
-
-class Name(BaseModel):
-    """The name slot in the ROR schema."""
-
-    value: str
-    types: list[str]
-    lang: str | None = None
-
-
-class Relationship(BaseModel):
-    """The relationship slot in the ROR schema."""
-
-    type: str
-    label: str
-    id: str
-
-
-class DateAnnotated(BaseModel):
-    """The annotated date slot in the ROR schema."""
-
-    date: datetime.date
-    schema_version: str
-
-
-class Admin(BaseModel):
-    """The admin slot in the ROR schema."""
-
-    created: DateAnnotated
-    last_modified: DateAnnotated
-
-
-Status: TypeAlias = Literal["active", "inactive", "withdrawn"]
-
-
-class Record(BaseModel):
-    """A ROR record."""
-
-    locations: list[Location]
-    established: int | None = None
-    external_ids: list[ExternalID]
-    id: str
-    domains: list[str]
-    links: list[Link]
-    names: list[Name]
-    relationships: list[Relationship]
-    status: Status
-    types: list[OrganizationType]
-    admin: Admin
-
-    def get_preferred_label(self) -> str | None:
-        """Get the preferred label."""
-        primary_name: str | None = None
-        for name in self.names:
-            if "ror_display" in name.types:
-                primary_name = name.value
-        if primary_name is None:
-            return None
-        primary_name = NAME_REMAPPING.get(primary_name, primary_name)
-        return primary_name
-
-
-_description_prefix = {
-    "education": "an educational organization",
-    "facility": "a facility",
-    "funder": "a funder",
-    "company": "a company",
-    "government": "a governmental organization",
-    "healthcare": "a healthcare organization",
-    "archive": "an archive",
-    "nonprofit": "a nonprofit organization",
-    "other": "an organization",
-}
-
-
-def _get_description(record: Record) -> str | None:
-    description = (
-        f"{_description_prefix[record.types[0]]} in {record.locations[0].geonames_details.name}"
-    )
-    if record.established:
-        description += f" established in {record.established}"
-    return description
-
-
 def iterate_ror_terms(*, force: bool = False) -> Iterable[Term]:
     """Iterate over terms in ROR."""
-    status, records = get_ror_records(force=force)
+    status, records = ror_downloader.get_organizations(force=force)
     unhandled_xref_prefixes: set[str] = set()
 
     seen_geonames_references = set()
@@ -249,7 +111,7 @@ def iterate_ror_terms(*, force: bool = False) -> Iterable[Term]:
         term = Term(
             reference=Reference(prefix=PREFIX, identifier=identifier, name=primary_name),
             type="Instance",
-            definition=_get_description(record),
+            definition=record.get_description(),
         )
         for organization_type in record.types:
             if organization_type in ROR_ORGANIZATION_TYPE_TO_OBI:
@@ -339,63 +201,6 @@ def iterate_ror_terms(*, force: bool = False) -> Iterable[Term]:
         yield geonames_term
 
 
-class RORStatus(NamedTuple):
-    """A version information tuple."""
-
-    version: str
-    url: str
-    path: Path
-
-
-def get_ror_status(*, force: bool = False, authenticate_zenodo: bool = True) -> RORStatus:
-    """Ensure the latest ROR record, metadata, and filepath.
-
-    :param force: Should the record be downloaded again? This almost
-        never needs to be true, since the data doesn't change for
-        a given version
-    :param authenticate_zenodo: Should Zenodo be authenticated?
-        This isn't required, but can help avoid rate limits
-    :return: A version information tuple
-
-    .. note::
-
-        this goes into the ``~/.data/zenodo/6347574`` folder,
-        because 6347574 is the super-record ID, which groups all
-        versions together. this is different from the value
-        for :data:`ROR_ZENODO_RECORD_ID`
-    """
-    client = zenodo_client.Zenodo()
-    latest_record_id = client.get_latest_record(
-        ROR_ZENODO_RECORD_ID, authenticate=authenticate_zenodo
-    )
-    response = client.get_record(latest_record_id, authenticate=authenticate_zenodo)
-    response_json = response.json()
-    version = response_json["metadata"]["version"].lstrip("v")
-    file_record = response_json["files"][0]
-    name = file_record["key"]
-    url = file_record["links"]["self"]
-    path = client.download(latest_record_id, name=name, force=force)
-    return RORStatus(version=version, url=url, path=path)
-
-
-@lru_cache
-def get_ror_records(
-    *, force: bool = False, authenticate_zenodo: bool = True
-) -> tuple[RORStatus, list[Record]]:
-    """Get the latest ROR metadata and records."""
-    status = get_ror_status(force=force, authenticate_zenodo=authenticate_zenodo)
-    with zipfile.ZipFile(status.path) as zf:
-        for zip_info in zf.filelist:
-            if zip_info.filename.endswith(".json"):
-                with zf.open(zip_info) as file:
-                    records = [
-                        Record.model_validate(record)
-                        for record in tqdm(json.load(file), unit_scale=True)
-                    ]
-                    return status, records
-    raise FileNotFoundError
-
-
 def get_ror_to_country_geonames(**kwargs: Any) -> dict[str, str]:
     """Get a mapping of ROR ids to GeoNames IDs for countries."""
     from pyobo.sources.geonames.geonames import get_city_to_country