wfr2/species-synonym-api
0
1"""2Abstract base class for biodiversity database API clients.3 4All external database connectors in ``apis_pipe`` subclass ``SpeciesAPI`` and5implement its three-phase pipeline contract. The three phases are enforced by6naming convention:7 8- ``_fetch_*`` — network calls only; return raw responses without parsing.9- ``_extract_*`` — pure string extraction from already-fetched data; no I/O.10- ``_compile_*`` — row assembly; call helpers and read dict keys, no cleaning11 or network calls.12 13The single public entry point is ``get_synonyms(name)``, which orchestrates14all three phases and returns a schema-validated ``pd.DataFrame``.15 16Set the ``APIS_PIPE_VERBOSE`` env var (or ``SpeciesAPI.VERBOSE = True`` at17runtime) to print a trace of data flowing through each pipeline stage and a18warning whenever a fetch call returns blank/empty data — most queries miss on19most portals (each source only covers part of the tree of life), so this is20off by default and intended for development/debugging.21"""22 23import inspect24import os25import re26import xml.etree.ElementTree as ET27from abc import ABC, abstractmethod28 29import pandas as pd30import requests31from dotenv import load_dotenv32 33from scripts.utils.normalize_query_string import normalize_query_string34from scripts.utils.schema import empty_synonym_table, make_synonym_row35 36load_dotenv()37 38 39class _Unset:40 """Sentinel for _format_row optional params that were not provided by the caller."""41 42 __slots__ = ()43 44 def __repr__(self) -> str:45 return "<NOT_PROVIDED>"46 47 48_UNSET = _Unset()49 50 51class SpeciesAPI(ABC):52 """53 Abstract base class establishing a unified contract for biodiversity database clients.54 55 Concrete subclasses must define a ``BASE_URL`` class attribute and implement56 the five abstract methods: ``_fetch_query_data``, ``_fetch_synonym_data``,57 ``_fetch_accepted_data``, ``_compile_synonyms``, and58 ``_compile_accepted``. Optional helpers may be implemented and/or59 overridden to customise behavior for a specific source.60 61 Attributes62 ----------63 HEADERS : dict64 HTTP headers sent with every request. Overrides the default65 ``requests`` User-Agent so that portals that reject bot agents respond66 normally.67 """68 69 HEADERS: dict = {"User-Agent": "Mozilla/5.0"}70 BASE_URL: str71 # Default request timeout (seconds) for all fetch helpers. Subclasses that72 # need longer can override this single attribute, e.g. ``_TIMEOUT = 60``.73 _TIMEOUT: int = 3074 # Verbose/debug logging toggle shared by all subclasses. Defaults from the75 # APIS_PIPE_VERBOSE env var; can also be flipped at runtime, e.g.76 # ``SpeciesAPI.VERBOSE = True`` (all clients) or ``GBIFAPI.VERBOSE = True``77 # (one client). When enabled, fetch wrappers print a warning whenever a78 # call returns blank/empty data, and get_synonyms() traces the raw data79 # flowing between pipeline stages.80 VERBOSE: bool = os.getenv("APIS_PIPE_VERBOSE", "").strip().lower() in (81 "1",82 "true",83 "yes",84 "on",85 )86 _INFRASPECIFIC_RE: re.Pattern = re.compile(87 r"\b(var\.|subsp\.|ssp\.|f\.|fo\.|subf\.|cv\.|sect\.|subsect\.|ser\.|subgen\.|subg\.)",88 # TODO: put this in a config, rather than having it inside this file so a user could add if needed89 re.IGNORECASE,90 )91 92 def __init_subclass__(cls, **kwargs):93 super().__init_subclass__(**kwargs)94 if not inspect.isabstract(cls) and not hasattr(cls, "BASE_URL"):95 raise TypeError(f"{cls.__name__} must define a BASE_URL class attribute.")96 97 # ------------------------------------------------------------------98 # Query methods (to be used by children to implement the required methods, can be optionally overridden but should work for most children as-is)99 # ------------------------------------------------------------------100 101 def _fetch(102 self, url: str, params: dict = {}, timeout: int | None = None103 ) -> requests.Response:104 """105 Make a GET request to the specified URL with error handling.106 107 Parameters108 ----------109 url : str110 The full URL to send the GET request to.111 params : dict, optional112 Query parameters to include in the request. Default is an empty dict.113 timeout : int, optional114 Request timeout in seconds. Defaults to the client's ``_TIMEOUT``115 (30s in the base class) when not provided.116 117 Returns118 -------119 requests.Response120 The response object if the request is successful.121 122 Raises123 ------124 requests.RequestException125 If the request times out, the source is unreachable, or the126 response has a non-2xx HTTP status.127 """128 if timeout is None:129 timeout = self._TIMEOUT130 try:131 response = requests.get(132 url, params=params, headers=self.HEADERS, timeout=timeout133 )134 response.raise_for_status()135 return response136 except requests.RequestException as e:137 print(f"{type(self).__name__} fetch error [{url}]: {e}")138 raise139 140 def _fetch_JSON(141 self, url: str, params: dict = {}, timeout: int | None = None142 ) -> dict:143 """144 Make a GET request to a REST JSON endpoint and return the parsed response.145 146 Used by children that query standard REST APIs returning JSON.147 148 Parameters149 ----------150 url : str151 Full URL of the endpoint.152 params : dict, optional153 URL query parameters.154 timeout : int, optional155 Request timeout in seconds. Defaults to the client's ``_TIMEOUT``156 (30s in the base class) when not provided.157 158 Returns159 -------160 dict161 Parsed JSON response, or ``{}`` on any error (network, HTTP, or a162 response body that is not valid JSON).163 """164 165 response = self._fetch(url, params=params, timeout=timeout)166 if response is None:167 return {}168 try:169 return response.json()170 except ValueError:171 print(f"{type(self).__name__} error parsing JSON.")172 return {}173 174 def _fetch_XML(175 self, url: str, params: dict = {}, timeout: int | None = None176 ) -> ET.Element:177 """178 Make a GET request and return the parsed XML root element.179 180 Used by children that consume XML responses. On a parse error of an181 otherwise successful response, prints a message and returns an empty182 ``ET.Element``.183 184 Parameters185 ----------186 url : str187 Full URL of the endpoint.188 params : dict, optional189 URL query parameters.190 timeout : int, optional191 Request timeout in seconds. Defaults to the client's ``_TIMEOUT``192 (30s in the base class) when not provided.193 194 Returns195 -------196 xml.etree.ElementTree.Element197 Parsed root element of the XML response, or an empty element198 if the response body could not be parsed as XML.199 200 Raises201 ------202 requests.RequestException203 If the underlying request fails. See ``_fetch``.204 """205 response = self._fetch(url, params=params, timeout=timeout)206 root = None207 if response is not None:208 try:209 root = ET.fromstring(response.text)210 except ET.ParseError:211 print(f"{type(self).__name__} error parsing XML.")212 if root is None:213 root = ET.Element(214 "empty"215 ) # tag name chosen to avoid confusion with valid root tags in responses, will be treated as empty by _is_empty()216 self._warn_blank(url, root, params)217 return root218 219 def _fetch_HTML(220 self, url: str, params: dict = {}, timeout: int | None = None221 ) -> str:222 """223 Make a GET request and return the raw HTML response text.224 225 Used by children that scrape HTML pages.226 227 Parameters228 ----------229 url : str230 Full URL of the endpoint.231 params : dict, optional232 URL query parameters.233 timeout : int, optional234 Request timeout in seconds. Defaults to the client's ``_TIMEOUT``235 (30s in the base class) when not provided.236 237 Returns238 -------239 str240 Raw HTML text of the response.241 242 Raises243 ------244 requests.RequestException245 If the underlying request fails. See ``_fetch``.246 """247 response = self._fetch(url, params=params, timeout=timeout)248 text = response.text if response is not None else ""249 self._warn_blank(url, text, params)250 return text251 252 # ------------------------------------------------------------------253 # Verbose/debug logging helpers (gated by VERBOSE; can be optionally overridden)254 # ------------------------------------------------------------------255 256 def _warn_blank(self, url: str, data, params: dict | None = None) -> None:257 """258 Print a warning when a fetch call returned blank/empty data.259 260 Only prints when ``VERBOSE`` is enabled. Called by the ``_fetch_JSON``,261 ``_fetch_XML``, and ``_fetch_HTML`` wrappers on whatever they are262 about to return, whether the request failed outright (``_fetch``263 also reports the network/HTTP error separately and unconditionally)264 or merely succeeded with no usable data (e.g. no match found for the265 query).266 267 Parameters268 ----------269 url : str270 The URL that was requested.271 data : list, str, dict, or xml.etree.ElementTree.Element272 The parsed response to check for emptiness.273 params : dict, optional274 Query parameters sent with the request, included in the warning275 for context.276 """277 if self.VERBOSE and self._is_empty(data):278 suffix = f" (params={params})" if params else ""279 print(280 f"[{type(self).__name__}] WARNING: blank/empty response from {url}{suffix}"281 )282 283 def _warn_if_blank(self, step: str, data) -> None:284 """285 Print a warning when a pipeline step's result is blank/empty.286 287 Only prints when ``VERBOSE`` is enabled. Unlike ``_warn_blank``288 (network-level, keyed by URL), this is used by ``get_synonyms`` for289 steps whose underlying fetch response was not itself blank, but290 which produced no usable data after extraction (e.g. an empty291 synonym list).292 293 Parameters294 ----------295 step : str296 Label identifying which pipeline stage produced *data*.297 data : list, str, dict, or xml.etree.ElementTree.Element298 The data to check for emptiness.299 """300 if self.VERBOSE and self._is_empty(data):301 print(f"[{type(self).__name__}] WARNING: {step} returned blank/empty data.")302 303 # ------------------------------------------------------------------304 # Boolean checker methods (to be used by children in their implementations of the required methods,can be optionally overridden but should work for most children as-is)305 # ------------------------------------------------------------------306 307 def _is_empty(self, input):308 """309 Return True if the input is blank, empty, or None.310 311 Parameters312 ----------313 input : list, str, dict, xml.etree.ElementTree.Element, or None314 The value to test for emptiness.315 316 Returns317 -------318 bool319 True if *input* is ``None``, ``""``, ``[]``, ``{}``, or an320 ``ET.Element`` with no children; False otherwise.321 """322 if input == {}:323 return True324 elif input == []:325 return True326 elif input == "":327 return True328 elif input is None:329 return True330 elif isinstance(input, ET.Element) and len(input) == 0:331 return True332 else:333 return False334 335 def _is_infraspecific(self, string: str) -> bool:336 """337 Return True if *string* is an infraspecific scientific name.338 339 Two checks are applied:340 341 1. Rank-marker check — detects explicit infraspecific abbreviations342 such as ``var.``, ``subsp.``, ``f.``, etc. via ``_INFRASPECIFIC_RE``.343 2. Bare-trinomial check — any name with three or more whitespace-344 delimited tokens (e.g. ``"Gadus morhua morhua"``) is treated as345 infraspecific even without a rank marker.346 347 Parameters348 ----------349 string : str350 A scientific name string to inspect.351 352 Returns353 -------354 bool355 True when either check matches, False otherwise.356 """357 return bool(self._INFRASPECIFIC_RE.search(string)) or len(string.split()) >= 3358 359 # ------------------------------------------------------------------360 # Extraction helper methods (to be overriden by children in their implementations of the required methods)361 # ------------------------------------------------------------------362 363 def _extract_publication_year(self, string: str) -> str:364 """365 Extract a four-digit publication year from a scientific name string.366 367 Parameters368 ----------369 string : str370 A scientific name or authorship string that may contain a year.371 372 Returns373 -------374 str375 Four-digit year string, or ``""`` if not found.376 377 Raises378 ------379 NotImplementedError380 When the child class has not provided an implementation.381 """382 raise NotImplementedError(383 f"{type(self).__name__} does not implement _extract_publication_year()."384 )385 386 def _extract_author(self, string: str) -> str:387 """388 Extract the authorship string from a string.389 390 Parameters391 ----------392 string : str393 A string that may contain an authorship component, such as a scientific name or a full citation.394 395 Returns396 -------397 str398 The authorship string (e.g. ``"(L.) Lam."``), or ``""`` if not found.399 400 Raises401 ------402 NotImplementedError403 When the child class has not provided an implementation.404 """405 raise NotImplementedError(406 f"{type(self).__name__} does not implement _extract_author()."407 )408 409 def _extract_publication_name(self, string: str) -> str:410 """411 Extract a publication name from a string.412 413 Parameters414 ----------415 string : str416 A string that may contain the title of the original publication of a species name, such as a citation of a journal or book.417 418 Returns419 -------420 str421 The publication name string, or ``""`` if not found.422 423 Raises424 ------425 NotImplementedError426 When the child class has not provided an implementation.427 """428 raise NotImplementedError(429 f"{type(self).__name__} does not implement _extract_publication_name()."430 )431 432 def _extract_status(self, string: str) -> str:433 """434 Map a raw status string to the schema's ``"Accepted"`` or ``"Synonym"``435 values by checking for those substrings.436 437 Parameters438 ----------439 string : str440 A raw status string from an API response, e.g. ``"accepted"``,441 ``"accepted name"``, ``"ambiguous synonym"``.442 443 Returns444 -------445 str446 ``"Accepted"``, ``"Synonym"``, or ``""`` if neither substring is found.447 """448 lower = string.lower()449 if "accepted" in lower:450 return "Accepted"451 if "synonym" in lower:452 return "Synonym"453 return ""454 455 def _extract_taxonomy(self, data: dict | list | str | ET.Element) -> dict[str, str]:456 """457 Extract taxonomy fields from a raw API response.458 459 Implementations should return a dict with any subset of the following460 keys, using ``class_`` (not ``class``) for the class rank to avoid the461 Python keyword conflict::462 463 {464 "kingdom": str,465 "phylum": str,466 "class_": str,467 "order": str,468 "family": str,469 "subfamily": str,470 }471 472 Ranks not implemented should be omitted from the dict rather than included as473 empty strings; ``_format_row`` treats absent keys as ``UNAVAILABLE``.474 The dict is typically unpacked with ``**taxonomy`` directly into a475 ``_format_row`` call.476 477 Parameters478 ----------479 data : any480 Raw API response data in the source's native format (varies by481 subclass — e.g. a ``dict``, ``list``, or482 ``xml.etree.ElementTree.Element``).483 484 Returns485 -------486 dict[str, str]487 Taxonomy field dict with string values for each rank present.488 489 Raises490 ------491 NotImplementedError492 When the child class has not provided an implementation.493 """494 raise NotImplementedError(495 f"{type(self).__name__} does not implement _extract_taxonomy()."496 )497 498 def _extract_genus_species(self, name: str) -> tuple[str, str]:499 """500 Parse a scientific name string into its genus and species components.501 502 Parameters503 ----------504 name : str505 A scientific name string whose first two whitespace-delimited506 tokens are the genus and species epithet (e.g.507 ``"Amanita muscaria"`` or ``"Amanita muscaria var. flavivolvata"``).508 509 Returns510 -------511 tuple[str, str]512 ``(genus, species)`` extracted from the first two tokens of *name*.513 514 Raises515 ------516 ValueError517 If *name* contains fewer than two whitespace-delimited tokens.518 """519 parts = name.split()520 if len(parts) < 2:521 raise ValueError(522 f"Expected at least two tokens in scientific name, got {name!r}"523 )524 return parts[0], parts[1]525 526 def _format_row(527 self,528 api_name: str,529 genus: str,530 species: str,531 api_internal_id: str,532 kingdom: str = _UNSET, # type: ignore[assignment]533 phylum: str = _UNSET, # type: ignore[assignment]534 class_: str = _UNSET, # type: ignore[assignment]535 order: str = _UNSET, # type: ignore[assignment]536 family: str = _UNSET, # type: ignore[assignment]537 subfamily: str = _UNSET, # type: ignore[assignment]538 author: str = _UNSET, # type: ignore[assignment]539 publication_name: str = _UNSET, # type: ignore[assignment]540 publication_year: str = _UNSET, # type: ignore[assignment]541 status: str = _UNSET, # type: ignore[assignment]542 original_source: str = _UNSET, # type: ignore[assignment]543 api_link: str = _UNSET, # type: ignore[assignment]544 ) -> dict:545 """546 Construct a validated pipeline-standard row record.547 548 Parameters549 ----------550 api_name : str551 The name of the API source (e.g. ``"GBIF"``). Must be one of the552 recognised values in ``schema._API_NAMES``.553 genus : str554 Taxonomic genus (single word, no whitespace).555 species : str556 Taxonomic species epithet (single word, no whitespace).557 api_internal_id : str558 Unique identifier for this record in the source database.559 kingdom, phylum, class_, family, subfamily : str, optional560 Taxonomic rank values. Each must be a single word. Use ``class_``561 for the class rank (``"class"`` is a Python keyword).562 author : str, optional563 Authorship string (e.g. ``"(L.) Lam."``).564 publication_name : str, optional565 Full publication citation string.566 publication_year : str, optional567 Four-digit publication year (e.g. ``"1783"``).568 status : str, optional569 Taxonomic status — ``"Accepted"``, ``"Synonym"``, or omit to570 leave as ``UNAVAILABLE``.571 original_source : str, optional572 Name of the original data source cited by the API.573 api_link : str, optional574 Direct URL to the taxon record in the source database.575 576 Returns577 -------578 dict579 A fully validated schema row with all columns from580 ``SYNONYM_COLUMNS``.581 """582 optional = {583 "kingdom": kingdom,584 "phylum": phylum,585 "class": class_,586 "order": order,587 "family": family,588 "subfamily": subfamily,589 "author": author,590 "publication_name": publication_name,591 "publication_year": publication_year,592 "status": status,593 "original_source": original_source,594 "api_link": api_link,595 }596 provided = {k: v for k, v in optional.items() if not isinstance(v, _Unset)}597 return make_synonym_row(598 api_name=api_name,599 genus=genus,600 species=species,601 api_internal_id=api_internal_id,602 **provided,603 )604 605 # ------------------------------------------------------------------606 # ID methods (not required, but one or the other is likely needed for most children)607 # ------------------------------------------------------------------608 609 def _extract_internal_id(self, raw_data: dict | list | str | ET.Element) -> str:610 """611 Resolve raw API response data to the source's internal database identifier.612 613 Parameters614 ----------615 raw_data : any616 The raw response data returned by the source API (type varies by617 subclass).618 619 Returns620 -------621 str622 The internal database identifier for the queried taxon.623 624 Raises625 ------626 NotImplementedError627 When the child class has not provided an implementation for this step.628 LookupError629 When the name cannot be resolved to an identifier.630 """631 raise NotImplementedError(632 f"{type(self).__name__} does not implement _extract_internal_id()."633 )634 635 def _extract_internal_accepted_id(636 self, raw_data: dict | list | str | ET.Element637 ) -> str:638 """639 Extract the internal identifier of the accepted taxon from API response data.640 641 An *accepted* taxon is the currently valid name that a synonym refers to.642 For some APIs the full synonym list is only accessible via the accepted643 name's record, so this method is needed when the initial search result644 may itself be a synonym.645 646 Parameters647 ----------648 raw_data : any649 Parsed API response data (type varies by subclass — e.g., ``dict``650 for GBIF, ``list`` for COL).651 652 Returns653 -------654 any655 The accepted taxon's internal identifier (type varies by subclass).656 657 Raises658 ------659 NotImplementedError660 When the child class has not provided an implementation.661 """662 raise NotImplementedError(663 f"{type(self).__name__} does not implement _extract_internal_accepted_id()."664 )665 666 # ------------------------------------------------------------------667 # Required methods (must be implemented by all children)668 # ------------------------------------------------------------------669 670 @abstractmethod671 def _fetch_query_data(self, name: str) -> dict | list | str | ET.Element:672 """673 Query the source for *name* and return the raw response.674 675 Parameters676 ----------677 name : str678 The scientific name to search (e.g. ``"Amanita muscaria"``).679 680 Returns681 -------682 dict or list or str or xml.etree.ElementTree.Element683 Raw query data in the source's native format (varies by subclass).684 """685 pass686 687 @abstractmethod688 def _fetch_synonym_data(689 self, raw_data: dict | list | ET.Element | str690 ) -> dict | list | str | ET.Element:691 """692 Fetch synonym records for the taxon resolved from *raw_data*.693 694 For sources that list synonyms under an accepted-name endpoint, this695 method extracts the accepted taxon's internal identifier from696 *raw_data* and issues a second request.697 698 Parameters699 ----------700 raw_data : dict or list or str or xml.etree.ElementTree.Element701 The response returned by ``_fetch_query_data``.702 703 Returns704 -------705 dict or list or str or xml.etree.ElementTree.Element706 Raw synonym data in the source's native format (varies by subclass).707 """708 pass709 710 @abstractmethod711 def _fetch_accepted_data(712 self,713 raw_data: dict | list | str | ET.Element,714 synonym_data: dict | list | str | ET.Element,715 ) -> dict | list | str | ET.Element:716 """717 Fetch metadata for the accepted name.718 719 Parameters720 ----------721 raw_data : dict or list or str or xml.etree.ElementTree.Element722 The response returned by ``_fetch_query_data``.723 synonym_data : dict or list or str or xml.etree.ElementTree.Element724 The response returned by ``_fetch_synonym_data``.725 726 Returns727 -------728 dict or list or str or xml.etree.ElementTree.Element729 Raw search term data in the source's native format (varies by subclass).730 """731 pass732 733 @abstractmethod734 def _compile_synonyms(735 self, synonym_data: dict | list | str | ET.Element736 ) -> list[dict]:737 """738 Convert raw synonym data into pipeline-standard synonym records.739 740 Parameters741 ----------742 synonym_data : dict or list or str or xml.etree.ElementTree.Element743 Raw synonym data as returned by ``_fetch_synonym_data``744 (type varies by subclass).745 746 Returns747 -------748 list of dict749 Pipeline-standard synonym records, each produced by ``_format_row``.750 """751 pass752 753 @abstractmethod754 def _compile_accepted(755 self, accepted_data: dict | list | str | ET.Element756 ) -> list[dict]:757 """758 Convert raw search term data into a one-item pipeline-standard record.759 760 Parameters761 ----------762 accepted_data : dict or list or str or xml.etree.ElementTree.Element763 Raw accepted as returned by764 ``_fetch_accepted_data`` (type varies by subclass).765 766 Returns767 -------768 list of dict769 A one-item list with the search term record, or ``[]`` if the name770 cannot be determined from ``accepted_data``.771 """772 pass773 774 # ------------------------------------------------------------------775 # Public methods (used by external callers, can be overrriden by children if needed but should work for most children as-is)776 # ------------------------------------------------------------------777 778 def get_synonyms(self, name: str) -> pd.DataFrame:779 """780 Retrieve taxonomic synonyms and publication metadata for a species name.781 782 Orchestrates the full pipeline: normalize the input, fetch raw query783 data, fetch synonym data, fetch accepted data (including taxonomy), and compile results into the standard format.784 785 This is the only public method and the main entry point for callers.786 787 Parameters788 ----------789 name : str790 The species name search query (e.g. ``"Amanita muscaria"``).791 792 Returns793 -------794 pd.DataFrame795 A DataFrame of accepted and synonym records in schema format, or an empty796 schema-format DataFrame if no results are found.797 """798 name = normalize_query_string(name)799 800 raw_data = self._fetch_query_data(name)801 if self._is_empty(raw_data):802 self._warn_if_blank("_fetch_query_data", raw_data)803 return empty_synonym_table()804 805 synonym_data = self._fetch_synonym_data(raw_data)806 self._warn_if_blank("_fetch_synonym_data", synonym_data)807 808 accepted_data = self._fetch_accepted_data(raw_data, synonym_data)809 self._warn_if_blank("_fetch_accepted_data", accepted_data)810 811 accepted = []812 synonyms = []813 814 # Compile accepted and synonym records only if their respective raw data is not empty.815 if not self._is_empty(synonym_data):816 synonyms = self._compile_synonyms(synonym_data)817 if not self._is_empty(accepted_data):818 accepted = self._compile_accepted(accepted_data)819 820 rows = accepted + synonyms821 if not rows:822 return empty_synonym_table()823 return pd.DataFrame(rows)824