From 63b59973f1209df1ea7359fc5e768de80a86a4aa Mon Sep 17 00:00:00 2001 From: Chris Mungall Date: Thu, 30 May 2024 16:17:00 -0700 Subject: [PATCH] format --- src/oaklib/query.py | 121 +++++++++++++----- src/oaklib/utilities/queries/dissector.py | 39 ++++-- tests/test_implementations/test_simple_obo.py | 2 +- 3 files changed, 121 insertions(+), 41 deletions(-) diff --git a/src/oaklib/query.py b/src/oaklib/query.py index 27f411505..6b3fe88c9 100644 --- a/src/oaklib/query.py +++ b/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,7 +259,8 @@ def gap_fill(term: Union[str, Query], predicates: Optional[List[PRED_CURIE]] = N ... print(r) ... - (('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: @@ -223,10 +268,17 @@ def gap_fill(term: Union[str, Query], predicates: Optional[List[PRED_CURIE]] = N :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"): diff --git a/src/oaklib/utilities/queries/dissector.py b/src/oaklib/utilities/queries/dissector.py index 8ec8d4ff1..f23e3a566 100644 --- a/src/oaklib/utilities/queries/dissector.py +++ b/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 - diff --git a/tests/test_implementations/test_simple_obo.py b/tests/test_implementations/test_simple_obo.py index 2e421af07..f5bce975f 100644 --- a/tests/test_implementations/test_simple_obo.py +++ b/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 (