Repository: SynBioDex/tyto (language: python)

Issue:
Need graph-based method for `is_subclass_of`
This would help validate SPARQL queries over an ontology to ensure that they return valid ontology terms when performing lookup.

Can probably borrow code from OPIL to do this.


Discussion / hints:

Fix patch:
diff --git a/tyto/endpoint/endpoint.py b/tyto/endpoint/endpoint.py
index 9cc5eb8..4cfc3dc 100644
--- a/tyto/endpoint/endpoint.py
+++ b/tyto/endpoint/endpoint.py
@@ -87,19 +87,33 @@ def get_uri_by_term(self, ontology: "Ontology", term: str) -> str:
         response = response[0]
         return response
 
-    def is_subclass_of(self, ontology: "Ontology", subclass_uri: str, superclass_uri: str) -> bool:
-        query = '''
-            SELECT distinct ?subclass 
+    def is_child_of(self, ontology: "Ontology", child_uri: str, parent_uri: str) -> bool:
+        query = f'''
+            SELECT distinct ?child 
             {{from_clause}}
             WHERE 
             {{{{
-                ?subclass rdf:type owl:Class .
-                ?subclass rdfs:subClassOf <{}>
+                ?child rdf:type owl:Class .
+                ?child rdfs:subClassOf <{parent_uri}>
             }}}}
-            '''.format(superclass_uri)
+            '''
+        error_msg = ''
+        child_terms = self.query(ontology, query, error_msg)
+        return child_uri in child_terms
+
+    def is_parent_of(self, ontology: "Ontology", parent_uri: str, child_uri: str) -> bool:
+        query = f'''
+            SELECT distinct ?parent
+            {{from_clause}}
+            WHERE 
+            {{{{
+                <{child_uri}> rdf:type owl:Class .
+                <{child_uri}> rdfs:subClassOf ?parent
+            }}}}
+            '''
         error_msg = ''
