format

src/oaklib/query.py

@@ -4,15 +4,27 @@
 import secrets
 import sys
 from enum import Enum
-from typing import Iterator, Iterable, Dict, IO, Optional, List, Tuple, Union, Any
+from typing import IO, Any, Dict, Iterable, Iterator, List, Optional, Tuple, Union
 
 from pydantic import BaseModel
 
 from oaklib import BasicOntologyInterface
 from oaklib.datamodels.search import create_search_configuration
-from oaklib.datamodels.vocabulary import OWL_CLASS, OWL_OBJECT_PROPERTY, IS_A, PART_OF, DEVELOPS_FROM, RDF_TYPE, \
-    EQUIVALENT_CLASS
-from oaklib.interfaces import OboGraphInterface, SearchInterface, OntologyInterface, SubsetterInterface
+from oaklib.datamodels.vocabulary import (
+    DEVELOPS_FROM,
+    EQUIVALENT_CLASS,
+    IS_A,
+    OWL_CLASS,
+    OWL_OBJECT_PROPERTY,
+    PART_OF,
+    RDF_TYPE,
+)
+from oaklib.interfaces import (
+    OboGraphInterface,
+    OntologyInterface,
+    SearchInterface,
+    SubsetterInterface,
+)
 from oaklib.interfaces.semsim_interface import SemanticSimilarityInterface
 from oaklib.types import CURIE, PRED_CURIE
 from oaklib.utilities.subsets.slimmer_utils import filter_redundant
@@ -24,6 +36,7 @@
 
 TERM = Union["Query", str, List[str]]
 
+
 class OperatorEnum(str, Enum):
     AND = "and"
     OR = "or"
@@ -46,13 +59,13 @@ class Query(BaseModel):
 
     description: Optional[str] = None
 
-    def __and__(self, other: "Expression"):
+    def __and__(self, other: "Query"):
         return BooleanQuery(operator="and", left=self, right=self._as_query_term(other))
 
-    def __or__(self, other: "Expression"):
+    def __or__(self, other: "Query"):
         return BooleanQuery(operator="or", left=self, right=self._as_query_term(other))
 
-    def __sub__(self, other: "Expression"):
+    def __sub__(self, other: "Query"):
         return BooleanQuery(operator="not", left=self, right=self._as_query_term(other))
 
     def execute(self, adapter: BasicOntologyInterface, labels=False, **kwargs):
@@ -66,7 +79,7 @@ def execute(self, adapter: BasicOntologyInterface, labels=False, **kwargs):
         """
         return onto_query(self, adapter, labels=labels)
 
-    def _as_query_term(self, other: "Expression") -> "Query":
+    def _as_query_term(self, other: "Query") -> "Query":
         if isinstance(other, Query):
             return other
         if isinstance(other, str):
@@ -109,11 +122,15 @@ def as_list(self):
             arg = arg.as_list()
         if self.function:
             if self.parameters:
+
                 def _flatten(v):
                     if isinstance(v, list):
                         return ",".join(v)
                     return v
-                param_str = "//" + "//".join([f"{k}={_flatten(v)}" for k, v in self.parameters.items()])
+
+                param_str = "//" + "//".join(
+                    [f"{k}={_flatten(v)}" for k, v in self.parameters.items()]
+                )
             else:
                 param_str = ""
             return [f".{self.function}{param_str}"] + [arg]
@@ -139,7 +156,9 @@ def subclass_of(term: TERM, description: Optional[str] = None) -> FunctionQuery:
     return FunctionQuery(function=FunctionEnum.SUBCLASS, argument=term, description=description)
 
 
-def descendant_of(term: TERM, predicates: Optional[List[PRED_CURIE]] = None, description: Optional[str] = None) -> FunctionQuery:
+def descendant_of(
+    term: TERM, predicates: Optional[List[PRED_CURIE]] = None, description: Optional[str] = None
+) -> FunctionQuery:
     """
     Returns a descendant query.
 
