SciPost Code Repository

Skip to content
Snippets Groups Projects
Commit 5de175df authored by George Katsikas's avatar George Katsikas :goat:
Browse files

refactor: :recycle:️ add typing and modernize ROR API handler

parent 52671ada
No related branches found
No related tags found
No related merge requests found
from typing import Any
import requests
import urllib
from urllib.parse import quote
class RORAPIHandler:
API_URL = "https://api.ror.org/v2/organizations"
def query(self, query_string, all_status=True):
def query(self, query_string: str, all_status: bool = True) -> dict[str, Any]:
# URL-encode query_string to make it safe for use in a URL
query_string = urllib.parse.quote(query_string)
query_string = quote(query_string)
url = f'{self.API_URL}?query="{query_string}"'
if all_status:
url += "&all_status"
response = requests.get(url)
data = response.json()
try:
data: dict[str, str] = response.json()
except requests.JSONDecodeError:
data = {}
items = list(map(self._process_organization, data["items"]))
items = list(map(self._map_ror_to_organization, data.get("items", [])))
return {**data, "items": items}
def fetch(self, ror_id):
def fetch(self, ror_id: str) -> dict[str, Any]:
"""
Query the ROR API for an organization with the given ROR ID
and return the JSON result.
......@@ -37,59 +41,67 @@ class RORAPIHandler:
response = requests.get(url)
try:
data = response.json()
response = self._process_organization(data)
except:
response = self._map_ror_to_organization(data)
except (requests.JSONDecodeError, KeyError):
response = {}
return response
@staticmethod
def organization_from_ror_id(ror_id):
def organization_from_ror_id(ror_id: str) -> dict[str, Any]:
"""
Returns a dictionary of Organization model fields as returned by the ROR API.
"""
def _first_name(result, **kwargs):
first_name = None
for name in result.get("names", []):
# Validate all kwarg filters
filters = []
def _first_name(result: dict[str, Any], **kwargs: str) -> str:
"""
Returns the first name in the list of names that matches all kwargs filters.
If no name matches, returns an empty string.
"""
first_name = ""
names: list[dict[str, str]] = result.get("names", [])
for name in names:
# Create filters for each kwarg
filters: list[bool] = []
for k, v in kwargs.items():
if k.endswith("__not"):
filters.append(name[k[:-5]] != v)
elif k.endswith("__in"):
filters.append(v in name[k[:-4]])
else:
filters.append(name[k] == v)
match k.split("__", 1):
case [key, "not"]:
filters.append(name[key] != v)
case [key, "in"]:
filters.append(v in name[key])
case _:
filters.append(name[k] == v)
first_name = name["value"] if all(filters) else first_name
return first_name
result = RORAPIHandler().fetch(ror_id)
if result == {}:
return {}
geonames = result.get("locations", [{}])[0].get("geonames_details", {})
location_details: dict[str, Any] = result.get("locations", [{}])[0].get(
"geonames_details", {}
)
organization_fields = {
"ror_json": result,
"name": _first_name(result, types__in="ror_display"),
"name_original": _first_name(result, types__in="label", lang__not="en"),
"acronym": _first_name(result, types__in="acronym"),
"country": geonames.get("country_code"),
"address": geonames.get("name"),
"country": location_details.get("country_code"),
"address": location_details.get("name"),
}
return organization_fields
@staticmethod
def _process_organization(organization):
def _map_ror_to_organization(data: dict[str, Any]) -> dict[str, Any]:
"""
Processes an organization from the ROR API into a dict that can be used
to create a new Organization object.
"""
# Remove url part from ROR ID
ror_link = organization["id"]
ror_link = data["id"]
ror_id = ror_link.split("/")[-1]
return {**organization, "id": ror_id, "ror_link": ror_link}
return {**data, "id": ror_id, "ror_link": ror_link}
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment