diff --git a/scipost_django/organizations/utils.py b/scipost_django/organizations/utils.py index 687114cdff29a1945396f14729d76e0afc97898a..b1e626712da7cdb9e8f7bdc028c8d98ab9c2ba34 100644 --- a/scipost_django/organizations/utils.py +++ b/scipost_django/organizations/utils.py @@ -1,24 +1,28 @@ +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}