Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions src/translator_tom/models/attribute.py
Original file line number Diff line number Diff line change
Expand Up @@ -172,6 +172,12 @@ def merge_attribute_lists(old: list[Attribute], new: list[Attribute]) -> None:
old.extend(attrs.values())


# Match "subject"/"object" only as whole words, so a free-form `id` such as
# "biolink:objective_*" is not mangled into "subjective_*" when reversing.
_SUBJECT_RE = re.compile(r"(?<![a-zA-Z])subject(?![a-zA-Z])")
_OBJECT_RE = re.compile(r"(?<![a-zA-Z])object(?![a-zA-Z])")


class AttributeConstraint(TOMBase):
"""Generic query constraint for a query node or query edge."""

Expand Down Expand Up @@ -330,3 +336,17 @@ def set_met_by(
any(c.met_by(attr) for attr in attrs_by_type.get(c.id, []))
for c in constraints
)

def get_inverse(self) -> AttributeConstraint:
"""Return a (SPO) inverse of the constraint, for reversing edges.

Flips subject/object for the few directional attribute types.
"""
new = self.model_copy()
if _OBJECT_RE.search(self.id):
new.id = _OBJECT_RE.sub("subject", self.id)
new.name = _OBJECT_RE.sub("subject", self.name)
elif _SUBJECT_RE.search(self.id):
new.id = _SUBJECT_RE.sub("object", self.id)
new.name = _SUBJECT_RE.sub("object", self.name)
return new
10 changes: 3 additions & 7 deletions src/translator_tom/models/query_graph.py
Original file line number Diff line number Diff line change
Expand Up @@ -244,18 +244,14 @@ def get_inverse(self) -> QEdge:
if len(failed_predicates) > 0:
raise ValueError(f"Cannot invert predicates {failed_predicates}.")

# TODO: attribute constraint inversion with respect to the edge direction
if len(self.attribute_constraints_list):
raise NotImplementedError(
"Attribute constraint inversion not yet implemented."
)

return QEdge(
knowledge_type=self.knowledge_type,
predicates=inverse_predicates or None,
subject=self.object,
object=self.subject,
attribute_constraints=self.attribute_constraints,
attribute_constraints=(
[ac.get_inverse() for ac in self.attribute_constraints_list] or None
),
qualifier_constraints=(
[qconstr.get_inverse() for qconstr in self.qualifier_constraints_list]
or None
Expand Down
78 changes: 78 additions & 0 deletions tests/test_models/test_attribute.py
Original file line number Diff line number Diff line change
Expand Up @@ -401,3 +401,81 @@ def test_constraint_with_no_matching_id_is_unsatisfied(self):
c = _con("==", 1, type_id="biolink:nonexistent")
attrs = [_attr(1, type_id="biolink:a")]
assert AttributeConstraint.set_met_by([c], attrs) is False


# ---- AttributeConstraint.get_inverse -----------------------------------------


class TestAttributeConstraintGetInverse:
def test_subject_to_object(self):
c = AttributeConstraint(
id="biolink:original_subject",
name="original subject",
operator="==",
value="X",
)
inv = c.get_inverse()
assert inv.id == "biolink:original_object"
assert inv.name == "original object"

def test_object_to_subject(self):
c = AttributeConstraint(
id="biolink:original_object",
name="original object",
operator="==",
value="X",
)
inv = c.get_inverse()
assert inv.id == "biolink:original_subject"
assert inv.name == "original subject"

def test_direction_independent_passes_through(self):
c = AttributeConstraint(
id="biolink:primary_knowledge_source",
name="primary knowledge source",
operator="==",
value="infores:foo",
)
inv = c.get_inverse()
assert inv.id == "biolink:primary_knowledge_source"
assert inv.name == "primary knowledge source"

def test_preserves_other_fields(self):
c = AttributeConstraint(
id="biolink:original_subject",
name="original subject",
operator=">",
value=5,
negated=True,
unit_id="UO:0000221",
unit_name="dalton",
)
inv = c.get_inverse()
assert inv.operator == ">"
assert inv.value == 5
assert inv.negated is True
assert inv.unit_id == "UO:0000221"
assert inv.unit_name == "dalton"

def test_does_not_mutate_original(self):
c = AttributeConstraint(
id="biolink:original_subject",
name="original subject",
operator="==",
value="X",
)
c.get_inverse()
assert c.id == "biolink:original_subject"
assert c.name == "original subject"

def test_no_false_substring_match(self):
# "object"/"subject" as part of a larger word must not be swapped.
c = AttributeConstraint(
id="biolink:objective_score",
name="objective score",
operator="==",
value=1,
)
inv = c.get_inverse()
assert inv.id == "biolink:objective_score"
assert inv.name == "objective score"
22 changes: 19 additions & 3 deletions tests/test_models/test_query_graph.py
Original file line number Diff line number Diff line change
Expand Up @@ -168,16 +168,32 @@ def test_raises_when_predicate_uninvertible(self):
with pytest.raises(ValueError, match="Cannot invert"):
e.get_inverse()

def test_raises_when_attribute_constraints_present(self):
def test_inverts_direction_scoped_attribute_constraints(self):
e = _qedge(
attribute_constraints=[
AttributeConstraint(
id="biolink:original_subject",
name="original subject",
operator="==",
value="X",
)
]
)
inv = e.get_inverse()
assert inv.attribute_constraints is not None
assert inv.attribute_constraints[0].id == "biolink:original_object"

def test_direction_independent_attribute_constraints_pass_through(self):
e = _qedge(
attribute_constraints=[
AttributeConstraint(
id="biolink:foo", name="foo", operator="==", value=1
)
]
)
with pytest.raises(NotImplementedError):
e.get_inverse()
inv = e.get_inverse()
assert inv.attribute_constraints is not None
assert inv.attribute_constraints[0].id == "biolink:foo"

def test_inverts_qualifier_constraints(self):
q = Qualifier(
Expand Down
Loading