Repository: SynBioDex/tyto (language: python)

Issue:
Adding support for tax_id's
Is there a way, through which we can add support for the **tax_id** as well, apart from the already existing ontologies?
(https://www.ncbi.nlm.nih.gov/taxonomy)


Discussion / hints:
Yes, this should be fairly straightforward to implement.  You can define your own interfaces for ontologies.  If you don't want to wait for me to implement it, follow the pattern in https://github.com/SynBioDex/tyto/blob/master/tyto/so.py

(You just need to know the URI for the NCBITaxonomy ontology in Ontobee.  Also, you do not need to include a local copy of the ontology...that is optional)

Fix patch:
diff --git a/tyto/endpoint/endpoint.py b/tyto/endpoint/endpoint.py
index 4cfc3dc..feab17a 100644
--- a/tyto/endpoint/endpoint.py
+++ b/tyto/endpoint/endpoint.py
@@ -115,6 +115,34 @@ def is_parent_of(self, ontology: "Ontology", parent_uri: str, child_uri: str) ->
         parent_terms = self.query(ontology, query, error_msg)
         return parent_uri in parent_terms
 
+    def is_ancestor_of(self, ontology: "Ontology", ancestor_uri: str, descendant_uri: str) -> bool:
+        query = f'''
+            SELECT distinct ?superclass
+            {{from_clause}}
+            WHERE 
+            {{{{
+                <{descendant_uri}> rdfs:subClassOf* ?superclass .
+                ?superclass rdf:type owl:Class .
+            }}}}
+            '''
+        error_msg = ''
+        ancestors = self.query(ontology, query, error_msg)
+        return ancestor_uri in ancestors
+
+    def is_descendant_of(self, ontology: "Ontology", descendant_uri: str, ancestor_uri: str) -> bool:
+        query = f'''
+            SELECT distinct ?descendant
+            {{from_clause}}
+            WHERE 
+            {{{{
+                ?descendant rdf:type owl:Class .
+                ?descendant rdfs:subClassOf* <{ancestor_uri}>
+            }}}}
+            '''
+        error_msg = ''
+        descendants = self.query(ontology, query, error_msg)
+        return descendant_uri in descendants
+
     def get_ontology(self):
         query = '''
             SELECT distinct ?ontology_uri
@@ -168,8 +196,8 @@ def convert(self, response):
         converted_response = []
         if response:
             response = response.convert()  # Convert http response to JSON
-            for var, binding in zip(response['head']['vars'],
-                                    response['results']['bindings']):
+            var = response['head']['vars'][0]
+            for binding in response['results']['bindings']:
                 if var in binding:
                     converted_response.append(binding[var]['value'])
         return converted_response
@@ -304,6 +332,32 @@ def get_children(self, ontology: "Ontology", uri: str):
              children = [ontology._reverse_sanitize_uri(iri) for iri in children]
         return children
 
+    def get_descendants(self, ontology: "Ontology", uri: str):
+        sanitized_uri = ontology._sanitize_uri(uri)
+        encoded_uri = urllib.parse.quote_plus(urllib.parse.quote_plus(sanitized_uri)) 
+        encoded_uri = urllib.parse.quote_plus(sanitized_uri)
+
+        request = '{url}/ontologies/{ontology}/descendants?id=' + encoded_uri
+        response = self._get_request(ontology, request)
+        descendants = []
+        if '_embedded' in response and 'terms' in response['_embedded']:
+             descendants = [term['iri'] for term in response['_embedded']['terms']]
+             descendants = [ontology._reverse_sanitize_uri(iri) for iri in descendants]
+        return descendants
+
+    def get_ancestors(self, ontology: "Ontology", uri: str):
+        sanitized_uri = ontology._sanitize_uri(uri)
+        encoded_uri = urllib.parse.quote_plus(urllib.parse.quote_plus(sanitized_uri)) 
+        encoded_uri = urllib.parse.quote_plus(sanitized_uri)
+
+        request = '{url}/ontologies/{ontology}/ancestors?id=' + encoded_uri
+        response = self._get_request(ontology, request)
+        descendants = []
+        if '_embedded' in response and 'terms' in response['_embedded']:
+             ancestors = [term['iri'] for term in response['_embedded']['terms']]
+             ancestors = [ontology._reverse_sanitize_uri(iri) for iri in ancestors]
+        return ancestors
+
     def is_parent_of(self, ontology: "Ontology", parent_uri: str, child_uri: str) -> bool:
         parent_uri = ontology._reverse_sanitize_uri(parent_uri)
         return parent_uri in self.get_parents(ontology, child_uri) 	
@@ -312,6 +366,15 @@ def is_child_of(self, ontology: "Ontology", child_uri: str, parent_uri: str) ->
         child_uri = ontology._reverse_sanitize_uri(child_uri)
         return child_uri in self.get_children(ontology, parent_uri)
 
+    def is_descendant_of(self, ontology: "Ontology", descendant_uri: str, ancestor: str) -> bool:
+        descendant_uri = ontology._reverse_sanitize_uri(descendant_uri)
+        return descendant_uri in self.get_descendants(ontology, ancestor) 	
+
+    def is_ancestor_of(self, ontology: "Ontology", ancestor_uri: str, descendant_uri: str) -> bool:
+        ancestor_uri = ontology._reverse_sanitize_uri(ancestor_uri)
+        return ancestor_uri in self.get_ancestors(ontology, descendant_uri)
+
+
     def convert(self, response):
         pass
 
diff --git a/tyto/ncbi_taxon.py b/tyto/ncbi_taxon.py
index 615678d..b7b6d4b 100644
--- a/tyto/ncbi_taxon.py
+++ b/tyto/ncbi_taxon.py
@@ -1,7 +1,7 @@
-from .tyto import Ontology, Ontobee, installation_path, multi_replace
+from .tyto import Ontology, EBIOntologyLookupService, installation_path, multi_replace
 
 
-NCBITaxon = Ontology(endpoints=[Ontobee], uri='http://purl.obolibrary.org/obo/ncbitaxon.owl')
+NCBITaxon = Ontology(endpoints=[EBIOntologyLookupService], uri='http://purl.obolibrary.org/obo/ncbitaxon.owl')
 
 # Translate URIs to and from the identifiers.org namespace
 NCBITaxon._sanitize_uri = lambda uri: uri.replace('https://identifiers.org/taxonomy:',
diff --git a/tyto/tyto.py b/tyto/tyto.py
index 64862d6..e55c736 100644
--- a/tyto/tyto.py
+++ b/tyto/tyto.py
@@ -2,7 +2,7 @@
 import logging
 from functools import lru_cache
 
-from .endpoint import Ontobee, EBIOntologyLookupService, GraphEndpoint, Endpoint
+from .endpoint import Ontobee, EBIOntologyLookupService, GraphEndpoint, QueryBackend
 
 
 LOGGER = logging.getLogger(__name__)
@@ -30,7 +30,7 @@ def __init__(self, path=None, endpoints=None, uri=None):
         self.endpoints = None
         self.uri = uri
         if endpoints:
-            if not type(endpoints) is list or not all([issubclass(type(e), Endpoint) for e in endpoints]):
+            if not type(endpoints) is list or not all([issubclass(type(e), QueryBackend) for e in endpoints]):
                 raise TypeError('The endpoints argument requires a list of Endpoints')
             self.endpoints = endpoints
         if path:
@@ -68,6 +68,7 @@ def _handler(self, method_name, exception, *args):
                         return response
                 except Exception as x:
                     LOGGER.error(x)
+                    raise
 
         # If the connection fails or nothing found, fall back and load the ontology locally
         if self.graph and not self.graph.is_loaded():
@@ -126,16 +127,32 @@ def __new__(cls, value: str, ontology: Ontology):
         term.ontology = ontology
         return term
 
-    def is_child_of(self, parent_uri: str):
+    def is_child_of(self, parent_uri: "URI"):
         child_uri = self.ontology._sanitize_uri(self)
         parent_uri = self.ontology._sanitize_uri(parent_uri)
         return self.ontology._handler('is_child_of', None, child_uri, parent_uri)
 
-    def is_parent_of(self, child_uri: str):
-        child_uri = self.ontology._sanitize_uri(child_uri)
+    def is_parent_of(self, child_uri: "URI"):
         parent_uri = self.ontology._sanitize_uri(self)
+        child_uri = self.ontology._sanitize_uri(child_uri)
         return self.ontology._handler('is_parent_of', None, parent_uri, child_uri)
 
+    def is_descendant_of(self, ancestor_uri: "URI"):
+        descendant_uri = self.ontology._sanitize_uri(self)
+        ancestor_uri = self.ontology._sanitize_uri(ancestor_uri)
+        return self.ontology._handler('is_descendant_of', None, descendant_uri, ancestor_uri)
+
+    def is_ancestor_of(self, descendant_uri: "URI"):
+        ancestor_uri = self.ontology._sanitize_uri(self)
+        descendant_uri = self.ontology._sanitize_uri(descendant_uri)
+        return self.ontology._handler('is_ancestor_of', None, ancestor_uri, descendant_uri)
+
+    def is_subtype_of(self, supertype: "URI"):
+        return self.is_descendant_of(supertype)
+
+    def is_supertype_of(self, subtype: "URI"):
+        return self.is_ancestor_of(subtype)
+  
 
 # Utility functions
 def installation_path(relative_path):


Test patch:
diff --git a/test/test_ontology.py b/test/test_ontology.py
index 025888c..a43cc1e 100644
--- a/test/test_ontology.py
+++ b/test/test_ontology.py
@@ -60,11 +60,19 @@ def test_child_of(self):
         self.assertTrue(type(SO.promoter) is URI)
         self.assertTrue(SO.inducible_promoter.is_child_of(SO.promoter))
         self.assertFalse(SO.promoter.is_child_of(SO.inducible_promoter))
+        self.assertTrue(NCBITaxon.Escherichia_coli.is_child_of(NCBITaxon.Escherichia))
 
     def test_parent_of(self):
         self.assertFalse(SO.inducible_promoter.is_parent_of(SO.promoter))
         self.assertTrue(SO.promoter.is_parent_of(SO.inducible_promoter))
 
+    def test_descendants(self):
+        self.assertTrue(SO.RNApol_III_promoter.is_descendant_of(SO.promoter))
+        self.assertFalse(SO.promoter.is_descendant_of(SO.RNApol_III_promoter))
+
+    def test_ancestors(self):
+        self.assertTrue(SO.promoter.is_ancestor_of(SO.RNApol_III_promoter))
+        self.assertFalse(SO.promoter.is_descendant_of(SO.RNApol_III_promoter))
 
 class TestOLS(unittest.TestCase):
 
@@ -105,9 +113,18 @@ def test_parents(self):
     def test_children(self):
         children = EBIOntologyLookupService.get_children(SO, SO.promoter) 
         self.assertIn(SO.inducible_promoter, children)
+        self.assertNotIn(SO.RNApol_III_promoter, children)  # is a descendant, not a child
         self.assertEqual(len(children), 7)
         self.assertTrue(SO.promoter.is_parent_of(SO.inducible_promoter))
 
+    def test_descendants(self):
+        descendants = EBIOntologyLookupService.get_descendants(SO, SO.promoter) 
+        self.assertIn(SO.RNApol_III_promoter, descendants)
+
+    def test_ancestors(self):
+        ancestors = EBIOntologyLookupService.get_ancestors(NCBITaxon, NCBITaxon.Escherichia_coli)
+        self.assertIn(NCBITaxon.Bacteria, ancestors)
+
     def test_SBO(self):
         uri = 'https://identifiers.org/SBO:0000241'
         self.assertEqual(SBO.get_term_by_uri(uri), 'functional entity')