@@ -156,10 +175,19 @@ def descendant_of(term: TERM, predicates: Optional[List[PRED_CURIE]] = None, des
     :param description:
     :return:
     """
-    return FunctionQuery(function=FunctionEnum.DESCENDANT, argument=term, description=description, parameters={"predicates": predicates})
-
-
-def ancestor_of(term: Union[str, Query], predicates: Optional[List[PRED_CURIE]] = None, description: Optional[str] = None) -> FunctionQuery:
+    return FunctionQuery(
+        function=FunctionEnum.DESCENDANT,
+        argument=term,
+        description=description,
+        parameters={"predicates": predicates},
+    )
+
+
+def ancestor_of(
+    term: Union[str, Query],
+    predicates: Optional[List[PRED_CURIE]] = None,
+    description: Optional[str] = None,
+) -> FunctionQuery:
     """
     Returns an ancestor query.
 
@@ -176,10 +204,17 @@ def ancestor_of(term: Union[str, Query], predicates: Optional[List[PRED_CURIE]]
     :param description:
     :return:
     """
-    return FunctionQuery(function=FunctionEnum.ANCESTOR, argument=term, description=description, parameters={"predicates": predicates})
+    return FunctionQuery(
+        function=FunctionEnum.ANCESTOR,
+        argument=term,
+        description=description,
+        parameters={"predicates": predicates},
+    )
 
 
-def non_redundant(term: TERM, predicates: Optional[List[PRED_CURIE]] = None,  description: Optional[str] = None) -> FunctionQuery:
+def non_redundant(
+    term: TERM, predicates: Optional[List[PRED_CURIE]] = None, description: Optional[str] = None
+) -> FunctionQuery:
     """
     Returns a query that when executed will return the non-redundant subset of the input set.
 
@@ -197,10 +232,19 @@ def non_redundant(term: TERM, predicates: Optional[List[PRED_CURIE]] = None,  de
     :param description:
     :return:
     """
-    return FunctionQuery(function=FunctionEnum.NON_REDUNDANT, argument=term, description=description, parameters={"predicates": predicates})
-
-
-def gap_fill(term: Union[str, Query], predicates: Optional[List[PRED_CURIE]] = None, description: Optional[str] = None) -> FunctionQuery:
+    return FunctionQuery(
+        function=FunctionEnum.NON_REDUNDANT,
+        argument=term,
+        description=description,
+        parameters={"predicates": predicates},
+    )
+
+
+def gap_fill(
+    term: Union[str, Query],
+    predicates: Optional[List[PRED_CURIE]] = None,
+    description: Optional[str] = None,
+) -> FunctionQuery:
     """
     Returns a query that when executed will fill in edges between nodes in the input set.
 
@@ -215,18 +259,26 @@ def gap_fill(term: Union[str, Query], predicates: Optional[List[PRED_CURIE]] = N
     ...     print(r)
     <BLANKLINE>
     ...
-    (('CL:0002169', 'basal cell of olfactory epithelium'), ('BFO:0000050', 'part_of'), ('UBERON:0001005', 'respiratory airway'))
+    (('CL:0002169', 'basal cell of olfactory epithelium'),
+     ('BFO:0000050', 'part_of'), ('UBERON:0001005', 'respiratory airway'))
     ...
 
     :param term:
     :param predicates:
     :param description:
     :return:
     """
-    return FunctionQuery(function=FunctionEnum.GAP_FILL, argument=term, description=description, parameters={"predicates": predicates})
+    return FunctionQuery(
+        function=FunctionEnum.GAP_FILL,
+        argument=term,
+        description=description,
+        parameters={"predicates": predicates},
+    )
 
 
-def onto_query(query_terms: Union[Query, NESTED_LIST], adapter: BasicOntologyInterface, labels=False) -> List[Union[CURIE, Tuple[CURIE, str]]]:
+def onto_query(
+    query_terms: Union[Query, NESTED_LIST], adapter: BasicOntologyInterface, labels=False
+) -> List[Union[CURIE, Tuple[CURIE, str]]]:
     """
     Turn list of tokens that represent a term query into a list of curies.
 
@@ -362,16 +414,24 @@ def onto_query(query_terms: Union[Query, NESTED_LIST], adapter: BasicOntologyInt
                 results = list(adapter.labels(results))
             else:
                 # get all distinct s, p, o in all results
-                all_ids = set([r[0] for r in results] + [r[1] for r in results] + [r[2] for r in results])
+                all_ids = set(
+                    [r[0] for r in results] + [r[1] for r in results] + [r[2] for r in results]
+                )
                 labels = {r[0]: r[1] for r in adapter.labels(all_ids)}
-                results = [((r[0], labels.get(r[0], None)),
-                             (r[1], labels.get(r[1], None)),
-                             (r[2], labels.get(r[2], None)))
-                            for r in results]
+                results = [
+                    (
+                        (r[0], labels.get(r[0], None)),
+                        (r[1], labels.get(r[1], None)),
+                        (r[2], labels.get(r[2], None)),
+                    )
+                    for r in results
+                ]
     return results
 
 
-def query_terms_iterator(query_terms: NESTED_LIST, adapter: BasicOntologyInterface) -> Iterator[CURIE]:
+def query_terms_iterator(
+    query_terms: NESTED_LIST, adapter: BasicOntologyInterface
+) -> Iterator[CURIE]:
     """
     Turn list of tokens that represent a term query into an iterator for curies.
 
@@ -568,7 +628,8 @@ def chain_results(v):
                 o for _s, _p, o in adapter.relationships(subjects=rest, predicates=this_predicates)
             ]
             sibs = [
-                s for s, _p, _o in adapter.relationships(objects=parents, predicates=this_predicates)
+                s
+                for s, _p, _o in adapter.relationships(objects=parents, predicates=this_predicates)
             ]
             chain_results(sibs)
         elif term.startswith(".anc"):

src/oaklib/utilities/queries/dissector.py

@@ -1,12 +1,12 @@
 import logging
-from typing import Union, List, Optional
+from typing import List, Optional, Union
 
 from pydantic import BaseModel
 
 from oaklib import BasicOntologyInterface
-from oaklib.datamodels.vocabulary import PART_OF, HAS_PART
+from oaklib.datamodels.vocabulary import HAS_PART, PART_OF
 from oaklib.interfaces import OboGraphInterface
-from oaklib.query import ancestor_of, descendant_of, Query, subclass_of
+from oaklib.query import Query, ancestor_of, descendant_of, subclass_of
 from oaklib.types import CURIE, PRED_CURIE
 
 ITEM = Union[CURIE, str]
@@ -25,7 +25,12 @@ class DissectedEntity(BaseModel):
     found_in: Optional[List[ClassReference]] = None
 
 
-def dissection_query(structure: ITEM, dissection_relation: PRED_OR_PREDS = HAS_PART, inverse_relation: PRED_OR_PREDS = PART_OF, entity_type: ITEM=None) -> Query:
+def dissection_query(
+    structure: ITEM,
+    dissection_relation: PRED_OR_PREDS = HAS_PART,
+    inverse_relation: PRED_OR_PREDS = PART_OF,
+    entity_type: ITEM = None,
+) -> Query:
     """
     Generate a query for the dissection of a structure.
 
@@ -51,13 +56,23 @@ def dissection_query(structure: ITEM, dissection_relation: PRED_OR_PREDS = HAS_P
     if isinstance(entity_type, str):
         # if user provides "neuron", assume we mean subtypes of neuron
         entity_type = subclass_of(entity_type)
-    dissect_q = ancestor_of(descendant_of(structure, predicates=inverse_relation), predicates=dissection_relation)
+    dissect_q = ancestor_of(
+        descendant_of(structure, predicates=inverse_relation), predicates=dissection_relation
+    )
     if entity_type:
         dissect_q = dissect_q & entity_type
     return dissect_q
 
 
-def dissect(adapter: BasicOntologyInterface, structure: ITEM, dissection_relation: PRED_OR_PREDS = HAS_PART, inverse_relation: PRED_OR_PREDS = PART_OF, entity_type: ITEM = None, complete=True, **kwargs) -> List[DissectedEntity]:
+def dissect(
+    adapter: BasicOntologyInterface,
+    structure: ITEM,
+    dissection_relation: PRED_OR_PREDS = HAS_PART,
+    inverse_relation: PRED_OR_PREDS = PART_OF,
+    entity_type: ITEM = None,
+    complete=True,
+    **kwargs,
+) -> List[DissectedEntity]:
     """
     Dissect a structure into its parts.
 
@@ -90,7 +105,12 @@ def dissect(adapter: BasicOntologyInterface, structure: ITEM, dissection_relatio
     :param entity_type:
     :return:
     """
-    q = dissection_query(structure, dissection_relation=dissection_relation, inverse_relation=inverse_relation, entity_type=entity_type)
+    q = dissection_query(
+        structure,
+        dissection_relation=dissection_relation,
+        inverse_relation=inverse_relation,
+        entity_type=entity_type,
+    )
     results = []
     for id, name in q.execute(adapter, labels=True, **kwargs):
         if not name:
@@ -102,9 +122,8 @@ def dissect(adapter: BasicOntologyInterface, structure: ITEM, dissection_relatio
     if complete:
         if not isinstance(adapter, OboGraphInterface):
             raise NotImplementedError
-        #ldefs = list(adapter.logical_definitions())
+        # ldefs = list(adapter.logical_definitions())
 
-        #for entity in results:
+        # for entity in results:
         #    entity.found_in =
     return results
-

tests/test_implementations/test_simple_obo.py

@@ -3,7 +3,6 @@
 from copy import deepcopy
 
 from kgcl_schema.datamodel import kgcl
-from oaklib.query import query_terms_iterator
 from oaklib.datamodels import obograph
 from oaklib.datamodels.search import SearchConfiguration
 from oaklib.datamodels.search_datamodel import SearchProperty, SearchTermSyntax
@@ -24,6 +23,7 @@
     TagValue,
     XrefList,
 )
+from oaklib.query import query_terms_iterator
 from oaklib.resource import OntologyResource
 from oaklib.utilities.kgcl_utilities import generate_change_id
 from oaklib.utilities.obograph_utils import (