-        subclasses = self.query(ontology, query, error_msg)
-        return subclass_uri in subclasses
+        parent_terms = self.query(ontology, query, error_msg)
+        return parent_uri in parent_terms
 
     def get_ontology(self):
         query = '''
@@ -125,6 +139,12 @@ class RESTEndpoint(QueryBackend, abc.ABC):
     def __init__(self, url):
         self.url = url
 
+    def _get_request(self, ontology: "Ontology", request: str):
+        response = requests.get(request)
+        if response.status_code == 200:
+            return response.json()
+        raise urllib.error.HTTPError(request, response.status_code, response.reason, response.headers, None)
+
 
 class SPARQLEndpoint(SPARQLBuilder, Endpoint):
 
@@ -219,6 +239,15 @@ def _load_ontology_ids(self):
             iri = o['config']['id']
             self.ontology_short_ids[iri] = short_id
 
+    def _get_request(self, ontology: "Ontology", get_request: str):
+        if not self.ontology_short_ids:
+            self._load_ontology_ids()
+        if ontology.uri not in self.ontology_short_ids:
+            raise LookupError(f'Ontology {ontology.uri} is not available at EBI Ontology Lookup Service')
+        short_id = self.ontology_short_ids[ontology.uri]
+        get_request = get_request.format(url=self.url, ontology=short_id)
+        return super()._get_request(ontology, get_request)
+
     def get_term_by_uri(self, ontology: "Ontology", uri: str):
         if not self.ontology_short_ids:
             self._load_ontology_ids()
@@ -249,6 +278,40 @@ def get_uri_by_term(self, ontology: "Ontology", term: str):
             return response['response']['docs'][0]['iri']
         raise urllib.error.HTTPError(get_query, response.status_code, response.reason, response.headers, None)
 
+    def get_parents(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}/parents?id=' + encoded_uri
+        response = self._get_request(ontology, request)
+        parents = []
+        if '_embedded' in response and 'terms' in response['_embedded']:
+             parents = [term['iri'] for term in response['_embedded']['terms']]
+             parents = [ontology._reverse_sanitize_uri(iri) for iri in parents]
+        return parents
+
+    def get_children(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}/children?id=' + encoded_uri
+        response = self._get_request(ontology, request)
+        children = []
+        if '_embedded' in response and 'terms' in response['_embedded']:
+             children = [term['iri'] for term in response['_embedded']['terms']]
+             children = [ontology._reverse_sanitize_uri(iri) for iri in children]
+        return children
+
+    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) 	
+
+    def is_child_of(self, ontology: "Ontology", child_uri: str, parent_uri: str) -> bool:
+        child_uri = ontology._reverse_sanitize_uri(child_uri)
+        return child_uri in self.get_children(ontology, parent_uri)
+
     def convert(self, response):
         pass
 
diff --git a/tyto/sbo.py b/tyto/sbo.py
index ed118ca..1ed124a 100644
--- a/tyto/sbo.py
+++ b/tyto/sbo.py
@@ -3,7 +3,7 @@
 
 SBO = Ontology(path=installation_path('ontologies/SBO_OWL.owl'),
                endpoints=[Ontobee],
-               uri='http://purl.obolibrary.org/obo/sbo.owl')
+               uri='http://biomodels.net/SBO/')
 """Ontology instance for Systems Biology Ontology"""
 
 
diff --git a/tyto/tyto.py b/tyto/tyto.py
index b156688..64862d6 100644
--- a/tyto/tyto.py
+++ b/tyto/tyto.py
@@ -6,7 +6,7 @@
 
 
 LOGGER = logging.getLogger(__name__)
-logging.basicConfig(format='tyto: %(levelname)s: %(message)s')
+logging.basicConfig(format='[%(levelname)s] %(filename)s %(lineno)d: %(message)s')
 
 
 class Ontology():
@@ -126,10 +126,15 @@ def __new__(cls, value: str, ontology: Ontology):
         term.ontology = ontology
         return term
 
-    def is_subclass_of(self, superclass_uri: str):
-        subclass_uri = self.ontology._sanitize_uri(self)
-        superclass_uri = self.ontology._sanitize_uri(superclass_uri)
-        return self.ontology._handler('is_subclass_of', None, subclass_uri, superclass_uri)
+    def is_child_of(self, parent_uri: str):
+        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)
+        parent_uri = self.ontology._sanitize_uri(self)
+        return self.ontology._handler('is_parent_of', None, parent_uri, child_uri)
 
 
 # Utility functions


Test patch:
diff --git a/test/test_ontology.py b/test/test_ontology.py
index 72c13fc..025888c 100644
--- a/test/test_ontology.py
+++ b/test/test_ontology.py
@@ -56,19 +56,38 @@ def test_OM(self):
         self.assertEqual(OM.get_term_by_uri('http://www.ontology-of-units-of-measure.org/resource/om-2/hour'),
                          'hour')
 
-    def test_subclass(self):
+    def test_child_of(self):
         self.assertTrue(type(SO.promoter) is URI)
-        self.assertTrue(SO.inducible_promoter.is_subclass_of(SO.promoter))
-        self.assertFalse(SO.promoter.is_subclass_of(SO.inducible_promoter))
+        self.assertTrue(SO.inducible_promoter.is_child_of(SO.promoter))
+        self.assertFalse(SO.promoter.is_child_of(SO.inducible_promoter))
+
+    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))
 
 
 class TestOLS(unittest.TestCase):
 
-    def test_SO(self):
-        restore_endpoints = SO.endpoints
-        restore_graph = SO.graph
+    SO_endpoints = SO.endpoints
+    SO_graph = SO.graph
+    SBO_endpoints = SBO.endpoints
+    SBO_graph = SBO.graph
+
+    @classmethod
+    def setUpClass(cls):
         SO.endpoints = [EBIOntologyLookupService]
         SO.graph = None
+        SBO.endpoints = [EBIOntologyLookupService]
+        SBO.graph = None
+
+    @classmethod
+    def tearDownClass(cls):
+        SO.endpoints = TestOLS.SO_endpoints
+        SO.graph = TestOLS.SO_graph
+        SBO.endpoints = TestOLS.SBO_endpoints
+        SBO.graph = TestOLS.SBO_graph
+
+    def test_SO(self):
         uri = 'https://identifiers.org/SO:0000167'
         self.assertEqual(SO.get_term_by_uri(uri), 'promoter')
         self.assertEqual(SO.promoter, uri)
@@ -77,12 +96,19 @@ def test_SO(self):
             self.assertEqual(SO.get_term_by_uri(uri), 'promoter')
         with self.assertRaises(LookupError):
             self.assertIsNone(SO.foo)
-        SO.endpoints = restore_endpoints
-        SO.graph = restore_graph
+
+    def test_parents(self):
+        self.assertCountEqual(EBIOntologyLookupService.get_parents(SO, SO.inducible_promoter),
+                              [SO.promoter])
+        self.assertTrue(SO.inducible_promoter.is_child_of(SO.promoter))
+
+    def test_children(self):
+        children = EBIOntologyLookupService.get_children(SO, SO.promoter) 
+        self.assertIn(SO.inducible_promoter, children)
+        self.assertEqual(len(children), 7)
+        self.assertTrue(SO.promoter.is_parent_of(SO.inducible_promoter))
 
     def test_SBO(self):
-        restore_endpoints = SBO.endpoints
-        restore_graph = SBO.graph
         uri = 'https://identifiers.org/SBO:0000241'
         self.assertEqual(SBO.get_term_by_uri(uri), 'functional entity')