Add type checks to terms, and ontology

src/hpotk/graph/_factory.py

@@ -91,7 +91,7 @@ def _build_adjacency_matrix(self, nodes: typing.Sequence[TermId],
 
 
 def get_array_of_unique_and_sorted_nodes(edge_list: typing.Sequence[DirectedEdge]) -> np.ndarray:
-    return np.fromiter(sorted(get_unique_nodes(edge_list)), dtype=object)
+    return np.array(list(sorted(get_unique_nodes(edge_list))))
 
 
 def get_list_of_unique_nodes(edge_list: typing.Sequence[DirectedEdge]):

src/hpotk/model/_term.py

@@ -1,3 +1,4 @@
+import hpotk
 import abc
 import enum
 import typing
@@ -15,10 +16,19 @@ class MinimalTerm(Identified, Named, metaclass=abc.ABCMeta):
     """
 
     @staticmethod
-    def create_minimal_term(term_id: TermId,
+    def create_minimal_term(term_id: typing.Union[TermId, str],
                             name: str,
-                            alt_term_ids: typing.Sequence[TermId],
+                            alt_term_ids: typing.Iterable[typing.Union[TermId, str]],
                             is_obsolete: bool):
+        """
+        Create a `MinimalTerm` from its components.
+
+        :param term_id: a `TermId` or a CURIE `str` (e.g. 'HP:0001250').
+        :param name: term name (e.g. Seizure) .
+        :param alt_term_ids: an iterable with `TermId`s that represent the alternative IDs of the term.
+        :param is_obsolete: `True` if the `MinimalTerm` has been obsoleted, or `False` otherwise.
+        :return:
+        """
         return DefaultMinimalTerm(term_id, name, alt_term_ids, is_obsolete)
 
     @property
@@ -106,8 +116,8 @@ def is_current(self) -> bool:
 class Synonym(Named):
 
     def __init__(self, name: str,
-                 synonym_category: typing.Optional[SynonymCategory],
-                 synonym_type: typing.Optional[SynonymType],
+                 synonym_category: typing.Optional[SynonymCategory] = None,
+                 synonym_type: typing.Optional[SynonymType] = None,
                  xrefs: typing.Optional[typing.Sequence[TermId]] = None):
         self._name = name
         self._scat = synonym_category
@@ -162,10 +172,10 @@ class Term(MinimalTerm, metaclass=abc.ABCMeta):
     # TODO - add the remaining attributes from phenol's Term?
 
     @staticmethod
-    def create_term(identifier: TermId,
+    def create_term(identifier: typing.Union[TermId, str],
                     name: str,
-                    alt_term_ids: typing.Sequence[TermId],
-                    is_obsolete: typing.Optional[bool],
+                    alt_term_ids: typing.Iterable[typing.Union[TermId, str]],
+                    is_obsolete: bool,
                     definition: typing.Optional[str],
                     comment: typing.Optional[str],
                     synonyms: typing.Optional[typing.Sequence[Synonym]],
@@ -254,19 +264,37 @@ def __str__(self):
                f'alt_term_ids="{self.alt_term_ids}")'
 
 
+def map_to_term_id(value: typing.Union[TermId, str]) -> TermId:
+    if isinstance(value, TermId):
+        return value
+    elif isinstance(value, str):
+        return TermId.from_curie(value)
+    else:
+        raise ValueError(f'Expected a `TermId` or `str` but got {type(value)}')
+
+
+def validate_name(name: typing.Optional[str]) -> str:
+    # Some obsolete nodes do not have labels in the Obographs format.
+    # We assign an empty string.
+    if name is None:
+        return ''
+    else:
+        return hpotk.util.validate_instance(name, str, 'name')
+
+
 class DefaultMinimalTerm(MinimalTerm):
 
-    def __init__(self, identifier: TermId,
+    def __init__(self, identifier: typing.Union[TermId, str],
                  name: str,
-                 alt_term_ids: typing.Sequence[TermId],
+                 alt_term_ids: typing.Iterable[typing.Union[TermId, str]],
                  is_obsolete: bool):
-        self._id = identifier
-        self._name = name
-        self._alts = alt_term_ids
-        self._is_obsolete = is_obsolete
+        self._id = hpotk.util.validate_instance(map_to_term_id(identifier), TermId, 'identifier')
+        self._name = validate_name(name)
+        self._alts = tuple(map(map_to_term_id, alt_term_ids))
+        self._is_obsolete = hpotk.util.validate_instance(is_obsolete, bool, 'is_obsolete')
 
     @property
-    def identifier(self):
+    def identifier(self) -> TermId:
         return self._id
 
     @property
@@ -289,40 +317,32 @@ def __repr__(self):
                f' alt_term_ids="{self._alts}")'
 
 
-class DefaultTerm(Term):
+def validate_synonyms(synonyms: typing.Optional[typing.Sequence[Synonym]]):
+    if synonyms is None:
+        return None
+    else:
+        validated = []
+        for i, s in enumerate(synonyms):
+            validated.append(hpotk.util.validate_instance(s, Synonym, f'synonym #{i}'))
+        return tuple(validated)
+
+
+class DefaultTerm(DefaultMinimalTerm, Term):
 
-    def __init__(self, identifier: TermId,
+    def __init__(self, identifier: typing.Union[TermId, str],
                  name: str,
-                 alt_term_ids: typing.Sequence[TermId],
-                 is_obsolete: typing.Optional[bool],
+                 alt_term_ids: typing.Iterable[typing.Union[TermId, str]],
+                 is_obsolete: bool,
                  definition: typing.Optional[str],
                  comment: typing.Optional[str],
                  synonyms: typing.Optional[typing.Sequence[Synonym]],
-                 xrefs: typing.Optional[typing.Sequence[TermId]]):
-        self._id = identifier
-        self._name = name
-        self._alt_term_ids = alt_term_ids
-        self._is_obsolete = False if is_obsolete is None else is_obsolete
-        self._definition = definition
-        self._comment = comment
-        self._synonyms = synonyms
-        self._xrefs = xrefs
-
-    @property
-    def identifier(self) -> TermId:
-        return self._id
-
-    @property
-    def name(self) -> str:
-        return self._name
-
-    @property
-    def alt_term_ids(self) -> typing.Sequence[TermId]:
-        return self._alt_term_ids
-
-    @property
-    def is_obsolete(self) -> bool:
-        return self._is_obsolete
+                 xrefs: typing.Optional[typing.Sequence[typing.Union[TermId, str]]]):
+        DefaultMinimalTerm.__init__(self, identifier=identifier, name=name,
+                                    alt_term_ids=alt_term_ids, is_obsolete=is_obsolete)
+        self._definition = hpotk.util.validate_optional_instance(definition, str, 'definition')
+        self._comment = hpotk.util.validate_optional_instance(comment, str, 'comment')
+        self._synonyms = validate_synonyms(synonyms)
+        self._xrefs = tuple(map(map_to_term_id, xrefs)) if xrefs is not None else None
 
     @property
     def definition(self) -> str:
@@ -349,4 +369,4 @@ def __repr__(self):
                f'synonyms={self._synonyms}, ' \
                f'xrefs={self._xrefs}, ' \
                f'is_obsolete={self._is_obsolete}, ' \
-               f'alt_term_ids="{self._alt_term_ids}")'
+               f'alt_term_ids="{self._alts}")'

src/hpotk/model/_term_id.py

@@ -15,9 +15,9 @@ def from_curie(curie: str):
         Create a `TermId` from a `str` where *prefix* and *id* are delimited either by a colon `:` (e.g. `HP:1234567`)
         or an underscore '_' (e.g. `NCIT_C3117`).
 
-        :param curie: CURIE to be parsed
-        :return: the created `TermId`
-        :raises: ValueError if the value is mis-formatted.
+        :param curie: a CURIE `str` to be parsed.
+        :return: the created `TermId`.
+        :raises: `ValueError` if the value is mis-formatted.
         """
         if curie is None:
             raise ValueError(f'Curie must not be None')

src/hpotk/ontology/_default.py

@@ -1,5 +1,6 @@
 import typing
 
+from hpotk.util import validate_instance, validate_optional_instance
 from hpotk.graph import OntologyGraph
 from hpotk.model import TermId
 from ._api import Ontology, MinimalOntology
@@ -12,10 +13,10 @@ def __init__(self, graph: OntologyGraph[ID],
                  current_terms: typing.Sequence[MINIMAL_TERM],
                  term_id_to_term: typing.Mapping[ID, MINIMAL_TERM],
                  version: typing.Optional[str] = None):
-        self._graph = graph
+        self._graph = validate_instance(graph, OntologyGraph, 'graph')
         self._current_terms = current_terms
         self._term_id_to_term = term_id_to_term
-        self._version = version
+        self._version = validate_optional_instance(version, str, 'version')
 
     @property
     def graph(self) -> OntologyGraph[ID]:
@@ -50,10 +51,10 @@ def __init__(self, graph: OntologyGraph[ID],
                  current_terms: typing.Sequence[TERM],
                  term_id_to_term: typing.Mapping[ID, TERM],
                  version: typing.Optional[str] = None):
-        self._graph = graph
+        self._graph = validate_instance(graph, OntologyGraph, 'graph')
         self._current_terms = current_terms
         self._term_id_to_term = term_id_to_term
-        self._version = version
+        self._version = validate_optional_instance(version, str, 'version')
 
     @property
     def graph(self) -> OntologyGraph[ID]:

tests/ontology/load/test_obographs.py

@@ -110,7 +110,7 @@ def test_term_properties(self):
         self.assertEqual(term.comment, 'The cardiovascular system consists of the heart, vasculature, and the '
                                        'lymphatic system.')
         self.assertEqual(term.is_obsolete, False)
-        self.assertListEqual(term.alt_term_ids, [TermId.from_curie('HP:0003116')])
+        self.assertTupleEqual(term.alt_term_ids, (TermId.from_curie('HP:0003116'),))
 
         synonyms = term.synonyms
         self.assertEqual(len(synonyms), 3)
@@ -133,9 +133,9 @@ def test_term_properties(self):
         self.assertEqual(three.synonym_type, hp.model.SynonymType.LAYPERSON_TERM)
         self.assertIsNone(three.xrefs)
 
-        self.assertEqual(term.xrefs, [TermId.from_curie(curie) for curie in ('UMLS:C0243050', 'UMLS:C0007222',
-                                                                             'MSH:D018376', 'SNOMEDCT_US:49601007',
-                                                                             'MSH:D002318')])
+        self.assertEqual(term.xrefs, tuple(TermId.from_curie(curie) for curie in ('UMLS:C0243050', 'UMLS:C0007222',
+                                                                                  'MSH:D018376', 'SNOMEDCT_US:49601007',
+                                                                                  'MSH:D002318')))
 
     def test_synonym_properties(self):
         term = TestTerms.ONTOLOGY.get_term('HP:0001627')

tests/test_obographs.py

@@ -33,7 +33,7 @@ def test_term_properties(self):
         self.assertEqual(term.comment, 'The cardiovascular system consists of the heart, vasculature, and the '
                                        'lymphatic system.')
         self.assertEqual(term.is_obsolete, False)
-        self.assertListEqual(term.alt_term_ids, [TermId.from_curie('HP:0003116')])
+        self.assertTupleEqual(term.alt_term_ids, (TermId.from_curie('HP:0003116'),))
 
         synonyms = term.synonyms
         self.assertEqual(len(synonyms), 3)
@@ -56,9 +56,10 @@ def test_term_properties(self):
         self.assertEqual(three.synonym_type, hp.model.SynonymType.LAYPERSON_TERM)
         self.assertIsNone(three.xrefs)
 
-        self.assertEqual(term.xrefs, [TermId.from_curie(curie) for curie in ('UMLS:C0243050', 'UMLS:C0007222',
-                                                                             'MSH:D018376', 'SNOMEDCT_US:49601007',
-                                                                             'MSH:D002318')])
+        self.assertEqual(term.xrefs, tuple(TermId.from_curie(curie)
+                                           for curie in ('UMLS:C0243050', 'UMLS:C0007222',
+                                                         'MSH:D018376', 'SNOMEDCT_US:49601007',
+                                                         'MSH:D002318')))
 
     def test_synonym_properties(self):
         term = TestTerms.ONTOLOGY.get_term('HP:0001627')

tests/test_term_id.py

@@ -13,6 +13,11 @@ def test_from_curie(self, curie):
         term_id = TermId.from_curie(curie)
         assert term_id.value == "HP:1234567"
 
+    def test_from_curie__unusual_input(self):
+        term_id = TermId.from_curie('SNOMEDCT_US:313307000')
+        assert term_id.prefix == 'SNOMEDCT_US'
+        assert term_id.id == '313307000'
+
     @ddt.data(
         ("HP1234567", "The CURIE HP1234567 has no colon `:` or underscore `_`"),
         (None, "Curie must not be None")