Document utilities and tighten type hints

Author: ubyndr

pandasaurus/query.py

@@ -256,7 +256,7 @@ def parent_enrichment(self):
         self.ancestor_enrichment(1)
 
     def synonym_lookup(self) -> pd.DataFrame:
-        """
+        """Return labels plus synonym rows for every seed term.
 
         Returns:
             A DataFrame containing labels and synonyms of the terms from the seed list.
@@ -284,7 +284,7 @@ def synonym_lookup(self) -> pd.DataFrame:
         return result_df
 
     def get_most_specific_objects(self, predicate: str, ontology: str):
-        """
+        """Return the most specific objects associated with the given predicate.
 
         Args:
             predicate: Relationship that wanted to be explored
@@ -296,6 +296,7 @@ def get_most_specific_objects(self, predicate: str, ontology: str):
                 - http://purl.obolibrary.org/obo/uberon.owl
 
         Returns:
+            DataFrame capturing subject, predicate, and object labels.
 
         """
         subject_list = [term.get_iri() for term in self._term_list]
@@ -310,7 +311,7 @@ def get_most_specific_objects(self, predicate: str, ontology: str):
         )
 
     def get_most_specific_subjects(self, predicate: str, ontology: str):
-        """
+        """Return the most specific subjects associated with the given predicate.
 
         Args:
             predicate: Relationship that wanted to be explored
@@ -322,6 +323,7 @@ def get_most_specific_subjects(self, predicate: str, ontology: str):
                 - http://purl.obolibrary.org/obo/uberon.owl
 
         Returns:
+            DataFrame capturing subject, predicate, and object labels.
 
         """
         object_list = [term.get_iri() for term in self._term_list]
@@ -354,7 +356,8 @@ def update_obsoleted_terms(self):
         """Replaces all obsoleted terms in the term list with the new term that obsoletes them."""
         [getattr(term, "update_obsoleted_term")() for term in self._term_list]
 
-    def mirror_enrichment_for_graph_generation(self, term_list: List[str]):
+    def mirror_enrichment_for_graph_generation(self, term_list: List[str]) -> None:
+        """Populate `graph_df` with all pairwise enrichment edges for graph output."""
         # TODO definitely need a refactoring later on
         s_result = []
         for s_chunk in chunks(term_list, 45):
@@ -373,7 +376,8 @@ def mirror_enrichment_for_graph_generation(self, term_list: List[str]):
             .reset_index(drop=True)
         )
 
-    def _generate_enrichment_graph(self, object_list):
+    def _generate_enrichment_graph(self, object_list: List[str]) -> None:
+        """Build the Graph representation backing the enrichment results."""
         self.mirror_enrichment_for_graph_generation(object_list)
         self.graph = GraphGenerator.generate_enrichment_graph(self.graph_df)
         self.graph = GraphGenerator.apply_transitive_reduction(self.graph, self.enriched_df["p"].unique().tolist())

pandasaurus/resources/term.py

@@ -6,20 +6,29 @@ class Term:
 
     def __init__(
         self,
-        label: str,
+        label: Optional[str],
         iri: str,
         is_valid: bool,
         new_label: Optional[str] = None,
         new_iri: Optional[str] = None,
     ):
+        """Initialize a term instance.
+
+        Args:
+            label: Human readable label; can be None for invalid CURIEs.
+            iri: The CURIE/IRI of the term.
+            is_valid: Whether the term exists in the ontology.
+            new_label: Replacement label for obsolete terms.
+            new_iri: Replacement IRI for obsolete terms.
+        """
         self.__label = label
         self.__iri = iri
         self.__is_valid = is_valid
         self.__new_label = new_label
         self.__new_iri = new_iri
         self.__is_obsolete: bool = True if new_label and new_iri else False
 
-    def get_label(self) -> str:
+    def get_label(self) -> Optional[str]:
         """Returns term label.
 
         Returns:
@@ -46,7 +55,7 @@ def get_is_valid(self) -> bool:
         """
         return self.__is_valid
 
-    def get_new_label(self) -> str:
+    def get_new_label(self) -> Optional[str]:
         """Returns new term label of obsoleted term.
 
         Returns:
@@ -55,7 +64,7 @@ def get_new_label(self) -> str:
         """
         return self.__new_label
 
-    def get_new_iri(self) -> str:
+    def get_new_iri(self) -> Optional[str]:
         """Returns new term IRI of obsoleted term.
 
         Returns:
@@ -64,7 +73,7 @@ def get_new_iri(self) -> str:
         """
         return self.__new_iri
 
-    def get_is_obsoleted(self) -> str:
+    def get_is_obsoleted(self) -> bool:
         """Returns term obsoletion status.
 
         Returns:

pandasaurus/slim_manager.py

@@ -52,5 +52,6 @@ def get_slim_members(slim_list: List[str]) -> List[str]:
         ]
 
     @staticmethod
-    def _get_ontology_list():
+    def _get_ontology_list() -> List[str]:
+        """Return ontology titles available in Ubergraph."""
         return [row.get("title") for row in run_sparql_query(get_ontology_list_query())]

pandasaurus/utils/logging_config.py

@@ -1,14 +1,18 @@
 import logging
 import sys
+from logging import LogRecord, Logger
 
 
-# Create a filter to exclude ERROR log records
 class NoErrorFilter(logging.Filter):
-    def filter(self, record):
+    """Filter that suppresses ERROR records, letting INFO/DEBUG through."""
+
+    def filter(self, record: LogRecord) -> bool:
+        """Return True when the log record is not an ERROR level entry."""
         return record.levelno != logging.ERROR
 
 
-def configure_logger():
+def configure_logger() -> Logger:
+    """Configure and return the shared pandasaurus logger."""
     logger = logging.getLogger(__name__)
     logger.setLevel(logging.INFO)
     # logger.propagate = False

pandasaurus/utils/query_utils.py

@@ -1,5 +1,5 @@
 import os
-from typing import Iterator
+from typing import Iterable, Iterator, List, Sequence, TypeVar
 
 import certifi
 from oaklib.implementations import UbergraphImplementation
@@ -9,21 +9,20 @@
 os.environ.setdefault("REQUESTS_CA_BUNDLE", certifi.where())
 
 oi = UbergraphImplementation()
+T = TypeVar("T")
 
 
 def run_sparql_query(query: str) -> Iterator:
+    """Execute a SPARQL query against Ubergraph."""
     return oi.query(query=query, prefixes=get_prefixes(query, oi.prefix_map().keys()))
 
 
-def chunks(lst, n):
-    for i in range(0, len(lst), n):
-        yield lst[i : i + n]
+def chunks(items: Sequence[T], size: int) -> Iterator[Sequence[T]]:
+    """Yield slices of `items` with at most `size` entries."""
+    for i in range(0, len(items), size):
+        yield items[i : i + size]
 
 
-def get_prefixes(text, prefix_map):
-    _prefixes = []
-    for prefix in prefix_map:
-        if prefix + ":" in text:
-            _prefixes.append(prefix)
-
-    return _prefixes
+def get_prefixes(text: str, prefix_map: Iterable[str]) -> List[str]:
+    """Return CURIE prefixes referenced in `text`."""
+    return [prefix for prefix in prefix_map if f"{prefix}:" in text]